From a241bc4fedefae3c0b0952c0a9d3a28d7747bfe3 Mon Sep 17 00:00:00 2001 From: David Rowley Date: Mon, 12 Oct 2026 22:41:35 +1300 Subject: [PATCH] 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 Reviewed-by: Tatsuya Kawata Reviewed-by: Chao Li Discussion: https://postgr.es/m/CAApHDvoUWCms-b27MpDKd5_RUFtAnVrxPgN9u3QoQbBihk+S6g@mail.gmail.com --- src/backend/commands/explain.c | 40 +++++++-- src/backend/executor/execParallel.c | 17 ++++ src/backend/executor/nodeMaterial.c | 110 +++++++++++++++++++++++++ src/include/executor/instrument_node.h | 21 +++++ src/include/executor/nodeMaterial.h | 7 ++ src/include/nodes/execnodes.h | 1 + src/tools/pgindent/typedefs.list | 2 + 7 files changed, 191 insertions(+), 7 deletions(-) diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c index c7b02a1479c..4b1ad4227ac 100644 --- a/src/backend/commands/explain.c +++ b/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); + } } /* diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c index 075bc5cd2c9..316dd3f67c6 100644 --- a/src/backend/executor/execParallel.c +++ b/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; } diff --git a/src/backend/executor/nodeMaterial.c b/src/backend/executor/nodeMaterial.c index 7f5aceaa3ca..72c8fe88e72 100644 --- a/src/backend/executor/nodeMaterial.c +++ b/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; +} diff --git a/src/include/executor/instrument_node.h b/src/include/executor/instrument_node.h index 41a7d33f19c..4759a628afe 100644 --- a/src/include/executor/instrument_node.h +++ b/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. * --------------------- diff --git a/src/include/executor/nodeMaterial.h b/src/include/executor/nodeMaterial.h index 19b254db4c8..e60b837923b 100644 --- a/src/include/executor/nodeMaterial.h +++ b/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 */ diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 0998c9b2b9d..e3cea52e7df 100644 --- a/src/include/nodes/execnodes.h +++ b/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; diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 3bed561f7f2..a7a8dbdd23f 100644 --- a/src/tools/pgindent/typedefs.list +++ b/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