Merge pull request #2807 from citusdata/2776_preparation

DistributedPlan: replace operation with modLevel

src/backend/distributed/executor/adaptive_executor.c

@@ -158,8 +158,8 @@
  */
 typedef struct DistributedExecution
 {
-	/* the corresponding distributed plan's operation */
+	/* the corresponding distributed plan's modLevel */
-	CmdType operation;
+	RowModifyLevel modLevel;
 
 	List *tasksToExecute;
 
@@ -504,7 +504,7 @@ int ExecutorSlowStartInterval = 10;
 
 
 /* local functions */
-static DistributedExecution * CreateDistributedExecution(CmdType operation,
+static DistributedExecution * CreateDistributedExecution(RowModifyLevel modLevel,
 														 List *taskList,
 														 bool hasReturning,
 														 ParamListInfo paramListInfo,
@@ -520,16 +520,16 @@ static void FinishDistributedExecution(DistributedExecution *execution);
 static void CleanUpSessions(DistributedExecution *execution);
 
 static void LockPartitionsForDistributedPlan(DistributedPlan *distributedPlan);
-static void AcquireExecutorShardLocks(DistributedExecution *execution);
+static void AcquireExecutorShardLocksForExecution(DistributedExecution *execution);
 static bool DistributedExecutionModifiesDatabase(DistributedExecution *execution);
 static bool DistributedPlanModifiesDatabase(DistributedPlan *plan);
-static bool TaskListModifiesDatabase(CmdType operation, List *taskList);
+static bool TaskListModifiesDatabase(RowModifyLevel modLevel, List *taskList);
 static bool DistributedExecutionRequiresRollback(DistributedExecution *execution);
-static bool SelectForUpdateOnReferenceTable(CmdType operation, List *taskList);
+static bool SelectForUpdateOnReferenceTable(RowModifyLevel modLevel, List *taskList);
 static void AssignTasksToConnections(DistributedExecution *execution);
 static void UnclaimAllSessionConnections(List *sessionList);
 static bool UseConnectionPerPlacement(void);
-static PlacementExecutionOrder ExecutionOrderForTask(CmdType operation, Task *task);
+static PlacementExecutionOrder ExecutionOrderForTask(RowModifyLevel modLevel, Task *task);
 static WorkerPool * FindOrCreateWorkerPool(DistributedExecution *execution,
 										   WorkerNode *workerNode);
 static WorkerSession * FindOrCreateWorkerSession(WorkerPool *workerPool,
@@ -610,7 +610,7 @@ AdaptiveExecutor(CustomScanState *node)
 		targetPoolSize = 1;
 	}
 
-	execution = CreateDistributedExecution(distributedPlan->operation, taskList,
+	execution = CreateDistributedExecution(distributedPlan->modLevel, taskList,
 										   distributedPlan->hasReturning,
 										   paramListInfo, tupleDescriptor,
 										   tupleStore, targetPoolSize);
@@ -626,7 +626,7 @@ AdaptiveExecutor(CustomScanState *node)
 		RunDistributedExecution(execution);
 	}
 
-	if (distributedPlan->operation != CMD_SELECT)
+	if (distributedPlan->modLevel != ROW_MODIFY_READONLY)
 	{
 		executorState->es_processed = execution->rowsProcessed;
 	}
@@ -654,14 +654,14 @@ ExecuteUtilityTaskListWithoutResults(List *taskList, int targetPoolSize,
 {
 	if (TaskExecutorType == MULTI_EXECUTOR_ADAPTIVE)
 	{
-		ExecuteTaskList(CMD_UTILITY, taskList, targetPoolSize);
+		ExecuteTaskList(ROW_MODIFY_NONE, taskList, targetPoolSize);
 	}
 	else
 	{
 		if (MultiShardConnectionType == SEQUENTIAL_CONNECTION ||
 			forceSequentialExecution)
 		{
-			ExecuteModifyTasksSequentiallyWithoutResults(taskList, CMD_UTILITY);
+			ExecuteModifyTasksSequentiallyWithoutResults(taskList, ROW_MODIFY_NONE);
 		}
 		else
 		{
@@ -676,13 +676,13 @@ ExecuteUtilityTaskListWithoutResults(List *taskList, int targetPoolSize,
  * for some of the arguments.
  */
 uint64
-ExecuteTaskList(CmdType operation, List *taskList, int targetPoolSize)
+ExecuteTaskList(RowModifyLevel modLevel, List *taskList, int targetPoolSize)
 {
 	TupleDesc tupleDescriptor = NULL;
 	Tuplestorestate *tupleStore = NULL;
 	bool hasReturning = false;
 
-	return ExecuteTaskListExtended(operation, taskList, tupleDescriptor,
+	return ExecuteTaskListExtended(modLevel, taskList, tupleDescriptor,
 								   tupleStore, hasReturning, targetPoolSize);
 }
 
@@ -692,7 +692,7 @@ ExecuteTaskList(CmdType operation, List *taskList, int targetPoolSize)
  * runs it.
  */
 uint64
-ExecuteTaskListExtended(CmdType operation, List *taskList,
+ExecuteTaskListExtended(RowModifyLevel modLevel, List *taskList,
 						TupleDesc tupleDescriptor, Tuplestorestate *tupleStore,
 						bool hasReturning, int targetPoolSize)
 {
@@ -705,7 +705,7 @@ ExecuteTaskListExtended(CmdType operation, List *taskList,
 	}
 
 	execution =
-		CreateDistributedExecution(operation, taskList, hasReturning, paramListInfo,
+		CreateDistributedExecution(modLevel, taskList, hasReturning, paramListInfo,
 								   tupleDescriptor, tupleStore, targetPoolSize);
 
 	StartDistributedExecution(execution);
@@ -721,14 +721,14 @@ ExecuteTaskListExtended(CmdType operation, List *taskList,
  * a distributed plan.
  */
 DistributedExecution *
-CreateDistributedExecution(CmdType operation, List *taskList, bool hasReturning,
+CreateDistributedExecution(RowModifyLevel modLevel, List *taskList, bool hasReturning,
 						   ParamListInfo paramListInfo, TupleDesc tupleDescriptor,
 						   Tuplestorestate *tupleStore, int targetPoolSize)
 {
 	DistributedExecution *execution =
 		(DistributedExecution *) palloc0(sizeof(DistributedExecution));
 
-	execution->operation = operation;
+	execution->modLevel = modLevel;
 	execution->tasksToExecute = taskList;
 	execution->hasReturning = hasReturning;
 
@@ -823,7 +823,7 @@ StartDistributedExecution(DistributedExecution *execution)
 	}
 
 	/* prevent unsafe concurrent modifications */
-	AcquireExecutorShardLocks(execution);
+	AcquireExecutorShardLocksForExecution(execution);
 }
 
 
@@ -834,7 +834,7 @@ StartDistributedExecution(DistributedExecution *execution)
 static bool
 DistributedExecutionModifiesDatabase(DistributedExecution *execution)
 {
-	return TaskListModifiesDatabase(execution->operation, execution->tasksToExecute);
+	return TaskListModifiesDatabase(execution->modLevel, execution->tasksToExecute);
 }
 
 
@@ -845,7 +845,7 @@ DistributedExecutionModifiesDatabase(DistributedExecution *execution)
 static bool
 DistributedPlanModifiesDatabase(DistributedPlan *plan)
 {
-	return TaskListModifiesDatabase(plan->operation, plan->workerJob->taskList);
+	return TaskListModifiesDatabase(plan->modLevel, plan->workerJob->taskList);
 }
 
 
@@ -854,22 +854,22 @@ DistributedPlanModifiesDatabase(DistributedPlan *plan)
  *  DistributedPlanModifiesDatabase.
  */
 static bool
-TaskListModifiesDatabase(CmdType operation, List *taskList)
+TaskListModifiesDatabase(RowModifyLevel modLevel, List *taskList)
 {
 	Task *firstTask = NULL;
 
-	if (operation == CMD_INSERT || operation == CMD_UPDATE || operation == CMD_DELETE)
+	if (modLevel > ROW_MODIFY_READONLY)
 	{
 		return true;
 	}
 
 	/*
-	 * If we cannot decide by only checking the operation, we should look closer
+	 * If we cannot decide by only checking the row modify level,
-	 * to the tasks.
+	 * we should look closer to the tasks.
 	 */
 	if (list_length(taskList) < 1)
 	{
-		/* does this ever possible? */
+		/* is this ever possible? */
 		return false;
 	}
 
@@ -968,12 +968,12 @@ DistributedExecutionRequiresRollback(DistributedExecution *execution)
  * that contains FOR UPDATE clause that locks any reference tables.
  */
 static bool
-SelectForUpdateOnReferenceTable(CmdType operation, List *taskList)
+SelectForUpdateOnReferenceTable(RowModifyLevel modLevel, List *taskList)
 {
 	Task *task = NULL;
 	ListCell *rtiLockCell = NULL;
 
-	if (operation != CMD_SELECT)
+	if (modLevel != ROW_MODIFY_READONLY)
 	{
 		return false;
 	}
@@ -1025,8 +1025,8 @@ LockPartitionsForDistributedPlan(DistributedPlan *distributedPlan)
 
 
 /*
- * AcquireExecutorShardLocks acquires advisory lock on shard IDs to prevent
+ * AcquireExecutorShardLocksForExecution acquires advisory lock on shard IDs
- * unsafe concurrent modifications of shards.
+ * to prevent unsafe concurrent modifications of shards.
  *
  * We prevent concurrent modifications of shards in two cases:
  * 1. Any non-commutative writes to a replicated table
@@ -1042,13 +1042,13 @@ LockPartitionsForDistributedPlan(DistributedPlan *distributedPlan)
  * TRUNCATE because the table locks already prevent concurrent access.
  */
 static void
-AcquireExecutorShardLocks(DistributedExecution *execution)
+AcquireExecutorShardLocksForExecution(DistributedExecution *execution)
 {
-	CmdType operation = execution->operation;
+	RowModifyLevel modLevel = execution->modLevel;
 	List *taskList = execution->tasksToExecute;
 
-	if (!(operation == CMD_INSERT || operation == CMD_UPDATE || operation == CMD_DELETE ||
+	if (modLevel <= ROW_MODIFY_READONLY &&
-		  SelectForUpdateOnReferenceTable(operation, taskList)))
+		!SelectForUpdateOnReferenceTable(modLevel, taskList))
 	{
 		/*
 		 * Executor locks only apply to DML commands and SELECT FOR UPDATE queries
@@ -1069,7 +1069,7 @@ AcquireExecutorShardLocks(DistributedExecution *execution)
 		{
 			Task *task = (Task *) lfirst(taskCell);
 
-			AcquireExecutorShardLock(task, operation);
+			AcquireExecutorShardLocks(task, modLevel);
 		}
 	}
 	else if (list_length(taskList) > 1)
@@ -1211,7 +1211,7 @@ UnclaimAllSessionConnections(List *sessionList)
 static void
 AssignTasksToConnections(DistributedExecution *execution)
 {
-	CmdType operation = execution->operation;
+	RowModifyLevel modLevel = execution->modLevel;
 	List *taskList = execution->tasksToExecute;
 	bool hasReturning = execution->hasReturning;
 
@@ -1233,13 +1233,15 @@ AssignTasksToConnections(DistributedExecution *execution)
 		shardCommandExecution =
 			(ShardCommandExecution *) palloc0(sizeof(ShardCommandExecution));
 		shardCommandExecution->task = task;
-		shardCommandExecution->executionOrder = ExecutionOrderForTask(operation, task);
+		shardCommandExecution->executionOrder = ExecutionOrderForTask(modLevel, task);
 		shardCommandExecution->executionState = TASK_EXECUTION_NOT_FINISHED;
 		shardCommandExecution->placementExecutions =
 			(TaskPlacementExecution **) palloc0(placementExecutionCount *
 												sizeof(TaskPlacementExecution *));
 		shardCommandExecution->placementExecutionCount = placementExecutionCount;
-		shardCommandExecution->expectResults = hasReturning || operation == CMD_SELECT;
+
+		shardCommandExecution->expectResults = hasReturning ||
+											   modLevel == ROW_MODIFY_READONLY;
 
 
 		foreach(taskPlacementCell, task->taskPlacementList)
@@ -1390,7 +1392,7 @@ UseConnectionPerPlacement(void)
  * ExecutionOrderForTask gives the appropriate execution order for a task.
  */
 static PlacementExecutionOrder
-ExecutionOrderForTask(CmdType operation, Task *task)
+ExecutionOrderForTask(RowModifyLevel modLevel, Task *task)
 {
 	switch (task->taskType)
 	{
@@ -1402,7 +1404,14 @@ ExecutionOrderForTask(CmdType operation, Task *task)
 
 		case MODIFY_TASK:
 		{
-			if (operation == CMD_INSERT && !task->upsertQuery)
+			/*
+			 * For non-commutative modifications we take aggressive locks, so
+			 * there is no risk of deadlock and we can run them in parallel.
+			 * When the modification is commutative, we take no additional
+			 * locks, so we take a conservative approach and execute sequentially
+			 * to avoid deadlocks.
+			 */
+			if (modLevel < ROW_MODIFY_NONCOMMUTATIVE)
 			{
 				return EXECUTION_ORDER_SEQUENTIAL;
 			}

src/backend/distributed/executor/citus_custom_scan.c

@@ -163,7 +163,7 @@ CitusBeginScan(CustomScanState *node, EState *estate, int eflags)
 
 	scanState = (CitusScanState *) node;
 	distributedPlan = scanState->distributedPlan;
-	if (distributedPlan->operation == CMD_SELECT ||
+	if (distributedPlan->modLevel == ROW_MODIFY_READONLY ||
 		distributedPlan->insertSelectSubquery != NULL)
 	{
 		/* no more action required */

src/backend/distributed/executor/insert_select_executor.c

@@ -133,7 +133,8 @@ CoordinatorInsertSelectExecScan(CustomScanState *node)
 					if (MultiShardConnectionType == SEQUENTIAL_CONNECTION)
 					{
 						ExecuteModifyTasksSequentially(scanState, prunedTaskList,
-													   CMD_INSERT, hasReturning);
+													   ROW_MODIFY_COMMUTATIVE,
+													   hasReturning);
 					}
 					else
 					{
@@ -151,7 +152,7 @@ CoordinatorInsertSelectExecScan(CustomScanState *node)
 					scanState->tuplestorestate =
 						tuplestore_begin_heap(randomAccess, interTransactions, work_mem);
 
-					ExecuteTaskListExtended(CMD_INSERT, prunedTaskList,
+					ExecuteTaskListExtended(ROW_MODIFY_COMMUTATIVE, prunedTaskList,
 											tupleDescriptor, scanState->tuplestorestate,
 											hasReturning, MaxAdaptiveExecutorPoolSize);
 				}

src/backend/distributed/executor/multi_router_executor.c

@@ -91,15 +91,17 @@ bool SortReturning = false;
 
 /* functions needed during run phase */
 static void AcquireMetadataLocks(List *taskList);
-static int64 ExecuteSingleModifyTask(CitusScanState *scanState, Task *task, CmdType
+static int64 ExecuteSingleModifyTask(CitusScanState *scanState, Task *task,
-									 operation, bool alwaysThrowErrorOnFailure, bool
+									 RowModifyLevel modLevel,
-									 expectResults);
+									 bool alwaysThrowErrorOnFailure, bool expectResults);
 static void ExecuteSingleSelectTask(CitusScanState *scanState, Task *task);
 static List * BuildPlacementAccessList(int32 groupId, List *relationShardList,
 									   ShardPlacementAccessType accessType);
 static List * GetModifyConnections(Task *task, bool markCritical);
 static int64 ExecuteModifyTasks(List *taskList, bool expectResults,
 								ParamListInfo paramListInfo, CitusScanState *scanState);
+static void AcquireExecutorShardLockForRowModify(Task *task, RowModifyLevel modLevel);
+static void AcquireExecutorShardLocksForRelationRowLockList(Task *task);
 static bool RequiresConsistentSnapshot(Task *task);
 static void RouterMultiModifyExecScan(CustomScanState *node);
 static void RouterSequentialModifyExecScan(CustomScanState *node);
@@ -138,27 +140,18 @@ AcquireMetadataLocks(List *taskList)
 }
 
 
-/*
+static void
- * AcquireExecutorShardLock acquires a lock on the shard for the given task and
+AcquireExecutorShardLockForRowModify(Task *task, RowModifyLevel modLevel)
- * command type if necessary to avoid divergence between multiple replicas of
- * the same shard. No lock is obtained when there is only one replica.
- *
- * The function determines the appropriate lock mode based on the commutativity
- * rule of the command. In each case, it uses a lock mode that enforces the
- * commutativity rule.
- *
- * The mapping is overridden when all_modifications_commutative is set to true.
- * In that case, all modifications are treated as commutative, which can be used
- * to communicate that the application is only generating commutative
- * UPDATE/DELETE/UPSERT commands and exclusive locks are unnecessary.
- */
-void
-AcquireExecutorShardLock(Task *task, CmdType commandType)
 {
 	LOCKMODE lockMode = NoLock;
 	int64 shardId = task->anchorShardId;
 
-	if (commandType == CMD_SELECT)
+	if (shardId == INVALID_SHARD_ID)
+	{
+		return;
+	}
+
+	if (modLevel <= ROW_MODIFY_READONLY)
 	{
 		/*
 		 * The executor shard lock is used to maintain consistency between
@@ -205,24 +198,7 @@ AcquireExecutorShardLock(Task *task, CmdType commandType)
 
 		lockMode = RowExclusiveLock;
 	}
-	else if (task->upsertQuery || commandType == CMD_UPDATE || commandType == CMD_DELETE)
+	else if (modLevel < ROW_MODIFY_NONCOMMUTATIVE)
-	{
-		/*
-		 * UPDATE/DELETE/UPSERT commands do not commute with other modifications
-		 * since the rows modified by one command may be affected by the outcome
-		 * of another command.
-		 *
-		 * We need to handle upsert before INSERT, because PostgreSQL models
-		 * upsert commands as INSERT with an ON CONFLICT section.
-		 *
-		 * ExclusiveLock conflicts with all lock types used by modifications
-		 * and therefore prevents other modifications from running
-		 * concurrently.
-		 */
-
-		lockMode = ExclusiveLock;
-	}
-	else if (commandType == CMD_INSERT)
 	{
 		/*
 		 * An INSERT commutes with other INSERT commands, since performing them
@@ -245,16 +221,34 @@ AcquireExecutorShardLock(Task *task, CmdType commandType)
 	}
 	else
 	{
-		ereport(ERROR, (errmsg("unrecognized operation code: %d", (int) commandType)));
+		/*
+		 * UPDATE/DELETE/UPSERT commands do not commute with other modifications
+		 * since the rows modified by one command may be affected by the outcome
+		 * of another command.
+		 *
+		 * We need to handle upsert before INSERT, because PostgreSQL models
+		 * upsert commands as INSERT with an ON CONFLICT section.
+		 *
+		 * ExclusiveLock conflicts with all lock types used by modifications
+		 * and therefore prevents other modifications from running
+		 * concurrently.
+		 */
+
+		lockMode = ExclusiveLock;
 	}
 
-	if (shardId != INVALID_SHARD_ID && lockMode != NoLock)
+	if (lockMode != NoLock)
 	{
 		ShardInterval *shardInterval = LoadShardInterval(shardId);
 
 		SerializeNonCommutativeWrites(list_make1(shardInterval), lockMode);
 	}
+}
 
+
+static void
+AcquireExecutorShardLocksForRelationRowLockList(Task *task)
+{
 	/*
 	 * If lock clause exists and it effects any reference table, we need to get
 	 * lock on shard resource. Type of lock is determined by the type of row lock
@@ -301,6 +295,28 @@ AcquireExecutorShardLock(Task *task, CmdType commandType)
 			}
 		}
 	}
+}
+
+
+/*
+ * AcquireExecutorShardLocks acquires locks on shards for the given task if
+ * necessary to avoid divergence between multiple replicas of the same shard.
+ * No lock is obtained when there is only one replica.
+ *
+ * The function determines the appropriate lock mode based on the commutativity
+ * rule of the command. In each case, it uses a lock mode that enforces the
+ * commutativity rule.
+ *
+ * The mapping is overridden when all_modifications_commutative is set to true.
+ * In that case, all modifications are treated as commutative, which can be used
+ * to communicate that the application is only generating commutative
+ * UPDATE/DELETE/UPSERT commands and exclusive locks are unnecessary.
+ */
+void
+AcquireExecutorShardLocks(Task *task, RowModifyLevel modLevel)
+{
+	AcquireExecutorShardLockForRowModify(task, modLevel);
+	AcquireExecutorShardLocksForRelationRowLockList(task);
 
 	/*
 	 * If the task has a subselect, then we may need to lock the shards from which
@@ -717,16 +733,16 @@ RouterSequentialModifyExecScan(CustomScanState *node)
 {
 	CitusScanState *scanState = (CitusScanState *) node;
 	DistributedPlan *distributedPlan = scanState->distributedPlan;
+	RowModifyLevel modLevel = distributedPlan->modLevel;
 	bool hasReturning = distributedPlan->hasReturning;
 	Job *workerJob = distributedPlan->workerJob;
 	List *taskList = workerJob->taskList;
 	EState *executorState = ScanStateGetExecutorState(scanState);
-	CmdType operation = scanState->distributedPlan->operation;
 
 	Assert(!scanState->finishedRemoteScan);
 
 	executorState->es_processed +=
-		ExecuteModifyTasksSequentially(scanState, taskList, operation, hasReturning);
+		ExecuteModifyTasksSequentially(scanState, taskList, modLevel, hasReturning);
 }
 
 
@@ -1056,7 +1072,7 @@ CreatePlacementAccess(ShardPlacement *placement, ShardPlacementAccessType access
  * The function returns affectedTupleCount if applicable.
  */
 static int64
-ExecuteSingleModifyTask(CitusScanState *scanState, Task *task, CmdType operation,
+ExecuteSingleModifyTask(CitusScanState *scanState, Task *task, RowModifyLevel modLevel,
 						bool alwaysThrowErrorOnFailure, bool expectResults)
 {
 	EState *executorState = NULL;
@@ -1107,10 +1123,9 @@ ExecuteSingleModifyTask(CitusScanState *scanState, Task *task, CmdType operation
 	 * acquire the necessary locks on the relations, and blocks any
 	 * unsafe concurrent operations.
 	 */
-	if (operation == CMD_INSERT || operation == CMD_UPDATE ||
+	if (modLevel > ROW_MODIFY_NONE)
-		operation == CMD_DELETE || operation == CMD_SELECT)
 	{
-		AcquireExecutorShardLock(task, operation);
+		AcquireExecutorShardLocks(task, modLevel);
 	}
 
 	/* try to execute modification on all placements */
@@ -1356,9 +1371,9 @@ ExecuteModifyTasksWithoutResults(List *taskList)
  * and ignores the results.
  */
 int64
-ExecuteModifyTasksSequentiallyWithoutResults(List *taskList, CmdType operation)
+ExecuteModifyTasksSequentiallyWithoutResults(List *taskList, RowModifyLevel modLevel)
 {
-	return ExecuteModifyTasksSequentially(NULL, taskList, operation, false);
+	return ExecuteModifyTasksSequentially(NULL, taskList, modLevel, false);
 }
 
 
@@ -1373,7 +1388,7 @@ ExecuteModifyTasksSequentiallyWithoutResults(List *taskList, CmdType operation)
  */
 int64
 ExecuteModifyTasksSequentially(CitusScanState *scanState, List *taskList,
-							   CmdType operation, bool hasReturning)
+							   RowModifyLevel modLevel, bool hasReturning)
 {
 	ListCell *taskCell = NULL;
 	bool multipleTasks = list_length(taskList) > 1;
@@ -1422,7 +1437,7 @@ ExecuteModifyTasksSequentially(CitusScanState *scanState, List *taskList,
 		bool expectResults = (hasReturning || task->relationRowLockList != NIL);
 
 		affectedTupleCount +=
-			ExecuteSingleModifyTask(scanState, task, operation, alwaysThrowErrorOnFailure,
+			ExecuteSingleModifyTask(scanState, task, modLevel, alwaysThrowErrorOnFailure,
 									expectResults);
 	}
 

src/backend/distributed/executor/multi_server_executor.c

@@ -95,7 +95,7 @@ JobExecutorType(DistributedPlan *distributedPlan)
 		return MULTI_EXECUTOR_COORDINATOR_INSERT_SELECT;
 	}
 
-	Assert(distributedPlan->operation == CMD_SELECT);
+	Assert(distributedPlan->modLevel == ROW_MODIFY_READONLY);
 
 	workerNodeList = ActiveReadableNodeList();
 	workerNodeCount = list_length(workerNodeList);

src/backend/distributed/master/master_stage_protocol.c

@@ -592,7 +592,7 @@ CreateShardsOnWorkersViaExecutor(Oid distributedRelationId, List *shardPlacement
 		poolSize = MaxAdaptiveExecutorPoolSize;
 	}
 
-	ExecuteTaskList(CMD_UTILITY, taskList, poolSize);
+	ExecuteTaskList(ROW_MODIFY_NONE, taskList, poolSize);
 }
 
 

src/backend/distributed/metadata/metadata_sync.c

@@ -69,7 +69,7 @@ PG_FUNCTION_INFO_V1(stop_metadata_sync_to_node);
  * start_metadata_sync_to_node function creates the metadata in a worker for preparing the
  * worker for accepting queries. The function first sets the localGroupId of the worker
  * so that the worker knows which tuple in pg_dist_node table represents itself. After
- * that, SQL statetemens for re-creating metadata of MX-eligible distributed tables are
+ * that, SQL statements for re-creating metadata of MX-eligible distributed tables are
  * sent to the worker. Finally, the hasmetadata column of the target node in pg_dist_node
  * is marked as true.
  */
@@ -1188,7 +1188,7 @@ DetachPartitionCommandList(void)
 
 	/*
 	 * We probably do not need this but as an extra precaution, we are enabling
-	 * DDL propagation to swtich back to original state.
+	 * DDL propagation to switch back to original state.
 	 */
 	detachPartitionCommandList = lappend(detachPartitionCommandList,
 										 ENABLE_DDL_PROPAGATION);

src/backend/distributed/planner/deparse_shard_query.c

@@ -70,7 +70,7 @@ RebuildQueryStrings(Query *originalQuery, List *taskList)
 
 			query = copyObject(originalQuery);
 
-			copiedInsertRte = ExtractInsertRangeTableEntry(query);
+			copiedInsertRte = ExtractResultRelationRTE(query);
 			copiedSubqueryRte = ExtractSelectRangeTableEntry(query);
 			copiedSubquery = copiedSubqueryRte->subquery;
 
@@ -92,7 +92,8 @@ RebuildQueryStrings(Query *originalQuery, List *taskList)
 
 			UpdateRelationToShardNames((Node *) copiedSubquery, relationShardList);
 		}
-		else if (task->upsertQuery || valuesRTE != NULL)
+		else if (query->commandType == CMD_INSERT && (query->onConflict != NULL ||
+													  valuesRTE != NULL))
 		{
 			RangeTblEntry *rangeTableEntry = NULL;
 

src/backend/distributed/planner/distributed_planner.c

@@ -50,8 +50,8 @@ int MultiTaskQueryLogLevel = MULTI_TASK_QUERY_INFO_OFF; /* multi-task query log
 static uint64 NextPlanId = 1;
 
 
-/* local function forward declarations */
 static bool ListContainsDistributedTableRTE(List *rangeTableList);
+static bool IsUpdateOrDelete(Query *query);
 static PlannedStmt * CreateDistributedPlannedStmt(uint64 planId, PlannedStmt *localPlan,
 												  Query *originalQuery, Query *query,
 												  ParamListInfo boundParams,
@@ -423,22 +423,6 @@ IsModifyCommand(Query *query)
 }
 
 
-/*
- * IsMultiShardModifyPlan returns true if the given plan was generated for
- * multi shard update or delete query.
- */
-bool
-IsMultiShardModifyPlan(DistributedPlan *distributedPlan)
-{
-	if (IsUpdateOrDelete(distributedPlan) && IsMultiTaskPlan(distributedPlan))
-	{
-		return true;
-	}
-
-	return false;
-}
-
-
 /*
  * IsMultiTaskPlan returns true if job contains multiple tasks.
  */
@@ -457,19 +441,13 @@ IsMultiTaskPlan(DistributedPlan *distributedPlan)
 
 
 /*
- * IsUpdateOrDelete returns true if the query performs update or delete.
+ * IsUpdateOrDelete returns true if the query performs an update or delete.
  */
 bool
-IsUpdateOrDelete(DistributedPlan *distributedPlan)
+IsUpdateOrDelete(Query *query)
 {
-	CmdType commandType = distributedPlan->operation;
+	return query->commandType == CMD_UPDATE ||
-
+		   query->commandType == CMD_DELETE;
-	if (commandType == CMD_UPDATE || commandType == CMD_DELETE)
-	{
-		return true;
-	}
-
-	return false;
 }
 
 
@@ -480,15 +458,7 @@ IsUpdateOrDelete(DistributedPlan *distributedPlan)
 bool
 IsModifyDistributedPlan(DistributedPlan *distributedPlan)
 {
-	bool isModifyDistributedPlan = false;
+	return distributedPlan->modLevel > ROW_MODIFY_READONLY;
-	CmdType operation = distributedPlan->operation;
-
-	if (operation == CMD_INSERT || operation == CMD_UPDATE || operation == CMD_DELETE)
-	{
-		isModifyDistributedPlan = true;
-	}
-
-	return isModifyDistributedPlan;
 }
 
 
@@ -569,11 +539,12 @@ CreateDistributedPlannedStmt(uint64 planId, PlannedStmt *localPlan, Query *origi
 	 * query planning failed (possibly) due to prepared statement parameters or
 	 * if it is planned as a multi shard modify query.
 	 */
-	if ((distributedPlan->planningError || IsMultiShardModifyPlan(distributedPlan)) &&
+	if ((distributedPlan->planningError ||
+		 (IsUpdateOrDelete(originalQuery) && IsMultiTaskPlan(distributedPlan))) &&
 		hasUnresolvedParams)
 	{
 		/*
-		 * Arbitraryly high cost, but low enough that it can be added up
+		 * Arbitrarily high cost, but low enough that it can be added up
 		 * without overflowing by choose_custom_plan().
 		 */
 		resultPlan->planTree->total_cost = FLT_MAX / 100000000;

src/backend/distributed/planner/insert_select_planner.c

@@ -85,7 +85,7 @@ InsertSelectIntoDistributedTable(Query *query)
 
 	if (insertSelectQuery)
 	{
-		RangeTblEntry *insertRte = ExtractInsertRangeTableEntry(query);
+		RangeTblEntry *insertRte = ExtractResultRelationRTE(query);
 		if (IsDistributedTable(insertRte->relid))
 		{
 			return true;
@@ -108,7 +108,7 @@ InsertSelectIntoLocalTable(Query *query)
 
 	if (insertSelectQuery)
 	{
-		RangeTblEntry *insertRte = ExtractInsertRangeTableEntry(query);
+		RangeTblEntry *insertRte = ExtractResultRelationRTE(query);
 		if (!IsDistributedTable(insertRte->relid))
 		{
 			return true;
@@ -220,7 +220,7 @@ CreateDistributedInsertSelectPlan(Query *originalQuery,
 	Job *workerJob = NULL;
 	uint64 jobId = INVALID_JOB_ID;
 	DistributedPlan *distributedPlan = CitusMakeNode(DistributedPlan);
-	RangeTblEntry *insertRte = ExtractInsertRangeTableEntry(originalQuery);
+	RangeTblEntry *insertRte = ExtractResultRelationRTE(originalQuery);
 	RangeTblEntry *subqueryRte = ExtractSelectRangeTableEntry(originalQuery);
 	Oid targetRelationId = insertRte->relid;
 	DistTableCacheEntry *targetCacheEntry = DistributedTableCacheEntry(targetRelationId);
@@ -230,7 +230,7 @@ CreateDistributedInsertSelectPlan(Query *originalQuery,
 	bool allReferenceTables = relationRestrictionContext->allReferenceTables;
 	bool allDistributionKeysInQueryAreEqual = false;
 
-	distributedPlan->operation = originalQuery->commandType;
+	distributedPlan->modLevel = RowModifyLevelForQuery(originalQuery);
 
 	/*
 	 * Error semantics for INSERT ... SELECT queries are different than regular
@@ -423,7 +423,7 @@ RouterModifyTaskForShardInterval(Query *originalQuery, ShardInterval *shardInter
 								 bool safeToPushdownSubquery)
 {
 	Query *copiedQuery = copyObject(originalQuery);
-	RangeTblEntry *copiedInsertRte = ExtractInsertRangeTableEntry(copiedQuery);
+	RangeTblEntry *copiedInsertRte = ExtractResultRelationRTE(copiedQuery);
 	RangeTblEntry *copiedSubqueryRte = ExtractSelectRangeTableEntry(copiedQuery);
 	Query *copiedSubquery = (Query *) copiedSubqueryRte->subquery;
 
@@ -443,7 +443,6 @@ RouterModifyTaskForShardInterval(Query *originalQuery, ShardInterval *shardInter
 	uint64 jobId = INVALID_JOB_ID;
 	List *insertShardPlacementList = NULL;
 	List *intersectedPlacementList = NULL;
-	bool upsertQuery = false;
 	bool replacePrunedQueryWithDummy = false;
 	bool allReferenceTables =
 		plannerRestrictionContext->relationRestrictionContext->allReferenceTables;
@@ -566,12 +565,6 @@ RouterModifyTaskForShardInterval(Query *originalQuery, ShardInterval *shardInter
 	/* this is required for correct deparsing of the query */
 	ReorderInsertSelectTargetLists(copiedQuery, copiedInsertRte, copiedSubqueryRte);
 
-	/* set the upsert flag */
-	if (originalQuery->onConflict != NULL)
-	{
-		upsertQuery = true;
-	}
-
 	/* setting an alias simplifies deparsing of RETURNING */
 	if (copiedInsertRte->alias == NULL)
 	{
@@ -589,7 +582,6 @@ RouterModifyTaskForShardInterval(Query *originalQuery, ShardInterval *shardInter
 	modifyTask->dependedTaskList = NULL;
 	modifyTask->anchorShardId = shardId;
 	modifyTask->taskPlacementList = insertShardPlacementList;
-	modifyTask->upsertQuery = upsertQuery;
 	modifyTask->relationShardList = relationShardList;
 	modifyTask->replicationModel = cacheEntry->replicationModel;
 
@@ -1138,11 +1130,11 @@ CreateCoordinatorInsertSelectPlan(uint64 planId, Query *parse)
 	Query *selectQuery = NULL;
 
 	RangeTblEntry *selectRte = ExtractSelectRangeTableEntry(insertSelectQuery);
-	RangeTblEntry *insertRte = ExtractInsertRangeTableEntry(insertSelectQuery);
+	RangeTblEntry *insertRte = ExtractResultRelationRTE(insertSelectQuery);
 	Oid targetRelationId = insertRte->relid;
 
 	DistributedPlan *distributedPlan = CitusMakeNode(DistributedPlan);
-	distributedPlan->operation = CMD_INSERT;
+	distributedPlan->modLevel = RowModifyLevelForQuery(insertSelectQuery);
 
 	distributedPlan->planningError =
 		CoordinatorInsertSelectSupported(insertSelectQuery);
@@ -1243,7 +1235,7 @@ CoordinatorInsertSelectSupported(Query *insertSelectQuery)
 		return deferredError;
 	}
 
-	insertRte = ExtractInsertRangeTableEntry(insertSelectQuery);
+	insertRte = ExtractResultRelationRTE(insertSelectQuery);
 	if (PartitionMethod(insertRte->relid) == DISTRIBUTE_BY_APPEND)
 	{
 		return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
@@ -1347,7 +1339,7 @@ TwoPhaseInsertSelectTaskList(Oid targetRelationId, Query *insertSelectQuery,
 	 * then deparse it.
 	 */
 	Query *insertResultQuery = copyObject(insertSelectQuery);
-	RangeTblEntry *insertRte = ExtractInsertRangeTableEntry(insertResultQuery);
+	RangeTblEntry *insertRte = ExtractResultRelationRTE(insertResultQuery);
 	RangeTblEntry *selectRte = ExtractSelectRangeTableEntry(insertResultQuery);
 
 	DistTableCacheEntry *targetCacheEntry = DistributedTableCacheEntry(targetRelationId);
@@ -1441,7 +1433,6 @@ TwoPhaseInsertSelectTaskList(Oid targetRelationId, Query *insertSelectQuery,
 		modifyTask->dependedTaskList = NULL;
 		modifyTask->anchorShardId = shardId;
 		modifyTask->taskPlacementList = insertShardPlacementList;
-		modifyTask->upsertQuery = insertResultQuery->onConflict != NULL;
 		modifyTask->relationShardList = list_make1(relationShard);
 		modifyTask->replicationModel = targetCacheEntry->replicationModel;
 

src/backend/distributed/planner/multi_physical_planner.c

@@ -228,7 +228,7 @@ CreatePhysicalDistributedPlan(MultiTreeRoot *multiTree,
 	distributedPlan->workerJob = workerJob;
 	distributedPlan->masterQuery = masterQuery;
 	distributedPlan->routerExecutable = DistributedPlanRouterExecutable(distributedPlan);
-	distributedPlan->operation = CMD_SELECT;
+	distributedPlan->modLevel = ROW_MODIFY_READONLY;
 
 	return distributedPlan;
 }
@@ -2430,7 +2430,6 @@ QueryPushdownTaskCreate(Query *originalQuery, int shardIndex,
 	subqueryTask->dependedTaskList = NULL;
 	subqueryTask->anchorShardId = anchorShardId;
 	subqueryTask->taskPlacementList = selectPlacementList;
-	subqueryTask->upsertQuery = false;
 	subqueryTask->relationShardList = relationShardList;
 
 	return subqueryTask;
@@ -4310,6 +4309,41 @@ GenerateSyntheticShardIntervalArray(int partitionCount)
 }
 
 
+/*
+ * Determine RowModifyLevel required for given query
+ */
+RowModifyLevel
+RowModifyLevelForQuery(Query *query)
+{
+	CmdType commandType = query->commandType;
+
+	if (commandType == CMD_SELECT)
+	{
+		return ROW_MODIFY_READONLY;
+	}
+
+	if (commandType == CMD_INSERT)
+	{
+		if (query->onConflict == NULL)
+		{
+			return ROW_MODIFY_COMMUTATIVE;
+		}
+		else
+		{
+			return ROW_MODIFY_NONCOMMUTATIVE;
+		}
+	}
+
+	if (commandType == CMD_UPDATE ||
+		commandType == CMD_DELETE)
+	{
+		return ROW_MODIFY_NONCOMMUTATIVE;
+	}
+
+	return ROW_MODIFY_NONE;
+}
+
+
 /*
  * ColumnName resolves the given column's name. The given column could belong to
  * a regular table or to an intermediate table formed to execute a distributed

src/backend/distributed/planner/multi_router_planner.c

@@ -165,7 +165,7 @@ static void ReorderTaskPlacementsByTaskAssignmentPolicy(Job *job,
 
 /*
  * CreateRouterPlan attempts to create a router executor plan for the given
- * SELECT statement.  ->planningError is set if planning fails.
+ * SELECT statement. ->planningError is set if planning fails.
  */
 DistributedPlan *
 CreateRouterPlan(Query *originalQuery, Query *query,
@@ -186,8 +186,8 @@ CreateRouterPlan(Query *originalQuery, Query *query,
 
 
 /*
- * CreateModifyPlan attempts to create a plan the given modification
+ * CreateModifyPlan attempts to create a plan for the given modification
- * statement.  If planning fails ->planningError is set to a description of
+ * statement. If planning fails ->planningError is set to a description of
  * the failure.
  */
 DistributedPlan *
@@ -198,7 +198,7 @@ CreateModifyPlan(Query *originalQuery, Query *query,
 	DistributedPlan *distributedPlan = CitusMakeNode(DistributedPlan);
 	bool multiShardQuery = false;
 
-	distributedPlan->operation = query->commandType;
+	distributedPlan->modLevel = RowModifyLevelForQuery(query);
 
 	distributedPlan->planningError = ModifyQuerySupported(query, originalQuery,
 														  multiShardQuery,
@@ -244,8 +244,8 @@ CreateModifyPlan(Query *originalQuery, Query *query,
  * CreateSingleTaskRouterPlan creates a physical plan for given query. The created plan is
  * either a modify task that changes a single shard, or a router task that returns
  * query results from a single worker. Supported modify queries (insert/update/delete)
- * are router plannable by default. If query is not router plannable then either NULL is
+ * are router plannable by default. If query is not router plannable the returned plan
- * returned, or the returned plan has planningError set to a description of the problem.
+ * has planningError set to a description of the problem.
  */
 static void
 CreateSingleTaskRouterPlan(DistributedPlan *distributedPlan, Query *originalQuery,
@@ -254,7 +254,7 @@ CreateSingleTaskRouterPlan(DistributedPlan *distributedPlan, Query *originalQuer
 {
 	Job *job = NULL;
 
-	distributedPlan->operation = query->commandType;
+	distributedPlan->modLevel = RowModifyLevelForQuery(query);
 
 	/* we cannot have multi shard update/delete query via this code path */
 	job = RouterJob(originalQuery, plannerRestrictionContext,
@@ -491,7 +491,7 @@ ModifyQueryResultRelationId(Query *query)
 						errmsg("input query is not a modification query")));
 	}
 
-	resultRte = ExtractInsertRangeTableEntry(query);
+	resultRte = ExtractResultRelationRTE(query);
 	Assert(OidIsValid(resultRte->relid));
 
 	return resultRte->relid;
@@ -512,20 +512,12 @@ ResultRelationOidForQuery(Query *query)
 
 
 /*
- * ExtractInsertRangeTableEntry returns the INSERT'ed table's range table entry.
+ * ExtractResultRelationRTE returns the table's resultRelation range table entry.
- * Note that the function expects and asserts that the input query be
- * an INSERT...SELECT query.
  */
 RangeTblEntry *
-ExtractInsertRangeTableEntry(Query *query)
+ExtractResultRelationRTE(Query *query)
 {
-	int resultRelation = query->resultRelation;
+	return rt_fetch(query->resultRelation, query->rtable);
-	List *rangeTableList = query->rtable;
-	RangeTblEntry *insertRTE = NULL;
-
-	insertRTE = rt_fetch(resultRelation, rangeTableList);
-
-	return insertRTE;
 }
 
 
@@ -1086,7 +1078,7 @@ HasDangerousJoinUsing(List *rtableList, Node *joinTreeNode)
 		{
 			/*
 			 * Yes, so check each join alias var to see if any of them are not
-			 * simple references to underlying columns.  If so, we have a
+			 * simple references to underlying columns. If so, we have a
 			 * dangerous situation and must pick unique aliases.
 			 */
 			RangeTblEntry *joinRTE = rt_fetch(joinExpr->rtindex, rtableList);
@@ -1129,14 +1121,8 @@ HasDangerousJoinUsing(List *rtableList, Node *joinTreeNode)
 bool
 UpdateOrDeleteQuery(Query *query)
 {
-	CmdType commandType = query->commandType;
+	return query->commandType == CMD_UPDATE ||
-
+		   query->commandType == CMD_DELETE;
-	if (commandType == CMD_UPDATE || commandType == CMD_DELETE)
-	{
-		return true;
-	}
-
-	return false;
 }
 
 
@@ -1538,11 +1524,6 @@ RouterInsertTaskList(Query *query, DeferredErrorMessage **planningError)
 		modifyTask->replicationModel = cacheEntry->replicationModel;
 		modifyTask->rowValuesLists = modifyRoute->rowValuesLists;
 
-		if (query->onConflict != NULL)
-		{
-			modifyTask->upsertQuery = true;
-		}
-
 		insertTaskList = lappend(insertTaskList, modifyTask);
 	}
 
@@ -1572,7 +1553,6 @@ CreateTask(TaskType taskType)
 	task->shardInterval = NULL;
 	task->assignmentConstrained = false;
 	task->taskExecution = NULL;
-	task->upsertQuery = false;
 	task->replicationModel = REPLICATION_MODEL_INVALID;
 	task->relationRowLockList = NIL;
 
@@ -1858,8 +1838,8 @@ SingleShardModifyTaskList(Query *query, uint64 jobId, List *relationShardList,
 
 
 /*
- * GetUpdateOrDeleteRTE checks query if it has an UPDATE or DELETE RTE. If it finds
+ * GetUpdateOrDeleteRTE checks query if it has an UPDATE or DELETE RTE.
- * it returns it.
+ * Returns that RTE if found.
  */
 static RangeTblEntry *
 GetUpdateOrDeleteRTE(Query *query)

src/backend/distributed/utils/citus_copyfuncs.c

@@ -103,12 +103,12 @@ CopyNodeDistributedPlan(COPYFUNC_ARGS)
 	DECLARE_FROM_AND_NEW_NODE(DistributedPlan);
 
 	COPY_SCALAR_FIELD(planId);
-	COPY_SCALAR_FIELD(operation);
+	COPY_SCALAR_FIELD(modLevel);
 	COPY_SCALAR_FIELD(hasReturning);
+	COPY_SCALAR_FIELD(routerExecutable);
 
 	COPY_NODE_FIELD(workerJob);
 	COPY_NODE_FIELD(masterQuery);
-	COPY_SCALAR_FIELD(routerExecutable);
 	COPY_SCALAR_FIELD(queryId);
 	COPY_NODE_FIELD(relationIdList);
 
@@ -257,7 +257,6 @@ CopyNodeTask(COPYFUNC_ARGS)
 	COPY_NODE_FIELD(shardInterval);
 	COPY_SCALAR_FIELD(assignmentConstrained);
 	COPY_NODE_FIELD(taskExecution);
-	COPY_SCALAR_FIELD(upsertQuery);
 	COPY_SCALAR_FIELD(replicationModel);
 	COPY_SCALAR_FIELD(modifyWithSubquery);
 	COPY_NODE_FIELD(relationShardList);

src/backend/distributed/utils/citus_outfuncs.c

@@ -176,12 +176,12 @@ OutDistributedPlan(OUTFUNC_ARGS)
 	WRITE_NODE_TYPE("DISTRIBUTEDPLAN");
 
 	WRITE_UINT64_FIELD(planId);
-	WRITE_INT_FIELD(operation);
+	WRITE_ENUM_FIELD(modLevel, RowModifyLevel);
 	WRITE_BOOL_FIELD(hasReturning);
+	WRITE_BOOL_FIELD(routerExecutable);
 
 	WRITE_NODE_FIELD(workerJob);
 	WRITE_NODE_FIELD(masterQuery);
-	WRITE_BOOL_FIELD(routerExecutable);
 	WRITE_UINT64_FIELD(queryId);
 	WRITE_NODE_FIELD(relationIdList);
 
@@ -467,7 +467,6 @@ OutTask(OUTFUNC_ARGS)
 	WRITE_NODE_FIELD(shardInterval);
 	WRITE_BOOL_FIELD(assignmentConstrained);
 	WRITE_NODE_FIELD(taskExecution);
-	WRITE_BOOL_FIELD(upsertQuery);
 	WRITE_CHAR_FIELD(replicationModel);
 	WRITE_BOOL_FIELD(modifyWithSubquery);
 	WRITE_NODE_FIELD(relationShardList);

src/backend/distributed/utils/citus_readfuncs.c

@@ -202,12 +202,12 @@ ReadDistributedPlan(READFUNC_ARGS)
 	READ_LOCALS(DistributedPlan);
 
 	READ_UINT64_FIELD(planId);
-	READ_INT_FIELD(operation);
+	READ_ENUM_FIELD(modLevel, RowModifyLevel);
 	READ_BOOL_FIELD(hasReturning);
+	READ_BOOL_FIELD(routerExecutable);
 
 	READ_NODE_FIELD(workerJob);
 	READ_NODE_FIELD(masterQuery);
-	READ_BOOL_FIELD(routerExecutable);
 	READ_UINT64_FIELD(queryId);
 	READ_NODE_FIELD(relationIdList);
 
@@ -381,7 +381,6 @@ ReadTask(READFUNC_ARGS)
 	READ_NODE_FIELD(shardInterval);
 	READ_BOOL_FIELD(assignmentConstrained);
 	READ_NODE_FIELD(taskExecution);
-	READ_BOOL_FIELD(upsertQuery);
 	READ_CHAR_FIELD(replicationModel);
 	READ_BOOL_FIELD(modifyWithSubquery);
 	READ_NODE_FIELD(relationShardList);

src/include/distributed/distributed_planner.h

@@ -108,12 +108,10 @@ extern void multi_join_restriction_hook(PlannerInfo *root,
 										JoinType jointype,
 										JoinPathExtraData *extra);
 extern bool IsModifyCommand(Query *query);
-extern bool IsUpdateOrDelete(struct DistributedPlan *distributedPlan);
 extern bool IsModifyDistributedPlan(struct DistributedPlan *distributedPlan);
 extern void EnsurePartitionTableNotReplicated(Oid relationId);
 extern Node * ResolveExternalParams(Node *inputNode, ParamListInfo boundParams);
 extern bool IsMultiTaskPlan(struct DistributedPlan *distributedPlan);
-extern bool IsMultiShardModifyPlan(struct DistributedPlan *distributedPlan);
 extern RangeTblEntry * RemoteScanRangeTableEntry(List *columnNameList);
 extern int GetRTEIdentity(RangeTblEntry *rte);
 

src/include/distributed/multi_executor.h

@@ -38,13 +38,14 @@ extern void CitusExecutorStart(QueryDesc *queryDesc, int eflags);
 extern void CitusExecutorRun(QueryDesc *queryDesc, ScanDirection direction, uint64 count,
 							 bool execute_once);
 extern TupleTableSlot * AdaptiveExecutor(CustomScanState *node);
-extern uint64 ExecuteTaskListExtended(CmdType operation, List *taskList,
+extern uint64 ExecuteTaskListExtended(RowModifyLevel modLevel, List *taskList,
 									  TupleDesc tupleDescriptor,
 									  Tuplestorestate *tupleStore,
 									  bool hasReturning, int targetPoolSize);
 extern void ExecuteUtilityTaskListWithoutResults(List *taskList, int targetPoolSize,
 												 bool forceSequentialExecution);
-extern uint64 ExecuteTaskList(CmdType operation, List *taskList, int targetPoolSize);
+extern uint64 ExecuteTaskList(RowModifyLevel modLevel, List *taskList, int
+							  targetPoolSize);
 extern TupleTableSlot * CitusExecScan(CustomScanState *node);
 extern TupleTableSlot * ReturnTupleFromTuplestore(CitusScanState *scanState);
 extern void LoadTuplesIntoTupleStore(CitusScanState *citusScanState, Job *workerJob);

src/include/distributed/multi_physical_planner.h

@@ -110,6 +110,15 @@ typedef enum
 } BoundaryNodeJobType;
 
 
+/* Enumeration that specifies extent of DML modifications */
+typedef enum RowModifyLevel
+{
+	ROW_MODIFY_NONE = 0,
+	ROW_MODIFY_READONLY = 1,
+	ROW_MODIFY_COMMUTATIVE = 2,
+	ROW_MODIFY_NONCOMMUTATIVE = 3
+} RowModifyLevel;
+
 /*
  * Job represents a logical unit of work that contains one set of data transfers
  * in our physical plan. The physical planner maps each SQL query into one or
@@ -183,7 +192,6 @@ typedef struct Task
 	ShardInterval *shardInterval;  /* only applies to merge tasks */
 	bool assignmentConstrained;    /* only applies to merge tasks */
 	TaskExecution *taskExecution;  /* used by task tracker executor */
-	bool upsertQuery;              /* only applies to modify tasks */
 	char replicationModel;         /* only applies to modify tasks */
 
 	List *relationRowLockList;
@@ -229,21 +237,21 @@ typedef struct DistributedPlan
 	/* unique identifier of the plan within the session */
 	uint64 planId;
 
-	/* type of command to execute (SELECT/INSERT/...) */
+	/* specifies nature of modifications in query */
-	CmdType operation;
+	RowModifyLevel modLevel;
 
 	/* specifies whether a DML command has a RETURNING */
 	bool hasReturning;
 
+	/* a router executable query is executed entirely on a worker */
+	bool routerExecutable;
+
 	/* job tree containing the tasks to be executed on workers */
 	Job *workerJob;
 
 	/* local query that merges results from the workers */
 	Query *masterQuery;
 
-	/* a router executable query is executed entirely on a worker */
-	bool routerExecutable;
-
 	/* query identifier (copied from the top-level PlannedStmt) */
 	uint64 queryId;
 
@@ -340,6 +348,7 @@ extern bool ShardIntervalsOverlap(ShardInterval *firstInterval,
 								  ShardInterval *secondInterval);
 extern bool CoPartitionedTables(Oid firstRelationId, Oid secondRelationId);
 extern ShardInterval ** GenerateSyntheticShardIntervalArray(int partitionCount);
+extern RowModifyLevel RowModifyLevelForQuery(Query *query);
 
 
 /* function declarations for Task and Task list operations */

src/include/distributed/multi_router_executor.h

@@ -46,15 +46,15 @@ extern void ExecuteMultipleTasks(CitusScanState *scanState, List *taskList,
 								 bool isModificationQuery, bool expectResults);
 
 int64 ExecuteModifyTasksSequentially(CitusScanState *scanState, List *taskList,
-									 CmdType operation, bool hasReturning);
+									 RowModifyLevel modLevel, bool hasReturning);
 extern int64 ExecuteModifyTasksWithoutResults(List *taskList);
 extern int64 ExecuteModifyTasksSequentiallyWithoutResults(List *taskList,
-														  CmdType operation);
+														  RowModifyLevel modLevel);
 extern ShardPlacementAccess * CreatePlacementAccess(ShardPlacement *placement,
 													ShardPlacementAccessType accessType);
 
 /* helper functions */
-extern void AcquireExecutorShardLock(Task *task, CmdType commandType);
+extern void AcquireExecutorShardLocks(Task *task, RowModifyLevel modLevel);
 extern void AcquireExecutorMultiShardLocks(List *taskList);
 extern ShardPlacementAccess * CreatePlacementAccess(ShardPlacement *placement,
 													ShardPlacementAccessType accessType);

src/include/distributed/multi_router_planner.h

@@ -61,7 +61,7 @@ extern RelationRestrictionContext * CopyRelationRestrictionContext(
 extern Oid ExtractFirstDistributedTableId(Query *query);
 extern RangeTblEntry * ExtractSelectRangeTableEntry(Query *query);
 extern Oid ModifyQueryResultRelationId(Query *query);
-extern RangeTblEntry * ExtractInsertRangeTableEntry(Query *query);
+extern RangeTblEntry * ExtractResultRelationRTE(Query *query);
 extern RangeTblEntry * ExtractDistributedInsertValuesRTE(Query *query);
 extern bool IsMultiRowInsert(Query *query);
 extern void AddShardIntervalRestrictionToSelect(Query *subqery,

src/test/regress/bin/normalized_tests.lst

@@ -17,6 +17,8 @@ multi_subtransactions
 multi_modifying_xacts
 multi_insert_select
 sql_procedure
+multi_reference_table
+multi_create_table_constraints
 
 # the following tests' output are
 # normalized for EXPLAIN outputs