diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c index 5059da8dce7c5..a84c4606abfc3 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,10 +34,8 @@ #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" #include "utils/acl.h" #include "utils/datum.h" #include "utils/fmgrprotos.h" @@ -53,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 @@ -60,66 +60,15 @@ */ 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; - - /* - * 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 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. - */ - int nparticipantsdone; - double reltuples; - double indtuples; + /* 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. + * BRIN-specific immutable state, not modified during the build. */ + BlockNumber pagesPerRange; } 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. */ @@ -235,13 +184,15 @@ 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); -static double _brin_parallel_merge(BrinBuildState *state); +static double _brin_parallel_heapscan(BrinBuildState *state, + 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); @@ -1223,7 +1174,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); } @@ -2400,29 +2351,20 @@ _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. + * Estimate size for the PARALLEL_KEY_TABLE_SCAN parallel heap scan, and + * our own PARALLEL_KEY_BRIN_SHARED workspace. */ - estbrinshared = _brin_parallel_estimate_shared(heap, snapshot); + 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); @@ -2445,14 +2387,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); @@ -2460,33 +2395,18 @@ _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; - - table_parallelscan_initialize(heap, - 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. @@ -2504,14 +2424,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 @@ -2563,23 +2476,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); } /* @@ -2589,34 +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. + * 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) +_brin_parallel_heapscan(BrinBuildState *state, IndexInfo *indexInfo) { 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; - - 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, + indexInfo, &state->bs_reltuples, + &state->bs_numtuples, NULL); return state->bs_reltuples; } @@ -2630,10 +2513,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 +2529,7 @@ _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); /* do the actual sort in the leader */ tuplesort_performsort(state->bs_sortstate); @@ -2775,18 +2660,6 @@ _brin_parallel_merge(BrinBuildState *state) 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. */ @@ -2805,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); } /* @@ -2820,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); @@ -2840,15 +2713,9 @@ _brin_parallel_scan_and_build(BrinBuildState *state, TUPLESORT_NONE); /* Join parallel scan */ - indexInfo = BuildIndexInfo(index); - indexInfo->ii_Concurrent = brinshared->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); @@ -2858,17 +2725,9 @@ _brin_parallel_scan_and_build(BrinBuildState *state, state->bs_reltuples += reltuples; - /* - * Done. Record ambuild statistics. - */ - SpinLockAcquire(&brinshared->mutex); - brinshared->nparticipantsdone++; - brinshared->reltuples += state->bs_reltuples; - brinshared->indtuples += state->bs_numtuples; - SpinLockRelease(&brinshared->mutex); - - /* Notify leader */ - ConditionVariableSignal(&brinshared->workersdonecv); + /* Done. Record ambuild statistics, and notify the leader. */ + ParallelIndexBuildReportScanDone(&brinshared->base, state->bs_reltuples, + state->bs_numtuples, false); tuplesort_end(state->bs_sortstate); } @@ -2879,14 +2738,11 @@ _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; Relation heapRel; Relation indexRel; - LOCKMODE heapLockmode; - LOCKMODE indexLockmode; WalUsage *walusage; BufferUsage *bufferusage; int sortmem; @@ -2898,34 +2754,14 @@ _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); - /* 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, @@ -2943,9 +2779,9 @@ _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, + _brin_parallel_scan_and_build(buildstate, brinshared, toc, sharedsort, heapRel, indexRel, sortmem, false); /* Report WAL/buffer usage during parallel execution */ @@ -2954,8 +2790,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..79f5c6fd479d4 --- /dev/null +++ b/src/backend/access/common/parallel_index_build.c @@ -0,0 +1,350 @@ +/*------------------------------------------------------------------------- + * + * 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 "catalog/index.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 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. + * + * Must be called before InitializeParallelDSM. + */ +void +ParallelIndexBuildEstimate(ParallelContext *pcxt, Relation heap, + Snapshot 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. 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(ParallelContext *pcxt, + ParallelIndexBuildShared *shared, + uint64 pscan_key, + Relation heap, Relation index, + bool isconcurrent, int scantuplesortstates, + Snapshot snapshot) +{ + ParallelTableScanDesc pscan; + + /* 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; + + /* 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. 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, IndexInfo *indexInfo, + double *reltuples, double *indtuples, + bool *havedead) +{ + for (;;) + { + SpinLockAcquire(&shared->mutex); + if (shared->nparticipantsdone == nparticipants) + { + *reltuples = shared->reltuples; + *indtuples = shared->indtuples; + if (havedead) + *havedead = shared->havedead; + if (shared->brokenhotchain) + indexInfo->ii_BrokenHotChain = true; + 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); +} + +/* + * 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 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) +{ + SpinLockAcquire(&shared->mutex); + shared->nparticipantsdone++; + shared->reltuples += reltuples; + shared->indtuples += indtuples; + if (havedead) + shared->havedead = 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 564b1509f1bf2..91c27c1bf876c 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,10 +29,8 @@ #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" #include "utils/datum.h" #include "utils/memutils.h" #include "utils/builtins.h" @@ -47,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 @@ -54,62 +54,10 @@ */ 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; - - /* - * 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. - */ - int nparticipantsdone; - double reltuples; - double indtuples; - - /* - * ParallelTableScanDescData data follows. Can't directly embed here, as - * implementations of the parallel table scan desc interface might need - * stronger alignment. - */ + /* Common parallel index build state (must be first) */ + ParallelIndexBuildShared base; } 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. */ @@ -187,13 +135,15 @@ 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); -static double _gin_parallel_merge(GinBuildState *state); +static double _gin_parallel_heapscan(GinBuildState *state, + 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); @@ -747,7 +697,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); } @@ -957,29 +907,20 @@ _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. + * Estimate size for the PARALLEL_KEY_TABLE_SCAN parallel heap scan, and + * our own PARALLEL_KEY_GIN_SHARED workspace. */ - estginshared = _gin_parallel_estimate_shared(heap, snapshot); + 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); @@ -1002,14 +943,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); @@ -1017,32 +951,16 @@ _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; - - ConditionVariableInit(&ginshared->workersdonecv); - SpinLockInit(&ginshared->mutex); - - /* Initialize mutable state */ - ginshared->nparticipantsdone = 0; - ginshared->reltuples = 0.0; - ginshared->indtuples = 0.0; - - table_parallelscan_initialize(heap, - 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. @@ -1056,14 +974,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 @@ -1115,23 +1026,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); } /* @@ -1141,34 +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. + * 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) +_gin_parallel_heapscan(GinBuildState *state, IndexInfo *indexInfo) { 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; - - 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, + indexInfo, &state->bs_reltuples, + &state->bs_numtuples, NULL); return state->bs_reltuples; } @@ -1642,10 +1523,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; @@ -1656,7 +1539,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); /* Execute the sort */ pgstat_progress_update_param(PROGRESS_CREATEIDX_SUBPHASE, @@ -1798,18 +1681,6 @@ _gin_parallel_merge(GinBuildState *state) 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. */ @@ -1828,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); } /* @@ -2024,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); @@ -2043,7 +1913,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, @@ -2058,15 +1928,10 @@ _gin_parallel_scan_and_build(GinBuildState *state, TUPLESORT_NONE); /* Join parallel scan */ - indexInfo = BuildIndexInfo(index); - indexInfo->ii_Concurrent = ginshared->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); @@ -2083,17 +1948,9 @@ _gin_parallel_scan_and_build(GinBuildState *state, state->bs_reltuples += reltuples; - /* - * Done. Record ambuild statistics. - */ - SpinLockAcquire(&ginshared->mutex); - ginshared->nparticipantsdone++; - ginshared->reltuples += state->bs_reltuples; - ginshared->indtuples += state->bs_numtuples; - SpinLockRelease(&ginshared->mutex); - - /* Notify leader */ - ConditionVariableSignal(&ginshared->workersdonecv); + /* Done. Record ambuild statistics, and notify the leader. */ + ParallelIndexBuildReportScanDone(&ginshared->base, state->bs_reltuples, + state->bs_numtuples, false); tuplesort_end(state->bs_sortstate); } @@ -2104,14 +1961,11 @@ _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; Relation heapRel; Relation indexRel; - LOCKMODE heapLockmode; - LOCKMODE indexLockmode; WalUsage *walusage; BufferUsage *bufferusage; int sortmem; @@ -2123,31 +1977,14 @@ _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); - /* Open relations using lock modes known to be obtained by index.c */ - if (!ginshared->isconcurrent) - { - heapLockmode = ShareLock; - indexLockmode = AccessExclusiveLock; - } - else - { - heapLockmode = ShareUpdateExclusiveLock; - indexLockmode = RowExclusiveLock; - } - - /* 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); @@ -2192,9 +2029,9 @@ _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, + _gin_parallel_scan_and_build(&buildstate, ginshared, toc, sharedsort, heapRel, indexRel, sortmem, false); /* Report WAL/buffer usage during parallel execution */ @@ -2203,8 +2040,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..1f65d9aecdeda 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" @@ -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 @@ -96,75 +97,14 @@ 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 - * 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. */ @@ -282,12 +222,12 @@ 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); + 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); @@ -482,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 @@ -1421,30 +1360,21 @@ _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 + * 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 = _bt_parallel_estimate_shared(btspool->heap, snapshot); + 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); @@ -1477,14 +1407,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); @@ -1492,34 +1415,19 @@ _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 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. @@ -1550,14 +1458,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 @@ -1610,35 +1511,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); } /* @@ -1648,38 +1522,24 @@ _bt_parallel_estimate_shared(Relation heap, Snapshot snapshot) * 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; - 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, + indexInfo, &reltuples, + &buildstate->indtuples, + &buildstate->havedead); return reltuples; } @@ -1725,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) @@ -1743,7 +1603,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; @@ -1751,8 +1610,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; @@ -1769,34 +1626,14 @@ _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); - /* 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,8 +1668,8 @@ _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; - _bt_parallel_scan_and_sort(btspool, btspool2, btshared, sharedsort, + sortmem = maintenance_work_mem / btshared->base.scantuplesortstates; + _bt_parallel_scan_and_sort(btspool, btspool2, btshared, toc, sharedsort, sharedsort2, sortmem, false); /* Report WAL/buffer usage during parallel execution */ @@ -1849,8 +1686,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); } /* @@ -1867,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); @@ -1925,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->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) @@ -1947,22 +1778,9 @@ _bt_parallel_scan_and_sort(BTSpool *btspool, BTSpool *btspool2, tuplesort_performsort(btspool2->sortstate); } - /* - * Done. Record ambuild statistics, and whether we encountered a broken - * HOT chain. - */ - 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); + /* Done. Record ambuild statistics, and notify the leader. */ + ParallelIndexBuildReportScanDone(&btshared->base, reltuples, + buildstate.indtuples, buildstate.havedead); /* We can end tuplesorts immediately */ tuplesort_end(btspool->sortstate); 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); diff --git a/src/include/access/parallel_index_build.h b/src/include/access/parallel_index_build.h new file mode 100644 index 0000000000000..f9fb7a6bf11b9 --- /dev/null +++ b/src/include/access/parallel_index_build.h @@ -0,0 +1,139 @@ +/*------------------------------------------------------------------------- + * + * 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/itemptr.h" +#include "storage/spin.h" +#include "utils/relcache.h" +#include "utils/snapshot.h" + +struct BufferUsage; +struct IndexInfo; +struct WalUsage; + +/* 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. 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 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; + 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 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, + Snapshot snapshot); +extern void ParallelIndexBuildWaitForWorkers(ParallelIndexBuildShared *shared, + int nparticipants, + struct IndexInfo *indexInfo, + double *reltuples, double *indtuples, + 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 (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); +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