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