mirror of https://github.com/citusdata/citus.git
reimplement ExecuteUtilityTaskListWithoutResults for local execution and refactor its usages
parent
afc942c6af
commit
ed66517e95
|
@ -715,8 +715,10 @@ ExecuteDistributedDDLJob(DDLJob *ddlJob)
|
||||||
SendCommandToWorkersWithMetadata((char *) ddlJob->commandString);
|
SendCommandToWorkersWithMetadata((char *) ddlJob->commandString);
|
||||||
}
|
}
|
||||||
|
|
||||||
/* use adaptive executor when enabled */
|
/* local execution is not implemented for this code path */
|
||||||
ExecuteUtilityTaskListWithoutResults(ddlJob->taskList);
|
bool tryLocalExecution = false;
|
||||||
|
|
||||||
|
ExecuteUtilityTaskListWithoutResults(ddlJob->taskList, localExecutionSupported);
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
|
@ -727,8 +729,11 @@ ExecuteDistributedDDLJob(DDLJob *ddlJob)
|
||||||
|
|
||||||
PG_TRY();
|
PG_TRY();
|
||||||
{
|
{
|
||||||
/* use adaptive executor when enabled */
|
/* local execution is not implemented for this code path */
|
||||||
ExecuteUtilityTaskListWithoutResults(ddlJob->taskList);
|
bool tryLocalExecution = false;
|
||||||
|
|
||||||
|
ExecuteUtilityTaskListWithoutResults(ddlJob->taskList,
|
||||||
|
localExecutionSupported);
|
||||||
|
|
||||||
if (shouldSyncMetadata)
|
if (shouldSyncMetadata)
|
||||||
{
|
{
|
||||||
|
|
|
@ -108,8 +108,10 @@ PostprocessVacuumStmt(VacuumStmt *vacuumStmt, const char *vacuumCommand)
|
||||||
List *vacuumColumnList = VacuumColumnList(vacuumStmt, relationIndex);
|
List *vacuumColumnList = VacuumColumnList(vacuumStmt, relationIndex);
|
||||||
List *taskList = VacuumTaskList(relationId, vacuumParams, vacuumColumnList);
|
List *taskList = VacuumTaskList(relationId, vacuumParams, vacuumColumnList);
|
||||||
|
|
||||||
/* use adaptive executor when enabled */
|
/* local execution is not implemented for VACUUM commands */
|
||||||
ExecuteUtilityTaskListWithoutResults(taskList);
|
bool localExecutionSupported = false;
|
||||||
|
|
||||||
|
ExecuteUtilityTaskListWithoutResults(taskList, localExecutionSupported);
|
||||||
executedVacuumCount++;
|
executedVacuumCount++;
|
||||||
}
|
}
|
||||||
relationIndex++;
|
relationIndex++;
|
||||||
|
|
|
@ -814,13 +814,50 @@ AdjustDistributedExecutionAfterLocalExecution(DistributedExecution *execution)
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* ExecuteUtilityTaskListWithoutResults is a wrapper around executing task
|
* ExecuteUtilityTaskListWithoutResults is a wrapper around executing task
|
||||||
* list for utility commands. It simply calls in adaptive executor's task
|
* list for utility commands. For remote tasks, it simply calls in adaptive
|
||||||
* execution function.
|
* executor's task execution function. For local tasks (if any), kicks Process
|
||||||
|
* Utility via CitusProcessUtility for utility commands. As some local utility
|
||||||
|
* commands can trigger udf calls, this function also processes those udf calls
|
||||||
|
* locally.
|
||||||
*/
|
*/
|
||||||
void
|
void
|
||||||
ExecuteUtilityTaskListWithoutResults(List *taskList)
|
ExecuteUtilityTaskListWithoutResults(List *taskList, bool localExecutionSupported)
|
||||||
{
|
{
|
||||||
ExecuteTaskList(ROW_MODIFY_NONE, taskList, MaxAdaptiveExecutorPoolSize);
|
RowModifyLevel rowModifyLevel = ROW_MODIFY_NONE;
|
||||||
|
|
||||||
|
List *localTaskList = NIL;
|
||||||
|
List *remoteTaskList = NIL;
|
||||||
|
|
||||||
|
/*
|
||||||
|
* Divide tasks into two if localExecutionSupported is set to true and execute
|
||||||
|
* the local tasks
|
||||||
|
*/
|
||||||
|
if (localExecutionSupported && ShouldExecuteTasksLocally(taskList))
|
||||||
|
{
|
||||||
|
/*
|
||||||
|
* Either we are executing a utility command or a UDF call triggered
|
||||||
|
* by such a command, it has to be a modifying one
|
||||||
|
*/
|
||||||
|
bool readOnlyPlan = false;
|
||||||
|
|
||||||
|
/* set local (if any) & remote tasks */
|
||||||
|
ExtractLocalAndRemoteTasks(readOnlyPlan, taskList, &localTaskList,
|
||||||
|
&remoteTaskList);
|
||||||
|
|
||||||
|
/* execute local tasks */
|
||||||
|
ExecuteLocalUtilityTaskList(localTaskList);
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
/* all tasks should be executed via remote connections */
|
||||||
|
remoteTaskList = taskList;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* execute remote tasks if any */
|
||||||
|
if (list_length(remoteTaskList) > 0)
|
||||||
|
{
|
||||||
|
ExecuteTaskList(rowModifyLevel, remoteTaskList, MaxAdaptiveExecutorPoolSize);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
|
@ -84,7 +84,8 @@ extern uint64 ExecuteTaskListIntoTupleStore(RowModifyLevel modLevel, List *taskL
|
||||||
TupleDesc tupleDescriptor,
|
TupleDesc tupleDescriptor,
|
||||||
Tuplestorestate *tupleStore,
|
Tuplestorestate *tupleStore,
|
||||||
bool hasReturning);
|
bool hasReturning);
|
||||||
extern void ExecuteUtilityTaskListWithoutResults(List *taskList);
|
extern void ExecuteUtilityTaskListWithoutResults(List *taskList, bool
|
||||||
|
localExecutionSupported);
|
||||||
extern uint64 ExecuteTaskList(RowModifyLevel modLevel, List *taskList, int
|
extern uint64 ExecuteTaskList(RowModifyLevel modLevel, List *taskList, int
|
||||||
targetPoolSize);
|
targetPoolSize);
|
||||||
extern bool IsCitusCustomState(PlanState *planState);
|
extern bool IsCitusCustomState(PlanState *planState);
|
||||||
|
|
Loading…
Reference in New Issue