From 8946374159099d28983d6363d32631ad66910cdf Mon Sep 17 00:00:00 2001 From: Lukas Fittl Date: Sat, 31 Oct 2026 12:30:52 -0700 Subject: [PATCH v2 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 7a133e0b2e9..a84c4606abf 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 36df09d6bc5..79f5c6fd479 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 2a1a344af9e..91c27c1bf87 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 56d204106e4..1f65d9aecde 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 12881719834..f9fb7a6bf11 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, -- 2.47.1