Avoid aggregating worker instrumentation multiple times.

Amit Kapila, per design ideas from me.

diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c
index eae13c5647752c79a42eff96406ed50b014edae5..6730037710912edc917913be650a820cfe6c2a91 100644 (file)
--- a/src/backend/executor/execParallel.c
+++ b/src/backend/executor/execParallel.c
@@ -277,13 +277,15 @@ ExecParallelSetupTupleQueues(ParallelContext *pcxt, bool reinitialize)
 }
 
 /*
- * Re-initialize the response queues for backend workers to return tuples
- * to the main backend and start the workers.
+ * Re-initialize the parallel executor info such that it can be reused by
+ * workers.
  */
-shm_mq_handle **
-ExecParallelReinitializeTupleQueues(ParallelContext *pcxt)
+void
+ExecParallelReinitialize(ParallelExecutorInfo *pei)
 {
-   return ExecParallelSetupTupleQueues(pcxt, true);
+   ReinitializeParallelDSM(pei->pcxt);
+   pei->tqueue = ExecParallelSetupTupleQueues(pei->pcxt, true);
+   pei->finished = false;
 }
 
 /*
@@ -308,6 +310,7 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, int nworkers)
 
    /* Allocate object for return value. */
    pei = palloc0(sizeof(ParallelExecutorInfo));
+   pei->finished = false;
    pei->planstate = planstate;
 
    /* Fix up and serialize plan to be sent to workers. */
@@ -469,6 +472,9 @@ ExecParallelFinish(ParallelExecutorInfo *pei)
 {
    int     i;
 
+   if (pei->finished)
+       return;
+
    /* First, wait for the workers to finish. */
    WaitForParallelWorkersToFinish(pei->pcxt);
 
@@ -480,6 +486,8 @@ ExecParallelFinish(ParallelExecutorInfo *pei)
    if (pei->instrumentation)
        ExecParallelRetrieveInstrumentation(pei->planstate,
                                            pei->instrumentation);
+
+   pei->finished = true;
 }
 
 /*

diff --git a/src/backend/executor/nodeGather.c b/src/backend/executor/nodeGather.c
index b368b48d01d839c76b6a871adc44595ddcc9aa96..b6e82d1664f83dedcdf1e1eb8e8e35071ffcff76 100644 (file)
--- a/src/backend/executor/nodeGather.c
+++ b/src/backend/executor/nodeGather.c
@@ -456,11 +456,7 @@ ExecReScanGather(GatherState *node)
    node->initialized = false;
 
    if (node->pei)
-   {
-       ReinitializeParallelDSM(node->pei->pcxt);
-       node->pei->tqueue =
-               ExecParallelReinitializeTupleQueues(node->pei->pcxt);
-   }
+       ExecParallelReinitialize(node->pei);
 
    ExecReScan(node->ps.lefttree);
 }

diff --git a/src/include/executor/execParallel.h b/src/include/executor/execParallel.h
index 23c29ebb9027de0c40794d82b951e09eda5d0958..b43af1dd2b3c220110d4d819ca61e207b541a0c4 100644 (file)
--- a/src/include/executor/execParallel.h
+++ b/src/include/executor/execParallel.h
@@ -27,12 +27,13 @@ typedef struct ParallelExecutorInfo
    BufferUsage *buffer_usage;
    SharedExecutorInstrumentation *instrumentation;
    shm_mq_handle **tqueue;
+   bool    finished;
 }  ParallelExecutorInfo;
 
 extern ParallelExecutorInfo *ExecInitParallelPlan(PlanState *planstate,
                     EState *estate, int nworkers);
 extern void ExecParallelFinish(ParallelExecutorInfo *pei);
 extern void ExecParallelCleanup(ParallelExecutorInfo *pei);
-extern shm_mq_handle **ExecParallelReinitializeTupleQueues(ParallelContext *pcxt);
+extern void ExecParallelReinitialize(ParallelExecutorInfo *pei);
 
 #endif   /* EXECPARALLEL_H */