external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INSERT/SELECT: Use ExecutePlan* instead of ExecuteSelect*

pull/3376/head
Hadi Moshayedi 2020-01-10 02:16:58 -08:00
parent a53b844939
commit d449c1857c
1 changed files with 35 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
src/backend/distributed/executor/insert_select_executor.c
View File

@ -52,14 +52,13 @@ static TupleTableSlot * CoordinatorInsertSelectExecScanInternal(CustomScanState
static Query * WrapSubquery(Query *subquery); static Query * WrapSubquery(Query *subquery);
static List * TwoPhaseInsertSelectTaskList(Oid targetRelationId, Query *insertSelectQuery, static List * TwoPhaseInsertSelectTaskList(Oid targetRelationId, Query *insertSelectQuery,
char *resultIdPrefix); char *resultIdPrefix);
static void ExecuteSelectIntoRelation(Oid targetRelationId, List *insertTargetList, static void ExecutePlanIntoRelation(Oid targetRelationId, List *insertTargetList,
Query *selectQuery, EState *executorState); PlannedStmt *selectPlan, EState *executorState);
static HTAB * ExecuteSelectIntoColocatedIntermediateResults(Oid targetRelationId, static HTAB * ExecutePlanIntoColocatedIntermediateResults(Oid targetRelationId,
List *insertTargetList, List *insertTargetList,
Query *selectQuery, PlannedStmt *selectPlan,
EState *executorState, EState *executorState,
char * char *intermediateResultIdPrefix);
intermediateResultIdPrefix);
static List * BuildColumnNameListFromTargetList(Oid targetRelationId, static List * BuildColumnNameListFromTargetList(Oid targetRelationId,
List *insertTargetList); List *insertTargetList);
static int PartitionColumnIndexFromColumnList(Oid relationId, List *columnNameList); static int PartitionColumnIndexFromColumnList(Oid relationId, List *columnNameList);
@ -106,6 +105,7 @@ CoordinatorInsertSelectExecScanInternal(CustomScanState *node)
if (!scanState->finishedRemoteScan) if (!scanState->finishedRemoteScan)
{ {
EState *executorState = ScanStateGetExecutorState(scanState); EState *executorState = ScanStateGetExecutorState(scanState);
ParamListInfo paramListInfo = executorState->es_param_list_info;
DistributedPlan *distributedPlan = scanState->distributedPlan; DistributedPlan *distributedPlan = scanState->distributedPlan;
Query *insertSelectQuery = copyObject(distributedPlan->insertSelectQuery); Query *insertSelectQuery = copyObject(distributedPlan->insertSelectQuery);
List *insertTargetList = insertSelectQuery->targetList; List *insertTargetList = insertSelectQuery->targetList;
@ -134,6 +134,18 @@ CoordinatorInsertSelectExecScanInternal(CustomScanState *node)
selectRte->subquery = selectQuery; selectRte->subquery = selectQuery;
ReorderInsertSelectTargetLists(insertSelectQuery, insertRte, selectRte); ReorderInsertSelectTargetLists(insertSelectQuery, insertRte, selectRte);
/*
* Make a copy of the query, since pg_plan_query may scribble on it and we
* want it to be replanned every time if it is stored in a prepared
* statement.
*/
selectQuery = copyObject(selectQuery);
/* plan the subquery, this may be another distributed query */
int cursorOptions = CURSOR_OPT_PARALLEL_OK;
PlannedStmt *selectPlan = pg_plan_query(selectQuery, cursorOptions,
paramListInfo);
/* /*
* If we are dealing with partitioned table, we also need to lock its * If we are dealing with partitioned table, we also need to lock its
* partitions. Here we only lock targetRelation, we acquire necessary * partitions. Here we only lock targetRelation, we acquire necessary
@ -156,10 +168,10 @@ CoordinatorInsertSelectExecScanInternal(CustomScanState *node)
ListCell *taskCell = NULL; ListCell *taskCell = NULL;
List *prunedTaskList = NIL; List *prunedTaskList = NIL;
shardStateHash = ExecuteSelectIntoColocatedIntermediateResults( shardStateHash = ExecutePlanIntoColocatedIntermediateResults(
targetRelationId, targetRelationId,
insertTargetList, insertTargetList,
selectQuery, selectPlan,
executorState, executorState,
intermediateResultIdPrefix); intermediateResultIdPrefix);
@ -209,8 +221,8 @@ CoordinatorInsertSelectExecScanInternal(CustomScanState *node)
} }
else else
{ {
ExecuteSelectIntoRelation(targetRelationId, insertTargetList, selectQuery, ExecutePlanIntoRelation(targetRelationId, insertTargetList, selectPlan,
executorState); executorState);
} }
scanState->finishedRemoteScan = true; scanState->finishedRemoteScan = true;
@ -438,17 +450,18 @@ TwoPhaseInsertSelectTaskList(Oid targetRelationId, Query *insertSelectQuery,
/* /*
* ExecuteSelectIntoColocatedIntermediateResults executes the given select query * ExecutePlanIntoColocatedIntermediateResults executes the given PlannedStmt
* and inserts tuples into a set of intermediate results that are colocated with * and inserts tuples into a set of intermediate results that are colocated with
* the target table for further processing of ON CONFLICT or RETURNING. It also * the target table for further processing of ON CONFLICT or RETURNING. It also
* returns the hash of shard states that were used to insert tuplesinto the target * returns the hash of shard states that were used to insert tuplesinto the target
* relation. * relation.
*/ */
static HTAB * static HTAB *
ExecuteSelectIntoColocatedIntermediateResults(Oid targetRelationId, ExecutePlanIntoColocatedIntermediateResults(Oid targetRelationId,
List *insertTargetList, List *insertTargetList,
Query *selectQuery, EState *executorState, PlannedStmt *selectPlan,
char *intermediateResultIdPrefix) EState *executorState,
char *intermediateResultIdPrefix)
{ {
ParamListInfo paramListInfo = executorState->es_param_list_info; ParamListInfo paramListInfo = executorState->es_param_list_info;
bool stopOnFailure = false; bool stopOnFailure = false;
@ -473,14 +486,7 @@ ExecuteSelectIntoColocatedIntermediateResults(Oid targetRelationId,
stopOnFailure, stopOnFailure,
intermediateResultIdPrefix); intermediateResultIdPrefix);
/* ExecutePlanIntoDestReceiver(selectPlan, paramListInfo, (DestReceiver *) copyDest);
* Make a copy of the query, since ExecuteQueryIntoDestReceiver may scribble on it
* and we want it to be replanned every time if it is stored in a prepared
* statement.
*/
Query *queryCopy = copyObject(selectQuery);
ExecuteQueryIntoDestReceiver(queryCopy, paramListInfo, (DestReceiver *) copyDest);
executorState->es_processed = copyDest->tuplesSent; executorState->es_processed = copyDest->tuplesSent;
@ -491,13 +497,13 @@ ExecuteSelectIntoColocatedIntermediateResults(Oid targetRelationId,
/* /*
* ExecuteSelectIntoRelation executes given SELECT query and inserts the * ExecutePlanIntoRelation executes the given plan and inserts the
* results into the target relation, which is assumed to be a distributed * results into the target relation, which is assumed to be a distributed
* table. * table.
*/ */
static void static void
ExecuteSelectIntoRelation(Oid targetRelationId, List *insertTargetList, ExecutePlanIntoRelation(Oid targetRelationId, List *insertTargetList,
Query *selectQuery, EState *executorState) PlannedStmt *selectPlan, EState *executorState)
{ {
ParamListInfo paramListInfo = executorState->es_param_list_info; ParamListInfo paramListInfo = executorState->es_param_list_info;
bool stopOnFailure = false; bool stopOnFailure = false;
@ -521,14 +527,7 @@ ExecuteSelectIntoRelation(Oid targetRelationId, List *insertTargetList,
executorState, executorState,
stopOnFailure, NULL); stopOnFailure, NULL);
/* ExecutePlanIntoDestReceiver(selectPlan, paramListInfo, (DestReceiver *) copyDest);
* Make a copy of the query, since ExecuteQueryIntoDestReceiver may scribble on it
* and we want it to be replanned every time if it is stored in a prepared
* statement.
*/
Query *queryCopy = copyObject(selectQuery);
ExecuteQueryIntoDestReceiver(queryCopy, paramListInfo, (DestReceiver *) copyDest);
executorState->es_processed = copyDest->tuplesSent; executorState->es_processed = copyDest->tuplesSent;