Move EnsureCompatibleLocalExecutionState to local_executor.c

src/backend/distributed/executor/adaptive_executor.c

@@ -1055,6 +1055,14 @@ ExecuteTaskListExtended(ExecutionParams *executionParams)
 			defaultTupleDest, &executionParams->xactProperties,
 			executionParams->jobIdList, executionParams->localExecutionSupported);
 
+	/*
+	 * If current transaction accessed local placements and task list includes
+	 * tasks that should be executed locally (accessing any of the local placements),
+	 * then we should error out as it would cause inconsistencies across the
+	 * remote connection and local execution.
+	 */
+	EnsureCompatibleLocalExecutionState(execution->remoteTaskList);
+
 	/* run the remote execution */
 	StartDistributedExecution(execution);
 	RunDistributedExecution(execution);

src/backend/distributed/executor/local_executor.c

@@ -915,6 +915,25 @@ TaskAccessesLocalNode(Task *task)
 }
 
 
+/*
+ * EnsureCompatibleLocalExecutionState makes sure that the tasks won't have
+ * any visibility problems because of local execution.
+ */
+void
+EnsureCompatibleLocalExecutionState(List *taskList)
+{
+	/*
+	 * We have LOCAL_EXECUTION_REQUIRED check here to avoid unnecessarily
+	 * iterating the task list in AnyTaskAccessesLocalNode.
+	 */
+	if (GetCurrentLocalExecutionStatus() == LOCAL_EXECUTION_REQUIRED &&
+		AnyTaskAccessesLocalNode(taskList))
+	{
+		ErrorIfTransactionAccessedPlacementsLocally();
+	}
+}
+
+
 /*
  * ErrorIfTransactionAccessedPlacementsLocally errors out if a local query
  * on any shard has already been executed in the same transaction.

src/include/distributed/local_executor.h

@@ -43,6 +43,7 @@ extern void ExecuteUtilityCommand(const char *utilityCommand);
 extern bool ShouldExecuteTasksLocally(List *taskList);
 extern bool AnyTaskAccessesLocalNode(List *taskList);
 extern bool TaskAccessesLocalNode(Task *task);
+extern void EnsureCompatibleLocalExecutionState(List *taskList);
 extern void ErrorIfTransactionAccessedPlacementsLocally(void);
 extern void DisableLocalExecution(void);
 extern void SetLocalExecutionStatus(LocalExecutionStatus newStatus);