From 6ed997724f61b61632cd2ec0745aa8d9787927dd Mon Sep 17 00:00:00 2001 From: Lukas Fittl Date: Sat, 31 Oct 2026 09:59:25 -0700 Subject: [PATCH v2 4/5] Unify parallel worker handling for query text Multiple parallel mechanisms outside of the executor (nbtree/GIN/BRIN index builds, vacuum) all had the same implementation for passing query text to the parallel workers. To reduce duplicated code, introduce new helper functions to: (1) estimate the needed shared memory space for the query text, (2) store the query text in shared memory, and (3) restore the query text in the worker. Author: Lukas Fittl Reviewed-by: Sami Imseih Reviewed-by: Michael Paquier Discussion: https://postgr.es/m/CAP53Pkw3WcVbg61G29AmN=6XkuxGryoWKnhTyC1JmkT08i8qVA@mail.gmail.com --- src/backend/access/brin/brin.c | 28 ++----------- src/backend/access/gin/gininsert.c | 28 ++----------- src/backend/access/nbtree/nbtsort.c | 27 ++----------- src/backend/access/transam/parallel.c | 57 +++++++++++++++++++++++++++ src/backend/commands/vacuumparallel.c | 25 ++---------- src/include/access/parallel.h | 5 +++ 6 files changed, 77 insertions(+), 93 deletions(-) diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c index 3b88bebf525..7a133e0b2e9 100644 --- a/src/backend/access/brin/brin.c +++ b/src/backend/access/brin/brin.c @@ -36,7 +36,6 @@ #include "storage/bufmgr.h" #include "storage/freespace.h" #include "storage/proc.h" -#include "tcop/tcopprot.h" #include "utils/acl.h" #include "utils/datum.h" #include "utils/fmgrprotos.h" @@ -2400,14 +2399,7 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index, shm_toc_estimate_keys(&pcxt->estimator, 1); /* Finally, estimate PARALLEL_KEY_QUERY_TEXT space */ - if (debug_query_string) - { - querylen = strlen(debug_query_string); - shm_toc_estimate_chunk(&pcxt->estimator, querylen + 1); - shm_toc_estimate_keys(&pcxt->estimator, 1); - } - else - querylen = 0; /* keep compiler quiet */ + querylen = EstimateParallelQueryText(pcxt); /* Everyone's had a chance to ask for space, so now create the DSM */ InitializeParallelDSM(pcxt); @@ -2445,14 +2437,7 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index, shm_toc_insert(pcxt->toc, PARALLEL_KEY_TUPLESORT, sharedsort); /* Store query string for workers */ - if (debug_query_string) - { - char *sharedquery; - - sharedquery = (char *) shm_toc_allocate(pcxt->toc, querylen + 1); - memcpy(sharedquery, debug_query_string, querylen + 1); - shm_toc_insert(pcxt->toc, PARALLEL_KEY_QUERY_TEXT, sharedquery); - } + StoreParallelQueryText(pcxt, PARALLEL_KEY_QUERY_TEXT, querylen); /* * Allocate space for each worker's WalUsage and BufferUsage; no need to @@ -2777,7 +2762,6 @@ _brin_parallel_scan_and_build(BrinBuildState *state, void _brin_parallel_build_main(dsm_segment *seg, shm_toc *toc) { - char *sharedquery; BrinShared *brinshared; Sharedsort *sharedsort; BrinBuildState *buildstate; @@ -2794,12 +2778,8 @@ _brin_parallel_build_main(dsm_segment *seg, shm_toc *toc) Assert((MyProc->statusFlags == 0) || (MyProc->statusFlags == PROC_IN_SAFE_IC)); - /* Set debug_query_string for individual workers first */ - sharedquery = shm_toc_lookup(toc, PARALLEL_KEY_QUERY_TEXT, true); - debug_query_string = sharedquery; - - /* Report the query string from leader */ - pgstat_report_activity(STATE_RUNNING, debug_query_string); + /* Set debug_query_string and report the query string from leader */ + RestoreParallelQueryText(toc, PARALLEL_KEY_QUERY_TEXT); /* Look up brin shared state */ brinshared = shm_toc_lookup(toc, PARALLEL_KEY_BRIN_SHARED, false); diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c index 7ecc4d3f98d..2a1a344af9e 100644 --- a/src/backend/access/gin/gininsert.c +++ b/src/backend/access/gin/gininsert.c @@ -31,7 +31,6 @@ #include "storage/bufmgr.h" #include "storage/proc.h" #include "storage/predicate.h" -#include "tcop/tcopprot.h" #include "utils/datum.h" #include "utils/memutils.h" #include "utils/builtins.h" @@ -956,14 +955,7 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index, shm_toc_estimate_keys(&pcxt->estimator, 1); /* Finally, estimate PARALLEL_KEY_QUERY_TEXT space */ - if (debug_query_string) - { - querylen = strlen(debug_query_string); - shm_toc_estimate_chunk(&pcxt->estimator, querylen + 1); - shm_toc_estimate_keys(&pcxt->estimator, 1); - } - else - querylen = 0; /* keep compiler quiet */ + querylen = EstimateParallelQueryText(pcxt); /* Everyone's had a chance to ask for space, so now create the DSM */ InitializeParallelDSM(pcxt); @@ -995,14 +987,7 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index, shm_toc_insert(pcxt->toc, PARALLEL_KEY_TUPLESORT, sharedsort); /* Store query string for workers */ - if (debug_query_string) - { - char *sharedquery; - - sharedquery = (char *) shm_toc_allocate(pcxt->toc, querylen + 1); - memcpy(sharedquery, debug_query_string, querylen + 1); - shm_toc_insert(pcxt->toc, PARALLEL_KEY_QUERY_TEXT, sharedquery); - } + StoreParallelQueryText(pcxt, PARALLEL_KEY_QUERY_TEXT, querylen); /* * Allocate space for each worker's WalUsage and BufferUsage; no need to @@ -1999,7 +1984,6 @@ _gin_parallel_scan_and_build(GinBuildState *state, void _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc) { - char *sharedquery; GinBuildShared *ginshared; Sharedsort *sharedsort; GinBuildState buildstate; @@ -2016,12 +2000,8 @@ _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc) Assert((MyProc->statusFlags == 0) || (MyProc->statusFlags == PROC_IN_SAFE_IC)); - /* Set debug_query_string for individual workers first */ - sharedquery = shm_toc_lookup(toc, PARALLEL_KEY_QUERY_TEXT, true); - debug_query_string = sharedquery; - - /* Report the query string from leader */ - pgstat_report_activity(STATE_RUNNING, debug_query_string); + /* Set debug_query_string and report the query string from leader */ + RestoreParallelQueryText(toc, PARALLEL_KEY_QUERY_TEXT); /* Look up gin shared state */ ginshared = shm_toc_lookup(toc, PARALLEL_KEY_GIN_SHARED, false); diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c index f049010150c..56d204106e4 100644 --- a/src/backend/access/nbtree/nbtsort.c +++ b/src/backend/access/nbtree/nbtsort.c @@ -1420,14 +1420,7 @@ _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request) shm_toc_estimate_keys(&pcxt->estimator, 1); /* Finally, estimate PARALLEL_KEY_QUERY_TEXT space */ - if (debug_query_string) - { - querylen = strlen(debug_query_string); - shm_toc_estimate_chunk(&pcxt->estimator, querylen + 1); - shm_toc_estimate_keys(&pcxt->estimator, 1); - } - else - querylen = 0; /* keep compiler quiet */ + querylen = EstimateParallelQueryText(pcxt); /* Everyone's had a chance to ask for space, so now create the DSM */ InitializeParallelDSM(pcxt); @@ -1479,14 +1472,7 @@ _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request) } /* Store query string for workers */ - if (debug_query_string) - { - char *sharedquery; - - sharedquery = (char *) shm_toc_allocate(pcxt->toc, querylen + 1); - memcpy(sharedquery, debug_query_string, querylen + 1); - shm_toc_insert(pcxt->toc, PARALLEL_KEY_QUERY_TEXT, sharedquery); - } + StoreParallelQueryText(pcxt, PARALLEL_KEY_QUERY_TEXT, querylen); /* * Allocate space for each worker's WalUsage and BufferUsage; no need to @@ -1629,7 +1615,6 @@ _bt_leader_participate_as_worker(BTBuildState *buildstate) void _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc) { - char *sharedquery; BTSpool *btspool; BTSpool *btspool2; BTShared *btshared; @@ -1653,12 +1638,8 @@ _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc) Assert((MyProc->statusFlags == 0) || (MyProc->statusFlags == PROC_IN_SAFE_IC)); - /* Set debug_query_string for individual workers first */ - sharedquery = shm_toc_lookup(toc, PARALLEL_KEY_QUERY_TEXT, true); - debug_query_string = sharedquery; - - /* Report the query string from leader */ - pgstat_report_activity(STATE_RUNNING, debug_query_string); + /* Set debug_query_string and report the query string from leader */ + RestoreParallelQueryText(toc, PARALLEL_KEY_QUERY_TEXT); /* Look up nbtree shared state */ btshared = shm_toc_lookup(toc, PARALLEL_KEY_BTREE_SHARED, false); diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index b51df2a4606..d7be9d84ce7 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -1034,6 +1034,63 @@ DestroyParallelContext(ParallelContext *pcxt) pfree(pcxt); } +/* + * Helpers for passing the current query text down to parallel workers. + */ + +/* + * Reserve DSM space for the query text, if any. + * + * Returns the length of the query text (or 0 if there is none), which the + * caller must pass on to StoreParallelQueryText, so that the space allocated + * always matches the space reserved. + */ +int +EstimateParallelQueryText(ParallelContext *pcxt) +{ + int querylen; + + if (!debug_query_string) + return 0; + + querylen = strlen(debug_query_string); + shm_toc_estimate_chunk(&pcxt->estimator, querylen + 1); + shm_toc_estimate_keys(&pcxt->estimator, 1); + + return querylen; +} + +/* + * Copy the query text into DSM under the given key, if any. querylen is the + * value returned by EstimateParallelQueryText. + */ +void +StoreParallelQueryText(ParallelContext *pcxt, uint64 key, int querylen) +{ + char *sharedquery; + + if (!debug_query_string) + return; + + Assert(strlen(debug_query_string) == querylen); + + sharedquery = (char *) shm_toc_allocate(pcxt->toc, querylen + 1); + memcpy(sharedquery, debug_query_string, querylen + 1); + shm_toc_insert(pcxt->toc, key, sharedquery); +} + +/* + * Restore the query text in a worker: set debug_query_string and report it as + * the current activity. The key is looked up with missing_ok, so when the + * leader stored no text this sets debug_query_string to NULL. + */ +void +RestoreParallelQueryText(shm_toc *toc, uint64 key) +{ + debug_query_string = shm_toc_lookup(toc, key, true); + pgstat_report_activity(STATE_RUNNING, debug_query_string); +} + /* * Are there any parallel contexts currently active? */ diff --git a/src/backend/commands/vacuumparallel.c b/src/backend/commands/vacuumparallel.c index c29f62c5a99..e900eaed428 100644 --- a/src/backend/commands/vacuumparallel.c +++ b/src/backend/commands/vacuumparallel.c @@ -383,14 +383,7 @@ parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes, shm_toc_estimate_keys(&pcxt->estimator, 1); /* Finally, estimate PARALLEL_VACUUM_KEY_QUERY_TEXT space */ - if (debug_query_string) - { - querylen = strlen(debug_query_string); - shm_toc_estimate_chunk(&pcxt->estimator, querylen + 1); - shm_toc_estimate_keys(&pcxt->estimator, 1); - } - else - querylen = 0; /* keep compiler quiet */ + querylen = EstimateParallelQueryText(pcxt); InitializeParallelDSM(pcxt); @@ -494,16 +487,7 @@ parallel_vacuum_init(Relation rel, Relation *indrels, int nindexes, pvs->wal_usage = wal_usage; /* Store query string for workers */ - if (debug_query_string) - { - char *sharedquery; - - sharedquery = (char *) shm_toc_allocate(pcxt->toc, querylen + 1); - memcpy(sharedquery, debug_query_string, querylen + 1); - sharedquery[querylen] = '\0'; - shm_toc_insert(pcxt->toc, - PARALLEL_VACUUM_KEY_QUERY_TEXT, sharedquery); - } + StoreParallelQueryText(pcxt, PARALLEL_VACUUM_KEY_QUERY_TEXT, querylen); /* Success -- return parallel vacuum state */ return pvs; @@ -1271,7 +1255,6 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc) BufferUsage *buffer_usage; WalUsage *wal_usage; int nindexes; - char *sharedquery; ErrorContextCallback errcallback; /* @@ -1296,9 +1279,7 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc) shared = (PVShared *) shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_SHARED, false); /* Set debug_query_string for individual workers */ - sharedquery = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_QUERY_TEXT, true); - debug_query_string = sharedquery; - pgstat_report_activity(STATE_RUNNING, debug_query_string); + RestoreParallelQueryText(toc, PARALLEL_VACUUM_KEY_QUERY_TEXT); /* Track query ID */ pgstat_report_query_id(shared->queryid, false); diff --git a/src/include/access/parallel.h b/src/include/access/parallel.h index 60f857675e0..77deab8bb4a 100644 --- a/src/include/access/parallel.h +++ b/src/include/access/parallel.h @@ -72,6 +72,11 @@ extern void WaitForParallelWorkersToFinish(ParallelContext *pcxt); extern void DestroyParallelContext(ParallelContext *pcxt); extern bool ParallelContextActive(void); +extern int EstimateParallelQueryText(ParallelContext *pcxt); +extern void StoreParallelQueryText(ParallelContext *pcxt, uint64 key, + int querylen); +extern void RestoreParallelQueryText(shm_toc *toc, uint64 key); + extern void HandleParallelMessageInterrupt(void); extern void ProcessParallelMessages(void); extern void AtEOXact_Parallel(bool isCommit); -- 2.47.1