Hi Tomas, Thanks for the detailed review, and no worries about the delay. > 1) I don't think the new ExecCustomScan callbacks can be this simple. > ... I suggest to make this more like the parallel query callbacks, so > that the core "owns" the allocation, and the CustomScan merely > estimates it, initializes it (after allocation, etc.).
Makes sense. Updated it accordingly.The wrappers in nodeCustom.c now mirror ExecCustomScanInitializeDS. The estimate callback returns a Size, and the executor does shm_toc_estimate_chunk/keys, shm_toc_allocate, shm_toc_insert and shm_toc_lookup itself, passing the chunk to the provider as "coordinate". The size is carried between the two phases in a new CustomScanState.pinstrument_len, like pscan_len. For the key I added PARALLEL_KEY_NODE_INSTRUMENTATION(plan_node_id) in execParallel.h, defined as 0xE100000000000000 + plan_node_id. That keeps it clear of plan_node_id (used for the parallel-aware chunk) and of the executor's fixed 0xE0.. keys, and the same macro serves ForeignScan. It lives in the header rather than in execParallel.c only because nodeCustom.c and nodeForeignscan.c both need it; happy to move or rename it if you prefer. > 2) AFAIK the callbacks are called even when instrumentation was not > requested, which seems a bit weird. Done. The wrappers now return early unless the node has instrumentation and there are workers, i.e. "if (!node->ss.ps.instrument || pcxt->nworkers == 0)", which is what nodeSort.c, nodeAgg.c and nodeMemoize.c do for their shared instrumentation. I went with that rather than SeqScan's INSTRUMENT_IO test since the executor cannot know which kind of stats a given provider collects. > 3) I'm not sure if the documentation should have more details about > what the callbacks should do Updated the documentation. > 4) There need to be some tests. Added tests. > 5) ... the EXPLAIN probably won't be able to print per-worker stats > because ExplainOpenWorker/ExplainCloseWorker are static in explain.c. Yep, the test needed them. I exported the two functions (declared in explain.h since that is where they are defined) rather than adding another callback; the provider's ExplainCustomScan then looks exactly like show_sort_info. Output of existing providers that do not use any of this is unchanged. > 6) I realized the same issue (lack of instrumentation) applies to > ForeignScan too. Took the liberty of adding a patch for ForeignScan as well (0002). It is the same change for FdwRoutine, sharing the key macro and the gating. I started on a test for it and realized it needs a bit more thought. There is no test FDW that is parallel-safe. file_fdw is, so a test would mean implementing the callbacks there and printing per-worker counts in its EXPLAIN output, which is a small user-visible feature for file_fdw rather than just a test. The alternative is a dummy parallel-safe FDW under src/test/modules - which would need a non trivial amount of scaffolding. I'm happy to do either as a follow-up once we agree which one; in the meantime the executor side is identical to the CustomScan paths that test_extensible covers. > 7) The naming of the new callbacks in CustomExecMethods seems a bit > inconsistent. Done. All four now end in "CustomScan": EstimateDSMInstrumentationCustomScan, InitializeDSMInstrumentationCustomScan, InitializeWorkerInstrumentationCustomScan, RetrieveInstrumentationCustomScan. Have attached two v5 patches, one for CustomScan and another for ForeignScan that pass make check and the test_extensible suite. Thanks, Siddharth
From 8347b9e6da6c0dd93d0a03b3f44f834699a2975d Mon Sep 17 00:00:00 2001 From: Siddharth Kothari <[email protected]> Date: Fri, 13 Nov 2026 12:25:16 +0000 Subject: [PATCH v5 1/2] Add parallel instrumentation hooks for CustomScan providers CustomScan providers currently lack a standard method to aggregate instrumentation data from parallel workers back to the leader process before the dynamic shared memory segment is destroyed. Furthermore, unlike built-in scan nodes, they are missing the callbacks required to allocate instrumentation shared memory when they are in the parallel part of a plan but not parallel-aware: execParallel.c only invokes the CustomScan DSM callbacks for parallel-aware nodes, and has no T_CustomScanState case when retrieving instrumentation. As a result, per-worker EXPLAIN ANALYZE data for such nodes is lost. This patch introduces four optional callbacks to CustomExecMethods: - EstimateDSMInstrumentationCustomScan - InitializeDSMInstrumentationCustomScan - InitializeWorkerInstrumentationCustomScan - RetrieveInstrumentationCustomScan They are modelled on the existing parallel-aware callbacks: the core executor owns the shm_toc chunk. The provider only returns the required size from the estimate callback and fills in the chunk it is handed as "coordinate"; the executor estimates, allocates, inserts and looks the chunk up under a new PARALLEL_KEY_NODE_INSTRUMENTATION(plan_node_id) key, which cannot collide with the plan_node_id key used for the node's parallel-aware state nor with the executor's fixed keys. A new CustomScanState.pinstrument_len field carries the estimated size from the estimate phase to the DSM initialization phase. The callbacks are invoked for any CustomScan node below a Gather, parallel-aware or not, but only when the node is instrumented and there is at least one worker, like built-in nodes do for their own shared instrumentation. RetrieveInstrumentationCustomScan is invoked once in the leader when the parallel context is torn down, and must copy the data out of DSM. ExplainOpenWorker() and ExplainCloseWorker() are exported so that a provider's ExplainCustomScan callback can print per-worker data in the same way as built-in nodes. Output of existing providers that do not use the new callbacks is unchanged. The lifecycle of the callbacks is documented in custom-scan.sgml, and src/test/modules/test_extensible is extended to implement them, count tuples per worker, and print the counts in EXPLAIN ANALYZE both serially and under a single-copy Gather. --- doc/src/sgml/custom-scan.sgml | 62 +++++++++- src/backend/commands/explain.c | 6 +- src/backend/executor/execParallel.c | 16 ++- src/backend/executor/nodeCustom.c | 77 ++++++++++++ src/include/commands/explain.h | 8 ++ src/include/executor/execParallel.h | 11 ++ src/include/executor/nodeCustom.h | 7 ++ src/include/nodes/execnodes.h | 1 + src/include/nodes/extensible.h | 15 +++ .../expected/test_extensible.out | 36 ++++++ .../test_extensible/sql/test_extensible.sql | 15 +++ .../modules/test_extensible/test_extensible.c | 115 ++++++++++++++++++ src/tools/pgindent/typedefs.list | 2 + 13 files changed, 365 insertions(+), 6 deletions(-) diff --git a/doc/src/sgml/custom-scan.sgml b/doc/src/sgml/custom-scan.sgml index a200d502cdd..70bc9f2e164 100644 --- a/doc/src/sgml/custom-scan.sgml +++ b/doc/src/sgml/custom-scan.sgml @@ -409,6 +409,63 @@ void (*ShutdownCustomScan) (CustomScanState *node); <para> <programlisting> +Size (*EstimateDSMInstrumentationCustomScan) (CustomScanState *node, + ParallelContext *pcxt); +</programlisting> + Estimate the amount of dynamic shared memory that will be required + for parallel operation's instrumentation. + This callback is optional, and need only be supplied if this custom + scan provider supports parallel execution and needs custom instrumentation + for <literal>EXPLAIN ANALYZE</literal>. It will only be invoked when + instrumentation is enabled and there is at least one worker. + </para> + + <para> +<programlisting> +void (*InitializeDSMInstrumentationCustomScan) (CustomScanState *node, + ParallelContext *pcxt, + void *coordinate); +</programlisting> + Initialize the dynamic shared memory that will be required for a custom + scan's parallel instrumentation. The core allocates the memory and maps it + to a <literal>shm_toc</literal> chunk, passing a pointer to the chunk in + <literal>coordinate</literal>. The provider should only initialize the chunk. + This callback is optional, and need only be supplied if this custom + scan provider supports parallel execution and needs custom instrumentation + for <literal>EXPLAIN ANALYZE</literal>. It will only be invoked when + instrumentation is enabled, there is at least one worker, and the chunk + size estimated by <function>EstimateDSMInstrumentationCustomScan</function> + is greater than zero. + </para> + + <para> +<programlisting> +void (*InitializeWorkerInstrumentationCustomScan) (CustomScanState *node, + ParallelWorkerContext *pwcxt, + void *coordinate); +</programlisting> + Initialize a parallel worker's local state based on the instrumentation + shared state set up by the leader. The same <literal>shm_toc</literal> + chunk allocated by the leader is passed in <literal>coordinate</literal>. + This callback is optional, and need only be supplied if this custom + scan provider supports parallel execution and needs custom instrumentation. + </para> + + <para> +<programlisting> +void (*RetrieveInstrumentationCustomScan) (CustomScanState *node); +</programlisting> + Retrieve parallel instrumentation data from dynamic shared memory into backend + memory. This callback is optional, and need only be supplied if this custom + scan provider supports parallel execution and needs custom instrumentation. + It is invoked once by the leader when the parallel context is being torn + down (after all workers have finished), not on every rescan. The custom + scan must copy out all data from the DSM segment into backend memory because + the DSM segment will be destroyed immediately afterwards. + </para> + + <para> +<programlisting> void (*ExplainCustomScan) (CustomScanState *node, List *ancestors, ExplainState *es); @@ -417,7 +474,10 @@ void (*ExplainCustomScan) (CustomScanState *node, plan node. This callback is optional. Common data stored in the <structname>ScanState</structname>, such as the target list and scan relation, will be shown even without this callback, but the callback allows the display - of additional, private state. + of additional, private state. It can also be used to output + per-worker instrumentation information by calling + <function>ExplainOpenWorker</function> and <function>ExplainCloseWorker</function>, + provided <literal>es->workers_state</literal> is not NULL. </para> </sect2> </sect1> diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c index e35e0a649b3..b7c9988a1c6 100644 --- a/src/backend/commands/explain.c +++ b/src/backend/commands/explain.c @@ -167,8 +167,6 @@ static void ExplainSubPlans(List *plans, List *ancestors, static void ExplainCustomChildren(CustomScanState *css, List *ancestors, ExplainState *es); static ExplainWorkersState *ExplainCreateWorkersState(int num_workers); -static void ExplainOpenWorker(int n, ExplainState *es); -static void ExplainCloseWorker(int n, ExplainState *es); static void ExplainFlushWorkersState(ExplainState *es); @@ -5203,7 +5201,7 @@ ExplainCreateWorkersState(int num_workers) /* * Begin or resume output into the set-aside group for worker N. */ -static void +void ExplainOpenWorker(int n, ExplainState *es) { ExplainWorkersState *wstate = es->workers_state; @@ -5265,7 +5263,7 @@ ExplainOpenWorker(int n, ExplainState *es) /* * End output for worker N --- must pair with previous ExplainOpenWorker call */ -static void +void ExplainCloseWorker(int n, ExplainState *es) { ExplainWorkersState *wstate = es->workers_state; diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c index 6b85508a696..73bdcc5786f 100644 --- a/src/backend/executor/execParallel.c +++ b/src/backend/executor/execParallel.c @@ -55,7 +55,9 @@ /* * Magic numbers for parallel executor communication. We use constants * greater than any 32-bit integer here so that values < 2^32 can be used - * by individual parallel nodes to store their own state. + * by individual parallel nodes to store their own state. For nodes that + * need additional keys beyond the plan_node_id (e.g. for parallel instrumentation + * chunks), see PARALLEL_KEY_NODE_INSTRUMENTATION in execParallel.h. */ #define PARALLEL_KEY_EXECUTOR_FIXED UINT64CONST(0xE000000000000001) #define PARALLEL_KEY_PLANNEDSTMT UINT64CONST(0xE000000000000002) @@ -304,6 +306,9 @@ ExecParallelEstimate(PlanState *planstate, ExecParallelEstimateContext *e) if (planstate->plan->parallel_aware) ExecCustomScanEstimate((CustomScanState *) planstate, e->pcxt); + /* even when not parallel-aware, for EXPLAIN ANALYZE */ + ExecCustomScanEstimateInstrumentation((CustomScanState *) planstate, + e->pcxt); break; case T_BitmapHeapScanState: if (planstate->plan->parallel_aware) @@ -552,6 +557,9 @@ ExecParallelInitializeDSM(PlanState *planstate, if (planstate->plan->parallel_aware) ExecCustomScanInitializeDSM((CustomScanState *) planstate, d->pcxt); + /* even when not parallel-aware, for EXPLAIN ANALYZE */ + ExecCustomScanInitializeDSMInstrumentation((CustomScanState *) planstate, + d->pcxt); break; case T_BitmapHeapScanState: if (planstate->plan->parallel_aware) @@ -1164,6 +1172,9 @@ ExecParallelRetrieveInstrumentation(PlanState *planstate, case T_TidRangeScanState: ExecTidRangeScanRetrieveInstrumentation((TidRangeScanState *) planstate); break; + case T_CustomScanState: + ExecCustomScanRetrieveInstrumentation((CustomScanState *) planstate); + break; default: break; } @@ -1449,6 +1460,9 @@ ExecParallelInitializeWorker(PlanState *planstate, ParallelWorkerContext *pwcxt) if (planstate->plan->parallel_aware) ExecCustomScanInitializeWorker((CustomScanState *) planstate, pwcxt); + /* even when not parallel-aware, for EXPLAIN ANALYZE */ + ExecCustomScanInitializeWorkerInstrumentation((CustomScanState *) planstate, + pwcxt); break; case T_BitmapHeapScanState: if (planstate->plan->parallel_aware) diff --git a/src/backend/executor/nodeCustom.c b/src/backend/executor/nodeCustom.c index b7cc890cd20..2d89d833a93 100644 --- a/src/backend/executor/nodeCustom.c +++ b/src/backend/executor/nodeCustom.c @@ -11,6 +11,7 @@ #include "postgres.h" #include "access/parallel.h" +#include "executor/execParallel.h" #include "executor/executor.h" #include "executor/nodeCustom.h" #include "miscadmin.h" @@ -217,6 +218,82 @@ ExecCustomScanInitializeWorker(CustomScanState *node, } } +/* ---------------------------------------------------------------- + * Parallel instrumentation support + * + * Unlike the parallel-aware callbacks above, these are invoked for + * any CustomScan node that runs in parallel workers, so that the + * provider can report per-worker statistics in EXPLAIN ANALYZE. The + * core executor owns the shm_toc chunk; the provider only estimates + * its size and fills it in. + * ---------------------------------------------------------------- + */ +void +ExecCustomScanEstimateInstrumentation(CustomScanState *node, + ParallelContext *pcxt) +{ + const CustomExecMethods *methods = node->methods; + + if (!node->ss.ps.instrument || pcxt->nworkers == 0 || + !methods->EstimateDSMInstrumentationCustomScan) + return; + + node->pinstrument_len = + methods->EstimateDSMInstrumentationCustomScan(node, pcxt); + if (node->pinstrument_len > 0) + { + shm_toc_estimate_chunk(&pcxt->estimator, node->pinstrument_len); + shm_toc_estimate_keys(&pcxt->estimator, 1); + } +} + +void +ExecCustomScanInitializeDSMInstrumentation(CustomScanState *node, + ParallelContext *pcxt) +{ + const CustomExecMethods *methods = node->methods; + void *coordinate; + + if (!node->ss.ps.instrument || pcxt->nworkers == 0 || + !methods->InitializeDSMInstrumentationCustomScan || + node->pinstrument_len == 0) + return; + + coordinate = shm_toc_allocate(pcxt->toc, node->pinstrument_len); + methods->InitializeDSMInstrumentationCustomScan(node, pcxt, coordinate); + shm_toc_insert(pcxt->toc, + PARALLEL_KEY_NODE_INSTRUMENTATION(node->ss.ps.plan->plan_node_id), + coordinate); +} + +void +ExecCustomScanInitializeWorkerInstrumentation(CustomScanState *node, + ParallelWorkerContext *pwcxt) +{ + const CustomExecMethods *methods = node->methods; + void *coordinate; + + if (!node->ss.ps.instrument || + !methods->InitializeWorkerInstrumentationCustomScan) + return; + + coordinate = shm_toc_lookup(pwcxt->toc, + PARALLEL_KEY_NODE_INSTRUMENTATION(node->ss.ps.plan->plan_node_id), + true); + if (coordinate) + methods->InitializeWorkerInstrumentationCustomScan(node, pwcxt, + coordinate); +} + +void +ExecCustomScanRetrieveInstrumentation(CustomScanState *node) +{ + const CustomExecMethods *methods = node->methods; + + if (methods->RetrieveInstrumentationCustomScan) + methods->RetrieveInstrumentationCustomScan(node); +} + void ExecShutdownCustomScan(CustomScanState *node) { diff --git a/src/include/commands/explain.h b/src/include/commands/explain.h index 472e141bba3..57c7b1237cf 100644 --- a/src/include/commands/explain.h +++ b/src/include/commands/explain.h @@ -82,4 +82,12 @@ extern void ExplainQueryText(ExplainState *es, QueryDesc *queryDesc); extern void ExplainQueryParameters(ExplainState *es, ParamListInfo params, int maxlen); +/* + * Extensions (e.g. CustomScan/ForeignScan providers) may use these from their + * EXPLAIN callback to print per-worker data. Callers must check that + * es->workers_state != NULL before attempting to open a worker group. + */ +extern void ExplainOpenWorker(int n, ExplainState *es); +extern void ExplainCloseWorker(int n, ExplainState *es); + #endif /* EXPLAIN_H */ diff --git a/src/include/executor/execParallel.h b/src/include/executor/execParallel.h index 5a2034811d5..29e32c2ae36 100644 --- a/src/include/executor/execParallel.h +++ b/src/include/executor/execParallel.h @@ -37,6 +37,17 @@ typedef struct ParallelExecutorInfo struct TupleQueueReader **reader; /* tuple reader/writer support */ } ParallelExecutorInfo; +/* + * shm_toc key for the parallel instrumentation chunk of a plan node whose + * shared state is opaque to the core executor (CustomScan, ForeignScan). + * plan_node_id itself is already used as the key for the node's + * parallel-aware shared state, so derive a distinct key from it. The + * fixed keys in execParallel.c live at 0xE000000000000000 and up, and + * plan_node_id is a non-negative int, so there can be no collision. + */ +#define PARALLEL_KEY_NODE_INSTRUMENTATION(plan_node_id) \ + (UINT64CONST(0xE100000000000000) + (uint64) (plan_node_id)) + extern ParallelExecutorInfo *ExecInitParallelPlan(PlanState *planstate, EState *estate, Bitmapset *sendParams, int nworkers, int64 tuples_needed); diff --git a/src/include/executor/nodeCustom.h b/src/include/executor/nodeCustom.h index fb0acc6e414..e56b88bb23c 100644 --- a/src/include/executor/nodeCustom.h +++ b/src/include/executor/nodeCustom.h @@ -37,6 +37,13 @@ extern void ExecCustomScanReInitializeDSM(CustomScanState *node, ParallelContext *pcxt); extern void ExecCustomScanInitializeWorker(CustomScanState *node, ParallelWorkerContext *pwcxt); +extern void ExecCustomScanEstimateInstrumentation(CustomScanState *node, + ParallelContext *pcxt); +extern void ExecCustomScanInitializeDSMInstrumentation(CustomScanState *node, + ParallelContext *pcxt); +extern void ExecCustomScanInitializeWorkerInstrumentation(CustomScanState *node, + ParallelWorkerContext *pwcxt); +extern void ExecCustomScanRetrieveInstrumentation(CustomScanState *node); extern void ExecShutdownCustomScan(CustomScanState *node); #endif /* NODECUSTOM_H */ diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index f0cb21444b2..ead605425df 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -2125,6 +2125,7 @@ typedef struct CustomScanState * nodes/extensible.h */ List *custom_ps; /* list of child PlanState nodes, if any */ Size pscan_len; /* size of parallel coordination information */ + Size pinstrument_len; /* size of parallel instrumentation chunk */ const struct CustomExecMethods *methods; const struct TupleTableSlotOps *slotOps; } CustomScanState; diff --git a/src/include/nodes/extensible.h b/src/include/nodes/extensible.h index b4fc6067807..0974974ddbc 100644 --- a/src/include/nodes/extensible.h +++ b/src/include/nodes/extensible.h @@ -153,6 +153,21 @@ typedef struct CustomExecMethods void *coordinate); void (*ShutdownCustomScan) (CustomScanState *node); + /* Optional: estimate parallel instrumentation size */ + Size (*EstimateDSMInstrumentationCustomScan) (CustomScanState *node, + ParallelContext *pcxt); + /* Optional: initialize parallel instrumentation DSM */ + void (*InitializeDSMInstrumentationCustomScan) (CustomScanState *node, + ParallelContext *pcxt, + void *coordinate); + /* Optional: initialize parallel instrumentation worker */ + void (*InitializeWorkerInstrumentationCustomScan) (CustomScanState *node, + ParallelWorkerContext *pwcxt, + void *coordinate); + + /* Optional: retrieve parallel instrumentation */ + void (*RetrieveInstrumentationCustomScan) (CustomScanState *node); + /* Optional: print additional information in EXPLAIN */ void (*ExplainCustomScan) (CustomScanState *node, List *ancestors, diff --git a/src/test/modules/test_extensible/expected/test_extensible.out b/src/test/modules/test_extensible/expected/test_extensible.out index acc0a1af7fa..2d3c9b356b9 100644 --- a/src/test/modules/test_extensible/expected/test_extensible.out +++ b/src/test/modules/test_extensible/expected/test_extensible.out @@ -245,5 +245,41 @@ SELECT id, val FROM ONLY test_extensible_tbl ORDER BY id; (6 rows) DROP TABLE test_extensible_child; +-- EXPLAIN ANALYZE instrumentation, serial: the leader counts the tuples +-- itself (disable debug_parallel_query in case it is forced on globally) +SET debug_parallel_query = off; +EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF) + SELECT id, val FROM ONLY test_extensible_tbl ORDER BY id; + QUERY PLAN +-------------------------------------------------------------------------------------- + Sort (actual rows=6.00 loops=1) + Sort Key: id + Sort Method: quicksort Memory: 25kB + -> Custom Scan (TestCustomScan) on test_extensible_tbl (actual rows=6.00 loops=1) + Tuples Emitted: 6 +(5 rows) + +-- EXPLAIN ANALYZE instrumentation, parallel: a single-copy Gather runs the +-- scan in one worker, so the count must come back through shared memory +SET max_parallel_workers_per_gather = 1; +SET debug_parallel_query = on; +EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF) + SELECT id, val FROM ONLY test_extensible_tbl ORDER BY id; + QUERY PLAN +-------------------------------------------------------------------------------------------- + Gather (actual rows=6.00 loops=1) + Workers Planned: 1 + Workers Launched: 1 + Single Copy: true + -> Sort (actual rows=6.00 loops=1) + Sort Key: id + Worker 0: Sort Method: quicksort Memory: 25kB + -> Custom Scan (TestCustomScan) on test_extensible_tbl (actual rows=6.00 loops=1) + Tuples Emitted: 0 + Worker 0: Tuples Emitted: 6 +(10 rows) + +RESET debug_parallel_query; +RESET max_parallel_workers_per_gather; DROP TABLE test_extensible_tbl; DROP EXTENSION test_extensible; diff --git a/src/test/modules/test_extensible/sql/test_extensible.sql b/src/test/modules/test_extensible/sql/test_extensible.sql index 72eb3cf82af..18d11102b01 100644 --- a/src/test/modules/test_extensible/sql/test_extensible.sql +++ b/src/test/modules/test_extensible/sql/test_extensible.sql @@ -82,6 +82,21 @@ EXPLAIN (COSTS OFF) SELECT id, val FROM ONLY test_extensible_tbl; SELECT id, val FROM ONLY test_extensible_tbl ORDER BY id; DROP TABLE test_extensible_child; +-- EXPLAIN ANALYZE instrumentation, serial: the leader counts the tuples +-- itself (disable debug_parallel_query in case it is forced on globally) +SET debug_parallel_query = off; +EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF) + SELECT id, val FROM ONLY test_extensible_tbl ORDER BY id; + +-- EXPLAIN ANALYZE instrumentation, parallel: a single-copy Gather runs the +-- scan in one worker, so the count must come back through shared memory +SET max_parallel_workers_per_gather = 1; +SET debug_parallel_query = on; +EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF, BUFFERS OFF) + SELECT id, val FROM ONLY test_extensible_tbl ORDER BY id; +RESET debug_parallel_query; +RESET max_parallel_workers_per_gather; + DROP TABLE test_extensible_tbl; DROP EXTENSION test_extensible; diff --git a/src/test/modules/test_extensible/test_extensible.c b/src/test/modules/test_extensible/test_extensible.c index dafcd772f56..b5b693624e6 100644 --- a/src/test/modules/test_extensible/test_extensible.c +++ b/src/test/modules/test_extensible/test_extensible.c @@ -17,6 +17,8 @@ #include "access/tableam.h" #include "catalog/namespace.h" #include "executor/executor.h" +#include "commands/explain.h" +#include "commands/explain_format.h" #include "fmgr.h" #include "miscadmin.h" #include "nodes/extensible.h" @@ -151,6 +153,19 @@ static const ExtensibleNodeMethods test_ext_node_methods = .nodeRead = test_ext_node_read_cb, }; +/* per-worker instrumentation, stored in shared memory */ +typedef struct TestScanInstrumentation +{ + int64 ntuples; /* tuples emitted by this worker */ +} TestScanInstrumentation; + +/* shared memory chunk holding instrumentation for all workers */ +typedef struct SharedTestScanInfo +{ + int num_workers; + TestScanInstrumentation sinstrument[FLEXIBLE_ARRAY_MEMBER]; +} SharedTestScanInfo; + /* * TestCustomScanState - execution state for the custom scan */ @@ -160,6 +175,9 @@ typedef struct TestCustomScanState TableScanDesc scandesc; int repeat_count; /* repeat_count from TestExtNode */ int repeats_left; /* how many more times to return current row */ + int64 ntuples; /* tuples emitted by this process */ + SharedTestScanInfo *shared_info; /* all-worker instrumentation */ + TestScanInstrumentation *sinstrument; /* this worker's slot, or NULL */ } TestCustomScanState; /* @@ -197,6 +215,9 @@ test_begin_custom_scan(CustomScanState *node, EState *estate, int eflags) tstate->repeat_count = tnode->repeat_count; tstate->repeats_left = 0; tstate->scandesc = NULL; + tstate->ntuples = 0; + tstate->shared_info = NULL; + tstate->sinstrument = NULL; /* A plain EXPLAIN never executes the plan, so skip */ if (eflags & EXEC_FLAG_EXPLAIN_ONLY) @@ -220,6 +241,7 @@ test_scan_next(ScanState *node) if (tstate->repeats_left > 0) { tstate->repeats_left--; + tstate->ntuples++; return slot; } @@ -227,6 +249,7 @@ test_scan_next(ScanState *node) return NULL; tstate->repeats_left = tstate->repeat_count - 1; + tstate->ntuples++; return slot; } @@ -251,6 +274,10 @@ test_end_custom_scan(CustomScanState *node) { TestCustomScanState *tstate = (TestCustomScanState *) node; + /* In a parallel worker, report our count to the leader. */ + if (tstate->sinstrument != NULL) + tstate->sinstrument->ntuples = tstate->ntuples; + /* No scan started under EXEC_FLAG_EXPLAIN_ONLY */ if (tstate->scandesc != NULL) table_endscan(tstate->scandesc); @@ -269,6 +296,89 @@ test_rescan_custom_scan(CustomScanState *node) ExecScanReScan(&node->ss); } +/* + * Parallel instrumentation callbacks. + * + * Each worker counts the tuples it emitted; at the end of the scan the count + * is written into a per-worker slot of a shared memory chunk owned by the + * core executor. The leader copies the chunk out of DSM in the Retrieve + * callback and prints it in EXPLAIN ANALYZE. + */ +static Size +test_estimate_dsm_custom_scan(CustomScanState *node, ParallelContext *pcxt) +{ + return offsetof(SharedTestScanInfo, sinstrument) + + pcxt->nworkers * sizeof(TestScanInstrumentation); +} + +static void +test_initialize_dsm_custom_scan(CustomScanState *node, ParallelContext *pcxt, + void *coordinate) +{ + TestCustomScanState *tstate = (TestCustomScanState *) node; + + tstate->shared_info = (SharedTestScanInfo *) coordinate; + memset(tstate->shared_info, 0, node->pinstrument_len); + tstate->shared_info->num_workers = pcxt->nworkers; +} + +static void +test_initialize_worker_custom_scan(CustomScanState *node, + ParallelWorkerContext *pwcxt, + void *coordinate) +{ + TestCustomScanState *tstate = (TestCustomScanState *) node; + + tstate->shared_info = (SharedTestScanInfo *) coordinate; + tstate->sinstrument = + &tstate->shared_info->sinstrument[ParallelWorkerNumber]; +} + +static void +test_retrieve_instrumentation_custom_scan(CustomScanState *node) +{ + TestCustomScanState *tstate = (TestCustomScanState *) node; + SharedTestScanInfo *si; + Size size; + + if (tstate->shared_info == NULL) + return; + + /* The DSM segment is about to go away, so copy the data out of it. */ + size = offsetof(SharedTestScanInfo, sinstrument) + + tstate->shared_info->num_workers * sizeof(TestScanInstrumentation); + si = palloc(size); + memcpy(si, tstate->shared_info, size); + tstate->shared_info = si; +} + +static void +test_explain_custom_scan(CustomScanState *node, List *ancestors, + ExplainState *es) +{ + TestCustomScanState *tstate = (TestCustomScanState *) node; + + if (!es->analyze) + return; + + ExplainPropertyInteger("Tuples Emitted", NULL, tstate->ntuples, es); + + if (tstate->shared_info != NULL && es->workers_state != NULL) + { + for (int n = 0; n < tstate->shared_info->num_workers; n++) + { + TestScanInstrumentation *sinstrument; + + sinstrument = &tstate->shared_info->sinstrument[n]; + + ExplainOpenWorker(n, es); + ExplainPropertyInteger("Tuples Emitted", NULL, + sinstrument->ntuples, es); + ExplainCloseWorker(n, es); + } + } +} + static const CustomExecMethods test_custom_exec_methods = { .CustomName = TEST_CUSTOM_SCAN_NAME, @@ -276,6 +386,11 @@ static const CustomExecMethods test_custom_exec_methods = .ExecCustomScan = test_exec_custom_scan, .EndCustomScan = test_end_custom_scan, .ReScanCustomScan = test_rescan_custom_scan, + .EstimateDSMInstrumentationCustomScan = test_estimate_dsm_custom_scan, + .InitializeDSMInstrumentationCustomScan = test_initialize_dsm_custom_scan, + .InitializeWorkerInstrumentationCustomScan = test_initialize_worker_custom_scan, + .RetrieveInstrumentationCustomScan = test_retrieve_instrumentation_custom_scan, + .ExplainCustomScan = test_explain_custom_scan, }; /* diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 7aedaafab90..e7b3beb05b2 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2872,6 +2872,7 @@ SharedRecordTableKey SharedRecordTypmodRegistry SharedSeqScanInstrumentation SharedSortInfo +SharedTestScanInfo SharedTidRangeScanInstrumentation SharedTuplestore SharedTuplestoreAccessor @@ -3167,6 +3168,7 @@ TestDSMRegistryStruct TestDecodingData TestDecodingTxnData TestExtNode +TestScanInstrumentation TestShmemData TestSpec TestValueType -- 2.56.0.611.g1030f0b940-goog
From 7ff334da9731d2a22f748e35d9ed21201ce2e2f1 Mon Sep 17 00:00:00 2001 From: Siddharth Kothari <[email protected]> Date: Fri, 13 Nov 2026 09:13:33 +0000 Subject: [PATCH v5 2/2] Add parallel instrumentation hooks for ForeignScan ForeignScan has the same gap as CustomScan: when a foreign scan runs in parallel workers but is not parallel-aware, there is no way for the FDW to allocate shared memory for its own instrumentation or to retrieve it from workers before the DSM segment is destroyed, so per-worker data is lost from EXPLAIN ANALYZE. Add four optional FdwRoutine callbacks mirroring the CustomScan ones: - EstimateDSMInstrumentationForeignScan - InitializeDSMInstrumentationForeignScan - InitializeWorkerInstrumentationForeignScan - RetrieveInstrumentationForeignScan The core executor owns the shm_toc chunk (estimate, allocate, insert, lookup under PARALLEL_KEY_NODE_INSTRUMENTATION) and only invokes the callbacks when instrumentation was requested and there are workers. A ForeignScanState.pinstrument_len field carries the estimated size between phases. Documented in fdwhandler.sgml. No in-tree FDW is parallel-safe, so there is no regression test for this yet; the code paths are identical to the tested CustomScan ones. --- doc/src/sgml/fdwhandler.sgml | 45 +++++++++++++++ src/backend/executor/execParallel.c | 12 ++++ src/backend/executor/nodeForeignscan.c | 78 ++++++++++++++++++++++++++ src/include/executor/nodeForeignscan.h | 7 +++ src/include/foreign/fdwapi.h | 15 +++++ src/include/nodes/execnodes.h | 1 + 6 files changed, 158 insertions(+) diff --git a/doc/src/sgml/fdwhandler.sgml b/doc/src/sgml/fdwhandler.sgml index 502441fefcb..db04633eb92 100644 --- a/doc/src/sgml/fdwhandler.sgml +++ b/doc/src/sgml/fdwhandler.sgml @@ -1612,6 +1612,51 @@ ShutdownForeignScan(ForeignScanState *node); wish to take some action before the DSM segment goes away should implement this method. </para> + + <para> +<programlisting> +Size +EstimateDSMInstrumentationForeignScan(ForeignScanState *node, ParallelContext *pcxt); +</programlisting> + Estimate the amount of dynamic shared memory that will be required + for a foreign scan's parallel instrumentation (for <literal>EXPLAIN ANALYZE</literal>). + This function is optional, and need only be supplied if this foreign + scan provider supports parallel execution and needs custom instrumentation. + </para> + + <para> +<programlisting> +void +InitializeDSMInstrumentationForeignScan(ForeignScanState *node, ParallelContext *pcxt, + void *coordinate); +</programlisting> + Initialize the dynamic shared memory that will be required for a foreign + scan's parallel instrumentation. + This function is optional, and need only be supplied if this foreign + scan provider supports parallel execution and needs custom instrumentation. + </para> + + <para> +<programlisting> +void +InitializeWorkerInstrumentationForeignScan(ForeignScanState *node, ParallelWorkerContext *pwcxt, + void *coordinate); +</programlisting> + Initialize a parallel worker's local state based on the instrumentation shared state + set up by the leader. + This function is optional, and need only be supplied if this foreign + scan provider supports parallel execution and needs custom instrumentation. + </para> + + <para> +<programlisting> +void +RetrieveInstrumentationForeignScan(ForeignScanState *node); +</programlisting> + Retrieve parallel instrumentation data from dynamic shared memory into backend memory. + This function is optional, and need only be supplied if this foreign + scan provider supports parallel execution and needs custom instrumentation. + </para> </sect2> <sect2 id="fdw-callbacks-async"> diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c index 73bdcc5786f..fdd93d09bcf 100644 --- a/src/backend/executor/execParallel.c +++ b/src/backend/executor/execParallel.c @@ -288,6 +288,9 @@ ExecParallelEstimate(PlanState *planstate, ExecParallelEstimateContext *e) if (planstate->plan->parallel_aware) ExecForeignScanEstimate((ForeignScanState *) planstate, e->pcxt); + /* even when not parallel-aware, for EXPLAIN ANALYZE */ + ExecForeignScanEstimateInstrumentation((ForeignScanState *) planstate, + e->pcxt); break; case T_TidRangeScanState: if (planstate->plan->parallel_aware) @@ -539,6 +542,9 @@ ExecParallelInitializeDSM(PlanState *planstate, if (planstate->plan->parallel_aware) ExecForeignScanInitializeDSM((ForeignScanState *) planstate, d->pcxt); + /* even when not parallel-aware, for EXPLAIN ANALYZE */ + ExecForeignScanInitializeDSMInstrumentation((ForeignScanState *) planstate, + d->pcxt); break; case T_TidRangeScanState: if (planstate->plan->parallel_aware) @@ -1175,6 +1181,9 @@ ExecParallelRetrieveInstrumentation(PlanState *planstate, case T_CustomScanState: ExecCustomScanRetrieveInstrumentation((CustomScanState *) planstate); break; + case T_ForeignScanState: + ExecForeignScanRetrieveInstrumentation((ForeignScanState *) planstate); + break; default: break; } @@ -1443,6 +1452,9 @@ ExecParallelInitializeWorker(PlanState *planstate, ParallelWorkerContext *pwcxt) if (planstate->plan->parallel_aware) ExecForeignScanInitializeWorker((ForeignScanState *) planstate, pwcxt); + /* even when not parallel-aware, for EXPLAIN ANALYZE */ + ExecForeignScanInitializeWorkerInstrumentation((ForeignScanState *) planstate, + pwcxt); break; case T_TidRangeScanState: if (planstate->plan->parallel_aware) diff --git a/src/backend/executor/nodeForeignscan.c b/src/backend/executor/nodeForeignscan.c index 6f0daddce07..c22f3269781 100644 --- a/src/backend/executor/nodeForeignscan.c +++ b/src/backend/executor/nodeForeignscan.c @@ -23,6 +23,7 @@ #include "postgres.h" #include "executor/executor.h" +#include "executor/execParallel.h" #include "executor/nodeForeignscan.h" #include "foreign/fdwapi.h" #include "utils/rel.h" @@ -493,3 +494,80 @@ ExecAsyncForeignScanNotify(AsyncRequest *areq) Assert(fdwroutine->ForeignAsyncNotify != NULL); fdwroutine->ForeignAsyncNotify(areq); } + +/* ---------------------------------------------------------------- + * Parallel instrumentation support + * + * Unlike the parallel-aware callbacks above, these are invoked for + * any ForeignScan node that runs in parallel workers, so that the + * FDW can report per-worker statistics in EXPLAIN ANALYZE. The + * core executor owns the shm_toc chunk; the FDW only estimates its + * size and fills it in. + * ---------------------------------------------------------------- + */ +void +ExecForeignScanEstimateInstrumentation(ForeignScanState *node, + ParallelContext *pcxt) +{ + FdwRoutine *fdwroutine = node->fdwroutine; + + if (!node->ss.ps.instrument || pcxt->nworkers == 0 || + !fdwroutine->EstimateDSMInstrumentationForeignScan) + return; + + node->pinstrument_len = + fdwroutine->EstimateDSMInstrumentationForeignScan(node, pcxt); + if (node->pinstrument_len > 0) + { + shm_toc_estimate_chunk(&pcxt->estimator, node->pinstrument_len); + shm_toc_estimate_keys(&pcxt->estimator, 1); + } +} + +void +ExecForeignScanInitializeDSMInstrumentation(ForeignScanState *node, + ParallelContext *pcxt) +{ + FdwRoutine *fdwroutine = node->fdwroutine; + void *coordinate; + + if (!node->ss.ps.instrument || pcxt->nworkers == 0 || + !fdwroutine->InitializeDSMInstrumentationForeignScan || + node->pinstrument_len == 0) + return; + + coordinate = shm_toc_allocate(pcxt->toc, node->pinstrument_len); + fdwroutine->InitializeDSMInstrumentationForeignScan(node, pcxt, coordinate); + shm_toc_insert(pcxt->toc, + PARALLEL_KEY_NODE_INSTRUMENTATION(node->ss.ps.plan->plan_node_id), + coordinate); +} + +void +ExecForeignScanInitializeWorkerInstrumentation(ForeignScanState *node, + ParallelWorkerContext *pwcxt) +{ + FdwRoutine *fdwroutine = node->fdwroutine; + void *coordinate; + + if (!node->ss.ps.instrument || + !fdwroutine->InitializeWorkerInstrumentationForeignScan) + return; + + coordinate = shm_toc_lookup(pwcxt->toc, + PARALLEL_KEY_NODE_INSTRUMENTATION(node->ss.ps.plan->plan_node_id), + true); + if (coordinate) + fdwroutine->InitializeWorkerInstrumentationForeignScan(node, pwcxt, + coordinate); +} + +void +ExecForeignScanRetrieveInstrumentation(ForeignScanState *node) +{ + FdwRoutine *fdwroutine = node->fdwroutine; + + if (fdwroutine->RetrieveInstrumentationForeignScan) + fdwroutine->RetrieveInstrumentationForeignScan(node); +} + diff --git a/src/include/executor/nodeForeignscan.h b/src/include/executor/nodeForeignscan.h index 4a14e67ed08..23151fa5376 100644 --- a/src/include/executor/nodeForeignscan.h +++ b/src/include/executor/nodeForeignscan.h @@ -29,6 +29,13 @@ extern void ExecForeignScanReInitializeDSM(ForeignScanState *node, ParallelContext *pcxt); extern void ExecForeignScanInitializeWorker(ForeignScanState *node, ParallelWorkerContext *pwcxt); +extern void ExecForeignScanEstimateInstrumentation(ForeignScanState *node, + ParallelContext *pcxt); +extern void ExecForeignScanInitializeDSMInstrumentation(ForeignScanState *node, + ParallelContext *pcxt); +extern void ExecForeignScanInitializeWorkerInstrumentation(ForeignScanState *node, + ParallelWorkerContext *pwcxt); +extern void ExecForeignScanRetrieveInstrumentation(ForeignScanState *node); extern void ExecShutdownForeignScan(ForeignScanState *node); extern void ExecAsyncForeignScanRequest(AsyncRequest *areq); diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h index abf59a0d8ad..ae55c572070 100644 --- a/src/include/foreign/fdwapi.h +++ b/src/include/foreign/fdwapi.h @@ -195,6 +195,16 @@ typedef void (*ForeignAsyncConfigureWait_function) (AsyncRequest *areq); typedef void (*ForeignAsyncNotify_function) (AsyncRequest *areq); +typedef Size (*EstimateDSMInstrumentationForeignScan_function) (ForeignScanState *node, + ParallelContext *pcxt); +typedef void (*InitializeDSMInstrumentationForeignScan_function) (ForeignScanState *node, + ParallelContext *pcxt, + void *coordinate); +typedef void (*InitializeWorkerInstrumentationForeignScan_function) (ForeignScanState *node, + ParallelWorkerContext *pwcxt, + void *coordinate); +typedef void (*RetrieveInstrumentationForeignScan_function) (ForeignScanState *node); + /* * FdwRoutine is the struct returned by a foreign-data wrapper's handler * function. It provides pointers to the callback functions needed by the @@ -275,6 +285,11 @@ typedef struct FdwRoutine InitializeWorkerForeignScan_function InitializeWorkerForeignScan; ShutdownForeignScan_function ShutdownForeignScan; + EstimateDSMInstrumentationForeignScan_function EstimateDSMInstrumentationForeignScan; + InitializeDSMInstrumentationForeignScan_function InitializeDSMInstrumentationForeignScan; + InitializeWorkerInstrumentationForeignScan_function InitializeWorkerInstrumentationForeignScan; + RetrieveInstrumentationForeignScan_function RetrieveInstrumentationForeignScan; + /* Support functions for path reparameterization. */ ReparameterizeForeignPathByChild_function ReparameterizeForeignPathByChild; diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index ead605425df..ac8749c535f 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -2097,6 +2097,7 @@ typedef struct ForeignScanState ScanState ss; /* its first field is NodeTag */ ExprState *fdw_recheck_quals; /* original quals not in ss.ps.qual */ Size pscan_len; /* size of parallel coordination information */ + Size pinstrument_len; /* size of parallel instrumentation chunk */ ResultRelInfo *resultRelInfo; /* result rel info, if UPDATE or DELETE */ /* use struct pointer to avoid including fdwapi.h here */ struct FdwRoutine *fdwroutine; -- 2.56.0.611.g1030f0b940-goog
