Show Material node Maximum Storage for parallel workers

1eff8279d did this for the leader, but neglected to consider that the
Material node may be below a Gather or Gather Merge and have parallel
workers also participate in the execution of the Material node.

Here we add EXPLAIN ANALYZE output to show Maximum Storage per worker
and all the required shared memory related code to get those details
back to the leader from each worker.

Because the prior code didn't have parallel worker processes report the
tuplestore usage details, this meant that the EXPLAIN ANALYZE output
didn't show the Maximum Storage when running debug_parallel_query =
regress.  You could argue that fixing that should be a bug fix, but
fixing it would start reporting the Maximum Storage for Material nodes
for parallel workers during parallel queries, which feels more like a
new feature than a bug fix.  The telemetery data going missing with
debug_parallel_query = regress isn't the end of the world, so let's just
fix this in master.

Author: David Rowley <dgrowleyml@gmail.com>
Reviewed-by: Tatsuya Kawata <kawatatatsuya0913@gmail.com>
Reviewed-by: Chao Li <li.evan.chao@gmail.com>
Discussion: https://postgr.es/m/CAApHDvoUWCms-b27MpDKd5_RUFtAnVrxPgN9u3QoQbBihk+S6g@mail.gmail.com

src/backend/commands/explain.c

@@ -3491,15 +3491,41 @@ show_material_info(MaterialState *mstate, ExplainState *es)
 
 	Tuplestorestate *tupstore = mstate->tuplestorestate;
 
-	/*
-	 * Nothing to show if ANALYZE option wasn't used or if execution didn't
-	 * get as far as creating the tuplestore.
-	 */
-	if (!es->analyze || tupstore == NULL)
+	if (!es->analyze)
 		return;
 
-	tuplestore_get_stats(tupstore, &usedDisk, &maxSpaceUsed);
-	show_storage_info(usedDisk, maxSpaceUsed, es);
+	/*
+	 * Nothing to show for the leader if execution didn't get as far as
+	 * creating the tuplestore.
+	 */
+	if (tupstore != NULL)
+	{
+		tuplestore_get_stats(tupstore, &usedDisk, &maxSpaceUsed);
+		show_storage_info(usedDisk, maxSpaceUsed, es);
+	}
+
+	if (mstate->shared_info == NULL)
+		return;
+
+	/* Show details from parallel workers */
+	for (int n = 0; n < mstate->shared_info->num_workers; n++)
+	{
+		MaterialInstrumentation *si;
+
+		si = &mstate->shared_info->sinstrument[n];
+
+		/* Skip workers that didn't create a tuplestore */
+		if (si->maxSpaceUsed == 0)
+			continue;
+
+		if (es->workers_state)
+			ExplainOpenWorker(n, es);
+
+		show_storage_info(si->usedDisk, si->maxSpaceUsed, es);
+
+		if (es->workers_state)
+			ExplainCloseWorker(n, es);
+	}
 }
 
 /*

src/backend/executor/execParallel.c

@@ -36,6 +36,7 @@
 #include "executor/nodeIncrementalSort.h"
 #include "executor/nodeIndexonlyscan.h"
 #include "executor/nodeIndexscan.h"
+#include "executor/nodeMaterial.h"
 #include "executor/nodeMemoize.h"
 #include "executor/nodeSeqscan.h"
 #include "executor/nodeSort.h"
@@ -338,6 +339,10 @@ ExecParallelEstimate(PlanState *planstate, ExecParallelEstimateContext *e)
 			/* even when not parallel-aware, for EXPLAIN ANALYZE */
 			ExecMemoizeEstimate((MemoizeState *) planstate, e->pcxt);
 			break;
+		case T_MaterialState:
+			/* even when not parallel-aware, for EXPLAIN ANALYZE */
+			ExecMaterialEstimate((MaterialState *) planstate, e->pcxt);
+			break;
 		default:
 			break;
 	}
@@ -586,6 +591,10 @@ ExecParallelInitializeDSM(PlanState *planstate,
 			/* even when not parallel-aware, for EXPLAIN ANALYZE */
 			ExecMemoizeInitializeDSM((MemoizeState *) planstate, d->pcxt);
 			break;
+		case T_MaterialState:
+			/* even when not parallel-aware, for EXPLAIN ANALYZE */
+			ExecMaterialInitializeDSM((MaterialState *) planstate, d->pcxt);
+			break;
 		default:
 			break;
 	}
@@ -1074,6 +1083,7 @@ ExecParallelReInitializeDSM(PlanState *planstate,
 		case T_SortState:
 		case T_IncrementalSortState:
 		case T_MemoizeState:
+		case T_MaterialState:
 			/* these nodes have DSM state, but no reinitialization is required */
 			break;
 
@@ -1155,6 +1165,9 @@ ExecParallelRetrieveInstrumentation(PlanState *planstate,
 		case T_MemoizeState:
 			ExecMemoizeRetrieveInstrumentation((MemoizeState *) planstate);
 			break;
+		case T_MaterialState:
+			ExecMaterialRetrieveInstrumentation((MaterialState *) planstate);
+			break;
 		case T_BitmapHeapScanState:
 			ExecBitmapHeapRetrieveInstrumentation((BitmapHeapScanState *) planstate);
 			break;
@@ -1484,6 +1497,10 @@ ExecParallelInitializeWorker(PlanState *planstate, ParallelWorkerContext *pwcxt)
 			/* even when not parallel-aware, for EXPLAIN ANALYZE */
 			ExecMemoizeInitializeWorker((MemoizeState *) planstate, pwcxt);
 			break;
+		case T_MaterialState:
+			/* even when not parallel-aware, for EXPLAIN ANALYZE */
+			ExecMaterialInitializeWorker((MaterialState *) planstate, pwcxt);
+			break;
 		default:
 			break;
 	}

src/backend/executor/nodeMaterial.c

@@ -197,6 +197,7 @@ ExecInitMaterial(Material *node, EState *estate, int eflags)
 
 	matstate->eof_underlying = false;
 	matstate->tuplestorestate = NULL;
+	matstate->shared_info = NULL;
 
 	/*
 	 * Miscellaneous initialization
@@ -240,6 +241,29 @@ ExecInitMaterial(Material *node, EState *estate, int eflags)
 void
 ExecEndMaterial(MaterialState *node)
 {
+	/*
+	 * When ending a parallel worker, copy the tuplestore usage details into
+	 * shared memory so that they can be picked up by the main process to
+	 * report in EXPLAIN ANALYZE.  The Gather node above us may be rescanned,
+	 * in which case a new set of workers will write to the same slots, so
+	 * merge with any existing values rather than overwriting them.
+	 */
+	if (node->shared_info != NULL && IsParallelWorker() &&
+		node->tuplestorestate != NULL)
+	{
+		MaterialInstrumentation *si;
+		bool		usedDisk;
+		int64		maxSpaceUsed;
+
+		Assert(ParallelWorkerNumber < node->shared_info->num_workers);
+
+		tuplestore_get_stats(node->tuplestorestate, &usedDisk, &maxSpaceUsed);
+
+		si = &node->shared_info->sinstrument[ParallelWorkerNumber];
+		si->maxSpaceUsed = Max(si->maxSpaceUsed, maxSpaceUsed);
+		si->usedDisk |= usedDisk;
+	}
+
 	/*
 	 * Release tuplestore resources
 	 */
@@ -365,3 +389,89 @@ ExecReScanMaterial(MaterialState *node)
 		node->eof_underlying = false;
 	}
 }
+
+/* ----------------------------------------------------------------
+ *						Parallel Query Support
+ * ----------------------------------------------------------------
+ */
+
+/* ----------------------------------------------------------------
+ *		ExecMaterialEstimate
+ *
+ *		Estimate space required to propagate material statistics.
+ * ----------------------------------------------------------------
+ */
+void
+ExecMaterialEstimate(MaterialState *node, ParallelContext *pcxt)
+{
+	Size		size;
+
+	/* don't need this if not instrumenting or no workers */
+	if (!node->ss.ps.instrument || pcxt->nworkers == 0)
+		return;
+
+	size = mul_size(pcxt->nworkers, sizeof(MaterialInstrumentation));
+	size = add_size(size, offsetof(SharedMaterialInfo, sinstrument));
+	shm_toc_estimate_chunk(&pcxt->estimator, size);
+	shm_toc_estimate_keys(&pcxt->estimator, 1);
+}
+
+/* ----------------------------------------------------------------
+ *		ExecMaterialInitializeDSM
+ *
+ *		Initialize DSM space for material statistics.
+ * ----------------------------------------------------------------
+ */
+void
+ExecMaterialInitializeDSM(MaterialState *node, ParallelContext *pcxt)
+{
+	Size		size;
+
+	/* don't need this if not instrumenting or no workers */
+	if (!node->ss.ps.instrument || pcxt->nworkers == 0)
+		return;
+
+	size = offsetof(SharedMaterialInfo, sinstrument)
+		+ pcxt->nworkers * sizeof(MaterialInstrumentation);
+	node->shared_info = shm_toc_allocate(pcxt->toc, size);
+	/* ensure any unfilled slots will contain zeroes */
+	memset(node->shared_info, 0, size);
+	node->shared_info->num_workers = pcxt->nworkers;
+	shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id,
+				   node->shared_info);
+}
+
+/* ----------------------------------------------------------------
+ *		ExecMaterialInitializeWorker
+ *
+ *		Attach worker to DSM space for material statistics.
+ * ----------------------------------------------------------------
+ */
+void
+ExecMaterialInitializeWorker(MaterialState *node, ParallelWorkerContext *pwcxt)
+{
+	node->shared_info =
+		shm_toc_lookup(pwcxt->toc, node->ss.ps.plan->plan_node_id, true);
+}
+
+/* ----------------------------------------------------------------
+ *		ExecMaterialRetrieveInstrumentation
+ *
+ *		Transfer material statistics from DSM to private memory.
+ * ----------------------------------------------------------------
+ */
+void
+ExecMaterialRetrieveInstrumentation(MaterialState *node)
+{
+	Size		size;
+	SharedMaterialInfo *si;
+
+	if (node->shared_info == NULL)
+		return;
+
+	size = offsetof(SharedMaterialInfo, sinstrument)
+		+ node->shared_info->num_workers * sizeof(MaterialInstrumentation);
+	si = palloc(size);
+	memcpy(si, node->shared_info, size);
+	node->shared_info = si;
+}

src/include/executor/instrument_node.h

@@ -173,6 +173,27 @@ typedef struct SharedMemoizeInfo
 } SharedMemoizeInfo;
 
 
+/* ---------------------
+ *	Instrumentation information for Material
+ * ---------------------
+ */
+typedef struct MaterialInstrumentation
+{
+	int64		maxSpaceUsed;	/* peak memory/disk usage in bytes, or 0 if no
+								 * tuplestore was created */
+	bool		usedDisk;		/* did the tuplestore spill to disk? */
+} MaterialInstrumentation;
+
+/*
+ * Shared memory container for per-worker material information
+ */
+typedef struct SharedMaterialInfo
+{
+	int			num_workers;
+	MaterialInstrumentation sinstrument[FLEXIBLE_ARRAY_MEMBER];
+} SharedMaterialInfo;
+
+
 /* ---------------------
  *	Instrumentation information for Sorts.
  * ---------------------

src/include/executor/nodeMaterial.h

@@ -14,6 +14,7 @@
 #ifndef NODEMATERIAL_H
 #define NODEMATERIAL_H
 
+#include "access/parallel.h"
 #include "nodes/execnodes.h"
 
 extern MaterialState *ExecInitMaterial(Material *node, EState *estate, int eflags);
@@ -21,5 +22,11 @@ extern void ExecEndMaterial(MaterialState *node);
 extern void ExecMaterialMarkPos(MaterialState *node);
 extern void ExecMaterialRestrPos(MaterialState *node);
 extern void ExecReScanMaterial(MaterialState *node);
+extern void ExecMaterialEstimate(MaterialState *node, ParallelContext *pcxt);
+extern void ExecMaterialInitializeDSM(MaterialState *node,
+									  ParallelContext *pcxt);
+extern void ExecMaterialInitializeWorker(MaterialState *node,
+										 ParallelWorkerContext *pwcxt);
+extern void ExecMaterialRetrieveInstrumentation(MaterialState *node);
 
 #endif							/* NODEMATERIAL_H */

src/include/nodes/execnodes.h

@@ -2259,6 +2259,7 @@ typedef struct MaterialState
 	int			eflags;			/* capability flags to pass to tuplestore */
 	bool		eof_underlying; /* reached end of underlying plan? */
 	Tuplestorestate *tuplestorestate;
+	SharedMaterialInfo *shared_info;	/* statistics for parallel workers */
 } MaterialState;
 
 struct MemoizeEntry;

src/tools/pgindent/typedefs.list

@@ -1726,6 +1726,7 @@ MVNDistinctItem
 ManyTestResource
 ManyTestResourceKind
 Material
+MaterialInstrumentation
 MaterialPath
 MaterialState
 MdPathStr
@@ -2861,6 +2862,7 @@ SharedInvalSmgrMsg
 SharedInvalSnapshotMsg
 SharedInvalidationMessage
 SharedJitInstrumentation
+SharedMaterialInfo
 SharedMemoizeInfo
 SharedRecordTableEntry
 SharedRecordTableKey