Hi,
Whilst stress testing the next version of the stack-based
instrumentation patch [0], I ran into what appears to be an ancient
bug in how we walk plan state for SubPlan nodes in parallel query
setup and instrumentation handling.
Specifically, the planstate_tree_walker callbacks in execParallel.c
assume that a given PlanState is only walked once, but that doesn't
hold true for hash joins since (I think, not tested) Postgres 10. We
create two separate references to the same subplan PlanState, one via
hashclauses and the other via hashkeys, each having their own
SubPlanState, but going to the same subplan PlanState. And
planstate_walk_subplans() descends into the subtree once per
reference.
I believe this is an issue since b8d7f053c5c (Postgres 10, CCing
Andres as the author of that one) changed how ExecInitHashJoin calls
ExecInitExpr, creating the second SubPlanState. More digging also
found the very old a64846f3ada that talked about a seemingly related
problem, but that must have happened before parallel query was in the
picture.
We can see this cause an assertion failure on master with the
following reproducer:
CREATE TABLE big (a int, b int);
INSERT INTO big SELECT g, g FROM generate_series(1, 10000) g;
CREATE TABLE small (x int PRIMARY KEY, y int);
INSERT INTO small SELECT g, g FROM generate_series(1, 1000) g;
CREATE TABLE other (k int PRIMARY KEY, f int);
INSERT INTO other SELECT g, g FROM generate_series(1, 1000) g;
ANALYZE big, small, other;
ALTER TABLE other SET (parallel_workers = 0);
SET parallel_setup_cost = 0;
SET parallel_tuple_cost = 0;
SET min_parallel_table_scan_size = 0;
SET enable_nestloop = off;
SET enable_mergejoin = off;
SET enable_parallel_hash = off;
SET parallel_leader_participation = off; -- to show loop count more clearly
EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT COUNT(*) FROM big b JOIN small s ON s.x =
CASE WHEN (b.a, b.b) NOT IN (SELECT o.k, o.f FROM other o ORDER BY o.f)
THEN 1 ELSE 2 END;
TRAP: failed Assert("toc->toc_entry[i].key != key"), File:
"shm_toc.c", Line: 190
Besides the assertion failure it also causes problems for
instrumentation. In this example we get an incorrect loop count (the
subplan is really only run in the two workers, so the loops=16 below
should be 4), and the Sort node is missing the Sort Method reporting:
QUERY PLAN
--------------------------------------------------------------
Finalize Aggregate (actual rows=1.00 loops=1)
-> Gather (actual rows=2.00 loops=1)
Workers Planned: 2
Workers Launched: 2
-> Partial Aggregate (actual rows=1.00 loops=2)
-> Hash Join (actual rows=5000.00 loops=2)
Hash Cond: (CASE WHEN (NOT (ANY ((b.a = (hashed
SubPlan any_1).col1) AND (b.b = (hashed SubPlan any_1).col2)))) THEN 1
ELSE 2 END = s.x)
-> Parallel Seq Scan on big b (actual
rows=5000.00 loops=2)
-> Hash (actual rows=1000.00 loops=2)
Buckets: 1024 Batches: 1 Memory Usage: 44kB
-> Seq Scan on small s (actual
rows=1000.00 loops=2)
SubPlan any_1
-> Sort (actual rows=1000.00 loops=16)
Sort Key: o.f
-> Seq Scan on other o (actual
rows=1000.00 loops=16)
The assertion failure happens because ExecParallelInitializeDSM visits
that Sort twice and tries to shm_toc_insert() the same plan_node_id
key. The instrumentation issue happens because (1) for loops, we
accumulate the worker instrumentation twice in
ExecParallelReportInstrumentation and then again in
ExecParallelRetrieveInstrumentation for the nodes underneath the
shared SubPlan, (2) for "Sort Method" the leader and workers end up
pointing at different (duplicate) SharedSortInfo slots and the leader
looks at the empty one.
To fix this, the least convoluted way I could think of is to invent a
new mechanism to collect a deduplicated list of plan states, and then
call the existing logic on each list member. This also allows some
refactoring to remove logic for counting how many nodes we've visited.
With the attached patch, the instrumentation now shows correctly, and
the added regression test no longer fails with an assertion failure.
...
SubPlan any_1
-> Sort (actual rows=1000.00 loops=4)
Sort Key: o.f
Worker 0: Sort Method: quicksort Memory: 48kB
Worker 1: Sort Method: quicksort Memory: 48kB
-> Seq Scan on other o (actual
rows=1000.00 loops=4)
We could alternatively deal with this in the existing walkers, but
then we'd have to modify the walker context in four different places
to carry the identities of already visited plan states. I also assume
that changing how ExecInitHashJoin works is off the table, but I admit
that logic goes a bit over my head.
Note that technically the ExecShutdownNode_walker suffers from the
same problem, but it doesn't appear to be a practical problem with
in-core code today, and it's a bit more complicated to fix with the
current nested InstrStartNode assumptions (that I intend to fix
separately in the stack-based instrumentation series).
Thoughts?
Thanks,
Lukas
[0]:
https://www.postgresql.org/message-id/flat/CAP53PkzdBK8VJ1fS4AZ481LgMN8f9mJiC39ZRHqkFUSYq6KWmg%40mail.gmail.com
--
Lukas Fittl
From be8b48a7d1a74d0a3c3a2485757add0344186d94 Mon Sep 17 00:00:00 2001
From: Lukas Fittl <[email protected]>
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 <[email protected]>
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
