Merge branch 'main' into ihalatci-persist-last-upgrade-version

Author: ihalatci

.devcontainer/src/test/regress/Pipfile

@@ -4,6 +4,7 @@ url = "https://pypi.python.org/simple"
 verify_ssl = true
 
 [packages]
+pyasn1 = ">=0.6.4"
 mitmproxy = {git = "https://github.com/citusdata/mitmproxy.git", ref = "321e6d203cf31e36d59ba8e4f9c6a3c4a5d6ddf0"}
 "aioquic" = ">=1.2.0,<1.3.0"
 "mitmproxy-rs" = ">=0.12.6,<0.13.0"

.devcontainer/src/test/regress/Pipfile.lock

@@ -36,7 +36,7 @@ jobs:
       style_checker_image_name: "ghcr.io/citusdata/stylechecker"
       style_checker_tools_version: "0.8.33"
       sql_snapshot_pg_version: "18.3"
-      image_suffix: "-va56062b"
+      image_suffix: "-v081d17c"
       pg16_version: '{ "major": "16", "full": "16.13" }'
       pg17_version: '{ "major": "17", "full": "17.9" }'
       pg18_version: '{ "major": "18", "full": "18.3" }'

.github/workflows/build_and_test.yml

@@ -1358,7 +1358,7 @@ CreateTableConversion(TableConversionParameters *params)
 	relation_close(relation, NoLock);
 	con->distributionKey =
 		BuildDistributionKeyFromColumnName(con->relationId, con->distributionColumn,
-										   NoLock);
+										   NoLock, false);
 
 	con->originalAccessMethod = NULL;
 	if (!PartitionedTable(con->relationId) && !IsForeignTable(con->relationId))

src/backend/distributed/commands/alter_table.c

@@ -535,7 +535,7 @@ CreateDistributedTableConcurrently(Oid relationId, char *distributionColumnName,
 	 */
 	Var *distributionColumn = BuildDistributionKeyFromColumnName(relationId,
 																 distributionColumnName,
-																 NoLock);
+																 NoLock, false);
 
 	/* get an advisory lock to serialize concurrent default group creations */
 	if (IsColocateWithDefault(colocateWithTableName))
@@ -704,7 +704,7 @@ EnsureColocateWithTableIsValid(Oid relationId, char distributionMethod,
 
 	Var *distributionColumn = BuildDistributionKeyFromColumnName(relationId,
 																 distributionColumnName,
-																 NoLock);
+																 AccessShareLock, true);
 	EnsureTableCanBeColocatedWith(relationId, replicationModel,
 								  distributionColumn, colocateWithTableId);
 }
@@ -1221,7 +1221,7 @@ CreateCitusTable(Oid relationId, CitusTableType tableType,
 		distributionColumn = BuildDistributionKeyFromColumnName(relationId,
 																distributedTableParams->
 																distributionColumnName,
-																NoLock);
+																NoLock, false);
 	}
 
 	CitusTableParams citusTableParams = DecideCitusTableParams(tableType,

src/backend/distributed/commands/create_distributed_table.c

@@ -3472,7 +3472,7 @@ citus_internal_add_partition_metadata(PG_FUNCTION_ARGS)
 
 		distributionColumnVar =
 			BuildDistributionKeyFromColumnName(relationId, distributionColumnString,
-											   AccessShareLock);
+											   AccessShareLock, false);
 		Assert(distributionColumnVar != NULL);
 	}
 

src/backend/distributed/metadata/metadata_sync.c

@@ -317,7 +317,7 @@ CreatePartitionedSplitCopyDestReceiver(EState *estate,
 	Oid shardRelationId = LookupShardRelationFromCatalog(shardId, missingOK);
 	Var *partitionColumn = BuildDistributionKeyFromColumnName(shardRelationId,
 															  partitionColumnName,
-															  AccessShareLock);
+															  AccessShareLock, false);
 
 	CitusTableCacheEntry *shardSearchInfo =
 		QueryTupleShardSearchInfo(minValuesArray, maxValuesArray,

src/backend/distributed/operations/worker_split_copy_udf.c

@@ -222,7 +222,7 @@ CreateShardSplitInfo(uint64 sourceShardIdToSplit,
 	/* determine the partition column in the tuple descriptor */
 	Var *partitionColumn = BuildDistributionKeyFromColumnName(sourceShardToSplitOid,
 															  partitionColumnName,
-															  AccessShareLock);
+															  AccessShareLock, false);
 	if (partitionColumn == NULL)
 	{
 		ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR),

src/backend/distributed/operations/worker_split_shard_replication_setup_udf.c

@@ -178,6 +178,7 @@ static bool ModifiesLocalTableWithRemoteCitusLocalTable(List *rangeTableList);
 static DeferredErrorMessage * DeferErrorIfUnsupportedLocalTableJoin(List *rangeTableList);
 static bool IsLocallyAccessibleCitusLocalTable(Oid relationId);
 static bool ConvertToQueryOnShard(Query *query, Oid relationID, Oid shardRelationId);
+static void ReplaceModifyingCteWithEmptyResult(Query *outerQuery);
 
 /*
  * CreateRouterPlan attempts to create a router executor plan for the given
@@ -284,6 +285,16 @@ CreateSingleTaskRouterSelectPlan(DistributedPlan *distributedPlan, Query *origin
 		return;
 	}
 
+	/*
+	 * RouterJob may rewrite a zero-shard modifying CTE inside a router SELECT
+	 * into a plain empty-result CTE (see ReplaceModifyingCteWithEmptyResult).
+	 * That clears originalQuery->hasModifyingCTE, so recompute modLevel from
+	 * the (possibly-transformed) job query -- otherwise the executor would
+	 * treat the plan as ROW_MODIFY_NONCOMMUTATIVE and attempt to acquire
+	 * shard-locks on the pruned-away shard.
+	 */
+	distributedPlan->modLevel = RowModifyLevelForQuery(job->jobQuery);
+
 	ereport(DEBUG2, (errmsg("Creating router plan")));
 
 	distributedPlan->workerJob = job;
@@ -2004,6 +2015,39 @@ RouterJob(Query *originalQuery, PlannerRestrictionContext *plannerRestrictionCon
 			}
 		}
 	}
+	else if (shardId == INVALID_SHARD_ID && originalQuery->hasModifyingCTE)
+	{
+		/*
+		 * Router SELECT that wraps a modifying CTE whose pruning
+		 * yielded zero shards (e.g. WITH u AS (UPDATE t ... WHERE
+		 * dist_key = X AND FALSE RETURNING ...) SELECT ... FROM u).
+		 *
+		 * PlanRouterQuery returned INVALID_SHARD_ID with a dummy
+		 * placement, and UpdateRelationToShardNames has already
+		 * flipped the CTE's target relation RTE to an empty-result
+		 * RTE_SUBQUERY. The direct UPDATE/DELETE escape above only
+		 * inspects the OUTER query's resultRelation (which is 0 for
+		 * a SELECT), so absent this branch the query would fall
+		 * through to SingleShardTaskList and be promoted to a
+		 * MODIFY_TASK with anchorShardId = 0 -- which then fails at
+		 * execution time in AcquireMetadataLocks ->
+		 * LookupShardIdCacheEntry(0) with "could not find valid
+		 * entry for shard 0".
+		 *
+		 * We cannot use the taskList = NIL contract that the direct
+		 * UPDATE/DELETE escape above uses, because the outer SELECT
+		 * has a consumer (e.g. SELECT count(*) FROM u must return
+		 * one row containing 0, not the empty resultset that an empty
+		 * task list would yield). Instead, rewrite each modifying CTE
+		 * in-place to a plain SELECT that returns no rows but retains
+		 * the CTE's output column shape, and clear hasModifyingCTE.
+		 * The rewritten query then flows through the normal
+		 * zero-shard router-SELECT machinery (READ_TASK on the dummy
+		 * placement), producing the correct empty CTE and letting the
+		 * outer aggregate emit the required single row.
+		 */
+		ReplaceModifyingCteWithEmptyResult(originalQuery);
+	}
 
 	if (isMultiShardModifyQuery)
 	{
@@ -2134,6 +2178,27 @@ CheckAndBuildDelayedFastPathPlan(DistributedPlanningContext *planContext,
 		return;
 	}
 
+	if (list_length(job->taskList) == 0)
+	{
+		/*
+		 * Planner-time shard pruning legitimately produced zero shards for a
+		 * delayed fast-path modification (e.g. WHERE dist_key = X AND FALSE,
+		 * or a fast-path fallback where the distribution-key type does not
+		 * match the literal type and PruneShards short-circuits on
+		 * ContainsFalseClause). GenerateSingleShardRouterTaskList already
+		 * established the terminal state job->taskList = NIL for this case
+		 * in the non-delayed fast-path route; mirror the deferred-pruning
+		 * branch above so we build a placeholder plan (no single-task local
+		 * shortcut, no deparse) and let the executor treat the empty task
+		 * list as a silent no-op, matching the non-fast-path direct
+		 * UPDATE/DELETE contract.
+		 */
+		planContext->plan = FastPathPlanner(planContext->originalQuery,
+											planContext->query,
+											planContext->boundParams);
+		return;
+	}
+
 	List *tasks = job->taskList;
 	Assert(list_length(tasks) == 1);
 	Task *task = (Task *) linitial(tasks);
@@ -2290,6 +2355,94 @@ ConvertToQueryOnShard(Query *query, Oid citusTableOid, Oid shardId)
 }
 
 
+/*
+ * ReplaceModifyingCteWithEmptyResult rewrites each modifying (UPDATE/DELETE)
+ * CTE in outerQuery->cteList so that its body becomes a plain SELECT whose
+ * jointree quals are constant FALSE and whose target list is a matching-shape
+ * list of NULL constants. This preserves the outer query's structural reference
+ * to the CTE (name, output column names/types/typmods/collations) while
+ * guaranteeing the CTE produces zero rows and requires no shard metadata.
+ *
+ * hasModifyingCTE is cleared only if no other modifying CTE (e.g. CMD_INSERT,
+ * CMD_MERGE) remains in the cteList after the rewrite; otherwise the flag is
+ * preserved so downstream routing (READ_TASK vs MODIFY_TASK selection,
+ * modLevel computation, lock acquisition) still treats the query as
+ * modifying. When the flag does end up cleared, the outer SELECT flows
+ * through the standard zero-shard router-SELECT machinery (READ_TASK on the
+ * dummy placement, no shard-metadata lookups, no MODIFY_TASK promotion in
+ * SingleShardTaskList).
+ *
+ * This helper mutates outerQuery in place. Callers must have already
+ * established that pruning yielded zero shards (shardId == INVALID_SHARD_ID);
+ * pruning-relevant fields (placementList, relationShardList, prunedShardIntervalListList)
+ * are unaffected.
+ */
+static void
+ReplaceModifyingCteWithEmptyResult(Query *outerQuery)
+{
+	CommonTableExpr *cte = NULL;
+	bool anyModifyingCteLeft = false;
+
+	foreach_declared_ptr(cte, outerQuery->cteList)
+	{
+		Query *cteQuery = (Query *) cte->ctequery;
+
+		if (cteQuery->commandType != CMD_UPDATE &&
+			cteQuery->commandType != CMD_DELETE)
+		{
+			/*
+			 * Non-UPDATE/DELETE CTE: leave it alone. If it is still a
+			 * modification (e.g. CMD_INSERT, CMD_MERGE) note that so we do
+			 * not clear the outer query's hasModifyingCTE flag below.
+			 */
+			if (cteQuery->commandType != CMD_SELECT)
+			{
+				anyModifyingCteLeft = true;
+			}
+			continue;
+		}
+
+		int columnCount = list_length(cte->ctecoltypes);
+		List *targetList = NIL;
+
+		for (int columnIndex = 0; columnIndex < columnCount; columnIndex++)
+		{
+			Oid coltype = list_nth_oid(cte->ctecoltypes, columnIndex);
+			int32 coltypmod = list_nth_int(cte->ctecoltypmods, columnIndex);
+			Oid colcoll = list_nth_oid(cte->ctecolcollations, columnIndex);
+			Node *colname = (Node *) list_nth(cte->ctecolnames, columnIndex);
+
+			Const *nullConst = makeNullConst(coltype, coltypmod, colcoll);
+
+			TargetEntry *targetEntry = makeNode(TargetEntry);
+			targetEntry->expr = (Expr *) nullConst;
+			targetEntry->resno = columnIndex + 1;
+			targetEntry->resname = pstrdup(strVal(colname));
+			targetEntry->resorigtbl = InvalidOid;
+			targetEntry->resorigcol = 0;
+			targetEntry->resjunk = false;
+
+			targetList = lappend(targetList, targetEntry);
+		}
+
+		FromExpr *joinTree = makeNode(FromExpr);
+		joinTree->fromlist = NIL;
+		joinTree->quals = (Node *) makeBoolConst(false, false);
+
+		Query *emptyCteQuery = makeNode(Query);
+		emptyCteQuery->commandType = CMD_SELECT;
+		emptyCteQuery->querySource = cteQuery->querySource;
+		emptyCteQuery->canSetTag = cteQuery->canSetTag;
+		emptyCteQuery->targetList = targetList;
+		emptyCteQuery->jointree = joinTree;
+
+		cte->ctequery = (Node *) emptyCteQuery;
+	}
+
+	outerQuery->hasModifyingCTE = anyModifyingCteLeft;
+}
+
+
 /*
  * GenerateSingleShardRouterTaskList is a wrapper around other corresponding task
  * list generation functions specific to single shard selects and modifications.

src/backend/distributed/planner/multi_router_planner.c

@@ -0,0 +1,4 @@
+-- citus--13.3-1--13.4-1
+-- bump version to 13.4-1
+
+#include "udfs/citus_internal_distribute_object/13.4-1.sql"