Fix accumulation of parallel worker instrumentation.

When a Gather or Gather Merge node is started and stopped multiple
times, the old code wouldn't reset the shared state between executions,
potentially resulting in dramatically inflated instrumentation data
for nodes beneath it.  (The per-worker instrumentation ended up OK,
I think, but the overall totals were inflated.)

Report by hubert depesz lubaczewski.  Analysis and fix by Amit Kapila,
reviewed and tweaked a bit by me.

Discussion: http://postgr.es/m/20171127175631.GA405@depesz.com

1 files changed, 38 insertions, 13 deletions

diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c
index 0aca00b0e68..ff5cff59b0f 100644
--- a/src/backend/executor/execParallel.c
+++ b/src/backend/executor/execParallel.c
@@ -808,6 +808,19 @@ ExecParallelReinitialize(PlanState *planstate,
 	/* Old workers must already be shut down */
 	Assert(pei->finished);
 
+	/* Clear the instrumentation space from the last round. */
+	if (pei->instrumentation)
+	{
+		Instrumentation *instrument;
+		SharedExecutorInstrumentation *sh_instr;
+		int			i;
+
+		sh_instr = pei->instrumentation;
+		instrument = GetInstrumentationArray(sh_instr);
+		for (i = 0; i < sh_instr->num_workers * sh_instr->num_plan_nodes; ++i)
+			InstrInit(&instrument[i], pei->planstate->state->es_instrument);
+	}
+
 	/* Force parameters we're going to pass to workers to be evaluated. */
 	ExecEvalParamExecParams(sendParams, estate);
 
@@ -925,21 +938,33 @@ ExecParallelRetrieveInstrumentation(PlanState *planstate,
 	for (n = 0; n < instrumentation->num_workers; ++n)
 		InstrAggNode(planstate->instrument, &instrument[n]);
 
-	/*
-	 * Also store the per-worker detail.
-	 *
-	 * Worker instrumentation should be allocated in the same context as the
-	 * regular instrumentation information, which is the per-query context.
-	 * Switch into per-query memory context.
-	 */
-	oldcontext = MemoryContextSwitchTo(planstate->state->es_query_cxt);
-	ibytes = mul_size(instrumentation->num_workers, sizeof(Instrumentation));
-	planstate->worker_instrument =
-		palloc(ibytes + offsetof(WorkerInstrumentation, instrument));
-	MemoryContextSwitchTo(oldcontext);
+	if (!planstate->worker_instrument)
+	{
+		/*
+		 * Allocate space for the per-worker detail.
+		 *
+		 * Worker instrumentation should be allocated in the same context as
+		 * the regular instrumentation information, which is the per-query
+		 * context. Switch into per-query memory context.
+		 */
+		oldcontext = MemoryContextSwitchTo(planstate->state->es_query_cxt);
+		ibytes =
+			mul_size(instrumentation->num_workers, sizeof(Instrumentation));
+		planstate->worker_instrument =
+			palloc(ibytes + offsetof(WorkerInstrumentation, instrument));
+		MemoryContextSwitchTo(oldcontext);
+
+		for (n = 0; n < instrumentation->num_workers; ++n)
+			InstrInit(&planstate->worker_instrument->instrument[n],
+					  planstate->state->es_instrument);
+	}
 
 	planstate->worker_instrument->num_workers = instrumentation->num_workers;
-	memcpy(&planstate->worker_instrument->instrument, instrument, ibytes);
+
+	/* Accumulate the per-worker detail. */
+	for (n = 0; n < instrumentation->num_workers; ++n)
+		InstrAggNode(&planstate->worker_instrument->instrument[n],
+					 &instrument[n]);
 
 	/* Perform any node-type-specific work that needs to be done. */
 	switch (nodeTag(planstate))