diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c index 316dd3f67c6e3..49e40e08a6b21 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 678b9a046ceb7..935f28118fe69 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 5a2034811d563..074d2ef67203b 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 696ffdebd4be4..46d3d040b830b 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 e13442156440f..32c3b2ab53557 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 71a75bc86ea76..b34d650fe1dbc 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 a7a8dbdd23f79..de67d4fc9d0a4 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