From fdb07aaa082580433046e184b2c0d6b470862786 Mon Sep 17 00:00:00 2001 From: Lukas Fittl Date: Sat, 31 Oct 2026 11:32:49 -0700 Subject: [PATCH 1/5] Fix query ID reporting in parallel workers for GIN index builds Commit 6fd5071909a made parallel workers for VACUUM, BRIN and btree index builds report the query ID of the leader, so that they show up with the correct query_id in pg_stat_activity. Parallel GIN index builds were added later by 8492feb98f6, but did not include this, so GIN build workers have always reported a query ID of 0. Fix by storing the leader's query ID in the shared build state and reporting it in the worker, like the other index AMs already do. Author: Lukas Fittl Discussion: https://postgr.es/m/CAP53Pkw3WcVbg61G29AmN=6XkuxGryoWKnhTyC1JmkT08i8qVA@mail.gmail.com --- src/backend/access/gin/gininsert.c | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c index 564b1509f1bf2..cbfaa5075b024 100644 --- a/src/backend/access/gin/gininsert.c +++ b/src/backend/access/gin/gininsert.c @@ -64,6 +64,9 @@ typedef struct GinBuildShared bool isconcurrent; int scantuplesortstates; + /* Query ID, for report in worker processes */ + int64 queryid; + /* * workersdonecv is used to monitor the progress of workers. All parallel * participants must indicate that they are done before leader can use @@ -1031,6 +1034,7 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index, ginshared->indexrelid = RelationGetRelid(index); ginshared->isconcurrent = isconcurrent; ginshared->scantuplesortstates = scantuplesortstates; + ginshared->queryid = pgstat_get_my_query_id(); ConditionVariableInit(&ginshared->workersdonecv); SpinLockInit(&ginshared->mutex); @@ -2145,6 +2149,9 @@ _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc) indexLockmode = RowExclusiveLock; } + /* Track query ID */ + pgstat_report_query_id(ginshared->queryid, false); + /* Open relations within worker */ heapRel = table_open(ginshared->heaprelid, heapLockmode); indexRel = index_open(ginshared->indexrelid, indexLockmode); From d3ccc28b07854f6333621f1413d05b8f7c1a8aed Mon Sep 17 00:00:00 2001 From: Lukas Fittl Date: Sat, 31 Oct 2026 11:53:59 -0700 Subject: [PATCH 2/5] Fix broken HOT chain handling in parallel BRIN and GIN index builds When a non-concurrent index build encounters a RECENTLY_DEAD tuple that was HOT-updated, heapam_index_build_range_scan() skips it, indexing only the live tuple at the end of the HOT chain, and sets ii_BrokenHotChain. index_build() then marks the index with indcheckxmin, so that transactions with a snapshot older than index creation don't use the index, as it may be missing tuple versions still visible to them. In parallel index builds this requires each worker to report back to the leader whether it encountered a broken HOT chain. btree correctly does this, but BRIN and GIN parallel index builds do not. As a result, indcheckxmin was never set, and a transaction holding an older snapshot could use the new index and get wrong results. Fix by tracking the brokenhotchain flag in the shared state of both AMs, like btree does, and copying it back into the ambuild IndexInfo in the leader after all participants are done. Oversight when parallel index builds were added for BRIN (b4375717147) and GIN (8492feb98f6) and backpatch to 17 and 18 respectively. Author: Lukas Fittl Reviewed-by: Discussion: https://postgr.es/m/CAP53Pkw3WcVbg61G29AmN=6XkuxGryoWKnhTyC1JmkT08i8qVA@mail.gmail.com --- src/backend/access/brin/brin.c | 33 ++++++++++++++++++++++-------- src/backend/access/gin/gininsert.c | 32 +++++++++++++++++++++-------- 2 files changed, 47 insertions(+), 18 deletions(-) diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c index 5059da8dce7c5..fedc0387ca0ad 100644 --- a/src/backend/access/brin/brin.c +++ b/src/backend/access/brin/brin.c @@ -99,10 +99,14 @@ typedef struct BrinShared * reltuples is the total number of input heap tuples. * * indtuples is the total number of tuples that made it into the index. + * + * brokenhotchain indicates if any worker detected a broken HOT chain + * during build. */ int nparticipantsdone; double reltuples; double indtuples; + bool brokenhotchain; /* * ParallelTableScanDescData data follows. Can't directly embed here, as @@ -236,8 +240,10 @@ static void _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Rela bool isconcurrent, int request); static void _brin_end_parallel(BrinLeader *brinleader, BrinBuildState *state); static Size _brin_parallel_estimate_shared(Relation heap, Snapshot snapshot); -static double _brin_parallel_heapscan(BrinBuildState *state); -static double _brin_parallel_merge(BrinBuildState *state); +static double _brin_parallel_heapscan(BrinBuildState *state, + bool *brokenhotchain); +static double _brin_parallel_merge(BrinBuildState *state, + IndexInfo *indexInfo); static void _brin_leader_participate_as_worker(BrinBuildState *buildstate, Relation heap, Relation index); static void _brin_parallel_scan_and_build(BrinBuildState *state, @@ -1223,7 +1229,7 @@ brinbuild(Relation heap, Relation index, IndexInfo *indexInfo) TUPLESORT_NONE); /* scan the relation and merge per-worker results */ - reltuples = _brin_parallel_merge(state); + reltuples = _brin_parallel_merge(state, indexInfo); _brin_end_parallel(state->bs_leader, state); } @@ -2483,6 +2489,7 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index, brinshared->nparticipantsdone = 0; brinshared->reltuples = 0.0; brinshared->indtuples = 0.0; + brinshared->brokenhotchain = false; table_parallelscan_initialize(heap, ParallelTableScanFromBrinShared(brinshared), @@ -2589,10 +2596,11 @@ _brin_end_parallel(BrinLeader *brinleader, BrinBuildState *state) * already be underway within worker processes (when leader participates * as a worker, we should end up here just as workers are finishing). * - * Returns the total number of heap tuples scanned. + * Returns the total number of heap tuples scanned. Also sets *brokenhotchain + * to whether any participant detected a broken HOT chain. */ static double -_brin_parallel_heapscan(BrinBuildState *state) +_brin_parallel_heapscan(BrinBuildState *state, bool *brokenhotchain) { BrinShared *brinshared = state->bs_leader->brinshared; int nparticipanttuplesorts; @@ -2606,6 +2614,7 @@ _brin_parallel_heapscan(BrinBuildState *state) /* copy the data into leader state */ state->bs_reltuples = brinshared->reltuples; state->bs_numtuples = brinshared->indtuples; + *brokenhotchain = brinshared->brokenhotchain; SpinLockRelease(&brinshared->mutex); break; @@ -2630,10 +2639,12 @@ _brin_parallel_heapscan(BrinBuildState *state) * summaries for the same page range, and also fill-in empty summaries for * ranges without any tuples. * - * Returns the total number of heap tuples scanned. + * Returns the total number of heap tuples scanned. Also sets + * indexInfo->ii_BrokenHotChain if any participant detected a broken HOT chain, + * so that index.c can mark the index as unsafe for older snapshots. */ static double -_brin_parallel_merge(BrinBuildState *state) +_brin_parallel_merge(BrinBuildState *state, IndexInfo *indexInfo) { BrinTuple *btup; BrinMemTuple *memtuple = NULL; @@ -2644,7 +2655,8 @@ _brin_parallel_merge(BrinBuildState *state) double reltuples; /* wait for workers to scan table and produce partial results */ - reltuples = _brin_parallel_heapscan(state); + reltuples = _brin_parallel_heapscan(state, + &indexInfo->ii_BrokenHotChain); /* do the actual sort in the leader */ tuplesort_performsort(state->bs_sortstate); @@ -2859,12 +2871,15 @@ _brin_parallel_scan_and_build(BrinBuildState *state, state->bs_reltuples += reltuples; /* - * Done. Record ambuild statistics. + * Done. Record ambuild statistics, and whether we encountered a broken + * HOT chain. */ SpinLockAcquire(&brinshared->mutex); brinshared->nparticipantsdone++; brinshared->reltuples += state->bs_reltuples; brinshared->indtuples += state->bs_numtuples; + if (indexInfo->ii_BrokenHotChain) + brinshared->brokenhotchain = true; SpinLockRelease(&brinshared->mutex); /* Notify leader */ diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c index cbfaa5075b024..f42c5bda4e473 100644 --- a/src/backend/access/gin/gininsert.c +++ b/src/backend/access/gin/gininsert.c @@ -92,10 +92,14 @@ typedef struct GinBuildShared * reltuples is the total number of input heap tuples. * * indtuples is the total number of tuples that made it into the index. + * + * brokenhotchain indicates if any worker detected a broken HOT chain + * during build. */ int nparticipantsdone; double reltuples; double indtuples; + bool brokenhotchain; /* * ParallelTableScanDescData data follows. Can't directly embed here, as @@ -191,8 +195,10 @@ static void _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relati bool isconcurrent, int request); static void _gin_end_parallel(GinLeader *ginleader, GinBuildState *state); static Size _gin_parallel_estimate_shared(Relation heap, Snapshot snapshot); -static double _gin_parallel_heapscan(GinBuildState *state); -static double _gin_parallel_merge(GinBuildState *state); +static double _gin_parallel_heapscan(GinBuildState *state, + bool *brokenhotchain); +static double _gin_parallel_merge(GinBuildState *state, + IndexInfo *indexInfo); static void _gin_leader_participate_as_worker(GinBuildState *buildstate, Relation heap, Relation index); static void _gin_parallel_scan_and_build(GinBuildState *state, @@ -750,7 +756,7 @@ ginbuild(Relation heap, Relation index, IndexInfo *indexInfo) TUPLESORT_NONE); /* scan the relation in parallel and merge per-worker results */ - reltuples = _gin_parallel_merge(state); + reltuples = _gin_parallel_merge(state, indexInfo); _gin_end_parallel(state->bs_leader, state); } @@ -1043,6 +1049,7 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index, ginshared->nparticipantsdone = 0; ginshared->reltuples = 0.0; ginshared->indtuples = 0.0; + ginshared->brokenhotchain = false; table_parallelscan_initialize(heap, ParallelTableScanFromGinBuildShared(ginshared), @@ -1145,10 +1152,11 @@ _gin_end_parallel(GinLeader *ginleader, GinBuildState *state) * already be underway within worker processes (when leader participates * as a worker, we should end up here just as workers are finishing). * - * Returns the total number of heap tuples scanned. + * Returns the total number of heap tuples scanned. Also sets *brokenhotchain + * to whether any participant detected a broken HOT chain. */ static double -_gin_parallel_heapscan(GinBuildState *state) +_gin_parallel_heapscan(GinBuildState *state, bool *brokenhotchain) { GinBuildShared *ginshared = state->bs_leader->ginshared; int nparticipanttuplesorts; @@ -1162,6 +1170,7 @@ _gin_parallel_heapscan(GinBuildState *state) /* copy the data into leader state */ state->bs_reltuples = ginshared->reltuples; state->bs_numtuples = ginshared->indtuples; + *brokenhotchain = ginshared->brokenhotchain; SpinLockRelease(&ginshared->mutex); break; @@ -1646,10 +1655,12 @@ GinBufferCanAddKey(GinBuffer *buffer, GinTuple *tup) * summaries for the same page range, and also fill-in empty summaries for * ranges without any tuples. * - * Returns the total number of heap tuples scanned. + * Returns the total number of heap tuples scanned. Also sets + * indexInfo->ii_BrokenHotChain if any participant detected a broken HOT chain, + * so that index.c can mark the index as unsafe for older snapshots. */ static double -_gin_parallel_merge(GinBuildState *state) +_gin_parallel_merge(GinBuildState *state, IndexInfo *indexInfo) { GinTuple *tup; Size tuplen; @@ -1660,7 +1671,7 @@ _gin_parallel_merge(GinBuildState *state) double numtuples = 0; /* wait for workers to scan table and produce partial results */ - reltuples = _gin_parallel_heapscan(state); + reltuples = _gin_parallel_heapscan(state, &indexInfo->ii_BrokenHotChain); /* Execute the sort */ pgstat_progress_update_param(PROGRESS_CREATEIDX_SUBPHASE, @@ -2088,12 +2099,15 @@ _gin_parallel_scan_and_build(GinBuildState *state, state->bs_reltuples += reltuples; /* - * Done. Record ambuild statistics. + * Done. Record ambuild statistics, and whether we encountered a broken + * HOT chain. */ SpinLockAcquire(&ginshared->mutex); ginshared->nparticipantsdone++; ginshared->reltuples += state->bs_reltuples; ginshared->indtuples += state->bs_numtuples; + if (indexInfo->ii_BrokenHotChain) + ginshared->brokenhotchain = true; SpinLockRelease(&ginshared->mutex); /* Notify leader */ From 50a6145214dbe5853a4e716dc47abf15db899eba Mon Sep 17 00:00:00 2001 From: Lukas Fittl Date: Sat, 31 Oct 2026 09:57:49 -0700 Subject: [PATCH 3/5] Unify parallel index build infrastructure The three in-tree index AMs that currently support parallel index builds (nbtree, gin and brin) each share very similar code for initializing parallel workers, allocating shared memory and communicating between the leader and the workers. To reduce duplication, introduce shared helpers for parallel index builds, unifying the existing code. 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 | 206 +++---------- src/backend/access/common/Makefile | 1 + src/backend/access/common/meson.build | 1 + .../access/common/parallel_index_build.c | 291 ++++++++++++++++++ src/backend/access/gin/gininsert.c | 208 ++----------- src/backend/access/nbtree/nbtsort.c | 212 +++---------- src/include/access/parallel_index_build.h | 123 ++++++++ src/tools/pgindent/typedefs.list | 1 + 8 files changed, 520 insertions(+), 523 deletions(-) create mode 100644 src/backend/access/common/parallel_index_build.c create mode 100644 src/include/access/parallel_index_build.h diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c index fedc0387ca0ad..3b88bebf52563 100644 --- a/src/backend/access/brin/brin.c +++ b/src/backend/access/brin/brin.c @@ -19,6 +19,7 @@ #include "access/brin_page.h" #include "access/brin_pageops.h" #include "access/brin_xlog.h" +#include "access/parallel_index_build.h" #include "access/relation.h" #include "access/reloptions.h" #include "access/relscan.h" @@ -33,7 +34,6 @@ #include "pgstat.h" #include "postmaster/autovacuum.h" #include "storage/bufmgr.h" -#include "storage/condition_variable.h" #include "storage/freespace.h" #include "storage/proc.h" #include "tcop/tcopprot.h" @@ -60,53 +60,13 @@ */ typedef struct BrinShared { - /* - * These fields are not modified during the build. They primarily exist - * for the benefit of worker processes that need to create state - * corresponding to that used by the leader. - */ - Oid heaprelid; - Oid indexrelid; - bool isconcurrent; - BlockNumber pagesPerRange; - int scantuplesortstates; - - /* Query ID, for report in worker processes */ - int64 queryid; + /* Common parallel index build state (must be first) */ + ParallelIndexBuildShared base; /* - * workersdonecv is used to monitor the progress of workers. All parallel - * participants must indicate that they are done before leader can use - * results built by the workers (and before leader can write the data into - * the index). + * BRIN-specific immutable state, not modified during the build. */ - ConditionVariable workersdonecv; - - /* - * mutex protects all fields before heapdesc. - * - * These fields contain status information of interest to BRIN index - * builds that must work just the same when an index is built in parallel. - */ - slock_t mutex; - - /* - * Mutable state that is maintained by workers, and reported back to - * leader at end of the scans. - * - * nparticipantsdone is number of worker processes finished. - * - * reltuples is the total number of input heap tuples. - * - * indtuples is the total number of tuples that made it into the index. - * - * brokenhotchain indicates if any worker detected a broken HOT chain - * during build. - */ - int nparticipantsdone; - double reltuples; - double indtuples; - bool brokenhotchain; + BlockNumber pagesPerRange; /* * ParallelTableScanDescData data follows. Can't directly embed here, as @@ -239,7 +199,6 @@ static void brin_fill_empty_ranges(BrinBuildState *state, static void _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index, bool isconcurrent, int request); static void _brin_end_parallel(BrinLeader *brinleader, BrinBuildState *state); -static Size _brin_parallel_estimate_shared(Relation heap, Snapshot snapshot); static double _brin_parallel_heapscan(BrinBuildState *state, bool *brokenhotchain); static double _brin_parallel_merge(BrinBuildState *state, @@ -2406,29 +2365,19 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index, * Enter parallel mode, and create context for parallel build of brin * index */ - EnterParallelMode(); - Assert(request > 0); - pcxt = CreateParallelContext("postgres", "_brin_parallel_build_main", - request); + pcxt = ParallelIndexBuildCreateContext("postgres", + "_brin_parallel_build_main", request); scantuplesortstates = leaderparticipates ? request + 1 : request; - /* - * Prepare for scan of the base relation. In a normal index build, we use - * SnapshotAny because we must retrieve all tuples and do our own time - * qual checks (because we have to index RECENTLY_DEAD tuples). In a - * concurrent build, we take a regular MVCC snapshot and index whatever's - * live according to that. - */ - if (!isconcurrent) - snapshot = SnapshotAny; - else - snapshot = RegisterSnapshot(GetTransactionSnapshot()); + /* Prepare for scan of the base relation. */ + snapshot = ParallelIndexBuildGetSnapshot(isconcurrent); /* * Estimate size for our own PARALLEL_KEY_BRIN_SHARED workspace. */ - estbrinshared = _brin_parallel_estimate_shared(heap, snapshot); + estbrinshared = ParallelIndexBuildEstimateShared(heap, snapshot, + sizeof(BrinShared)); shm_toc_estimate_chunk(&pcxt->estimator, estbrinshared); estsort = tuplesort_estimate_shared(scantuplesortstates); shm_toc_estimate_chunk(&pcxt->estimator, estsort); @@ -2466,34 +2415,19 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index, /* If no DSM segment was available, back out (do serial build) */ if (pcxt->seg == NULL) { - if (IsMVCCSnapshot(snapshot)) - UnregisterSnapshot(snapshot); - DestroyParallelContext(pcxt); - ExitParallelMode(); + ParallelIndexBuildAbort(pcxt, snapshot); return; } /* Store shared build state, for which we reserved space */ brinshared = (BrinShared *) shm_toc_allocate(pcxt->toc, estbrinshared); - /* Initialize immutable state */ - brinshared->heaprelid = RelationGetRelid(heap); - brinshared->indexrelid = RelationGetRelid(index); - brinshared->isconcurrent = isconcurrent; - brinshared->scantuplesortstates = scantuplesortstates; + /* Initialize BRIN-specific immutable state */ brinshared->pagesPerRange = buildstate->bs_pagesPerRange; - brinshared->queryid = pgstat_get_my_query_id(); - ConditionVariableInit(&brinshared->workersdonecv); - SpinLockInit(&brinshared->mutex); - - /* Initialize mutable state */ - brinshared->nparticipantsdone = 0; - brinshared->reltuples = 0.0; - brinshared->indtuples = 0.0; - brinshared->brokenhotchain = false; - - table_parallelscan_initialize(heap, - ParallelTableScanFromBrinShared(brinshared), - snapshot); + /* Initialize common state, and the parallel scan that follows the struct */ + ParallelIndexBuildInitShared(&brinshared->base, heap, index, isconcurrent, + scantuplesortstates, + ParallelTableScanFromBrinShared(brinshared), + snapshot); /* * Store shared tuplesort-private state, for which we reserved space. @@ -2570,23 +2504,8 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index, static void _brin_end_parallel(BrinLeader *brinleader, BrinBuildState *state) { - int i; - - /* Shutdown worker processes */ - WaitForParallelWorkersToFinish(brinleader->pcxt); - - /* - * Next, accumulate WAL usage. (This must wait for the workers to finish, - * or we might get incomplete data.) - */ - for (i = 0; i < brinleader->pcxt->nworkers_launched; i++) - InstrAccumParallelQuery(&brinleader->bufferusage[i], &brinleader->walusage[i]); - - /* Free last reference to MVCC snapshot, if one was used */ - if (IsMVCCSnapshot(brinleader->snapshot)) - UnregisterSnapshot(brinleader->snapshot); - DestroyParallelContext(brinleader->pcxt); - ExitParallelMode(); + ParallelIndexBuildEnd(brinleader->pcxt, brinleader->bufferusage, + brinleader->walusage, brinleader->snapshot); } /* @@ -2603,29 +2522,12 @@ static double _brin_parallel_heapscan(BrinBuildState *state, bool *brokenhotchain) { BrinShared *brinshared = state->bs_leader->brinshared; - int nparticipanttuplesorts; - nparticipanttuplesorts = state->bs_leader->nparticipanttuplesorts; - for (;;) - { - SpinLockAcquire(&brinshared->mutex); - if (brinshared->nparticipantsdone == nparticipanttuplesorts) - { - /* copy the data into leader state */ - state->bs_reltuples = brinshared->reltuples; - state->bs_numtuples = brinshared->indtuples; - *brokenhotchain = brinshared->brokenhotchain; - - SpinLockRelease(&brinshared->mutex); - break; - } - SpinLockRelease(&brinshared->mutex); - - ConditionVariableSleep(&brinshared->workersdonecv, - WAIT_EVENT_PARALLEL_CREATE_INDEX_SCAN); - } - - ConditionVariableCancelSleep(); + /* Wait for all participants to finish, and collect their results */ + ParallelIndexBuildWaitForWorkers(&brinshared->base, + state->bs_leader->nparticipanttuplesorts, + &state->bs_reltuples, &state->bs_numtuples, + NULL, brokenhotchain); return state->bs_reltuples; } @@ -2787,18 +2689,6 @@ _brin_parallel_merge(BrinBuildState *state, IndexInfo *indexInfo) return reltuples; } -/* - * Returns size of shared memory required to store state for a parallel - * brin index build based on the snapshot its parallel scan will use. - */ -static Size -_brin_parallel_estimate_shared(Relation heap, Snapshot snapshot) -{ - /* c.f. shm_toc_allocate as to why BUFFERALIGN is used */ - return add_size(BUFFERALIGN(sizeof(BrinShared)), - table_parallelscan_estimate(heap, snapshot)); -} - /* * Within leader, participate as a parallel worker. */ @@ -2853,7 +2743,7 @@ _brin_parallel_scan_and_build(BrinBuildState *state, /* Join parallel scan */ indexInfo = BuildIndexInfo(index); - indexInfo->ii_Concurrent = brinshared->isconcurrent; + indexInfo->ii_Concurrent = brinshared->base.isconcurrent; scan = table_beginscan_parallel(heap, ParallelTableScanFromBrinShared(brinshared), @@ -2871,19 +2761,12 @@ _brin_parallel_scan_and_build(BrinBuildState *state, state->bs_reltuples += reltuples; /* - * Done. Record ambuild statistics, and whether we encountered a broken - * HOT chain. + * Done. Record ambuild statistics (including whether we encountered a + * broken HOT chain), and notify the leader. */ - SpinLockAcquire(&brinshared->mutex); - brinshared->nparticipantsdone++; - brinshared->reltuples += state->bs_reltuples; - brinshared->indtuples += state->bs_numtuples; - if (indexInfo->ii_BrokenHotChain) - brinshared->brokenhotchain = true; - SpinLockRelease(&brinshared->mutex); - - /* Notify leader */ - ConditionVariableSignal(&brinshared->workersdonecv); + ParallelIndexBuildReportScanDone(&brinshared->base, state->bs_reltuples, + state->bs_numtuples, false, + indexInfo->ii_BrokenHotChain); tuplesort_end(state->bs_sortstate); } @@ -2900,8 +2783,6 @@ _brin_parallel_build_main(dsm_segment *seg, shm_toc *toc) BrinBuildState *buildstate; Relation heapRel; Relation indexRel; - LOCKMODE heapLockmode; - LOCKMODE indexLockmode; WalUsage *walusage; BufferUsage *bufferusage; int sortmem; @@ -2923,24 +2804,8 @@ _brin_parallel_build_main(dsm_segment *seg, shm_toc *toc) /* Look up brin shared state */ brinshared = shm_toc_lookup(toc, PARALLEL_KEY_BRIN_SHARED, false); - /* Open relations using lock modes known to be obtained by index.c */ - if (!brinshared->isconcurrent) - { - heapLockmode = ShareLock; - indexLockmode = AccessExclusiveLock; - } - else - { - heapLockmode = ShareUpdateExclusiveLock; - indexLockmode = RowExclusiveLock; - } - - /* Track query ID */ - pgstat_report_query_id(brinshared->queryid, false); - - /* Open relations within worker */ - heapRel = table_open(brinshared->heaprelid, heapLockmode); - indexRel = index_open(brinshared->indexrelid, indexLockmode); + /* Open relations within worker, using the leader's lock modes */ + ParallelIndexBuildOpenRelations(&brinshared->base, &heapRel, &indexRel); buildstate = initialize_brin_buildstate(indexRel, NULL, brinshared->pagesPerRange, @@ -2958,7 +2823,7 @@ _brin_parallel_build_main(dsm_segment *seg, shm_toc *toc) * (when requested number of workers were not launched, this will be * somewhat higher than it is for other workers). */ - sortmem = maintenance_work_mem / brinshared->scantuplesortstates; + sortmem = maintenance_work_mem / brinshared->base.scantuplesortstates; _brin_parallel_scan_and_build(buildstate, brinshared, sharedsort, heapRel, indexRel, sortmem, false); @@ -2969,8 +2834,7 @@ _brin_parallel_build_main(dsm_segment *seg, shm_toc *toc) InstrEndParallelQuery(&bufferusage[ParallelWorkerNumber], &walusage[ParallelWorkerNumber]); - index_close(indexRel, indexLockmode); - table_close(heapRel, heapLockmode); + ParallelIndexBuildCloseRelations(&brinshared->base, heapRel, indexRel); } /* diff --git a/src/backend/access/common/Makefile b/src/backend/access/common/Makefile index e78de312659ed..0d4769b96af57 100644 --- a/src/backend/access/common/Makefile +++ b/src/backend/access/common/Makefile @@ -18,6 +18,7 @@ OBJS = \ detoast.o \ heaptuple.o \ indextuple.o \ + parallel_index_build.o \ printsimple.o \ printtup.o \ relation.o \ diff --git a/src/backend/access/common/meson.build b/src/backend/access/common/meson.build index 35e89b5ea67d5..7d6247b97542d 100644 --- a/src/backend/access/common/meson.build +++ b/src/backend/access/common/meson.build @@ -6,6 +6,7 @@ backend_sources += files( 'detoast.c', 'heaptuple.c', 'indextuple.c', + 'parallel_index_build.c', 'printsimple.c', 'printtup.c', 'relation.c', diff --git a/src/backend/access/common/parallel_index_build.c b/src/backend/access/common/parallel_index_build.c new file mode 100644 index 0000000000000..36df09d6bc5e4 --- /dev/null +++ b/src/backend/access/common/parallel_index_build.c @@ -0,0 +1,291 @@ +/*------------------------------------------------------------------------- + * + * parallel_index_build.c + * Shared infrastructure for parallel index builds. + * + * This file contains the access-method-independent parts of the parallel + * index build lifecycle shared by different index access methods: setting up + * the parallel context and shared state, and opening/closing relations and + * tearing everything down again in the worker and leader. + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * IDENTIFICATION + * src/backend/access/common/parallel_index_build.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include "access/genam.h" +#include "access/parallel_index_build.h" +#include "access/table.h" +#include "access/tableam.h" +#include "access/xact.h" +#include "executor/instrument.h" +#include "pgstat.h" +#include "utils/rel.h" +#include "utils/snapmgr.h" +#include "utils/wait_event.h" + +/* + * Enter parallel mode and create a parallel context for an index build, using + * the named worker entry point in the named library (as for + * CreateParallelContext; in-core access methods pass "postgres"). + */ +ParallelContext * +ParallelIndexBuildCreateContext(const char *library_name, + const char *worker_function, int nworkers) +{ + Assert(nworkers > 0); + + EnterParallelMode(); + + return CreateParallelContext(library_name, worker_function, nworkers); +} + +/* + * Choose the snapshot for the heap scan of an index build. + * + * In a normal index build, we use SnapshotAny because we must retrieve all + * tuples and do our own time qual checks (because we have to index + * RECENTLY_DEAD tuples). In a concurrent build, we take a regular MVCC + * snapshot and index whatever's live according to that. The caller is + * responsible for unregistering an MVCC snapshot by calling + * ParallelIndexBuildEnd. + */ +Snapshot +ParallelIndexBuildGetSnapshot(bool isconcurrent) +{ + if (!isconcurrent) + return SnapshotAny; + + return RegisterSnapshot(GetTransactionSnapshot()); +} + +/* + * Estimate the DSM space for the shared state struct of a parallel index + * build, including the parallel table scan descriptor that trails it. + * + * am_shared_size is sizeof() the access method's whole shared struct (which + * embeds ParallelIndexBuildShared as its first member). + */ +Size +ParallelIndexBuildEstimateShared(Relation heap, Snapshot snapshot, + Size am_shared_size) +{ + /* c.f. shm_toc_allocate as to why BUFFERALIGN is used */ + return add_size(BUFFERALIGN(am_shared_size), + table_parallelscan_estimate(heap, snapshot)); +} + +/* + * Initialize the common shared state for a parallel index build. + * + * The caller has already allocated the embedding shared struct in DSM; this + * fills in the common header and initializes the parallel heap scan + * descriptor that follows the (whole) embedding struct, which the caller + * passes as pscan. AM-specific fields are the caller's responsibility. + */ +void +ParallelIndexBuildInitShared(ParallelIndexBuildShared *shared, + Relation heap, Relation index, + bool isconcurrent, int scantuplesortstates, + ParallelTableScanDesc pscan, Snapshot snapshot) +{ + /* Initialize immutable state */ + shared->heaprelid = RelationGetRelid(heap); + shared->indexrelid = RelationGetRelid(index); + shared->isconcurrent = isconcurrent; + shared->scantuplesortstates = scantuplesortstates; + shared->queryid = pgstat_get_my_query_id(); + ConditionVariableInit(&shared->workersdonecv); + SpinLockInit(&shared->mutex); + + /* Initialize mutable state */ + shared->nparticipantsdone = 0; + shared->reltuples = 0.0; + shared->indtuples = 0.0; + shared->brokenhotchain = false; + shared->havedead = false; + + table_parallelscan_initialize(heap, pscan, snapshot); +} + +/* + * Wait, in the leader, for all participants to finish their portion of the + * scan, then read back the accumulated per-build results. + * + * reltuples and indtuples receive the totals. brokenhotchain receives whether + * any participant saw a broken HOT chain; the caller must propagate it to the + * IndexInfo passed to ambuild. havedead is optional (pass NULL when the AM + * does not track it): it reports whether any worker saw RECENTLY_DEAD tuples. + */ +void +ParallelIndexBuildWaitForWorkers(ParallelIndexBuildShared *shared, + int nparticipants, + double *reltuples, double *indtuples, + bool *havedead, bool *brokenhotchain) +{ + for (;;) + { + SpinLockAcquire(&shared->mutex); + if (shared->nparticipantsdone == nparticipants) + { + *reltuples = shared->reltuples; + *indtuples = shared->indtuples; + if (havedead) + *havedead = shared->havedead; + if (brokenhotchain) + *brokenhotchain = shared->brokenhotchain; + SpinLockRelease(&shared->mutex); + break; + } + SpinLockRelease(&shared->mutex); + + ConditionVariableSleep(&shared->workersdonecv, + WAIT_EVENT_PARALLEL_CREATE_INDEX_SCAN); + } + + ConditionVariableCancelSleep(); +} + +/* + * Back out of a parallel index build before any workers were launched, for + * example because InitializeParallelDSM could not create a DSM segment. The + * caller then proceeds with a serial build. + * + * Releases the MVCC snapshot if one was used, destroys the parallel context, + * and exits parallel mode. Once workers have been launched, use + * ParallelIndexBuildEnd instead. + */ +void +ParallelIndexBuildAbort(ParallelContext *pcxt, Snapshot snapshot) +{ + Assert(pcxt->nworkers_launched == 0); + + if (IsMVCCSnapshot(snapshot)) + UnregisterSnapshot(snapshot); + DestroyParallelContext(pcxt); + ExitParallelMode(); +} + +/* + * Shut down the workers and tear down a parallel index build. + * + * Waits for all workers to finish, accumulates their buffer/WAL usage into the + * leader's stats, releases the MVCC snapshot if one was used, and exits + * parallel mode. bufferusage and walusage are the per-worker arrays the + * leader allocated in DSM, which each worker filled in via + * InstrEndParallelQuery. + */ +void +ParallelIndexBuildEnd(ParallelContext *pcxt, BufferUsage *bufferusage, + WalUsage *walusage, Snapshot snapshot) +{ + /* Shutdown worker processes */ + WaitForParallelWorkersToFinish(pcxt); + + /* + * Next, accumulate WAL and buffer usage. (This must wait for the workers + * to finish, or we might get incomplete data.) + */ + for (int i = 0; i < pcxt->nworkers_launched; i++) + InstrAccumParallelQuery(&bufferusage[i], &walusage[i]); + + /* Free last reference to MVCC snapshot, if one was used */ + if (IsMVCCSnapshot(snapshot)) + UnregisterSnapshot(snapshot); + DestroyParallelContext(pcxt); + ExitParallelMode(); +} + +/* + * Select the lock modes used by a parallel index build worker. These match + * the lock modes known to be obtained by index.c in the leader, which differ + * for concurrent builds. + */ +static void +ParallelIndexBuildLockModes(ParallelIndexBuildShared *shared, + LOCKMODE *heapLockmode, LOCKMODE *indexLockmode) +{ + if (!shared->isconcurrent) + { + *heapLockmode = ShareLock; + *indexLockmode = AccessExclusiveLock; + } + else + { + *heapLockmode = ShareUpdateExclusiveLock; + *indexLockmode = RowExclusiveLock; + } +} + +/* + * Open the heap and index relations in a parallel index build worker. + * + * Reports the query ID and opens both relations with the lock modes used by + * the leader. Must be paired with ParallelIndexBuildCloseRelations. + */ +void +ParallelIndexBuildOpenRelations(ParallelIndexBuildShared *shared, + Relation *heapRel, Relation *indexRel) +{ + LOCKMODE heapLockmode; + LOCKMODE indexLockmode; + + ParallelIndexBuildLockModes(shared, &heapLockmode, &indexLockmode); + + /* Track query ID */ + pgstat_report_query_id(shared->queryid, false); + + /* Open relations within worker */ + *heapRel = table_open(shared->heaprelid, heapLockmode); + *indexRel = index_open(shared->indexrelid, indexLockmode); +} + +/* + * Close the relations opened by ParallelIndexBuildOpenRelations. + */ +void +ParallelIndexBuildCloseRelations(ParallelIndexBuildShared *shared, + Relation heapRel, Relation indexRel) +{ + LOCKMODE heapLockmode; + LOCKMODE indexLockmode; + + ParallelIndexBuildLockModes(shared, &heapLockmode, &indexLockmode); + + index_close(indexRel, indexLockmode); + table_close(heapRel, heapLockmode); +} + +/* + * Report that a participant (a worker, or the leader when it participates as + * a worker) has finished its portion of the scan. + * + * Accumulates the participant's tuple counts into the shared totals, ORs in + * the havedead/brokenhotchain flags, and signals the leader. brokenhotchain + * must be the ii_BrokenHotChain of the IndexInfo used for the participant's + * heap scan; havedead is only tracked by nbtree (others pass false). Must be + * paired with ParallelIndexBuildWaitForWorkers in the leader. + */ +void +ParallelIndexBuildReportScanDone(ParallelIndexBuildShared *shared, + double reltuples, double indtuples, + bool havedead, bool brokenhotchain) +{ + SpinLockAcquire(&shared->mutex); + shared->nparticipantsdone++; + shared->reltuples += reltuples; + shared->indtuples += indtuples; + if (havedead) + shared->havedead = true; + if (brokenhotchain) + shared->brokenhotchain = true; + SpinLockRelease(&shared->mutex); + + /* Notify leader */ + ConditionVariableSignal(&shared->workersdonecv); +} diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c index f42c5bda4e473..7ecc4d3f98d5b 100644 --- a/src/backend/access/gin/gininsert.c +++ b/src/backend/access/gin/gininsert.c @@ -17,6 +17,7 @@ #include "access/gin_private.h" #include "access/gin_tuple.h" #include "access/parallel.h" +#include "access/parallel_index_build.h" #include "access/table.h" #include "access/tableam.h" #include "access/xloginsert.h" @@ -28,7 +29,6 @@ #include "nodes/execnodes.h" #include "pgstat.h" #include "storage/bufmgr.h" -#include "storage/condition_variable.h" #include "storage/proc.h" #include "storage/predicate.h" #include "tcop/tcopprot.h" @@ -54,52 +54,8 @@ */ typedef struct GinBuildShared { - /* - * These fields are not modified during the build. They primarily exist - * for the benefit of worker processes that need to create state - * corresponding to that used by the leader. - */ - Oid heaprelid; - Oid indexrelid; - bool isconcurrent; - int scantuplesortstates; - - /* Query ID, for report in worker processes */ - int64 queryid; - - /* - * workersdonecv is used to monitor the progress of workers. All parallel - * participants must indicate that they are done before leader can use - * results built by the workers (and before leader can write the data into - * the index). - */ - ConditionVariable workersdonecv; - - /* - * mutex protects all following fields - * - * These fields contain status information of interest to GIN index builds - * that must work just the same when an index is built in parallel. - */ - slock_t mutex; - - /* - * Mutable state that is maintained by workers, and reported back to - * leader at end of the scans. - * - * nparticipantsdone is number of worker processes finished. - * - * reltuples is the total number of input heap tuples. - * - * indtuples is the total number of tuples that made it into the index. - * - * brokenhotchain indicates if any worker detected a broken HOT chain - * during build. - */ - int nparticipantsdone; - double reltuples; - double indtuples; - bool brokenhotchain; + /* Common parallel index build state (must be first) */ + ParallelIndexBuildShared base; /* * ParallelTableScanDescData data follows. Can't directly embed here, as @@ -194,7 +150,6 @@ typedef struct static void _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index, bool isconcurrent, int request); static void _gin_end_parallel(GinLeader *ginleader, GinBuildState *state); -static Size _gin_parallel_estimate_shared(Relation heap, Snapshot snapshot); static double _gin_parallel_heapscan(GinBuildState *state, bool *brokenhotchain); static double _gin_parallel_merge(GinBuildState *state, @@ -966,29 +921,19 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index, /* * Enter parallel mode, and create context for parallel build of gin index */ - EnterParallelMode(); - Assert(request > 0); - pcxt = CreateParallelContext("postgres", "_gin_parallel_build_main", - request); + pcxt = ParallelIndexBuildCreateContext("postgres", + "_gin_parallel_build_main", request); scantuplesortstates = leaderparticipates ? request + 1 : request; - /* - * Prepare for scan of the base relation. In a normal index build, we use - * SnapshotAny because we must retrieve all tuples and do our own time - * qual checks (because we have to index RECENTLY_DEAD tuples). In a - * concurrent build, we take a regular MVCC snapshot and index whatever's - * live according to that. - */ - if (!isconcurrent) - snapshot = SnapshotAny; - else - snapshot = RegisterSnapshot(GetTransactionSnapshot()); + /* Prepare for scan of the base relation. */ + snapshot = ParallelIndexBuildGetSnapshot(isconcurrent); /* * Estimate size for our own PARALLEL_KEY_GIN_SHARED workspace. */ - estginshared = _gin_parallel_estimate_shared(heap, snapshot); + estginshared = ParallelIndexBuildEstimateShared(heap, snapshot, + sizeof(GinBuildShared)); shm_toc_estimate_chunk(&pcxt->estimator, estginshared); estsort = tuplesort_estimate_shared(scantuplesortstates); shm_toc_estimate_chunk(&pcxt->estimator, estsort); @@ -1026,34 +971,17 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index, /* If no DSM segment was available, back out (do serial build) */ if (pcxt->seg == NULL) { - if (IsMVCCSnapshot(snapshot)) - UnregisterSnapshot(snapshot); - DestroyParallelContext(pcxt); - ExitParallelMode(); + ParallelIndexBuildAbort(pcxt, snapshot); return; } /* Store shared build state, for which we reserved space */ ginshared = (GinBuildShared *) shm_toc_allocate(pcxt->toc, estginshared); - /* Initialize immutable state */ - ginshared->heaprelid = RelationGetRelid(heap); - ginshared->indexrelid = RelationGetRelid(index); - ginshared->isconcurrent = isconcurrent; - ginshared->scantuplesortstates = scantuplesortstates; - ginshared->queryid = pgstat_get_my_query_id(); - - ConditionVariableInit(&ginshared->workersdonecv); - SpinLockInit(&ginshared->mutex); - - /* Initialize mutable state */ - ginshared->nparticipantsdone = 0; - ginshared->reltuples = 0.0; - ginshared->indtuples = 0.0; - ginshared->brokenhotchain = false; - - table_parallelscan_initialize(heap, - ParallelTableScanFromGinBuildShared(ginshared), - snapshot); + /* Initialize common state, and the parallel scan that follows the struct */ + ParallelIndexBuildInitShared(&ginshared->base, heap, index, isconcurrent, + scantuplesortstates, + ParallelTableScanFromGinBuildShared(ginshared), + snapshot); /* * Store shared tuplesort-private state, for which we reserved space. @@ -1126,23 +1054,8 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index, static void _gin_end_parallel(GinLeader *ginleader, GinBuildState *state) { - int i; - - /* Shutdown worker processes */ - WaitForParallelWorkersToFinish(ginleader->pcxt); - - /* - * Next, accumulate WAL usage. (This must wait for the workers to finish, - * or we might get incomplete data.) - */ - for (i = 0; i < ginleader->pcxt->nworkers_launched; i++) - InstrAccumParallelQuery(&ginleader->bufferusage[i], &ginleader->walusage[i]); - - /* Free last reference to MVCC snapshot, if one was used */ - if (IsMVCCSnapshot(ginleader->snapshot)) - UnregisterSnapshot(ginleader->snapshot); - DestroyParallelContext(ginleader->pcxt); - ExitParallelMode(); + ParallelIndexBuildEnd(ginleader->pcxt, ginleader->bufferusage, + ginleader->walusage, ginleader->snapshot); } /* @@ -1159,29 +1072,12 @@ static double _gin_parallel_heapscan(GinBuildState *state, bool *brokenhotchain) { GinBuildShared *ginshared = state->bs_leader->ginshared; - int nparticipanttuplesorts; - nparticipanttuplesorts = state->bs_leader->nparticipanttuplesorts; - for (;;) - { - SpinLockAcquire(&ginshared->mutex); - if (ginshared->nparticipantsdone == nparticipanttuplesorts) - { - /* copy the data into leader state */ - state->bs_reltuples = ginshared->reltuples; - state->bs_numtuples = ginshared->indtuples; - *brokenhotchain = ginshared->brokenhotchain; - - SpinLockRelease(&ginshared->mutex); - break; - } - SpinLockRelease(&ginshared->mutex); - - ConditionVariableSleep(&ginshared->workersdonecv, - WAIT_EVENT_PARALLEL_CREATE_INDEX_SCAN); - } - - ConditionVariableCancelSleep(); + /* Wait for all participants to finish, and collect their results */ + ParallelIndexBuildWaitForWorkers(&ginshared->base, + state->bs_leader->nparticipanttuplesorts, + &state->bs_reltuples, &state->bs_numtuples, + NULL, brokenhotchain); return state->bs_reltuples; } @@ -1813,18 +1709,6 @@ _gin_parallel_merge(GinBuildState *state, IndexInfo *indexInfo) return reltuples; } -/* - * Returns size of shared memory required to store state for a parallel - * gin index build based on the snapshot its parallel scan will use. - */ -static Size -_gin_parallel_estimate_shared(Relation heap, Snapshot snapshot) -{ - /* c.f. shm_toc_allocate as to why BUFFERALIGN is used */ - return add_size(BUFFERALIGN(sizeof(GinBuildShared)), - table_parallelscan_estimate(heap, snapshot)); -} - /* * Within leader, participate as a parallel worker. */ @@ -2058,7 +1942,7 @@ _gin_parallel_scan_and_build(GinBuildState *state, state->work_mem = (sortmem / 2); /* remember how many workers participate in the build */ - state->bs_num_workers = ginshared->scantuplesortstates; + state->bs_num_workers = ginshared->base.scantuplesortstates; /* Begin "partial" tuplesort */ state->bs_sortstate = tuplesort_begin_index_gin(heap, index, @@ -2074,7 +1958,7 @@ _gin_parallel_scan_and_build(GinBuildState *state, /* Join parallel scan */ indexInfo = BuildIndexInfo(index); - indexInfo->ii_Concurrent = ginshared->isconcurrent; + indexInfo->ii_Concurrent = ginshared->base.isconcurrent; scan = table_beginscan_parallel(heap, ParallelTableScanFromGinBuildShared(ginshared), @@ -2099,19 +1983,12 @@ _gin_parallel_scan_and_build(GinBuildState *state, state->bs_reltuples += reltuples; /* - * Done. Record ambuild statistics, and whether we encountered a broken - * HOT chain. + * Done. Record ambuild statistics (including whether we encountered a + * broken HOT chain), and notify the leader. */ - SpinLockAcquire(&ginshared->mutex); - ginshared->nparticipantsdone++; - ginshared->reltuples += state->bs_reltuples; - ginshared->indtuples += state->bs_numtuples; - if (indexInfo->ii_BrokenHotChain) - ginshared->brokenhotchain = true; - SpinLockRelease(&ginshared->mutex); - - /* Notify leader */ - ConditionVariableSignal(&ginshared->workersdonecv); + ParallelIndexBuildReportScanDone(&ginshared->base, state->bs_reltuples, + state->bs_numtuples, false, + indexInfo->ii_BrokenHotChain); tuplesort_end(state->bs_sortstate); } @@ -2128,8 +2005,6 @@ _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc) GinBuildState buildstate; Relation heapRel; Relation indexRel; - LOCKMODE heapLockmode; - LOCKMODE indexLockmode; WalUsage *walusage; BufferUsage *bufferusage; int sortmem; @@ -2151,24 +2026,8 @@ _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc) /* Look up gin shared state */ ginshared = shm_toc_lookup(toc, PARALLEL_KEY_GIN_SHARED, false); - /* Open relations using lock modes known to be obtained by index.c */ - if (!ginshared->isconcurrent) - { - heapLockmode = ShareLock; - indexLockmode = AccessExclusiveLock; - } - else - { - heapLockmode = ShareUpdateExclusiveLock; - indexLockmode = RowExclusiveLock; - } - - /* Track query ID */ - pgstat_report_query_id(ginshared->queryid, false); - - /* Open relations within worker */ - heapRel = table_open(ginshared->heaprelid, heapLockmode); - indexRel = index_open(ginshared->indexrelid, indexLockmode); + /* Open relations within worker, using the leader's lock modes */ + ParallelIndexBuildOpenRelations(&ginshared->base, &heapRel, &indexRel); /* initialize the GIN build state */ initGinState(&buildstate.ginstate, indexRel); @@ -2213,7 +2072,7 @@ _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc) * (when requested number of workers were not launched, this will be * somewhat higher than it is for other workers). */ - sortmem = maintenance_work_mem / ginshared->scantuplesortstates; + sortmem = maintenance_work_mem / ginshared->base.scantuplesortstates; _gin_parallel_scan_and_build(&buildstate, ginshared, sharedsort, heapRel, indexRel, sortmem, false); @@ -2224,8 +2083,7 @@ _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc) InstrEndParallelQuery(&bufferusage[ParallelWorkerNumber], &walusage[ParallelWorkerNumber]); - index_close(indexRel, indexLockmode); - table_close(heapRel, heapLockmode); + ParallelIndexBuildCloseRelations(&ginshared->base, heapRel, indexRel); } /* diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c index 756dfa3dcf47e..f049010150ccf 100644 --- a/src/backend/access/nbtree/nbtsort.c +++ b/src/backend/access/nbtree/nbtsort.c @@ -42,6 +42,7 @@ #include "access/nbtree.h" #include "access/parallel.h" +#include "access/parallel_index_build.h" #include "access/relscan.h" #include "access/table.h" #include "access/tableam.h" @@ -52,7 +53,6 @@ #include "miscadmin.h" #include "pgstat.h" #include "storage/bulk_write.h" -#include "storage/condition_variable.h" #include "storage/proc.h" #include "tcop/tcopprot.h" #include "utils/rel.h" @@ -96,58 +96,12 @@ typedef struct BTSpool */ typedef struct BTShared { - /* - * These fields are not modified during the sort. They primarily exist - * for the benefit of worker processes that need to create BTSpool state - * corresponding to that used by the leader. - */ - Oid heaprelid; - Oid indexrelid; + /* Common parallel index build state (must be first) */ + ParallelIndexBuildShared base; + + /* nbtree-specific immutable state, not modified during the sort */ bool isunique; bool nulls_not_distinct; - bool isconcurrent; - int scantuplesortstates; - - /* Query ID, for report in worker processes */ - int64 queryid; - - /* - * workersdonecv is used to monitor the progress of workers. All parallel - * participants must indicate that they are done before leader can use - * mutable state that workers maintain during scan (and before leader can - * proceed to tuplesort_performsort()). - */ - ConditionVariable workersdonecv; - - /* - * mutex protects all fields before heapdesc. - * - * These fields contain status information of interest to B-Tree index - * builds that must work just the same when an index is built in parallel. - */ - slock_t mutex; - - /* - * Mutable state that is maintained by workers, and reported back to - * leader at end of parallel scan. - * - * nparticipantsdone is number of worker processes finished. - * - * reltuples is the total number of input heap tuples. - * - * havedead indicates if RECENTLY_DEAD tuples were encountered during - * build. - * - * indtuples is the total number of tuples that made it into the index. - * - * brokenhotchain indicates if any worker detected a broken HOT chain - * during build. - */ - int nparticipantsdone; - double reltuples; - bool havedead; - double indtuples; - bool brokenhotchain; /* * ParallelTableScanDescData data follows. Can't directly embed here, as @@ -282,7 +236,6 @@ static void _bt_load(BTWriteState *wstate, static void _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request); static void _bt_end_parallel(BTLeader *btleader); -static Size _bt_parallel_estimate_shared(Relation heap, Snapshot snapshot); static double _bt_parallel_heapscan(BTBuildState *buildstate, bool *brokenhotchain); static void _bt_leader_participate_as_worker(BTBuildState *buildstate); @@ -1421,30 +1374,20 @@ _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request) * Enter parallel mode, and create context for parallel build of btree * index */ - EnterParallelMode(); - Assert(request > 0); - pcxt = CreateParallelContext("postgres", "_bt_parallel_build_main", - request); + pcxt = ParallelIndexBuildCreateContext("postgres", + "_bt_parallel_build_main", request); scantuplesortstates = leaderparticipates ? request + 1 : request; - /* - * Prepare for scan of the base relation. In a normal index build, we use - * SnapshotAny because we must retrieve all tuples and do our own time - * qual checks (because we have to index RECENTLY_DEAD tuples). In a - * concurrent build, we take a regular MVCC snapshot and index whatever's - * live according to that. - */ - if (!isconcurrent) - snapshot = SnapshotAny; - else - snapshot = RegisterSnapshot(GetTransactionSnapshot()); + /* Prepare for scan of the base relation. */ + snapshot = ParallelIndexBuildGetSnapshot(isconcurrent); /* * Estimate size for our own PARALLEL_KEY_BTREE_SHARED workspace, and * PARALLEL_KEY_TUPLESORT tuplesort workspace */ - estbtshared = _bt_parallel_estimate_shared(btspool->heap, snapshot); + estbtshared = ParallelIndexBuildEstimateShared(btspool->heap, snapshot, + sizeof(BTShared)); shm_toc_estimate_chunk(&pcxt->estimator, estbtshared); estsort = tuplesort_estimate_shared(scantuplesortstates); shm_toc_estimate_chunk(&pcxt->estimator, estsort); @@ -1492,34 +1435,20 @@ _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request) /* If no DSM segment was available, back out (do serial build) */ if (pcxt->seg == NULL) { - if (IsMVCCSnapshot(snapshot)) - UnregisterSnapshot(snapshot); - DestroyParallelContext(pcxt); - ExitParallelMode(); + ParallelIndexBuildAbort(pcxt, snapshot); return; } /* Store shared build state, for which we reserved space */ btshared = (BTShared *) shm_toc_allocate(pcxt->toc, estbtshared); - /* Initialize immutable state */ - btshared->heaprelid = RelationGetRelid(btspool->heap); - btshared->indexrelid = RelationGetRelid(btspool->index); + /* Initialize nbtree-specific immutable state */ btshared->isunique = btspool->isunique; btshared->nulls_not_distinct = btspool->nulls_not_distinct; - btshared->isconcurrent = isconcurrent; - btshared->scantuplesortstates = scantuplesortstates; - btshared->queryid = pgstat_get_my_query_id(); - ConditionVariableInit(&btshared->workersdonecv); - SpinLockInit(&btshared->mutex); - /* Initialize mutable state */ - btshared->nparticipantsdone = 0; - btshared->reltuples = 0.0; - btshared->havedead = false; - btshared->indtuples = 0.0; - btshared->brokenhotchain = false; - table_parallelscan_initialize(btspool->heap, - ParallelTableScanFromBTShared(btshared), - snapshot); + /* Initialize common state, and the parallel scan that follows the struct */ + ParallelIndexBuildInitShared(&btshared->base, btspool->heap, btspool->index, + isconcurrent, scantuplesortstates, + ParallelTableScanFromBTShared(btshared), + snapshot); /* * Store shared tuplesort-private state, for which we reserved space. @@ -1610,35 +1539,8 @@ _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request) static void _bt_end_parallel(BTLeader *btleader) { - int i; - - /* Shutdown worker processes */ - WaitForParallelWorkersToFinish(btleader->pcxt); - - /* - * Next, accumulate WAL usage. (This must wait for the workers to finish, - * or we might get incomplete data.) - */ - for (i = 0; i < btleader->pcxt->nworkers_launched; i++) - InstrAccumParallelQuery(&btleader->bufferusage[i], &btleader->walusage[i]); - - /* Free last reference to MVCC snapshot, if one was used */ - if (IsMVCCSnapshot(btleader->snapshot)) - UnregisterSnapshot(btleader->snapshot); - DestroyParallelContext(btleader->pcxt); - ExitParallelMode(); -} - -/* - * Returns size of shared memory required to store state for a parallel - * btree index build based on the snapshot its parallel scan will use. - */ -static Size -_bt_parallel_estimate_shared(Relation heap, Snapshot snapshot) -{ - /* c.f. shm_toc_allocate as to why BUFFERALIGN is used */ - return add_size(BUFFERALIGN(sizeof(BTShared)), - table_parallelscan_estimate(heap, snapshot)); + ParallelIndexBuildEnd(btleader->pcxt, btleader->bufferusage, + btleader->walusage, btleader->snapshot); } /* @@ -1657,29 +1559,13 @@ static double _bt_parallel_heapscan(BTBuildState *buildstate, bool *brokenhotchain) { BTShared *btshared = buildstate->btleader->btshared; - int nparticipanttuplesorts; double reltuples; - nparticipanttuplesorts = buildstate->btleader->nparticipanttuplesorts; - for (;;) - { - SpinLockAcquire(&btshared->mutex); - if (btshared->nparticipantsdone == nparticipanttuplesorts) - { - buildstate->havedead = btshared->havedead; - buildstate->indtuples = btshared->indtuples; - *brokenhotchain = btshared->brokenhotchain; - reltuples = btshared->reltuples; - SpinLockRelease(&btshared->mutex); - break; - } - SpinLockRelease(&btshared->mutex); - - ConditionVariableSleep(&btshared->workersdonecv, - WAIT_EVENT_PARALLEL_CREATE_INDEX_SCAN); - } - - ConditionVariableCancelSleep(); + /* Wait for all participants to finish, and collect their results */ + ParallelIndexBuildWaitForWorkers(&btshared->base, + buildstate->btleader->nparticipanttuplesorts, + &reltuples, &buildstate->indtuples, + &buildstate->havedead, brokenhotchain); return reltuples; } @@ -1751,8 +1637,6 @@ _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc) Sharedsort *sharedsort2; Relation heapRel; Relation indexRel; - LOCKMODE heapLockmode; - LOCKMODE indexLockmode; WalUsage *walusage; BufferUsage *bufferusage; int sortmem; @@ -1779,24 +1663,8 @@ _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc) /* Look up nbtree shared state */ btshared = shm_toc_lookup(toc, PARALLEL_KEY_BTREE_SHARED, false); - /* Open relations using lock modes known to be obtained by index.c */ - if (!btshared->isconcurrent) - { - heapLockmode = ShareLock; - indexLockmode = AccessExclusiveLock; - } - else - { - heapLockmode = ShareUpdateExclusiveLock; - indexLockmode = RowExclusiveLock; - } - - /* Track query ID */ - pgstat_report_query_id(btshared->queryid, false); - - /* Open relations within worker */ - heapRel = table_open(btshared->heaprelid, heapLockmode); - indexRel = index_open(btshared->indexrelid, indexLockmode); + /* Open relations within worker, using the leader's lock modes */ + ParallelIndexBuildOpenRelations(&btshared->base, &heapRel, &indexRel); /* Initialize worker's own spool */ btspool = palloc0_object(BTSpool); @@ -1831,7 +1699,7 @@ _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc) InstrStartParallelQuery(); /* Perform sorting of spool, and possibly a spool2 */ - sortmem = maintenance_work_mem / btshared->scantuplesortstates; + sortmem = maintenance_work_mem / btshared->base.scantuplesortstates; _bt_parallel_scan_and_sort(btspool, btspool2, btshared, sharedsort, sharedsort2, sortmem, false); @@ -1849,8 +1717,7 @@ _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc) } #endif /* BTREE_BUILD_STATS */ - index_close(indexRel, indexLockmode); - table_close(heapRel, heapLockmode); + ParallelIndexBuildCloseRelations(&btshared->base, heapRel, indexRel); } /* @@ -1926,7 +1793,7 @@ _bt_parallel_scan_and_sort(BTSpool *btspool, BTSpool *btspool2, /* Join parallel scan */ indexInfo = BuildIndexInfo(btspool->index); - indexInfo->ii_Concurrent = btshared->isconcurrent; + indexInfo->ii_Concurrent = btshared->base.isconcurrent; scan = table_beginscan_parallel(btspool->heap, ParallelTableScanFromBTShared(btshared), SO_NONE); @@ -1948,21 +1815,12 @@ _bt_parallel_scan_and_sort(BTSpool *btspool, BTSpool *btspool2, } /* - * Done. Record ambuild statistics, and whether we encountered a broken - * HOT chain. + * Done. Record ambuild statistics (including whether we encountered a + * broken HOT chain), and notify the leader. */ - SpinLockAcquire(&btshared->mutex); - btshared->nparticipantsdone++; - btshared->reltuples += reltuples; - if (buildstate.havedead) - btshared->havedead = true; - btshared->indtuples += buildstate.indtuples; - if (indexInfo->ii_BrokenHotChain) - btshared->brokenhotchain = true; - SpinLockRelease(&btshared->mutex); - - /* Notify leader */ - ConditionVariableSignal(&btshared->workersdonecv); + ParallelIndexBuildReportScanDone(&btshared->base, reltuples, + buildstate.indtuples, buildstate.havedead, + indexInfo->ii_BrokenHotChain); /* We can end tuplesorts immediately */ tuplesort_end(btspool->sortstate); diff --git a/src/include/access/parallel_index_build.h b/src/include/access/parallel_index_build.h new file mode 100644 index 0000000000000..12881719834ea --- /dev/null +++ b/src/include/access/parallel_index_build.h @@ -0,0 +1,123 @@ +/*------------------------------------------------------------------------- + * + * parallel_index_build.h + * Shared infrastructure for parallel index builds. + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/include/access/parallel_index_build.h + * + *------------------------------------------------------------------------- + */ +#ifndef PARALLEL_INDEX_BUILD_H +#define PARALLEL_INDEX_BUILD_H + +#include "access/parallel.h" +#include "storage/condition_variable.h" +#include "storage/spin.h" +#include "utils/relcache.h" +#include "utils/snapshot.h" + +struct BufferUsage; +struct WalUsage; +struct ParallelTableScanDescData; + +/* + * Shared state common to all parallel index builds. + * + * Access methods embed this as the first member of their own shared state + * and append any AM-specific fields, followed by the ParallelTableScanDescData + * for the heap scan. The immutable fields are set once by the leader before + * workers are launched; the mutable fields are maintained by the workers + * under mutex and read back by the leader once all workers have signalled + * completion. + */ +typedef struct ParallelIndexBuildShared +{ + /* Immutable state, set by the leader before launching workers */ + Oid heaprelid; + Oid indexrelid; + bool isconcurrent; + int scantuplesortstates; + + /* Query ID, for report in worker processes */ + int64 queryid; + + /* + * workersdonecv is used to monitor the progress of workers. All parallel + * participants must indicate that they are done before the leader can use + * the results built by the workers. + */ + ConditionVariable workersdonecv; + + /* + * mutex protects the mutable fields below. + */ + slock_t mutex; + + /* + * Mutable state that is maintained by workers for all index types, and + * reported back to leader at end of the scans. + * + * nparticipantsdone is the number of worker processes finished. + * + * reltuples is the total number of input heap tuples. + * + * indtuples is the total number of tuples that made it into the index. + * + * brokenhotchain indicates if any worker detected a broken HOT chain + * during build. Every AM must report this from the IndexInfo used for + * its heap scan, so that the leader can propagate it to index_build() and + * the index gets marked with indcheckxmin. + */ + int nparticipantsdone; + double reltuples; + double indtuples; + bool brokenhotchain; + + /* + * Mutable state that is only maintained by nbtree, and left false by the + * other AMs. + * + * havedead indicates if RECENTLY_DEAD tuples were encountered during + * build. When building a unique index, nbtree sorts dead tuples in a + * separate spool to keep them out of the uniqueness check, and uses this + * to skip sorting and merging that spool if it turned out to be empty. + */ + bool havedead; +} ParallelIndexBuildShared; + +/* Leader-side helpers */ +extern ParallelContext *ParallelIndexBuildCreateContext(const char *library_name, + const char *worker_function, + int nworkers); +extern Snapshot ParallelIndexBuildGetSnapshot(bool isconcurrent); +extern Size ParallelIndexBuildEstimateShared(Relation heap, Snapshot snapshot, + Size am_shared_size); +extern void ParallelIndexBuildInitShared(ParallelIndexBuildShared *shared, + Relation heap, Relation index, + bool isconcurrent, + int scantuplesortstates, + struct ParallelTableScanDescData *pscan, + Snapshot snapshot); +extern void ParallelIndexBuildWaitForWorkers(ParallelIndexBuildShared *shared, + int nparticipants, + double *reltuples, double *indtuples, + bool *havedead, bool *brokenhotchain); +extern void ParallelIndexBuildAbort(ParallelContext *pcxt, Snapshot snapshot); +extern void ParallelIndexBuildEnd(ParallelContext *pcxt, + struct BufferUsage *bufferusage, + struct WalUsage *walusage, + Snapshot snapshot); + +/* Worker-side helpers */ +extern void ParallelIndexBuildReportScanDone(ParallelIndexBuildShared *shared, + double reltuples, double indtuples, + bool havedead, bool brokenhotchain); +extern void ParallelIndexBuildOpenRelations(ParallelIndexBuildShared *shared, + Relation *heapRel, Relation *indexRel); +extern void ParallelIndexBuildCloseRelations(ParallelIndexBuildShared *shared, + Relation heapRel, Relation indexRel); + +#endif /* PARALLEL_INDEX_BUILD_H */ diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index b964e5ee90770..1ba0c9564d74d 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2141,6 +2141,7 @@ ParallelHashGrowth ParallelHashJoinBatch ParallelHashJoinBatchAccessor ParallelHashJoinState +ParallelIndexBuildShared ParallelIndexScanDesc ParallelSlot ParallelSlotArray From 6ed997724f61b61632cd2ec0745aa8d9787927dd Mon Sep 17 00:00:00 2001 From: Lukas Fittl Date: Sat, 31 Oct 2026 09:59:25 -0700 Subject: [PATCH 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 3b88bebf52563..7a133e0b2e972 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 7ecc4d3f98d5b..2a1a344af9e9c 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 f049010150ccf..56d204106e48f 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 b51df2a460683..d7be9d84ce713 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 c29f62c5a9910..e900eaed4285c 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 60f857675e05a..77deab8bb4a15 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); From 8946374159099d28983d6363d32631ad66910cdf Mon Sep 17 00:00:00 2001 From: Lukas Fittl Date: Sat, 31 Oct 2026 12:30:52 -0700 Subject: [PATCH 5/5] Unify parallel index build scan handling The parallel index build helpers left it to each AM to pass the ii_BrokenHotChain flag from the IndexInfo used for its heap scan into ParallelIndexBuildReportScanDone(), and to read it back out of ParallelIndexBuildWaitForWorkers() into the IndexInfo passed to ambuild. This is brittle and both GIN and BRIN have gotten this wrong in the past, and there extensions that handle this incorrectly today. Instead, introduce ParallelIndexBuildScan(), which builds the IndexInfo, joins the shared parallel scan, runs table_index_build_scan() and records a broken HOT chain itself. ParallelTableScanDescData is now stored in a separate shmem TOC entry for easier handling. Author: Lukas Fittl Reviewed-by: Discussion: https://postgr.es/m/CAP53Pkw3WcVbg61G29AmN=6XkuxGryoWKnhTyC1JmkT08i8qVA@mail.gmail.com --- src/backend/access/brin/brin.c | 78 ++++-------- .../access/common/parallel_index_build.c | 119 +++++++++++++----- src/backend/access/gin/gininsert.c | 79 +++++------- src/backend/access/nbtree/nbtsort.c | 87 +++++-------- src/include/access/parallel_index_build.h | 50 +++++--- 5 files changed, 210 insertions(+), 203 deletions(-) diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c index 7a133e0b2e972..a84c4606abfc3 100644 --- a/src/backend/access/brin/brin.c +++ b/src/backend/access/brin/brin.c @@ -52,6 +52,7 @@ #define PARALLEL_KEY_QUERY_TEXT UINT64CONST(0xB000000000000003) #define PARALLEL_KEY_WAL_USAGE UINT64CONST(0xB000000000000004) #define PARALLEL_KEY_BUFFER_USAGE UINT64CONST(0xB000000000000005) +#define PARALLEL_KEY_TABLE_SCAN UINT64CONST(0xB000000000000006) /* * Status for index builds performed in parallel. This is allocated in a @@ -66,23 +67,8 @@ typedef struct BrinShared * BRIN-specific immutable state, not modified during the build. */ BlockNumber pagesPerRange; - - /* - * ParallelTableScanDescData data follows. Can't directly embed here, as - * implementations of the parallel table scan desc interface might need - * stronger alignment. - */ } BrinShared; -/* - * Return pointer to a BrinShared's parallel table scan. - * - * c.f. shm_toc_allocate as to why BUFFERALIGN is used, rather than just - * MAXALIGN. - */ -#define ParallelTableScanFromBrinShared(shared) \ - (ParallelTableScanDesc) ((char *) (shared) + BUFFERALIGN(sizeof(BrinShared))) - /* * Status for leader in parallel index build. */ @@ -199,13 +185,14 @@ static void _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Rela bool isconcurrent, int request); static void _brin_end_parallel(BrinLeader *brinleader, BrinBuildState *state); static double _brin_parallel_heapscan(BrinBuildState *state, - bool *brokenhotchain); + IndexInfo *indexInfo); static double _brin_parallel_merge(BrinBuildState *state, IndexInfo *indexInfo); static void _brin_leader_participate_as_worker(BrinBuildState *buildstate, Relation heap, Relation index); static void _brin_parallel_scan_and_build(BrinBuildState *state, BrinShared *brinshared, + shm_toc *toc, Sharedsort *sharedsort, Relation heap, Relation index, int sortmem, bool progress); @@ -2373,10 +2360,11 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index, snapshot = ParallelIndexBuildGetSnapshot(isconcurrent); /* - * Estimate size for our own PARALLEL_KEY_BRIN_SHARED workspace. + * Estimate size for the PARALLEL_KEY_TABLE_SCAN parallel heap scan, and + * our own PARALLEL_KEY_BRIN_SHARED workspace. */ - estbrinshared = ParallelIndexBuildEstimateShared(heap, snapshot, - sizeof(BrinShared)); + ParallelIndexBuildEstimate(pcxt, heap, snapshot); + estbrinshared = sizeof(BrinShared); shm_toc_estimate_chunk(&pcxt->estimator, estbrinshared); estsort = tuplesort_estimate_shared(scantuplesortstates); shm_toc_estimate_chunk(&pcxt->estimator, estsort); @@ -2415,11 +2403,10 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index, brinshared = (BrinShared *) shm_toc_allocate(pcxt->toc, estbrinshared); /* Initialize BRIN-specific immutable state */ brinshared->pagesPerRange = buildstate->bs_pagesPerRange; - /* Initialize common state, and the parallel scan that follows the struct */ - ParallelIndexBuildInitShared(&brinshared->base, heap, index, isconcurrent, - scantuplesortstates, - ParallelTableScanFromBrinShared(brinshared), - snapshot); + /* Initialize common state, and the parallel heap scan */ + ParallelIndexBuildInitShared(pcxt, &brinshared->base, + PARALLEL_KEY_TABLE_SCAN, heap, index, + isconcurrent, scantuplesortstates, snapshot); /* * Store shared tuplesort-private state, for which we reserved space. @@ -2500,19 +2487,19 @@ _brin_end_parallel(BrinLeader *brinleader, BrinBuildState *state) * already be underway within worker processes (when leader participates * as a worker, we should end up here just as workers are finishing). * - * Returns the total number of heap tuples scanned. Also sets *brokenhotchain - * to whether any participant detected a broken HOT chain. + * Returns the total number of heap tuples scanned. Also sets + * indexInfo->ii_BrokenHotChain if any participant detected a broken HOT chain. */ static double -_brin_parallel_heapscan(BrinBuildState *state, bool *brokenhotchain) +_brin_parallel_heapscan(BrinBuildState *state, IndexInfo *indexInfo) { BrinShared *brinshared = state->bs_leader->brinshared; /* Wait for all participants to finish, and collect their results */ ParallelIndexBuildWaitForWorkers(&brinshared->base, state->bs_leader->nparticipanttuplesorts, - &state->bs_reltuples, &state->bs_numtuples, - NULL, brokenhotchain); + indexInfo, &state->bs_reltuples, + &state->bs_numtuples, NULL); return state->bs_reltuples; } @@ -2542,8 +2529,7 @@ _brin_parallel_merge(BrinBuildState *state, IndexInfo *indexInfo) double reltuples; /* wait for workers to scan table and produce partial results */ - reltuples = _brin_parallel_heapscan(state, - &indexInfo->ii_BrokenHotChain); + reltuples = _brin_parallel_heapscan(state, indexInfo); /* do the actual sort in the leader */ tuplesort_performsort(state->bs_sortstate); @@ -2692,7 +2678,8 @@ _brin_leader_participate_as_worker(BrinBuildState *buildstate, Relation heap, Re /* Perform work common to all participants */ _brin_parallel_scan_and_build(buildstate, brinleader->brinshared, - brinleader->sharedsort, heap, index, sortmem, true); + brinleader->pcxt->toc, brinleader->sharedsort, + heap, index, sortmem, true); } /* @@ -2707,14 +2694,13 @@ _brin_leader_participate_as_worker(BrinBuildState *buildstate, Relation heap, Re */ static void _brin_parallel_scan_and_build(BrinBuildState *state, - BrinShared *brinshared, Sharedsort *sharedsort, + BrinShared *brinshared, shm_toc *toc, + Sharedsort *sharedsort, Relation heap, Relation index, int sortmem, bool progress) { SortCoordinate coordinate; - TableScanDesc scan; double reltuples; - IndexInfo *indexInfo; /* Initialize local tuplesort coordination state */ coordinate = palloc0_object(SortCoordinateData); @@ -2727,15 +2713,9 @@ _brin_parallel_scan_and_build(BrinBuildState *state, TUPLESORT_NONE); /* Join parallel scan */ - indexInfo = BuildIndexInfo(index); - indexInfo->ii_Concurrent = brinshared->base.isconcurrent; - - scan = table_beginscan_parallel(heap, - ParallelTableScanFromBrinShared(brinshared), - SO_NONE); - - reltuples = table_index_build_scan(heap, index, indexInfo, true, true, - brinbuildCallbackParallel, state, scan); + reltuples = ParallelIndexBuildScan(&brinshared->base, toc, + PARALLEL_KEY_TABLE_SCAN, heap, index, + true, brinbuildCallbackParallel, state); /* insert the last item */ form_and_spill_tuple(state); @@ -2745,13 +2725,9 @@ _brin_parallel_scan_and_build(BrinBuildState *state, state->bs_reltuples += reltuples; - /* - * Done. Record ambuild statistics (including whether we encountered a - * broken HOT chain), and notify the leader. - */ + /* Done. Record ambuild statistics, and notify the leader. */ ParallelIndexBuildReportScanDone(&brinshared->base, state->bs_reltuples, - state->bs_numtuples, false, - indexInfo->ii_BrokenHotChain); + state->bs_numtuples, false); tuplesort_end(state->bs_sortstate); } @@ -2805,7 +2781,7 @@ _brin_parallel_build_main(dsm_segment *seg, shm_toc *toc) */ sortmem = maintenance_work_mem / brinshared->base.scantuplesortstates; - _brin_parallel_scan_and_build(buildstate, brinshared, sharedsort, + _brin_parallel_scan_and_build(buildstate, brinshared, toc, sharedsort, heapRel, indexRel, sortmem, false); /* Report WAL/buffer usage during parallel execution */ diff --git a/src/backend/access/common/parallel_index_build.c b/src/backend/access/common/parallel_index_build.c index 36df09d6bc5e4..79f5c6fd479d4 100644 --- a/src/backend/access/common/parallel_index_build.c +++ b/src/backend/access/common/parallel_index_build.c @@ -23,6 +23,7 @@ #include "access/table.h" #include "access/tableam.h" #include "access/xact.h" +#include "catalog/index.h" #include "executor/instrument.h" #include "pgstat.h" #include "utils/rel.h" @@ -65,35 +66,43 @@ ParallelIndexBuildGetSnapshot(bool isconcurrent) } /* - * Estimate the DSM space for the shared state struct of a parallel index - * build, including the parallel table scan descriptor that trails it. + * Estimate the DSM space and keys this module needs for a parallel index + * build, namely the parallel table scan descriptor for the heap scan. The + * access method's own shared struct (which embeds ParallelIndexBuildShared) + * is estimated by the caller, like any other AM-specific chunk. * - * am_shared_size is sizeof() the access method's whole shared struct (which - * embeds ParallelIndexBuildShared as its first member). + * Must be called before InitializeParallelDSM. */ -Size -ParallelIndexBuildEstimateShared(Relation heap, Snapshot snapshot, - Size am_shared_size) +void +ParallelIndexBuildEstimate(ParallelContext *pcxt, Relation heap, + Snapshot snapshot) { - /* c.f. shm_toc_allocate as to why BUFFERALIGN is used */ - return add_size(BUFFERALIGN(am_shared_size), - table_parallelscan_estimate(heap, snapshot)); + shm_toc_estimate_chunk(&pcxt->estimator, + table_parallelscan_estimate(heap, snapshot)); + shm_toc_estimate_keys(&pcxt->estimator, 1); } /* * Initialize the common shared state for a parallel index build. * * The caller has already allocated the embedding shared struct in DSM; this - * fills in the common header and initializes the parallel heap scan - * descriptor that follows the (whole) embedding struct, which the caller - * passes as pscan. AM-specific fields are the caller's responsibility. + * fills in the common header. It also allocates and initializes the parallel + * heap scan descriptor, for which ParallelIndexBuildEstimate reserved space, + * and stores it in the TOC under pscan_key (one of the caller's own keys). + * Participants join the scan through ParallelIndexBuildScan(), which also + * takes care of reporting broken HOT chains. AM-specific fields are the + * caller's responsibility. */ void -ParallelIndexBuildInitShared(ParallelIndexBuildShared *shared, +ParallelIndexBuildInitShared(ParallelContext *pcxt, + ParallelIndexBuildShared *shared, + uint64 pscan_key, Relation heap, Relation index, bool isconcurrent, int scantuplesortstates, - ParallelTableScanDesc pscan, Snapshot snapshot) + Snapshot snapshot) { + ParallelTableScanDesc pscan; + /* Initialize immutable state */ shared->heaprelid = RelationGetRelid(heap); shared->indexrelid = RelationGetRelid(index); @@ -110,23 +119,29 @@ ParallelIndexBuildInitShared(ParallelIndexBuildShared *shared, shared->brokenhotchain = false; shared->havedead = false; + /* Set up the parallel heap scan, in its own TOC entry */ + pscan = (ParallelTableScanDesc) + shm_toc_allocate(pcxt->toc, + table_parallelscan_estimate(heap, snapshot)); table_parallelscan_initialize(heap, pscan, snapshot); + shm_toc_insert(pcxt->toc, pscan_key, pscan); } /* * Wait, in the leader, for all participants to finish their portion of the * scan, then read back the accumulated per-build results. * - * reltuples and indtuples receive the totals. brokenhotchain receives whether - * any participant saw a broken HOT chain; the caller must propagate it to the - * IndexInfo passed to ambuild. havedead is optional (pass NULL when the AM - * does not track it): it reports whether any worker saw RECENTLY_DEAD tuples. + * reltuples and indtuples receive the totals. indexInfo must be the IndexInfo + * passed to ambuild: if any participant's scan detected a broken HOT chain, + * its ii_BrokenHotChain is set, so that index_build() marks the index with + * indcheckxmin. havedead is optional (pass NULL when the AM does not track + * it): it reports whether any worker saw RECENTLY_DEAD tuples. */ void ParallelIndexBuildWaitForWorkers(ParallelIndexBuildShared *shared, - int nparticipants, + int nparticipants, IndexInfo *indexInfo, double *reltuples, double *indtuples, - bool *havedead, bool *brokenhotchain) + bool *havedead) { for (;;) { @@ -137,8 +152,8 @@ ParallelIndexBuildWaitForWorkers(ParallelIndexBuildShared *shared, *indtuples = shared->indtuples; if (havedead) *havedead = shared->havedead; - if (brokenhotchain) - *brokenhotchain = shared->brokenhotchain; + if (shared->brokenhotchain) + indexInfo->ii_BrokenHotChain = true; SpinLockRelease(&shared->mutex); break; } @@ -261,20 +276,66 @@ ParallelIndexBuildCloseRelations(ParallelIndexBuildShared *shared, table_close(heapRel, heapLockmode); } +/* + * Run a participant's portion of the parallel heap scan. + * + * Builds the IndexInfo for the scan, joins the shared parallel scan and runs + * table_index_build_scan() with the given callback, as a serial build would. + * Returns the number of heap tuples this participant scanned. + * + * toc is the parallel build's table of contents: the one passed to the worker + * entry point in a worker, or the ParallelContext's in the leader. pscan_key + * must be the key passed to ParallelIndexBuildInitShared. + * + * If the scan detected a broken HOT chain, that is recorded in the shared + * state here, so that ParallelIndexBuildWaitForWorkers() can propagate it to + * the IndexInfo passed to ambuild. Doing this here rather than in + * ParallelIndexBuildReportScanDone() means an AM cannot forget to report it. + * It is safe to set the flag before the participant reports done, since the + * leader only reads it once every participant has done so. + */ +double +ParallelIndexBuildScan(ParallelIndexBuildShared *shared, shm_toc *toc, + uint64 pscan_key, Relation heap, Relation index, + bool progress, IndexBuildCallback callback, + void *callback_state) +{ + ParallelTableScanDesc pscan; + IndexInfo *indexInfo; + TableScanDesc scan; + double reltuples; + + indexInfo = BuildIndexInfo(index); + indexInfo->ii_Concurrent = shared->isconcurrent; + + pscan = shm_toc_lookup(toc, pscan_key, false); + scan = table_beginscan_parallel(heap, pscan, SO_NONE); + reltuples = table_index_build_scan(heap, index, indexInfo, true, progress, + callback, callback_state, scan); + + if (indexInfo->ii_BrokenHotChain) + { + SpinLockAcquire(&shared->mutex); + shared->brokenhotchain = true; + SpinLockRelease(&shared->mutex); + } + + return reltuples; +} + /* * Report that a participant (a worker, or the leader when it participates as * a worker) has finished its portion of the scan. * * Accumulates the participant's tuple counts into the shared totals, ORs in - * the havedead/brokenhotchain flags, and signals the leader. brokenhotchain - * must be the ii_BrokenHotChain of the IndexInfo used for the participant's - * heap scan; havedead is only tracked by nbtree (others pass false). Must be - * paired with ParallelIndexBuildWaitForWorkers in the leader. + * the havedead flag (only tracked by nbtree; others pass false), and signals + * the leader. Must be paired with ParallelIndexBuildWaitForWorkers in the + * leader. Broken HOT chains are reported by ParallelIndexBuildScan() instead. */ void ParallelIndexBuildReportScanDone(ParallelIndexBuildShared *shared, double reltuples, double indtuples, - bool havedead, bool brokenhotchain) + bool havedead) { SpinLockAcquire(&shared->mutex); shared->nparticipantsdone++; @@ -282,8 +343,6 @@ ParallelIndexBuildReportScanDone(ParallelIndexBuildShared *shared, shared->indtuples += indtuples; if (havedead) shared->havedead = true; - if (brokenhotchain) - shared->brokenhotchain = true; SpinLockRelease(&shared->mutex); /* Notify leader */ diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c index 2a1a344af9e9c..91c27c1bf876c 100644 --- a/src/backend/access/gin/gininsert.c +++ b/src/backend/access/gin/gininsert.c @@ -46,6 +46,7 @@ #define PARALLEL_KEY_QUERY_TEXT UINT64CONST(0xB000000000000003) #define PARALLEL_KEY_WAL_USAGE UINT64CONST(0xB000000000000004) #define PARALLEL_KEY_BUFFER_USAGE UINT64CONST(0xB000000000000005) +#define PARALLEL_KEY_TABLE_SCAN UINT64CONST(0xB000000000000006) /* * Status for index builds performed in parallel. This is allocated in a @@ -55,23 +56,8 @@ typedef struct GinBuildShared { /* Common parallel index build state (must be first) */ ParallelIndexBuildShared base; - - /* - * ParallelTableScanDescData data follows. Can't directly embed here, as - * implementations of the parallel table scan desc interface might need - * stronger alignment. - */ } GinBuildShared; -/* - * Return pointer to a GinBuildShared's parallel table scan. - * - * c.f. shm_toc_allocate as to why BUFFERALIGN is used, rather than just - * MAXALIGN. - */ -#define ParallelTableScanFromGinBuildShared(shared) \ - (ParallelTableScanDesc) ((char *) (shared) + BUFFERALIGN(sizeof(GinBuildShared))) - /* * Status for leader in parallel index build. */ @@ -150,13 +136,14 @@ static void _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relati bool isconcurrent, int request); static void _gin_end_parallel(GinLeader *ginleader, GinBuildState *state); static double _gin_parallel_heapscan(GinBuildState *state, - bool *brokenhotchain); + IndexInfo *indexInfo); static double _gin_parallel_merge(GinBuildState *state, IndexInfo *indexInfo); static void _gin_leader_participate_as_worker(GinBuildState *buildstate, Relation heap, Relation index); static void _gin_parallel_scan_and_build(GinBuildState *state, GinBuildShared *ginshared, + shm_toc *toc, Sharedsort *sharedsort, Relation heap, Relation index, int sortmem, bool progress); @@ -929,10 +916,11 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index, snapshot = ParallelIndexBuildGetSnapshot(isconcurrent); /* - * Estimate size for our own PARALLEL_KEY_GIN_SHARED workspace. + * Estimate size for the PARALLEL_KEY_TABLE_SCAN parallel heap scan, and + * our own PARALLEL_KEY_GIN_SHARED workspace. */ - estginshared = ParallelIndexBuildEstimateShared(heap, snapshot, - sizeof(GinBuildShared)); + ParallelIndexBuildEstimate(pcxt, heap, snapshot); + estginshared = sizeof(GinBuildShared); shm_toc_estimate_chunk(&pcxt->estimator, estginshared); estsort = tuplesort_estimate_shared(scantuplesortstates); shm_toc_estimate_chunk(&pcxt->estimator, estsort); @@ -969,11 +957,10 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index, /* Store shared build state, for which we reserved space */ ginshared = (GinBuildShared *) shm_toc_allocate(pcxt->toc, estginshared); - /* Initialize common state, and the parallel scan that follows the struct */ - ParallelIndexBuildInitShared(&ginshared->base, heap, index, isconcurrent, - scantuplesortstates, - ParallelTableScanFromGinBuildShared(ginshared), - snapshot); + /* Initialize common state, and the parallel heap scan */ + ParallelIndexBuildInitShared(pcxt, &ginshared->base, PARALLEL_KEY_TABLE_SCAN, + heap, index, isconcurrent, + scantuplesortstates, snapshot); /* * Store shared tuplesort-private state, for which we reserved space. @@ -1050,19 +1037,19 @@ _gin_end_parallel(GinLeader *ginleader, GinBuildState *state) * already be underway within worker processes (when leader participates * as a worker, we should end up here just as workers are finishing). * - * Returns the total number of heap tuples scanned. Also sets *brokenhotchain - * to whether any participant detected a broken HOT chain. + * Returns the total number of heap tuples scanned. Also sets + * indexInfo->ii_BrokenHotChain if any participant detected a broken HOT chain. */ static double -_gin_parallel_heapscan(GinBuildState *state, bool *brokenhotchain) +_gin_parallel_heapscan(GinBuildState *state, IndexInfo *indexInfo) { GinBuildShared *ginshared = state->bs_leader->ginshared; /* Wait for all participants to finish, and collect their results */ ParallelIndexBuildWaitForWorkers(&ginshared->base, state->bs_leader->nparticipanttuplesorts, - &state->bs_reltuples, &state->bs_numtuples, - NULL, brokenhotchain); + indexInfo, &state->bs_reltuples, + &state->bs_numtuples, NULL); return state->bs_reltuples; } @@ -1552,7 +1539,7 @@ _gin_parallel_merge(GinBuildState *state, IndexInfo *indexInfo) double numtuples = 0; /* wait for workers to scan table and produce partial results */ - reltuples = _gin_parallel_heapscan(state, &indexInfo->ii_BrokenHotChain); + reltuples = _gin_parallel_heapscan(state, indexInfo); /* Execute the sort */ pgstat_progress_update_param(PROGRESS_CREATEIDX_SUBPHASE, @@ -1712,8 +1699,8 @@ _gin_leader_participate_as_worker(GinBuildState *buildstate, Relation heap, Rela /* Perform work common to all participants */ _gin_parallel_scan_and_build(buildstate, ginleader->ginshared, - ginleader->sharedsort, heap, index, - sortmem, true); + ginleader->pcxt->toc, ginleader->sharedsort, + heap, index, sortmem, true); } /* @@ -1908,14 +1895,13 @@ _gin_process_worker_data(GinBuildState *state, Tuplesortstate *worker_sort, */ static void _gin_parallel_scan_and_build(GinBuildState *state, - GinBuildShared *ginshared, Sharedsort *sharedsort, + GinBuildShared *ginshared, shm_toc *toc, + Sharedsort *sharedsort, Relation heap, Relation index, int sortmem, bool progress) { SortCoordinate coordinate; - TableScanDesc scan; double reltuples; - IndexInfo *indexInfo; /* Initialize local tuplesort coordination state */ coordinate = palloc0_object(SortCoordinateData); @@ -1942,15 +1928,10 @@ _gin_parallel_scan_and_build(GinBuildState *state, TUPLESORT_NONE); /* Join parallel scan */ - indexInfo = BuildIndexInfo(index); - indexInfo->ii_Concurrent = ginshared->base.isconcurrent; - - scan = table_beginscan_parallel(heap, - ParallelTableScanFromGinBuildShared(ginshared), - SO_NONE); - - reltuples = table_index_build_scan(heap, index, indexInfo, true, progress, - ginBuildCallbackParallel, state, scan); + reltuples = ParallelIndexBuildScan(&ginshared->base, toc, + PARALLEL_KEY_TABLE_SCAN, heap, index, + progress, ginBuildCallbackParallel, + state); /* write remaining accumulated entries */ ginFlushBuildState(state, index); @@ -1967,13 +1948,9 @@ _gin_parallel_scan_and_build(GinBuildState *state, state->bs_reltuples += reltuples; - /* - * Done. Record ambuild statistics (including whether we encountered a - * broken HOT chain), and notify the leader. - */ + /* Done. Record ambuild statistics, and notify the leader. */ ParallelIndexBuildReportScanDone(&ginshared->base, state->bs_reltuples, - state->bs_numtuples, false, - indexInfo->ii_BrokenHotChain); + state->bs_numtuples, false); tuplesort_end(state->bs_sortstate); } @@ -2054,7 +2031,7 @@ _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc) */ sortmem = maintenance_work_mem / ginshared->base.scantuplesortstates; - _gin_parallel_scan_and_build(&buildstate, ginshared, sharedsort, + _gin_parallel_scan_and_build(&buildstate, ginshared, toc, sharedsort, heapRel, indexRel, sortmem, false); /* Report WAL/buffer usage during parallel execution */ diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c index 56d204106e48f..1f65d9aecdeda 100644 --- a/src/backend/access/nbtree/nbtsort.c +++ b/src/backend/access/nbtree/nbtsort.c @@ -68,6 +68,7 @@ #define PARALLEL_KEY_QUERY_TEXT UINT64CONST(0xA000000000000004) #define PARALLEL_KEY_WAL_USAGE UINT64CONST(0xA000000000000005) #define PARALLEL_KEY_BUFFER_USAGE UINT64CONST(0xA000000000000006) +#define PARALLEL_KEY_TABLE_SCAN UINT64CONST(0xA000000000000007) /* * DISABLE_LEADER_PARTICIPATION disables the leader's participation in @@ -102,23 +103,8 @@ typedef struct BTShared /* nbtree-specific immutable state, not modified during the sort */ bool isunique; bool nulls_not_distinct; - - /* - * ParallelTableScanDescData data follows. Can't directly embed here, as - * implementations of the parallel table scan desc interface might need - * stronger alignment. - */ } BTShared; -/* - * Return pointer to a BTShared's parallel table scan. - * - * c.f. shm_toc_allocate as to why BUFFERALIGN is used, rather than just - * MAXALIGN. - */ -#define ParallelTableScanFromBTShared(shared) \ - (ParallelTableScanDesc) ((char *) (shared) + BUFFERALIGN(sizeof(BTShared))) - /* * Status for leader in parallel index build. */ @@ -237,10 +223,11 @@ static void _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request); static void _bt_end_parallel(BTLeader *btleader); static double _bt_parallel_heapscan(BTBuildState *buildstate, - bool *brokenhotchain); + IndexInfo *indexInfo); static void _bt_leader_participate_as_worker(BTBuildState *buildstate); static void _bt_parallel_scan_and_sort(BTSpool *btspool, BTSpool *btspool2, - BTShared *btshared, Sharedsort *sharedsort, + BTShared *btshared, shm_toc *toc, + Sharedsort *sharedsort, Sharedsort *sharedsort2, int sortmem, bool progress); @@ -435,8 +422,7 @@ _bt_spools_heapscan(Relation heap, Relation index, BTBuildState *buildstate, _bt_build_callback, buildstate, NULL); else - reltuples = _bt_parallel_heapscan(buildstate, - &indexInfo->ii_BrokenHotChain); + reltuples = _bt_parallel_heapscan(buildstate, indexInfo); /* * Set the progress target for the next phase. Reset the block number @@ -1383,11 +1369,12 @@ _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request) snapshot = ParallelIndexBuildGetSnapshot(isconcurrent); /* - * Estimate size for our own PARALLEL_KEY_BTREE_SHARED workspace, and - * PARALLEL_KEY_TUPLESORT tuplesort workspace + * Estimate size for the PARALLEL_KEY_TABLE_SCAN parallel heap scan, our + * own PARALLEL_KEY_BTREE_SHARED workspace, and PARALLEL_KEY_TUPLESORT + * tuplesort workspace */ - estbtshared = ParallelIndexBuildEstimateShared(btspool->heap, snapshot, - sizeof(BTShared)); + ParallelIndexBuildEstimate(pcxt, btspool->heap, snapshot); + estbtshared = sizeof(BTShared); shm_toc_estimate_chunk(&pcxt->estimator, estbtshared); estsort = tuplesort_estimate_shared(scantuplesortstates); shm_toc_estimate_chunk(&pcxt->estimator, estsort); @@ -1437,11 +1424,10 @@ _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request) /* Initialize nbtree-specific immutable state */ btshared->isunique = btspool->isunique; btshared->nulls_not_distinct = btspool->nulls_not_distinct; - /* Initialize common state, and the parallel scan that follows the struct */ - ParallelIndexBuildInitShared(&btshared->base, btspool->heap, btspool->index, - isconcurrent, scantuplesortstates, - ParallelTableScanFromBTShared(btshared), - snapshot); + /* Initialize common state, and the parallel heap scan */ + ParallelIndexBuildInitShared(pcxt, &btshared->base, PARALLEL_KEY_TABLE_SCAN, + btspool->heap, btspool->index, isconcurrent, + scantuplesortstates, snapshot); /* * Store shared tuplesort-private state, for which we reserved space. @@ -1536,13 +1522,14 @@ _bt_end_parallel(BTLeader *btleader) * already be underway within worker processes (when leader participates * as a worker, we should end up here just as workers are finishing). * - * Fills in fields needed for ambuild statistics, and lets caller set - * field indicating that some worker encountered a broken HOT chain. + * Fills in fields needed for ambuild statistics, and sets + * indexInfo->ii_BrokenHotChain if some participant encountered a broken HOT + * chain. * * Returns the total number of heap tuples scanned. */ static double -_bt_parallel_heapscan(BTBuildState *buildstate, bool *brokenhotchain) +_bt_parallel_heapscan(BTBuildState *buildstate, IndexInfo *indexInfo) { BTShared *btshared = buildstate->btleader->btshared; double reltuples; @@ -1550,8 +1537,9 @@ _bt_parallel_heapscan(BTBuildState *buildstate, bool *brokenhotchain) /* Wait for all participants to finish, and collect their results */ ParallelIndexBuildWaitForWorkers(&btshared->base, buildstate->btleader->nparticipanttuplesorts, - &reltuples, &buildstate->indtuples, - &buildstate->havedead, brokenhotchain); + indexInfo, &reltuples, + &buildstate->indtuples, + &buildstate->havedead); return reltuples; } @@ -1597,8 +1585,8 @@ _bt_leader_participate_as_worker(BTBuildState *buildstate) /* Perform work common to all participants */ _bt_parallel_scan_and_sort(leaderworker, leaderworker2, btleader->btshared, - btleader->sharedsort, btleader->sharedsort2, - sortmem, true); + btleader->pcxt->toc, btleader->sharedsort, + btleader->sharedsort2, sortmem, true); #ifdef BTREE_BUILD_STATS if (log_btree_build_stats) @@ -1681,7 +1669,7 @@ _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc) /* Perform sorting of spool, and possibly a spool2 */ sortmem = maintenance_work_mem / btshared->base.scantuplesortstates; - _bt_parallel_scan_and_sort(btspool, btspool2, btshared, sharedsort, + _bt_parallel_scan_and_sort(btspool, btspool2, btshared, toc, sharedsort, sharedsort2, sortmem, false); /* Report WAL/buffer usage during parallel execution */ @@ -1715,14 +1703,13 @@ _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc) */ static void _bt_parallel_scan_and_sort(BTSpool *btspool, BTSpool *btspool2, - BTShared *btshared, Sharedsort *sharedsort, - Sharedsort *sharedsort2, int sortmem, bool progress) + BTShared *btshared, shm_toc *toc, + Sharedsort *sharedsort, Sharedsort *sharedsort2, + int sortmem, bool progress) { SortCoordinate coordinate; BTBuildState buildstate; - TableScanDesc scan; double reltuples; - IndexInfo *indexInfo; /* Initialize local tuplesort coordination state */ coordinate = palloc0_object(SortCoordinateData); @@ -1773,14 +1760,10 @@ _bt_parallel_scan_and_sort(BTSpool *btspool, BTSpool *btspool2, buildstate.btleader = NULL; /* Join parallel scan */ - indexInfo = BuildIndexInfo(btspool->index); - indexInfo->ii_Concurrent = btshared->base.isconcurrent; - scan = table_beginscan_parallel(btspool->heap, - ParallelTableScanFromBTShared(btshared), - SO_NONE); - reltuples = table_index_build_scan(btspool->heap, btspool->index, indexInfo, - true, progress, _bt_build_callback, - &buildstate, scan); + reltuples = ParallelIndexBuildScan(&btshared->base, toc, + PARALLEL_KEY_TABLE_SCAN, btspool->heap, + btspool->index, progress, + _bt_build_callback, &buildstate); /* Execute this worker's part of the sort */ if (progress) @@ -1795,13 +1778,9 @@ _bt_parallel_scan_and_sort(BTSpool *btspool, BTSpool *btspool2, tuplesort_performsort(btspool2->sortstate); } - /* - * Done. Record ambuild statistics (including whether we encountered a - * broken HOT chain), and notify the leader. - */ + /* Done. Record ambuild statistics, and notify the leader. */ ParallelIndexBuildReportScanDone(&btshared->base, reltuples, - buildstate.indtuples, buildstate.havedead, - indexInfo->ii_BrokenHotChain); + buildstate.indtuples, buildstate.havedead); /* We can end tuplesorts immediately */ tuplesort_end(btspool->sortstate); diff --git a/src/include/access/parallel_index_build.h b/src/include/access/parallel_index_build.h index 12881719834ea..f9fb7a6bf11b9 100644 --- a/src/include/access/parallel_index_build.h +++ b/src/include/access/parallel_index_build.h @@ -15,23 +15,31 @@ #include "access/parallel.h" #include "storage/condition_variable.h" +#include "storage/itemptr.h" #include "storage/spin.h" #include "utils/relcache.h" #include "utils/snapshot.h" struct BufferUsage; +struct IndexInfo; struct WalUsage; -struct ParallelTableScanDescData; + +/* Avoid including tableam.h here */ +typedef void (*IndexBuildCallback) (Relation index, + ItemPointer tid, + Datum *values, + bool *isnull, + bool tupleIsAlive, + void *state); /* * Shared state common to all parallel index builds. * * Access methods embed this as the first member of their own shared state - * and append any AM-specific fields, followed by the ParallelTableScanDescData - * for the heap scan. The immutable fields are set once by the leader before - * workers are launched; the mutable fields are maintained by the workers - * under mutex and read back by the leader once all workers have signalled - * completion. + * and append any AM-specific fields. The immutable fields are set once by + * the leader before workers are launched; the mutable fields are maintained + * by the workers under mutex and read back by the leader once all workers + * have signalled completion. */ typedef struct ParallelIndexBuildShared { @@ -66,10 +74,10 @@ typedef struct ParallelIndexBuildShared * * indtuples is the total number of tuples that made it into the index. * - * brokenhotchain indicates if any worker detected a broken HOT chain - * during build. Every AM must report this from the IndexInfo used for - * its heap scan, so that the leader can propagate it to index_build() and - * the index gets marked with indcheckxmin. + * brokenhotchain indicates if any participant detected a broken HOT chain + * during build. Set by ParallelIndexBuildScan() and propagated to the + * IndexInfo passed to ambuild by ParallelIndexBuildWaitForWorkers(), so + * that index_build() marks the index with indcheckxmin. */ int nparticipantsdone; double reltuples; @@ -93,28 +101,36 @@ extern ParallelContext *ParallelIndexBuildCreateContext(const char *library_name const char *worker_function, int nworkers); extern Snapshot ParallelIndexBuildGetSnapshot(bool isconcurrent); -extern Size ParallelIndexBuildEstimateShared(Relation heap, Snapshot snapshot, - Size am_shared_size); -extern void ParallelIndexBuildInitShared(ParallelIndexBuildShared *shared, +extern void ParallelIndexBuildEstimate(ParallelContext *pcxt, Relation heap, + Snapshot snapshot); +extern void ParallelIndexBuildInitShared(ParallelContext *pcxt, + ParallelIndexBuildShared *shared, + uint64 pscan_key, Relation heap, Relation index, bool isconcurrent, int scantuplesortstates, - struct ParallelTableScanDescData *pscan, Snapshot snapshot); extern void ParallelIndexBuildWaitForWorkers(ParallelIndexBuildShared *shared, int nparticipants, + struct IndexInfo *indexInfo, double *reltuples, double *indtuples, - bool *havedead, bool *brokenhotchain); + bool *havedead); extern void ParallelIndexBuildAbort(ParallelContext *pcxt, Snapshot snapshot); extern void ParallelIndexBuildEnd(ParallelContext *pcxt, struct BufferUsage *bufferusage, struct WalUsage *walusage, Snapshot snapshot); -/* Worker-side helpers */ +/* Worker-side helpers (also used by the leader when it participates) */ +extern double ParallelIndexBuildScan(ParallelIndexBuildShared *shared, + shm_toc *toc, uint64 pscan_key, + Relation heap, Relation index, + bool progress, + IndexBuildCallback callback, + void *callback_state); extern void ParallelIndexBuildReportScanDone(ParallelIndexBuildShared *shared, double reltuples, double indtuples, - bool havedead, bool brokenhotchain); + bool havedead); extern void ParallelIndexBuildOpenRelations(ParallelIndexBuildShared *shared, Relation *heapRel, Relation *indexRel); extern void ParallelIndexBuildCloseRelations(ParallelIndexBuildShared *shared,