From be8b48a7d1a74d0a3c3a2485757add0344186d94 Mon Sep 17 00:00:00 2001 From: Lukas Fittl Date: Mon, 26 Oct 2026 07:17:12 -0700 Subject: [PATCH v1] Process each PlanState only once during parallel query setup Parallel query walks the PlanState tree several times, to size and initialize DSM state and to collect instrumentation. These walks assumed each PlanState is visited once, but that does not hold for SubPlans: when the same SubPlan is referenced by more than one expression (for example a hash clause and the hash keys derived from it), each reference has its own SubPlanState pointing at the same subplan PlanState, so planstate_walk_subplans() descends into that subtree once per reference. The repeated visit makes ExecParallelInitializeDSM() insert a duplicate shm_toc key for each node under the subplan, which causes an assertion failure in shm_toc_insert() when doing EXPLAIN ANALYZE with such plans. Additionally it causes instrumentation data to be incorrect, inflating loop counts and losing per-worker details such as "Sort Method". Fix by adding the planstate_tree_collect() helper, which returns the distinct PlanStates of a tree, and having the parallel setup and instrumentation code iterate over that list instead of recursing per reference. ExecShutdownNode() has the same structure but is left unchanged: all in-core shutdown callbacks are idempotent, and reworking its nested instrumentation accounting is better done separately. Author: Lukas Fittl Reviewed-by: Discussion: --- src/backend/executor/execParallel.c | 245 ++++++++---------- src/backend/nodes/nodeFuncs.c | 52 ++++ src/include/executor/execParallel.h | 2 + src/include/nodes/nodeFuncs.h | 2 + src/test/regress/expected/select_parallel.out | 66 +++++ src/test/regress/sql/select_parallel.sql | 46 ++++ src/tools/pgindent/typedefs.list | 3 +- 7 files changed, 273 insertions(+), 143 deletions(-) diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c index 316dd3f67c6..49e40e08a6b 100644 --- a/src/backend/executor/execParallel.c +++ b/src/backend/executor/execParallel.c @@ -115,33 +115,12 @@ struct SharedExecutorInstrumentation (StaticAssertVariableIsOfTypeMacro(sei, SharedExecutorInstrumentation *), \ (NodeInstrumentation *) (((char *) sei) + sei->instrument_offset)) -/* Context object for ExecParallelEstimate. */ -typedef struct ExecParallelEstimateContext -{ - ParallelContext *pcxt; - int nnodes; -} ExecParallelEstimateContext; - -/* Context object for ExecParallelInitializeDSM. */ -typedef struct ExecParallelInitializeDSMContext -{ - ParallelContext *pcxt; - SharedExecutorInstrumentation *instrumentation; - int nnodes; -} ExecParallelInitializeDSMContext; - /* Helper functions that run in the parallel leader. */ static char *ExecSerializePlan(Plan *plan, EState *estate); -static bool ExecParallelEstimate(PlanState *planstate, - ExecParallelEstimateContext *e); -static bool ExecParallelInitializeDSM(PlanState *planstate, - ExecParallelInitializeDSMContext *d); static shm_mq_handle **ExecParallelSetupTupleQueues(ParallelContext *pcxt, bool reinitialize); -static bool ExecParallelReInitializeDSM(PlanState *planstate, +static void ExecParallelReInitializeDSM(PlanState *planstate, ParallelContext *pcxt); -static bool ExecParallelRetrieveInstrumentation(PlanState *planstate, - SharedExecutorInstrumentation *instrumentation); /* Helper function that runs in the parallel worker. */ static DestReceiver *ExecParallelGetReceiver(dsm_segment *seg, shm_toc *toc); @@ -239,115 +218,104 @@ ExecSerializePlan(Plan *plan, EState *estate) * which is shared across all parallel workers. Before we size the DSM, give * them a chance to call shm_toc_estimate_chunk or shm_toc_estimate_keys on * &pcxt->estimator. - * - * While we're at it, count the number of PlanState nodes in the tree, so - * we know how many Instrumentation structures we need. */ -static bool -ExecParallelEstimate(PlanState *planstate, ExecParallelEstimateContext *e) +static void +ExecParallelEstimate(PlanState *planstate, ParallelContext *pcxt) { - if (planstate == NULL) - return false; - - /* Count this node. */ - e->nnodes++; - switch (nodeTag(planstate)) { case T_SeqScanState: if (planstate->plan->parallel_aware) ExecSeqScanEstimate((SeqScanState *) planstate, - e->pcxt); + pcxt); /* even when not parallel-aware, for EXPLAIN ANALYZE */ ExecSeqScanInstrumentEstimate((SeqScanState *) planstate, - e->pcxt); + pcxt); break; case T_IndexScanState: if (planstate->plan->parallel_aware) ExecIndexScanEstimate((IndexScanState *) planstate, - e->pcxt); + pcxt); /* even when not parallel-aware, for EXPLAIN ANALYZE */ ExecIndexScanInstrumentEstimate((IndexScanState *) planstate, - e->pcxt); + pcxt); break; case T_IndexOnlyScanState: if (planstate->plan->parallel_aware) ExecIndexOnlyScanEstimate((IndexOnlyScanState *) planstate, - e->pcxt); + pcxt); /* even when not parallel-aware, for EXPLAIN ANALYZE */ ExecIndexOnlyScanInstrumentEstimate((IndexOnlyScanState *) planstate, - e->pcxt); + pcxt); break; case T_BitmapIndexScanState: /* even when not parallel-aware, for EXPLAIN ANALYZE */ ExecBitmapIndexScanEstimate((BitmapIndexScanState *) planstate, - e->pcxt); + pcxt); break; case T_ForeignScanState: if (planstate->plan->parallel_aware) ExecForeignScanEstimate((ForeignScanState *) planstate, - e->pcxt); + pcxt); break; case T_TidRangeScanState: if (planstate->plan->parallel_aware) ExecTidRangeScanEstimate((TidRangeScanState *) planstate, - e->pcxt); + pcxt); /* even when not parallel-aware, for EXPLAIN ANALYZE */ ExecTidRangeScanInstrumentEstimate((TidRangeScanState *) planstate, - e->pcxt); + pcxt); break; case T_AppendState: if (planstate->plan->parallel_aware) ExecAppendEstimate((AppendState *) planstate, - e->pcxt); + pcxt); break; case T_CustomScanState: if (planstate->plan->parallel_aware) ExecCustomScanEstimate((CustomScanState *) planstate, - e->pcxt); + pcxt); break; case T_BitmapHeapScanState: if (planstate->plan->parallel_aware) ExecBitmapHeapEstimate((BitmapHeapScanState *) planstate, - e->pcxt); + pcxt); /* even when not parallel-aware, for EXPLAIN ANALYZE */ ExecBitmapHeapInstrumentEstimate((BitmapHeapScanState *) planstate, - e->pcxt); + pcxt); break; case T_HashJoinState: if (planstate->plan->parallel_aware) ExecHashJoinEstimate((HashJoinState *) planstate, - e->pcxt); + pcxt); break; case T_HashState: /* even when not parallel-aware, for EXPLAIN ANALYZE */ - ExecHashEstimate((HashState *) planstate, e->pcxt); + ExecHashEstimate((HashState *) planstate, pcxt); break; case T_SortState: /* even when not parallel-aware, for EXPLAIN ANALYZE */ - ExecSortEstimate((SortState *) planstate, e->pcxt); + ExecSortEstimate((SortState *) planstate, pcxt); break; case T_IncrementalSortState: /* even when not parallel-aware, for EXPLAIN ANALYZE */ - ExecIncrementalSortEstimate((IncrementalSortState *) planstate, e->pcxt); + ExecIncrementalSortEstimate((IncrementalSortState *) planstate, pcxt); break; case T_AggState: /* even when not parallel-aware, for EXPLAIN ANALYZE */ - ExecAggEstimate((AggState *) planstate, e->pcxt); + ExecAggEstimate((AggState *) planstate, pcxt); break; case T_MemoizeState: /* even when not parallel-aware, for EXPLAIN ANALYZE */ - ExecMemoizeEstimate((MemoizeState *) planstate, e->pcxt); + ExecMemoizeEstimate((MemoizeState *) planstate, pcxt); break; case T_MaterialState: /* even when not parallel-aware, for EXPLAIN ANALYZE */ - ExecMaterialEstimate((MaterialState *) planstate, e->pcxt); + ExecMaterialEstimate((MaterialState *) planstate, pcxt); break; default: break; } - - return planstate_tree_walker(planstate, ExecParallelEstimate, e); } /* @@ -481,21 +449,9 @@ RestoreParamExecParams(char *start_address, EState *estate) * Initialize the dynamic shared memory segment that will be used to control * parallel execution. */ -static bool -ExecParallelInitializeDSM(PlanState *planstate, - ExecParallelInitializeDSMContext *d) +static void +ExecParallelInitializeDSM(PlanState *planstate, ParallelContext *pcxt) { - if (planstate == NULL) - return false; - - /* If instrumentation is enabled, initialize slot for this node. */ - if (d->instrumentation != NULL) - d->instrumentation->plan_node_id[d->nnodes] = - planstate->plan->plan_node_id; - - /* Count this node. */ - d->nnodes++; - /* * Call initializers for DSM-using plan nodes. * @@ -510,96 +466,94 @@ ExecParallelInitializeDSM(PlanState *planstate, case T_SeqScanState: if (planstate->plan->parallel_aware) ExecSeqScanInitializeDSM((SeqScanState *) planstate, - d->pcxt); + pcxt); /* even when not parallel-aware, for EXPLAIN ANALYZE */ ExecSeqScanInstrumentInitDSM((SeqScanState *) planstate, - d->pcxt); + pcxt); break; case T_IndexScanState: if (planstate->plan->parallel_aware) ExecIndexScanInitializeDSM((IndexScanState *) planstate, - d->pcxt); + pcxt); /* even when not parallel-aware, for EXPLAIN ANALYZE */ ExecIndexScanInstrumentInitDSM((IndexScanState *) planstate, - d->pcxt); + pcxt); break; case T_IndexOnlyScanState: if (planstate->plan->parallel_aware) ExecIndexOnlyScanInitializeDSM((IndexOnlyScanState *) planstate, - d->pcxt); + pcxt); /* even when not parallel-aware, for EXPLAIN ANALYZE */ ExecIndexOnlyScanInstrumentInitDSM((IndexOnlyScanState *) planstate, - d->pcxt); + pcxt); break; case T_BitmapIndexScanState: /* even when not parallel-aware, for EXPLAIN ANALYZE */ - ExecBitmapIndexScanInitializeDSM((BitmapIndexScanState *) planstate, d->pcxt); + ExecBitmapIndexScanInitializeDSM((BitmapIndexScanState *) planstate, pcxt); break; case T_ForeignScanState: if (planstate->plan->parallel_aware) ExecForeignScanInitializeDSM((ForeignScanState *) planstate, - d->pcxt); + pcxt); break; case T_TidRangeScanState: if (planstate->plan->parallel_aware) ExecTidRangeScanInitializeDSM((TidRangeScanState *) planstate, - d->pcxt); + pcxt); /* even when not parallel-aware, for EXPLAIN ANALYZE */ ExecTidRangeScanInstrumentInitDSM((TidRangeScanState *) planstate, - d->pcxt); + pcxt); break; case T_AppendState: if (planstate->plan->parallel_aware) ExecAppendInitializeDSM((AppendState *) planstate, - d->pcxt); + pcxt); break; case T_CustomScanState: if (planstate->plan->parallel_aware) ExecCustomScanInitializeDSM((CustomScanState *) planstate, - d->pcxt); + pcxt); break; case T_BitmapHeapScanState: if (planstate->plan->parallel_aware) ExecBitmapHeapInitializeDSM((BitmapHeapScanState *) planstate, - d->pcxt); + pcxt); /* even when not parallel-aware, for EXPLAIN ANALYZE */ ExecBitmapHeapInstrumentInitDSM((BitmapHeapScanState *) planstate, - d->pcxt); + pcxt); break; case T_HashJoinState: if (planstate->plan->parallel_aware) ExecHashJoinInitializeDSM((HashJoinState *) planstate, - d->pcxt); + pcxt); break; case T_HashState: /* even when not parallel-aware, for EXPLAIN ANALYZE */ - ExecHashInitializeDSM((HashState *) planstate, d->pcxt); + ExecHashInitializeDSM((HashState *) planstate, pcxt); break; case T_SortState: /* even when not parallel-aware, for EXPLAIN ANALYZE */ - ExecSortInitializeDSM((SortState *) planstate, d->pcxt); + ExecSortInitializeDSM((SortState *) planstate, pcxt); break; case T_IncrementalSortState: /* even when not parallel-aware, for EXPLAIN ANALYZE */ - ExecIncrementalSortInitializeDSM((IncrementalSortState *) planstate, d->pcxt); + ExecIncrementalSortInitializeDSM((IncrementalSortState *) planstate, pcxt); break; case T_AggState: /* even when not parallel-aware, for EXPLAIN ANALYZE */ - ExecAggInitializeDSM((AggState *) planstate, d->pcxt); + ExecAggInitializeDSM((AggState *) planstate, pcxt); break; case T_MemoizeState: /* even when not parallel-aware, for EXPLAIN ANALYZE */ - ExecMemoizeInitializeDSM((MemoizeState *) planstate, d->pcxt); + ExecMemoizeInitializeDSM((MemoizeState *) planstate, pcxt); break; case T_MaterialState: /* even when not parallel-aware, for EXPLAIN ANALYZE */ - ExecMaterialInitializeDSM((MaterialState *) planstate, d->pcxt); + ExecMaterialInitializeDSM((MaterialState *) planstate, pcxt); break; default: break; } - - return planstate_tree_walker(planstate, ExecParallelInitializeDSM, d); } /* @@ -664,8 +618,9 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, { ParallelExecutorInfo *pei; ParallelContext *pcxt; - ExecParallelEstimateContext e; - ExecParallelInitializeDSMContext d; + List *nodes; + ListCell *lc; + int nnodes; FixedParallelExecutorState *fpes; char *pstmt_data; char *pstmt_space; @@ -758,24 +713,28 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, shm_toc_estimate_keys(&pcxt->estimator, 1); /* - * Give parallel-aware nodes a chance to add to the estimates, and get a - * count of how many PlanState nodes there are. + * Give parallel-aware nodes a chance to add to the estimates. The node + * count determines how many Instrumentation structures we need. The node + * list is kept around for the later walks (DSM reinitialization and + * instrumentation retrieval). */ - e.pcxt = pcxt; - e.nnodes = 0; - ExecParallelEstimate(planstate, &e); + nodes = planstate_tree_collect(planstate); + nnodes = list_length(nodes); + pei->planstate_nodes = nodes; + foreach(lc, nodes) + ExecParallelEstimate((PlanState *) lfirst(lc), pcxt); /* Estimate space for instrumentation, if required. */ if (estate->es_instrument) { instrumentation_len = offsetof(SharedExecutorInstrumentation, plan_node_id) + - sizeof(int) * e.nnodes; + sizeof(int) * nnodes; instrumentation_len = MAXALIGN(instrumentation_len); instrument_offset = instrumentation_len; instrumentation_len += mul_size(sizeof(NodeInstrumentation), - mul_size(e.nnodes, nworkers)); + mul_size(nnodes, nworkers)); shm_toc_estimate_chunk(&pcxt->estimator, instrumentation_len); shm_toc_estimate_keys(&pcxt->estimator, 1); @@ -867,9 +826,9 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, instrumentation->instrument_options = estate->es_instrument; instrumentation->instrument_offset = instrument_offset; instrumentation->num_workers = nworkers; - instrumentation->num_plan_nodes = e.nnodes; + instrumentation->num_plan_nodes = nnodes; instrument = GetInstrumentationArray(instrumentation); - for (i = 0; i < nworkers * e.nnodes; ++i) + for (i = 0; i < nworkers * nnodes; ++i) InstrInitNode(&instrument[i], estate->es_instrument, false); shm_toc_insert(pcxt->toc, PARALLEL_KEY_INSTRUMENTATION, instrumentation); @@ -922,21 +881,25 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, * This also initializes the elements of instrumentation->ps_instrument, * if it exists. */ - d.pcxt = pcxt; - d.instrumentation = instrumentation; - d.nnodes = 0; - - /* Install our DSA area while initializing the plan. */ - estate->es_query_dsa = pei->area; - ExecParallelInitializeDSM(planstate, &d); - estate->es_query_dsa = NULL; /* - * Make sure that the world hasn't shifted under our feet. This could - * probably just be an Assert(), but let's be conservative for now. + * Install our DSA area while initializing the plan. Fill the + * instrumentation array's plan_node_id slots in the same node order that + * ExecParallelRetrieveInstrumentation and the workers will see, since all + * of them iterate over planstate_tree_collect results. */ - if (e.nnodes != d.nnodes) - elog(ERROR, "inconsistent count of PlanState nodes"); + estate->es_query_dsa = pei->area; + foreach(lc, nodes) + { + PlanState *ps = (PlanState *) lfirst(lc); + + if (instrumentation != NULL) + instrumentation->plan_node_id[foreach_current_index(lc)] = + ps->plan->plan_node_id; + + ExecParallelInitializeDSM(ps, pcxt); + } + estate->es_query_dsa = NULL; /* OK, we're ready to rock and roll. */ return pei; @@ -979,6 +942,7 @@ ExecParallelReinitialize(PlanState *planstate, Bitmapset *sendParams) { EState *estate = planstate->state; + ListCell *lc; FixedParallelExecutorState *fpes; /* Old workers must already be shut down */ @@ -1015,20 +979,18 @@ ExecParallelReinitialize(PlanState *planstate, /* Traverse plan tree and let each child node reset associated state. */ estate->es_query_dsa = pei->area; - ExecParallelReInitializeDSM(planstate, pei->pcxt); + foreach(lc, pei->planstate_nodes) + ExecParallelReInitializeDSM((PlanState *) lfirst(lc), pei->pcxt); estate->es_query_dsa = NULL; } /* - * Traverse plan tree to reinitialize per-node dynamic shared memory state + * Reinitialize per-node dynamic shared memory state */ -static bool +static void ExecParallelReInitializeDSM(PlanState *planstate, ParallelContext *pcxt) { - if (planstate == NULL) - return false; - /* * Call reinitializers for DSM-using plan nodes. */ @@ -1091,14 +1053,13 @@ ExecParallelReInitializeDSM(PlanState *planstate, break; } - return planstate_tree_walker(planstate, ExecParallelReInitializeDSM, pcxt); } /* * Copy instrumentation information about this node and its descendants from * dynamic shared memory. */ -static bool +static void ExecParallelRetrieveInstrumentation(PlanState *planstate, SharedExecutorInstrumentation *instrumentation) { @@ -1181,8 +1142,6 @@ ExecParallelRetrieveInstrumentation(PlanState *planstate, break; } - return planstate_tree_walker(planstate, ExecParallelRetrieveInstrumentation, - instrumentation); } /* @@ -1284,10 +1243,13 @@ ExecParallelFinish(ParallelExecutorInfo *pei) void ExecParallelCleanup(ParallelExecutorInfo *pei) { + ListCell *lc; + /* Accumulate instrumentation, if any. */ if (pei->instrumentation) - ExecParallelRetrieveInstrumentation(pei->planstate, - pei->instrumentation); + foreach(lc, pei->planstate_nodes) + ExecParallelRetrieveInstrumentation((PlanState *) lfirst(lc), + pei->instrumentation); /* Accumulate JIT instrumentation, if any. */ if (pei->jit_instrumentation) @@ -1310,6 +1272,7 @@ ExecParallelCleanup(ParallelExecutorInfo *pei) DestroyParallelContext(pei->pcxt); pei->pcxt = NULL; } + list_free(pei->planstate_nodes); pfree(pei); } @@ -1365,7 +1328,7 @@ ExecParallelGetQueryDesc(shm_toc *toc, DestReceiver *receiver, * Copy instrumentation information from this node and its descendants into * dynamic shared memory, so that the parallel leader can retrieve it. */ -static bool +static void ExecParallelReportInstrumentation(PlanState *planstate, SharedExecutorInstrumentation *instrumentation) { @@ -1397,8 +1360,6 @@ ExecParallelReportInstrumentation(PlanState *planstate, Assert(ParallelWorkerNumber < instrumentation->num_workers); InstrAggNode(&instrument[ParallelWorkerNumber], planstate->instrument); - return planstate_tree_walker(planstate, ExecParallelReportInstrumentation, - instrumentation); } /* @@ -1406,12 +1367,10 @@ ExecParallelReportInstrumentation(PlanState *planstate, * retrieved from shared memory. This has to be done once the PlanState * is allocated and initialized by executor; that is, after ExecutorStart(). */ -static bool -ExecParallelInitializeWorker(PlanState *planstate, ParallelWorkerContext *pwcxt) +static void +ExecParallelInitializeWorker(PlanState *planstate, + ParallelWorkerContext *pwcxt) { - if (planstate == NULL) - return false; - switch (nodeTag(planstate)) { case T_SeqScanState: @@ -1505,8 +1464,6 @@ ExecParallelInitializeWorker(PlanState *planstate, ParallelWorkerContext *pwcxt) break; } - return planstate_tree_walker(planstate, ExecParallelInitializeWorker, - pwcxt); } /* @@ -1535,6 +1492,8 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc) QueryDesc *queryDesc; SharedExecutorInstrumentation *instrumentation; SharedJitInstrumentation *jit_instrumentation; + List *nodes; + ListCell *lc; int instrument_options = 0; void *area_space; dsa_area *area; @@ -1577,7 +1536,9 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc) } pwcxt.toc = toc; pwcxt.seg = seg; - ExecParallelInitializeWorker(queryDesc->planstate, &pwcxt); + nodes = planstate_tree_collect(queryDesc->planstate); + foreach(lc, nodes) + ExecParallelInitializeWorker((PlanState *) lfirst(lc), &pwcxt); /* Pass down any tuple bound */ ExecSetTupleBound(fpes->tuples_needed, queryDesc->planstate); @@ -1610,8 +1571,10 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc) /* Report instrumentation data if any instrumentation options are set. */ if (instrumentation != NULL) - ExecParallelReportInstrumentation(queryDesc->planstate, - instrumentation); + foreach(lc, nodes) + ExecParallelReportInstrumentation((PlanState *) lfirst(lc), + instrumentation); + list_free(nodes); /* Report JIT instrumentation data if any */ if (queryDesc->estate->es_jit && jit_instrumentation != NULL) diff --git a/src/backend/nodes/nodeFuncs.c b/src/backend/nodes/nodeFuncs.c index 678b9a046ce..935f28118fe 100644 --- a/src/backend/nodes/nodeFuncs.c +++ b/src/backend/nodes/nodeFuncs.c @@ -4860,3 +4860,55 @@ planstate_walk_members(PlanState **planstates, int nplans, return false; } + +typedef struct planstate_tree_collect_context +{ + List *nodes; /* result, in depth-first pre-order */ + Bitmapset *seen; /* plan_node_ids already in 'nodes' */ +} planstate_tree_collect_context; + +static bool +planstate_tree_collect_walker(PlanState *planstate, + planstate_tree_collect_context *context) +{ + /* plan_node_id is unique across the whole PlannedStmt */ + int plan_node_id = planstate->plan->plan_node_id; + + if (bms_is_member(plan_node_id, context->seen)) + return false; + context->seen = bms_add_member(context->seen, plan_node_id); + context->nodes = lappend(context->nodes, planstate); + + return planstate_tree_walker(planstate, planstate_tree_collect_walker, + context); +} + +/* + * planstate_tree_collect --- list all PlanStates of a plan state tree + * + * Returns a freshly-allocated list of every distinct PlanState reachable + * from 'planstate', including 'planstate' itself, in the order a + * planstate_tree_walker traversal would first visit them (depth-first, + * pre-order). + * + * Unlike a recursive walk with planstate_tree_walker, each node appears + * exactly once in the result. That matters for SubPlans: the same SubPlan + * (identified by plan_id) can be referenced from several expressions, e.g. + * a hash clause along with the hash key expressions derived from it. + * + * The caller is responsible for freeing the list. + */ +List * +planstate_tree_collect(PlanState *planstate) +{ + planstate_tree_collect_context context; + + context.nodes = NIL; + context.seen = NULL; + + (void) planstate_tree_collect_walker(planstate, &context); + + bms_free(context.seen); + + return context.nodes; +} diff --git a/src/include/executor/execParallel.h b/src/include/executor/execParallel.h index 5a2034811d5..074d2ef6720 100644 --- a/src/include/executor/execParallel.h +++ b/src/include/executor/execParallel.h @@ -24,6 +24,8 @@ typedef struct SharedExecutorInstrumentation SharedExecutorInstrumentation; typedef struct ParallelExecutorInfo { PlanState *planstate; /* plan subtree we're running in parallel */ + List *planstate_nodes; /* all distinct nodes of that subtree, + * from planstate_tree_collect */ ParallelContext *pcxt; /* parallel context we're using */ BufferUsage *buffer_usage; /* points to bufusage area in DSM */ WalUsage *wal_usage; /* walusage area in DSM */ diff --git a/src/include/nodes/nodeFuncs.h b/src/include/nodes/nodeFuncs.h index 696ffdebd4b..46d3d040b83 100644 --- a/src/include/nodes/nodeFuncs.h +++ b/src/include/nodes/nodeFuncs.h @@ -221,4 +221,6 @@ extern bool planstate_tree_walker_impl(PlanState *planstate, planstate_tree_walker_callback walker, void *context); +extern List *planstate_tree_collect(PlanState *planstate); + #endif /* NODEFUNCS_H */ diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out index e1344215644..32c3b2ab535 100644 --- a/src/test/regress/expected/select_parallel.out +++ b/src/test/regress/expected/select_parallel.out @@ -334,6 +334,72 @@ explain (costs off) -> Seq Scan on tenk2 (4 rows) +-- test instrumentation of a subplan shared by several expressions (here a +-- hash clause and the hash keys derived from it), which the parallel-query +-- walks must process only once +set enable_nestloop = off; +set enable_mergejoin = off; +set enable_indexscan = off; +set enable_indexonlyscan = off; +set parallel_leader_participation = off; +create function explain_parallel_shared_subplan() returns setof text +language plpgsql as +$$ +declare ln text; +declare prev text; +begin + for ln in + explain (analyze, timing off, summary off, costs off, buffers off) + select count(*) from tenk1 t1 join tenk1 t2 + on t2.unique1 = case when t1.hundred not in + (select thousand from tenk2 order by twothousand limit 50) + then 1 else 2 end + loop + ln := regexp_replace(ln, 'actual rows=[0-9.]+ loops=[0-9]+', + 'actual rows=N loops=N'); + ln := regexp_replace(ln, 'Memory: \S*', 'Memory: xxx'); + ln := regexp_replace(ln, 'Memory Usage: \S*', 'Memory Usage: xxx'); + -- the launched worker count and the number of per-worker "Sort + -- Method" lines vary with how many workers start and get rows; + -- normalize both so the test only confirms that per-worker sort + -- statistics are reported at all (they vanish without the fix) + ln := regexp_replace(ln, 'Workers Launched: \d+', 'Workers Launched: N'); + ln := regexp_replace(ln, 'Worker \d+:', 'Worker N:'); + if ln is distinct from prev then + return next ln; + end if; + prev := ln; + end loop; +end; +$$; +select * from explain_parallel_shared_subplan(); + explain_parallel_shared_subplan +---------------------------------------------------------------------------------------------------------------------------------- + Finalize Aggregate (actual rows=N loops=N) + -> Gather (actual rows=N loops=N) + Workers Planned: 4 + Workers Launched: N + -> Partial Aggregate (actual rows=N loops=N) + -> Parallel Hash Join (actual rows=N loops=N) + Hash Cond: (CASE WHEN (NOT (ANY (t1.hundred = (hashed SubPlan any_1).col1))) THEN 1 ELSE 2 END = t2.unique1) + -> Parallel Seq Scan on tenk1 t1 (actual rows=N loops=N) + -> Parallel Hash (actual rows=N loops=N) + Buckets: 16384 Batches: 1 Memory Usage: xxx + -> Parallel Seq Scan on tenk1 t2 (actual rows=N loops=N) + SubPlan any_1 + -> Limit (actual rows=N loops=N) + -> Sort (actual rows=N loops=N) + Sort Key: tenk2.twothousand + Worker N: Sort Method: top-N heapsort Memory: xxx + -> Seq Scan on tenk2 (actual rows=N loops=N) +(17 rows) + +drop function explain_parallel_shared_subplan(); +reset parallel_leader_participation; +reset enable_indexscan; +reset enable_indexonlyscan; +reset enable_nestloop; +reset enable_mergejoin; alter table tenk2 reset (parallel_workers); -- test parallel plan for a query containing initplan. set enable_indexscan = off; diff --git a/src/test/regress/sql/select_parallel.sql b/src/test/regress/sql/select_parallel.sql index 71a75bc86ea..b34d650fe1d 100644 --- a/src/test/regress/sql/select_parallel.sql +++ b/src/test/regress/sql/select_parallel.sql @@ -122,6 +122,52 @@ select count(*) from tenk1 where (two, four) not in explain (costs off) select * from tenk1 where (unique1 + random())::integer not in (select ten from tenk2); + +-- test instrumentation of a subplan shared by several expressions (here a +-- hash clause and the hash keys derived from it), which the parallel-query +-- walks must process only once +set enable_nestloop = off; +set enable_mergejoin = off; +set enable_indexscan = off; +set enable_indexonlyscan = off; +set parallel_leader_participation = off; +create function explain_parallel_shared_subplan() returns setof text +language plpgsql as +$$ +declare ln text; +declare prev text; +begin + for ln in + explain (analyze, timing off, summary off, costs off, buffers off) + select count(*) from tenk1 t1 join tenk1 t2 + on t2.unique1 = case when t1.hundred not in + (select thousand from tenk2 order by twothousand limit 50) + then 1 else 2 end + loop + ln := regexp_replace(ln, 'actual rows=[0-9.]+ loops=[0-9]+', + 'actual rows=N loops=N'); + ln := regexp_replace(ln, 'Memory: \S*', 'Memory: xxx'); + ln := regexp_replace(ln, 'Memory Usage: \S*', 'Memory Usage: xxx'); + -- the launched worker count and the number of per-worker "Sort + -- Method" lines vary with how many workers start and get rows; + -- normalize both so the test only confirms that per-worker sort + -- statistics are reported at all (they vanish without the fix) + ln := regexp_replace(ln, 'Workers Launched: \d+', 'Workers Launched: N'); + ln := regexp_replace(ln, 'Worker \d+:', 'Worker N:'); + if ln is distinct from prev then + return next ln; + end if; + prev := ln; + end loop; +end; +$$; +select * from explain_parallel_shared_subplan(); +drop function explain_parallel_shared_subplan(); +reset parallel_leader_participation; +reset enable_indexscan; +reset enable_indexonlyscan; +reset enable_nestloop; +reset enable_mergejoin; alter table tenk2 reset (parallel_workers); -- test parallel plan for a query containing initplan. diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index a7a8dbdd23f..de67d4fc9d0 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -776,8 +776,6 @@ ExecForeignDelete_function ExecForeignInsert_function ExecForeignTruncate_function ExecForeignUpdate_function -ExecParallelEstimateContext -ExecParallelInitializeDSMContext ExecPhraseData ExecProcNodeMtd ExecRowMark @@ -4138,6 +4136,7 @@ pivot_field planner_hook_type planner_setup_hook_type planner_shutdown_hook_type +planstate_tree_collect_context planstate_tree_walker_callback plperl_array_info plperl_call_data -- 2.53.0