From 3cd39ef87485d2f1acdd712b146f2a9c3898ad8c Mon Sep 17 00:00:00 2001 From: Mikhail Nikalayeu Date: Mon, 12 Oct 2026 12:00:49 +0200 Subject: [PATCH v8 5/7] Allow snapshot resets in parallel concurrent index builds The leader must keep its initial snapshot until every worker has restored it. Have workers signal through the shared scan descriptor when they have joined the scan with their own snapshot. The leader waits for those signals before it starts scanning and can release the original snapshot. Each participant then replaces its snapshot independently. Use the build's active snapshot as the initial scan snapshot, avoiding a separately registered snapshot that would keep xmin pinned. This enables resets for parallel GIN, BRIN, and non-unique btree builds. Worker attachment now happens before the leader joins the scan in all parallel builds. Reviewed-by: Matthias van de Meent --- doc/src/sgml/indexam.sgml | 2 + src/backend/access/brin/brin.c | 49 ++++++++------- src/backend/access/gin/gininsert.c | 49 ++++++++------- src/backend/access/heap/heapam.c | 12 ++-- src/backend/access/heap/heapam_handler.c | 3 + src/backend/access/nbtree/nbtsort.c | 50 ++++++++++----- src/backend/access/table/tableam.c | 62 +++++++++++++++++-- src/backend/catalog/index.c | 3 +- src/backend/executor/nodeSeqscan.c | 3 +- src/backend/executor/nodeTidrangescan.c | 3 +- .../utils/activity/wait_event_names.txt | 1 + src/backend/utils/time/snapmgr.c | 5 +- src/include/access/relscan.h | 6 ++ src/include/access/tableam.h | 15 ++++- .../expected/cic_reset_snapshots.out | 39 ++++++++++++ .../sql/cic_reset_snapshots.sql | 8 ++- 16 files changed, 228 insertions(+), 82 deletions(-) diff --git a/doc/src/sgml/indexam.sgml b/doc/src/sgml/indexam.sgml index c58efc639d..d110b2f636 100644 --- a/doc/src/sgml/indexam.sgml +++ b/doc/src/sgml/indexam.sgml @@ -322,6 +322,8 @@ ambuild (Relation heapRelation, after the build callback returns, fetching external TOAST data rather than retaining pointers to it. The access method must not hold an additional snapshot across the scan. + Access methods with parallel builds must also follow the parallel table + scan's snapshot reset protocol. diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c index eeaa788859..3ee9fd94d4 100644 --- a/src/backend/access/brin/brin.c +++ b/src/backend/access/brin/brin.c @@ -145,12 +145,9 @@ typedef struct BrinLeader * * brinshared is the shared state for entire build. sharedsort is the * shared, tuplesort-managed state passed to each process tuplesort. - * snapshot is the snapshot used by the scan iff an MVCC snapshot is - * required. */ BrinShared *brinshared; Sharedsort *sharedsort; - Snapshot snapshot; WalUsage *walusage; BufferUsage *bufferusage; } BrinLeader; @@ -1255,6 +1252,8 @@ brinbuild(Relation heap, Relation index, IndexInfo *indexInfo) * for btree, but not for BRIN, which can do with much less memory. So * maybe make that somehow less strict, optionally? */ + /* Concurrent builds support snapshot resets. */ + Assert(indexInfo->ii_ResetSnapshot == indexInfo->ii_Concurrent); if (indexInfo->ii_ParallelWorkers > 0) _brin_begin_parallel(state, heap, index, indexInfo->ii_Concurrent, indexInfo->ii_ParallelWorkers); @@ -2492,16 +2491,16 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index, 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. + * Use SnapshotAny for non-concurrent builds so the scan can include + * RECENTLY_DEAD tuples. Concurrent builds use the active snapshot from + * index_build(), which parallel initialization serializes for the + * workers. Do not register another copy, as it would pin xmin across + * snapshot resets. */ if (!isconcurrent) snapshot = SnapshotAny; else - snapshot = RegisterSnapshot(GetTransactionSnapshot()); + snapshot = InvalidSnapshot; /* * Estimate size for our own PARALLEL_KEY_BRIN_SHARED workspace. @@ -2544,8 +2543,6 @@ _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(); return; @@ -2570,7 +2567,8 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index, table_parallelscan_initialize(heap, ParallelTableScanFromBrinShared(brinshared), - snapshot); + snapshot, + isconcurrent); /* * Store shared tuplesort-private state, for which we reserved space. @@ -2616,7 +2614,6 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index, brinleader->nparticipanttuplesorts++; brinleader->brinshared = brinshared; brinleader->sharedsort = sharedsort; - brinleader->snapshot = snapshot; brinleader->walusage = walusage; brinleader->bufferusage = bufferusage; @@ -2630,15 +2627,19 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index, /* Save leader state now that it's clear build will be parallel */ buildstate->bs_leader = brinleader; + /* + * Wait for workers to attach before joining the scan, so startup failures + * are reported before we wait for their snapshots. The leader must keep + * its initial snapshot until every worker has restored it. + */ + WaitForParallelWorkersToAttach(pcxt); + if (isconcurrent) + table_parallelscan_wait_restored(ParallelTableScanFromBrinShared(brinshared), + pcxt->nworkers_launched); + /* Join heap scan ourselves */ if (leaderparticipates) _brin_leader_participate_as_worker(buildstate, heap, index); - - /* - * Caller needs to wait for all launched workers when we return. Make - * sure that the failure-to-start case will not hang forever. - */ - WaitForParallelWorkersToAttach(pcxt); } /* @@ -2659,9 +2660,6 @@ _brin_end_parallel(BrinLeader *brinleader, BrinBuildState *state) 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(); } @@ -2861,7 +2859,7 @@ _brin_parallel_merge(BrinBuildState *state) /* * Returns size of shared memory required to store state for a parallel - * brin index build based on the snapshot its parallel scan will use. + * brin index build. */ static Size _brin_parallel_estimate_shared(Relation heap, Snapshot snapshot) @@ -2910,6 +2908,7 @@ _brin_parallel_scan_and_build(BrinBuildState *state, { SortCoordinate coordinate; TableScanDesc scan; + ParallelTableScanDesc pscan; double reltuples; IndexInfo *indexInfo; @@ -2924,12 +2923,14 @@ _brin_parallel_scan_and_build(BrinBuildState *state, TUPLESORT_NONE); /* Join parallel scan */ + pscan = ParallelTableScanFromBrinShared(brinshared); indexInfo = BuildIndexInfo(index); indexInfo->ii_Concurrent = brinshared->isconcurrent; + indexInfo->ii_ResetSnapshot = pscan->phs_reset_snapshot; state->bs_attvarlena = brin_build_attvarlena(heap, indexInfo); scan = table_beginscan_parallel(heap, - ParallelTableScanFromBrinShared(brinshared), + pscan, SO_NONE); reltuples = table_index_build_scan(heap, index, indexInfo, true, true, diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c index 564b1509f1..95a601bc58 100644 --- a/src/backend/access/gin/gininsert.c +++ b/src/backend/access/gin/gininsert.c @@ -132,12 +132,9 @@ typedef struct GinLeader * * GinBuildShared is the shared state for entire build. sharedsort is the * shared, tuplesort-managed state passed to each process tuplesort. - * snapshot is the snapshot used by the scan iff an MVCC snapshot is - * required. */ GinBuildShared *ginshared; Sharedsort *sharedsort; - Snapshot snapshot; WalUsage *walusage; BufferUsage *bufferusage; } GinLeader; @@ -696,6 +693,8 @@ ginbuild(Relation heap, Relation index, IndexInfo *indexInfo) * ignore the memory used to accumulate data in memory (set by work_mem), * but there is no way to communicate that to plan_create_index_workers. */ + /* Concurrent builds support snapshot resets. */ + Assert(indexInfo->ii_ResetSnapshot == indexInfo->ii_Concurrent); if (indexInfo->ii_ParallelWorkers > 0) _gin_begin_parallel(state, heap, index, indexInfo->ii_Concurrent, indexInfo->ii_ParallelWorkers); @@ -965,16 +964,16 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index, 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. + * Use SnapshotAny for non-concurrent builds so the scan can include + * RECENTLY_DEAD tuples. Concurrent builds use the active snapshot from + * index_build(), which parallel initialization serializes for the + * workers. Do not register another copy, as it would pin xmin across + * snapshot resets. */ if (!isconcurrent) snapshot = SnapshotAny; else - snapshot = RegisterSnapshot(GetTransactionSnapshot()); + snapshot = InvalidSnapshot; /* * Estimate size for our own PARALLEL_KEY_GIN_SHARED workspace. @@ -1017,8 +1016,6 @@ _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(); return; @@ -1042,7 +1039,8 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index, table_parallelscan_initialize(heap, ParallelTableScanFromGinBuildShared(ginshared), - snapshot); + snapshot, + isconcurrent); /* * Store shared tuplesort-private state, for which we reserved space. @@ -1084,7 +1082,6 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index, ginleader->nparticipanttuplesorts++; ginleader->ginshared = ginshared; ginleader->sharedsort = sharedsort; - ginleader->snapshot = snapshot; ginleader->walusage = walusage; ginleader->bufferusage = bufferusage; @@ -1098,15 +1095,19 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index, /* Save leader state now that it's clear build will be parallel */ buildstate->bs_leader = ginleader; + /* + * Wait for workers to attach before joining the scan, so startup failures + * are reported before we wait for their snapshots. The leader must keep + * its initial snapshot until every worker has restored it. + */ + WaitForParallelWorkersToAttach(pcxt); + if (isconcurrent) + table_parallelscan_wait_restored(ParallelTableScanFromGinBuildShared(ginshared), + pcxt->nworkers_launched); + /* Join heap scan ourselves */ if (leaderparticipates) _gin_leader_participate_as_worker(buildstate, heap, index); - - /* - * Caller needs to wait for all launched workers when we return. Make - * sure that the failure-to-start case will not hang forever. - */ - WaitForParallelWorkersToAttach(pcxt); } /* @@ -1127,9 +1128,6 @@ _gin_end_parallel(GinLeader *ginleader, GinBuildState *state) 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(); } @@ -1800,7 +1798,7 @@ _gin_parallel_merge(GinBuildState *state) /* * Returns size of shared memory required to store state for a parallel - * gin index build based on the snapshot its parallel scan will use. + * gin index build. */ static Size _gin_parallel_estimate_shared(Relation heap, Snapshot snapshot) @@ -2030,6 +2028,7 @@ _gin_parallel_scan_and_build(GinBuildState *state, { SortCoordinate coordinate; TableScanDesc scan; + ParallelTableScanDesc pscan; double reltuples; IndexInfo *indexInfo; @@ -2058,11 +2057,13 @@ _gin_parallel_scan_and_build(GinBuildState *state, TUPLESORT_NONE); /* Join parallel scan */ + pscan = ParallelTableScanFromGinBuildShared(ginshared); indexInfo = BuildIndexInfo(index); indexInfo->ii_Concurrent = ginshared->isconcurrent; + indexInfo->ii_ResetSnapshot = pscan->phs_reset_snapshot; scan = table_beginscan_parallel(heap, - ParallelTableScanFromGinBuildShared(ginshared), + pscan, SO_NONE); reltuples = table_index_build_scan(heap, index, indexInfo, true, progress, diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index c374ffa1ae..c871d32899 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -1220,13 +1220,17 @@ heap_beginscan(Relation relation, Snapshot snapshot, if (flags & SO_RESET_SNAPSHOT) { - /* Not for parallel scans yet; see index_build_resets_snapshots(). */ - Assert(parallel_scan == NULL); - /* The scan snapshot is required to be the active one. */ Assert(GetActiveSnapshot() == snapshot); Assert(!(flags & SO_TEMP_SNAPSHOT)); - INJECTION_POINT("heap_beginscan_reset_snapshot", NULL); +#ifdef USE_INJECTION_POINTS + /* + * Parallel scans fire table_parallelscan_initialize_reset_snapshot + * once during initialization, rather than once per participant here. + */ + if (parallel_scan == NULL) + INJECTION_POINT("heap_beginscan_reset_snapshot", NULL); +#endif } /* diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index 7c406e3959..f1bad64758 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -1255,6 +1255,9 @@ heapam_index_build_range_scan(Relation heapRelation, Assert(!IsBootstrapProcessingMode()); Assert(allow_sync); snapshot = scan->rs_snapshot; + /* Workers copied the flag from the shared scan descriptor. */ + Assert(indexInfo->ii_ResetSnapshot == + ((scan->rs_flags & SO_RESET_SNAPSHOT) != 0)); } hscan = (HeapScanDesc) scan; diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c index 756dfa3dcf..6f9db97d99 100644 --- a/src/backend/access/nbtree/nbtsort.c +++ b/src/backend/access/nbtree/nbtsort.c @@ -87,6 +87,7 @@ typedef struct BTSpool Relation index; bool isunique; bool nulls_not_distinct; + bool reset_snapshot; /* from ii_ResetSnapshot; leader only */ } BTSpool; /* @@ -383,6 +384,7 @@ _bt_spools_heapscan(Relation heap, Relation index, BTBuildState *buildstate, btspool->index = index; btspool->isunique = indexInfo->ii_Unique; btspool->nulls_not_distinct = indexInfo->ii_NullsNotDistinct; + btspool->reset_snapshot = indexInfo->ii_ResetSnapshot; /* Save as primary spool */ buildstate->spool = btspool; @@ -1411,6 +1413,7 @@ _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request) WalUsage *walusage; BufferUsage *bufferusage; bool leaderparticipates = true; + bool reset_snapshot; int querylen; #ifdef DISABLE_LEADER_PARTICIPATION @@ -1428,15 +1431,19 @@ _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request) scantuplesortstates = leaderparticipates ? request + 1 : request; + reset_snapshot = buildstate->spool->reset_snapshot; + /* - * 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. + * Use SnapshotAny for non-concurrent builds so the scan can include + * RECENTLY_DEAD tuples. Concurrent builds that reset snapshots use the + * active snapshot from index_build(), which parallel initialization + * serializes for the workers. Register a fixed MVCC snapshot for + * concurrent unique builds, which do not support resets yet. */ if (!isconcurrent) snapshot = SnapshotAny; + else if (reset_snapshot) + snapshot = InvalidSnapshot; else snapshot = RegisterSnapshot(GetTransactionSnapshot()); @@ -1492,7 +1499,7 @@ _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)) + if (snapshot != InvalidSnapshot && IsMVCCSnapshot(snapshot)) UnregisterSnapshot(snapshot); DestroyParallelContext(pcxt); ExitParallelMode(); @@ -1519,7 +1526,8 @@ _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request) btshared->brokenhotchain = false; table_parallelscan_initialize(btspool->heap, ParallelTableScanFromBTShared(btshared), - snapshot); + snapshot, + reset_snapshot); /* * Store shared tuplesort-private state, for which we reserved space. @@ -1593,15 +1601,19 @@ _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request) /* Save leader state now that it's clear build will be parallel */ buildstate->btleader = btleader; + /* + * Wait for workers to attach before joining the scan, so startup failures + * are reported before we wait for their snapshots. The leader must keep + * its initial snapshot until every worker has restored it. + */ + WaitForParallelWorkersToAttach(pcxt); + if (reset_snapshot) + table_parallelscan_wait_restored(ParallelTableScanFromBTShared(btshared), + pcxt->nworkers_launched); + /* Join heap scan ourselves */ if (leaderparticipates) _bt_leader_participate_as_worker(buildstate); - - /* - * Caller needs to wait for all launched workers when we return. Make - * sure that the failure-to-start case will not hang forever. - */ - WaitForParallelWorkersToAttach(pcxt); } /* @@ -1623,15 +1635,16 @@ _bt_end_parallel(BTLeader *btleader) InstrAccumParallelQuery(&btleader->bufferusage[i], &btleader->walusage[i]); /* Free last reference to MVCC snapshot, if one was used */ - if (IsMVCCSnapshot(btleader->snapshot)) + if (btleader->snapshot != InvalidSnapshot && 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. + * Estimate shared memory for a parallel btree build. Pass InvalidSnapshot + * for scans that reset snapshots, since they use the snapshot serialized by + * parallel initialization. */ static Size _bt_parallel_estimate_shared(Relation heap, Snapshot snapshot) @@ -1873,6 +1886,7 @@ _bt_parallel_scan_and_sort(BTSpool *btspool, BTSpool *btspool2, SortCoordinate coordinate; BTBuildState buildstate; TableScanDesc scan; + ParallelTableScanDesc pscan; double reltuples; IndexInfo *indexInfo; @@ -1925,10 +1939,12 @@ _bt_parallel_scan_and_sort(BTSpool *btspool, BTSpool *btspool2, buildstate.btleader = NULL; /* Join parallel scan */ + pscan = ParallelTableScanFromBTShared(btshared); indexInfo = BuildIndexInfo(btspool->index); indexInfo->ii_Concurrent = btshared->isconcurrent; + indexInfo->ii_ResetSnapshot = pscan->phs_reset_snapshot; scan = table_beginscan_parallel(btspool->heap, - ParallelTableScanFromBTShared(btshared), + pscan, SO_NONE); reltuples = table_index_build_scan(btspool->heap, btspool->index, indexInfo, true, progress, _bt_build_callback, diff --git a/src/backend/access/table/tableam.c b/src/backend/access/table/tableam.c index 3a1ff9a1d4..6d7ed5d12b 100644 --- a/src/backend/access/table/tableam.c +++ b/src/backend/access/table/tableam.c @@ -30,6 +30,8 @@ #include "storage/bufmgr.h" #include "storage/shmem.h" #include "storage/smgr.h" +#include "utils/injection_point.h" +#include "utils/wait_event.h" /* * Constants to control the behavior of block allocation to parallel workers @@ -132,10 +134,10 @@ table_parallelscan_estimate(Relation rel, Snapshot snapshot) { Size sz = 0; - if (IsMVCCSnapshot(snapshot)) + if (snapshot != InvalidSnapshot && IsMVCCSnapshot(snapshot)) sz = add_size(sz, EstimateSnapshotSpace(snapshot)); else - Assert(snapshot == SnapshotAny); + Assert(snapshot == SnapshotAny || snapshot == InvalidSnapshot); sz = add_size(sz, rel->rd_tableam->parallelscan_estimate(rel)); @@ -144,24 +146,60 @@ table_parallelscan_estimate(Relation rel, Snapshot snapshot) void table_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan, - Snapshot snapshot) + Snapshot snapshot, bool reset_snapshot) { Size snapshot_off = rel->rd_tableam->parallelscan_initialize(rel, pscan); pscan->phs_snapshot_off = snapshot_off; - if (IsMVCCSnapshot(snapshot)) + /* + * Serialize a fixed MVCC snapshot for ordinary scans. Scans that reset + * snapshots use each participant's active snapshot instead. + */ + if (reset_snapshot) + { + Assert(snapshot == InvalidSnapshot); + pscan->phs_snapshot_any = false; + pscan->phs_reset_snapshot = true; + pg_atomic_init_u32(&pscan->phs_nrestored, 0); + ConditionVariableInit(&pscan->phs_restored_cv); + INJECTION_POINT("table_parallelscan_initialize_reset_snapshot", NULL); + } + else if (IsMVCCSnapshot(snapshot)) { SerializeSnapshot(snapshot, (char *) pscan + pscan->phs_snapshot_off); pscan->phs_snapshot_any = false; + pscan->phs_reset_snapshot = false; } else { Assert(snapshot == SnapshotAny); + Assert(!reset_snapshot); pscan->phs_snapshot_any = true; + pscan->phs_reset_snapshot = false; } } +/* + * Wait until nparticipants have joined the scan with their own snapshots. + * Until then, the leader must retain the snapshot serialized for workers. + * + * Call WaitForParallelWorkersToAttach() first so a worker startup failure + * reports an error instead of leaving this wait blocked. + */ +void +table_parallelscan_wait_restored(ParallelTableScanDesc pscan, + int nparticipants) +{ + Assert(pscan->phs_reset_snapshot); + + ConditionVariablePrepareToSleep(&pscan->phs_restored_cv); + while (pg_atomic_read_u32(&pscan->phs_nrestored) < (uint32) nparticipants) + ConditionVariableSleep(&pscan->phs_restored_cv, + WAIT_EVENT_PARALLEL_CREATE_INDEX_SNAPSHOT); + ConditionVariableCancelSleep(); +} + TableScanDesc table_beginscan_parallel(Relation relation, ParallelTableScanDesc pscan, uint32 flags) @@ -172,7 +210,20 @@ table_beginscan_parallel(Relation relation, ParallelTableScanDesc pscan, Assert(RelFileLocatorEquals(relation->rd_locator, pscan->phs_locator)); - if (!pscan->phs_snapshot_any) + /* + * Scans that reset snapshots start with the active snapshot. + */ + if (pscan->phs_reset_snapshot) + { + Assert(ActiveSnapshotSet()); + internal_flags |= SO_RESET_SNAPSHOT; + snapshot = GetActiveSnapshot(); + + /* Tell table_parallelscan_wait_restored() we joined the scan. */ + pg_atomic_add_fetch_u32(&pscan->phs_nrestored, 1); + ConditionVariableBroadcast(&pscan->phs_restored_cv); + } + else if (!pscan->phs_snapshot_any) { /* Snapshot was serialized -- restore it */ snapshot = RestoreSnapshot((char *) pscan + pscan->phs_snapshot_off); @@ -203,6 +254,7 @@ table_beginscan_parallel_tidrange(Relation relation, /* disable syncscan in parallel tid range scan. */ pscan->phs_syncscan = false; + Assert(!pscan->phs_reset_snapshot); if (!pscan->phs_snapshot_any) { /* Snapshot was serialized -- restore it */ diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c index e87ddfa0ac..c8b0684dd7 100644 --- a/src/backend/catalog/index.c +++ b/src/backend/catalog/index.c @@ -3125,14 +3125,13 @@ index_update_stats(Relation rel, * The index AM must opt in with amcanresetsnapshot. The scan decides whether * to perform resets; see SO_RESET_SNAPSHOT. * - * Only serial, non-unique builds support resets at this point. + * Unique builds do not support resets yet. */ static bool index_build_resets_snapshots(Relation indexRelation, IndexInfo *indexInfo) { return indexInfo->ii_Concurrent && !indexInfo->ii_Unique && - indexInfo->ii_ParallelWorkers == 0 && indexRelation->rd_indam->amcanresetsnapshot; } diff --git a/src/backend/executor/nodeSeqscan.c b/src/backend/executor/nodeSeqscan.c index b8c528ca08..711af493fa 100644 --- a/src/backend/executor/nodeSeqscan.c +++ b/src/backend/executor/nodeSeqscan.c @@ -404,7 +404,8 @@ ExecSeqScanInitializeDSM(SeqScanState *node, pscan = shm_toc_allocate(pcxt->toc, node->pscan_len); table_parallelscan_initialize(node->ss.ss_currentRelation, pscan, - estate->es_snapshot); + estate->es_snapshot, + false); shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id, pscan); node->ss.ss_currentScanDesc = diff --git a/src/backend/executor/nodeTidrangescan.c b/src/backend/executor/nodeTidrangescan.c index b387ed6c30..07e575164a 100644 --- a/src/backend/executor/nodeTidrangescan.c +++ b/src/backend/executor/nodeTidrangescan.c @@ -488,7 +488,8 @@ ExecTidRangeScanInitializeDSM(TidRangeScanState *node, ParallelContext *pcxt) pscan = shm_toc_allocate(pcxt->toc, node->trss_pscanlen); table_parallelscan_initialize(node->ss.ss_currentRelation, pscan, - estate->es_snapshot); + estate->es_snapshot, + false); shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id, pscan); node->ss.ss_currentScanDesc = table_beginscan_parallel_tidrange(node->ss.ss_currentRelation, diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt index 3d366fd111..b5aa62348d 100644 --- a/src/backend/utils/activity/wait_event_names.txt +++ b/src/backend/utils/activity/wait_event_names.txt @@ -145,6 +145,7 @@ MESSAGE_QUEUE_SEND "Waiting to send bytes to a shared message queue." MULTIXACT_CREATION "Waiting for a multixact creation to complete." PARALLEL_BITMAP_SCAN "Waiting for parallel bitmap scan to become initialized." PARALLEL_CREATE_INDEX_SCAN "Waiting for parallel CREATE INDEX workers to finish heap scan." +PARALLEL_CREATE_INDEX_SNAPSHOT "Waiting for parallel CREATE INDEX workers to restore the scan snapshot." PARALLEL_FINISH "Waiting for parallel workers to finish computing." PROCARRAY_GROUP_UPDATE "Waiting for the group leader to clear the transaction ID at transaction end." PROC_SIGNAL_BARRIER "Waiting for a barrier event to be processed by all backends." diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c index 84d8a4e001..f065792b94 100644 --- a/src/backend/utils/time/snapmgr.c +++ b/src/backend/utils/time/snapmgr.c @@ -429,6 +429,9 @@ SnapshotRotationDue(SnapshotRotation *rotation) * * Push the new snapshot and unregister it, leaving the active-stack entry as * the operation's only reference for the next reset. + * + * Parallel builds allow participants to replace their snapshots + * independently; see table_parallelscan_initialize(). */ void RotateSnapshot(SnapshotRotation *rotation) @@ -445,8 +448,6 @@ RotateSnapshot(SnapshotRotation *rotation) Assert(!HistoricSnapshotActive()); /* The operation is running, so the transaction snapshot was taken. */ Assert(FirstSnapshotSet); - /* Not in parallel mode yet; see index_build_resets_snapshots(). */ - Assert(!IsInParallelMode()); newsnap = RegisterSnapshot(GetSnapshotData(&SecondarySnapshotData)); diff --git a/src/include/access/relscan.h b/src/include/access/relscan.h index d761410473..5edc305393 100644 --- a/src/include/access/relscan.h +++ b/src/include/access/relscan.h @@ -19,6 +19,7 @@ #include "access/sdir.h" #include "nodes/tidbitmap.h" #include "port/atomics.h" +#include "storage/condition_variable.h" #include "storage/relfilelocator.h" #include "utils/relcache.h" @@ -88,7 +89,12 @@ typedef struct ParallelTableScanDescData RelFileLocator phs_locator; /* physical relation to scan */ bool phs_syncscan; /* report location to syncscan logic? */ bool phs_snapshot_any; /* SnapshotAny, not phs_snapshot_data? */ + bool phs_reset_snapshot; /* use SO_RESET_SNAPSHOT? */ Size phs_snapshot_off; /* data for snapshot */ + /* remaining fields are used only when phs_reset_snapshot */ + pg_atomic_uint32 phs_nrestored; /* # participants that joined the scan + * with their own snapshot */ + ConditionVariable phs_restored_cv; /* signaled on phs_nrestored change */ } ParallelTableScanDescData; typedef struct ParallelTableScanDescData *ParallelTableScanDesc; diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h index 4e9b7208be..be3f93c984 100644 --- a/src/include/access/tableam.h +++ b/src/include/access/tableam.h @@ -1200,10 +1200,23 @@ extern Size table_parallelscan_estimate(Relation rel, Snapshot snapshot); * relation. `pscan` needs to be sized according to parallelscan_estimate() * for the same relation. Call this just once in the leader process; then, * individual workers attach via table_beginscan_parallel. + * + * `reset_snapshot` indicates that the scan resets its snapshot periodically; + * `snapshot` must then be InvalidSnapshot, and every participant starts with + * its active snapshot instead. */ extern void table_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan, - Snapshot snapshot); + Snapshot snapshot, + bool reset_snapshot); + +/* + * Wait for nparticipants to join a scan with their own snapshots before the + * leader releases the snapshot serialized for workers. Call + * WaitForParallelWorkersToAttach() first. + */ +extern void table_parallelscan_wait_restored(ParallelTableScanDesc pscan, + int nparticipants); /* * Begin a parallel scan. `pscan` needs to have been initialized with diff --git a/src/test/modules/injection_points/expected/cic_reset_snapshots.out b/src/test/modules/injection_points/expected/cic_reset_snapshots.out index 972e7836fa..78b6d56c35 100644 --- a/src/test/modules/injection_points/expected/cic_reset_snapshots.out +++ b/src/test/modules/injection_points/expected/cic_reset_snapshots.out @@ -25,6 +25,12 @@ SELECT injection_points_attach('heap_beginscan_reset_snapshot', 'notice'); (1 row) +SELECT injection_points_attach('table_parallelscan_initialize_reset_snapshot', 'notice'); + injection_points_attach +------------------------- + +(1 row) + CREATE SCHEMA cic_reset_snap; -- Pad rows to fit exactly seven per page at any block size. A scan of -- 35 rows then resets its snapshot five times with a zero interval. @@ -157,10 +163,14 @@ CREATE UNIQUE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl(i); REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; DROP INDEX CONCURRENTLY cic_reset_snap.idx; CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl(i); +NOTICE: notice triggered for injection point table_parallelscan_initialize_reset_snapshot REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; +NOTICE: notice triggered for injection point table_parallelscan_initialize_reset_snapshot DROP INDEX CONCURRENTLY cic_reset_snap.idx; CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl(MOD(i, 2), j) WHERE MOD(i, 2) = 0; +NOTICE: notice triggered for injection point table_parallelscan_initialize_reset_snapshot REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; +NOTICE: notice triggered for injection point table_parallelscan_initialize_reset_snapshot DROP INDEX CONCURRENTLY cic_reset_snap.idx; CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl(i, j) WHERE cic_reset_snap.predicate_stable(i); NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot @@ -175,17 +185,40 @@ NOTICE: notice triggered for injection point heap_reset_scan_snapshot_xid_assig NOTICE: notice triggered for injection point heap_reset_scan_snapshot_xid_assigned NOTICE: notice triggered for injection point heap_reset_scan_snapshot_xid_assigned DROP INDEX CONCURRENTLY cic_reset_snap.idx; +-- Constant folding of the predicate assigns an xid before the build. +-- Parallel work distribution makes the leader's reset count variable, +-- so do not check the count. +SELECT injection_points_detach('heap_reset_scan_snapshot_xid_assigned'); + injection_points_detach +------------------------- + +(1 row) + CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl(i, j) WHERE cic_reset_snap.predicate_stable_no_param(); +NOTICE: notice triggered for injection point table_parallelscan_initialize_reset_snapshot REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; +NOTICE: notice triggered for injection point table_parallelscan_initialize_reset_snapshot DROP INDEX CONCURRENTLY cic_reset_snap.idx; +SELECT injection_points_attach('heap_reset_scan_snapshot_xid_assigned', 'notice'); + injection_points_attach +------------------------- + +(1 row) + CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl(i DESC NULLS LAST); +NOTICE: notice triggered for injection point table_parallelscan_initialize_reset_snapshot REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; +NOTICE: notice triggered for injection point table_parallelscan_initialize_reset_snapshot DROP INDEX CONCURRENTLY cic_reset_snap.idx; CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl USING BRIN(i); +NOTICE: notice triggered for injection point table_parallelscan_initialize_reset_snapshot REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; +NOTICE: notice triggered for injection point table_parallelscan_initialize_reset_snapshot DROP INDEX CONCURRENTLY cic_reset_snap.idx; CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl USING GIN((ARRAY[i, j])); +NOTICE: notice triggered for injection point table_parallelscan_initialize_reset_snapshot REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; +NOTICE: notice triggered for injection point table_parallelscan_initialize_reset_snapshot DROP INDEX CONCURRENTLY cic_reset_snap.idx; BEGIN TRANSACTION; CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl(i); @@ -231,4 +264,10 @@ SELECT injection_points_detach('heap_beginscan_reset_snapshot'); (1 row) +SELECT injection_points_detach('table_parallelscan_initialize_reset_snapshot'); + injection_points_detach +------------------------- + +(1 row) + DROP EXTENSION injection_points; diff --git a/src/test/modules/injection_points/sql/cic_reset_snapshots.sql b/src/test/modules/injection_points/sql/cic_reset_snapshots.sql index a0da9c33e0..23939f5412 100644 --- a/src/test/modules/injection_points/sql/cic_reset_snapshots.sql +++ b/src/test/modules/injection_points/sql/cic_reset_snapshots.sql @@ -6,7 +6,7 @@ SET concurrent_index_reset_snapshot_interval = 0; SELECT injection_points_attach('heap_reset_scan_snapshot_xmin_pinned', 'error'); SELECT injection_points_attach('heap_reset_scan_snapshot_xid_assigned', 'notice'); SELECT injection_points_attach('heap_beginscan_reset_snapshot', 'notice'); - +SELECT injection_points_attach('table_parallelscan_initialize_reset_snapshot', 'notice'); CREATE SCHEMA cic_reset_snap; -- Pad rows to fit exactly seven per page at any block size. A scan of @@ -116,9 +116,14 @@ CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl(i, j) WHERE cic_reset_snap.p REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; DROP INDEX CONCURRENTLY cic_reset_snap.idx; +-- Constant folding of the predicate assigns an xid before the build. +-- Parallel work distribution makes the leader's reset count variable, +-- so do not check the count. +SELECT injection_points_detach('heap_reset_scan_snapshot_xid_assigned'); CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl(i, j) WHERE cic_reset_snap.predicate_stable_no_param(); REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; DROP INDEX CONCURRENTLY cic_reset_snap.idx; +SELECT injection_points_attach('heap_reset_scan_snapshot_xid_assigned', 'notice'); CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl(i DESC NULLS LAST); REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; @@ -155,5 +160,6 @@ SELECT injection_points_detach('heap_reset_scan_snapshot_swap'); DROP SCHEMA cic_reset_snap CASCADE; SELECT injection_points_detach('heap_beginscan_reset_snapshot'); +SELECT injection_points_detach('table_parallelscan_initialize_reset_snapshot'); DROP EXTENSION injection_points; -- 2.43.0