From 441faa3c13d8fa3e7e04fcf38023af431f821ae5 Mon Sep 17 00:00:00 2001 From: Mikhail Nikalayeu Date: Sun, 11 Oct 2026 20:05:22 +0200 Subject: [PATCH v7 7/7] Reset snapshots after the heap scan in concurrent index builds After the heap scan, merging and writing index tuples can still take a long time while holding the scan's last snapshot and pinning xmin. Add index_build_rotate_snapshot() and call it between tuples in the remaining btree, hash, GIN, BRIN, and GiST build loops. Use the same reset interval as the scan and keep an active snapshot available for support functions that execute SQL and heap lookups that need xmin protection. Sorting and intermediate merges inside tuplesort still hold one snapshot for their duration. The validation phase also keeps its fixed snapshot. Test resets after the scan, including buffered GiST builds and opclasses whose support functions execute SQL in serial and parallel builds. Reviewed-by: Matthias van de Meent --- doc/src/sgml/config.sgml | 12 +- doc/src/sgml/indexam.sgml | 6 + src/backend/access/brin/brin.c | 20 +-- src/backend/access/gin/gininsert.c | 28 +++-- src/backend/access/gist/gistbuild.c | 11 ++ src/backend/access/hash/hash.c | 2 +- src/backend/access/hash/hashsort.c | 4 +- src/backend/access/heap/heapam.c | 8 +- src/backend/access/nbtree/nbtsort.c | 11 +- src/backend/catalog/index.c | 28 +++++ src/backend/utils/misc/guc_parameters.dat | 2 +- src/include/access/hash.h | 3 +- src/include/catalog/index.h | 1 + src/include/nodes/execnodes.h | 8 +- .../expected/cic_reset_snapshots.out | 118 +++++++++++++++++- .../sql/cic_reset_snapshots.sql | 95 ++++++++++++++ 16 files changed, 318 insertions(+), 39 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 58c55299655..27604e58093 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -12783,11 +12783,13 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' - Sets the minimum time between snapshot resets during the first heap - scan of CREATE INDEX CONCURRENTLY and - REINDEX CONCURRENTLY. The scan resets its snapshot - at the next page boundary after this interval, allowing the xmin - horizon to advance. + Sets the minimum time between snapshot resets during + CREATE INDEX CONCURRENTLY and + REINDEX CONCURRENTLY, allowing the xmin horizon to + advance. The first heap scan resets its snapshot at page boundaries. + Later build phases check for resets between tuples. Sorting and + intermediate merges inside tuplesort do not reset the snapshot, and + the validation scan keeps a fixed snapshot. If no unit is specified, the value is in milliseconds. The default is 100ms. A value of 0 resets at every opportunity, including the start of the scan, and is useful diff --git a/doc/src/sgml/indexam.sgml b/doc/src/sgml/indexam.sgml index d110b2f636c..e8441084b2a 100644 --- a/doc/src/sgml/indexam.sgml +++ b/doc/src/sgml/indexam.sgml @@ -326,6 +326,12 @@ ambuild (Relation heapRelation, scan's snapshot reset protocol. + + After the scan, call index_build_rotate_snapshot + between tuples to continue resets at the same interval. Do not call it + while the table scan is open, since the scan still uses the active snapshot. + + void diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c index 3ee9fd94d4f..13590e9a535 100644 --- a/src/backend/access/brin/brin.c +++ b/src/backend/access/brin/brin.c @@ -231,7 +231,7 @@ static void brin_vacuum_scan(Relation idxrel, BufferAccessStrategy strategy); static bool add_values_to_range(Relation idxRel, BrinDesc *bdesc, BrinMemTuple *dtup, const Datum *values, const bool *nulls); static bool check_null_keys(BrinValues *bval, ScanKey *nullkeys, int nnullkeys); -static void brin_fill_empty_ranges(BrinBuildState *state, +static void brin_fill_empty_ranges(BrinBuildState *state, IndexInfo *indexInfo, BlockNumber prevRange, BlockNumber nextRange); /* parallel index builds */ @@ -240,7 +240,7 @@ static void _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Rela 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_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, @@ -1304,7 +1304,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); } @@ -1335,7 +1335,7 @@ brinbuild(Relation heap, Relation index, IndexInfo *indexInfo) * index with a predicate like WHERE (nonnull_column IS NULL), or * other very selective predicates. */ - brin_fill_empty_ranges(state, + brin_fill_empty_ranges(state, indexInfo, state->bs_currRangeStart, state->bs_maxRangeStart); } @@ -2715,7 +2715,7 @@ _brin_parallel_heapscan(BrinBuildState *state) * Returns the total number of heap tuples scanned. */ static double -_brin_parallel_merge(BrinBuildState *state) +_brin_parallel_merge(BrinBuildState *state, IndexInfo *indexInfo) { BrinTuple *btup; BrinMemTuple *memtuple = NULL; @@ -2757,6 +2757,8 @@ _brin_parallel_merge(BrinBuildState *state) */ while ((btup = tuplesort_getbrintuple(state->bs_sortstate, &tuplen, true)) != NULL) { + index_build_rotate_snapshot(indexInfo); + /* Ranges should be multiples of pages_per_range for the index. */ Assert(btup->bt_blkno % state->bs_leader->brinshared->pagesPerRange == 0); @@ -2822,7 +2824,7 @@ _brin_parallel_merge(BrinBuildState *state) } /* Fill empty ranges for all ranges missing in the tuplesort. */ - brin_fill_empty_ranges(state, prevblkno, btup->bt_blkno); + brin_fill_empty_ranges(state, indexInfo, prevblkno, btup->bt_blkno); prevblkno = btup->bt_blkno; } @@ -2845,7 +2847,7 @@ _brin_parallel_merge(BrinBuildState *state) } /* Fill empty ranges at the end, for all ranges missing in the tuplesort. */ - brin_fill_empty_ranges(state, prevblkno, state->bs_maxRangeStart); + brin_fill_empty_ranges(state, indexInfo, prevblkno, state->bs_maxRangeStart); /* * Switch back to the original memory context, and destroy the one we @@ -3093,7 +3095,7 @@ brin_build_empty_tuple(BrinBuildState *state, BlockNumber blkno) * The empty tuple is built only once, and then reused for all future calls. */ static void -brin_fill_empty_ranges(BrinBuildState *state, +brin_fill_empty_ranges(BrinBuildState *state, IndexInfo *indexInfo, BlockNumber prevRange, BlockNumber nextRange) { BlockNumber blkno; @@ -3107,6 +3109,8 @@ brin_fill_empty_ranges(BrinBuildState *state, /* Generate empty ranges until we hit the next non-empty range. */ while (blkno < nextRange) { + index_build_rotate_snapshot(indexInfo); + /* Did we already build the empty tuple? If not, do it now. */ brin_build_empty_tuple(state, blkno); diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c index 95a601bc582..8b6f8afa410 100644 --- a/src/backend/access/gin/gininsert.c +++ b/src/backend/access/gin/gininsert.c @@ -186,7 +186,7 @@ static void _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relati 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_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, @@ -497,9 +497,12 @@ ginBuildCallback(Relation index, ItemPointer tid, Datum *values, * We don't need to worry about overflowing maintenance_work_mem. We can't * build chunks larger than work_mem, and that limit was set so that workers * produce sufficiently small chunks. + * + * Pass NULL for indexInfo while the heap scan is open. */ static void -ginFlushBuildState(GinBuildState *buildstate, Relation index) +ginFlushBuildState(GinBuildState *buildstate, Relation index, + IndexInfo *indexInfo) { ItemPointerData *list; Datum key; @@ -523,6 +526,9 @@ ginFlushBuildState(GinBuildState *buildstate, Relation index) /* start of the chunk */ uint32 offset = 0; + if (indexInfo) + index_build_rotate_snapshot(indexInfo); + /* split the entry into smaller chunk with up to maxlen items */ while (offset < nlist) { @@ -591,7 +597,7 @@ ginBuildCallbackParallel(Relation index, ItemPointer tid, Datum *values, * must have wrapped around. */ if (ItemPointerCompare(tid, &buildstate->tid) < 0) - ginFlushBuildState(buildstate, index); + ginFlushBuildState(buildstate, index, NULL); /* remember the TID we're about to process */ buildstate->tid = *tid; @@ -606,7 +612,7 @@ ginBuildCallbackParallel(Relation index, ItemPointer tid, Datum *values, * the other half is used for the tuplesort. */ if (buildstate->accum.allocatedMemory >= buildstate->work_mem * (Size) 1024) - ginFlushBuildState(buildstate, index); + ginFlushBuildState(buildstate, index, NULL); MemoryContextSwitchTo(oldCtx); } @@ -746,7 +752,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); } @@ -767,6 +773,7 @@ ginbuild(Relation heap, Relation index, IndexInfo *indexInfo) { /* there could be many entries, so be willing to abort here */ CHECK_FOR_INTERRUPTS(); + index_build_rotate_snapshot(indexInfo); ginEntryInsert(&buildstate.ginstate, attnum, key, category, list, nlist, &buildstate.buildStats); } @@ -1643,7 +1650,7 @@ GinBufferCanAddKey(GinBuffer *buffer, GinTuple *tup) * Returns the total number of heap tuples scanned. */ static double -_gin_parallel_merge(GinBuildState *state) +_gin_parallel_merge(GinBuildState *state, IndexInfo *indexInfo) { GinTuple *tup; Size tuplen; @@ -1704,6 +1711,8 @@ _gin_parallel_merge(GinBuildState *state) { MemoryContext oldCtx; + index_build_rotate_snapshot(indexInfo); + CHECK_FOR_INTERRUPTS(); /* @@ -1846,7 +1855,7 @@ _gin_leader_participate_as_worker(GinBuildState *buildstate, Relation heap, Rela */ static void _gin_process_worker_data(GinBuildState *state, Tuplesortstate *worker_sort, - bool progress) + IndexInfo *indexInfo, bool progress) { GinTuple *tup; Size tuplen; @@ -1881,6 +1890,7 @@ _gin_process_worker_data(GinBuildState *state, Tuplesortstate *worker_sort, { CHECK_FOR_INTERRUPTS(); + index_build_rotate_snapshot(indexInfo); /* * If the buffer can accept the new GIN tuple, just store it there and @@ -2070,14 +2080,14 @@ _gin_parallel_scan_and_build(GinBuildState *state, ginBuildCallbackParallel, state, scan); /* write remaining accumulated entries */ - ginFlushBuildState(state, index); + ginFlushBuildState(state, index, indexInfo); /* * Do the first phase of in-worker processing - sort the data produced by * the callback, and combine them into much larger chunks and place that * into the shared tuplestore for leader to process. */ - _gin_process_worker_data(state, state->bs_worker_sort, progress); + _gin_process_worker_data(state, state->bs_worker_sort, indexInfo, progress); /* sort the GIN tuples built by this worker */ tuplesort_performsort(state->bs_sortstate); diff --git a/src/backend/access/gist/gistbuild.c b/src/backend/access/gist/gistbuild.c index 07bad403b2d..2185c370047 100644 --- a/src/backend/access/gist/gistbuild.c +++ b/src/backend/access/gist/gistbuild.c @@ -38,6 +38,7 @@ #include "access/gist_private.h" #include "access/tableam.h" #include "access/xloginsert.h" +#include "catalog/index.h" #include "miscadmin.h" #include "nodes/execnodes.h" #include "optimizer/optimizer.h" @@ -108,6 +109,9 @@ typedef struct BlockNumber pages_allocated; BulkWriteState *bulkstate; + + IndexInfo *indexInfo; + bool scan_done; } GISTBuildState; #define GIST_SORTED_BUILD_PAGE_NUM 4 @@ -196,6 +200,8 @@ gistbuild(Relation heap, Relation index, IndexInfo *indexInfo) buildstate.indexrel = index; buildstate.heaprel = heap; + buildstate.indexInfo = indexInfo; + buildstate.scan_done = false; buildstate.sortstate = NULL; buildstate.giststate = initGISTstate(index); @@ -321,6 +327,7 @@ gistbuild(Relation heap, Relation index, IndexInfo *indexInfo) if (buildstate.buildMode == GIST_BUFFERING_ACTIVE) { elog(DEBUG1, "all tuples processed, emptying buffers"); + buildstate.scan_done = true; gistEmptyAllBuffers(&buildstate); gistFreeBuildBuffers(buildstate.gfbb); } @@ -419,6 +426,7 @@ gist_indexsortbuild(GISTBuildState *state) */ while ((itup = tuplesort_getindextuple(state->sortstate, true)) != NULL) { + index_build_rotate_snapshot(state->indexInfo); gist_indexsortbuild_levelstate_add(state, levelstate, itup); MemoryContextReset(state->giststate->tempCxt); } @@ -1336,6 +1344,9 @@ gistProcessEmptyingQueue(GISTBuildState *buildstate) if (!gistPopItupFromNodeBuffer(gfbb, emptyingNodeBuffer, &itup)) break; + if (buildstate->scan_done) + index_build_rotate_snapshot(buildstate->indexInfo); + /* * Run it down to the underlying node buffer or leaf page. * diff --git a/src/backend/access/hash/hash.c b/src/backend/access/hash/hash.c index 584fb5b78e1..2a40464c4c1 100644 --- a/src/backend/access/hash/hash.c +++ b/src/backend/access/hash/hash.c @@ -200,7 +200,7 @@ hashbuild(Relation heap, Relation index, IndexInfo *indexInfo) if (buildstate.spool) { /* sort the tuples and insert them into the index */ - _h_indexbuild(buildstate.spool, buildstate.heapRel); + _h_indexbuild(buildstate.spool, buildstate.heapRel, indexInfo); _h_spooldestroy(buildstate.spool); } diff --git a/src/backend/access/hash/hashsort.c b/src/backend/access/hash/hashsort.c index 77bbfaa461b..fc6a1f9f776 100644 --- a/src/backend/access/hash/hashsort.c +++ b/src/backend/access/hash/hashsort.c @@ -26,6 +26,7 @@ #include "postgres.h" #include "access/hash.h" +#include "catalog/index.h" #include "commands/progress.h" #include "miscadmin.h" #include "pgstat.h" @@ -117,7 +118,7 @@ _h_spool(HSpool *hspool, const ItemPointerData *self, const Datum *values, const * create an entire index. */ void -_h_indexbuild(HSpool *hspool, Relation heapRel) +_h_indexbuild(HSpool *hspool, Relation heapRel, IndexInfo *indexInfo) { IndexTuple itup; int64 tups_done = 0; @@ -150,6 +151,7 @@ _h_indexbuild(HSpool *hspool, Relation heapRel) /* allow insertion phase to be interrupted, and track progress */ CHECK_FOR_INTERRUPTS(); + index_build_rotate_snapshot(indexInfo); pgstat_progress_update_param(PROGRESS_CREATEIDX_TUPLES_DONE, ++tups_done); diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index ab4aff2509e..1bce66c363e 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -709,9 +709,6 @@ heap_prepare_pagescan(TableScanDesc sscan) * * The scan uses the active snapshot without registering another copy, * allowing RotateSnapshot() to release it and advance xmin. - * - * The caller leaves it active for the rest of the build and pops it when the - * build ends. */ static inline void heap_reset_scan_snapshot(HeapScanDesc scan) @@ -1494,10 +1491,7 @@ heap_endscan(TableScanDesc sscan) if (scan->rs_base.rs_flags & SO_RESET_SNAPSHOT) { - /* - * Leave the snapshot active for the rest of the build. The caller is - * responsible for popping it. - */ + /* The caller owns the active snapshot. */ Assert(!(scan->rs_base.rs_flags & SO_TEMP_SNAPSHOT)); Assert(GetActiveSnapshot() == sscan->rs_snapshot); } diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c index b3568d77550..74dcddad4ee 100644 --- a/src/backend/access/nbtree/nbtsort.c +++ b/src/backend/access/nbtree/nbtsort.c @@ -262,6 +262,7 @@ typedef struct BTWriteState BulkWriteState *bulkstate; BTScanInsert inskey; /* generic insertion scankey */ BlockNumber btws_pages_alloced; /* # pages allocated */ + IndexInfo *indexInfo; } BTWriteState; @@ -270,7 +271,8 @@ static double _bt_spools_heapscan(Relation heap, Relation index, static void _bt_spooldestroy(BTSpool *btspool); static void _bt_spool(BTSpool *btspool, const ItemPointerData *self, const Datum *values, const bool *isnull); -static void _bt_leafbuild(BTSpool *btspool, BTSpool *btspool2); +static void _bt_leafbuild(BTSpool *btspool, BTSpool *btspool2, + IndexInfo *indexInfo); static void _bt_build_callback(Relation index, ItemPointer tid, Datum *values, bool *isnull, bool tupleIsAlive, void *state); static BulkWriteBuffer _bt_blnewpage(BTWriteState *wstate, uint32 level); @@ -337,7 +339,7 @@ btbuild(Relation heap, Relation index, IndexInfo *indexInfo) * inserting the sorted tuples into btree pages and (3) building the upper * levels. Finally, it may also be necessary to end use of parallelism. */ - _bt_leafbuild(buildstate.spool, buildstate.spool2); + _bt_leafbuild(buildstate.spool, buildstate.spool2, indexInfo); _bt_spooldestroy(buildstate.spool); if (buildstate.spool2) _bt_spooldestroy(buildstate.spool2); @@ -558,7 +560,7 @@ _bt_spool(BTSpool *btspool, const ItemPointerData *self, const Datum *values, co * create an entire btree. */ static void -_bt_leafbuild(BTSpool *btspool, BTSpool *btspool2) +_bt_leafbuild(BTSpool *btspool, BTSpool *btspool2, IndexInfo *indexInfo) { BTWriteState wstate; @@ -583,6 +585,7 @@ _bt_leafbuild(BTSpool *btspool, BTSpool *btspool2) wstate.heap = btspool->heap; wstate.index = btspool->index; + wstate.indexInfo = indexInfo; wstate.inskey = _bt_mkscankey(wstate.index, NULL); /* _bt_mkscankey() won't set allequalimage without metapage */ wstate.inskey->allequalimage = _bt_allequalimage(wstate.index, true); @@ -822,6 +825,7 @@ _bt_buildadd(BTWriteState *wstate, BTPageState *state, IndexTuple itup, * load phase of index creation. */ CHECK_FOR_INTERRUPTS(); + index_build_rotate_snapshot(wstate->indexInfo); nbuf = state->btps_buf; npage = (Page) nbuf; @@ -1279,6 +1283,7 @@ _bt_load(BTWriteState *wstate, BTSpool *btspool, BTSpool *btspool2) * interrupts and snapshot resets here too. */ CHECK_FOR_INTERRUPTS(); + index_build_rotate_snapshot(wstate->indexInfo); } /* diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c index 6ccbec8e0ef..28681f69b59 100644 --- a/src/backend/catalog/index.c +++ b/src/backend/catalog/index.c @@ -74,6 +74,7 @@ #include "utils/builtins.h" #include "utils/fmgroids.h" #include "utils/guc.h" +#include "utils/injection_point.h" #include "utils/inval.h" #include "utils/lsyscache.h" #include "utils/memutils.h" @@ -3132,6 +3133,30 @@ index_build_resets_snapshots(Relation indexRelation, IndexInfo *indexInfo) indexRelation->rd_indam->amcanresetsnapshot; } +/* + * index_build_rotate_snapshot -- reset the snapshot after the heap scan + * + * Call between tuples in later build phases. Do not call while the heap + * scan is open: it still holds a pointer to the active snapshot. + */ +void +index_build_rotate_snapshot(IndexInfo *indexInfo) +{ + SnapshotRotation *rotation = &indexInfo->ii_SnapshotRotation; + + if (!indexInfo->ii_ResetSnapshot) + return; + + if (rotation->snapshot == NULL) + StartSnapshotRotation(rotation); + + if (!SnapshotRotationDue(rotation)) + return; + + RotateSnapshot(rotation); + INJECTION_POINT("index_build_rotate_snapshot", NULL); +} + /* * index_build - invoke access-method-specific index build procedure * @@ -3350,7 +3375,10 @@ index_build(Relation heapRelation, IndexCheckExclusion(heapRelation, indexRelation, indexInfo); if (indexInfo->ii_Concurrent) + { PopActiveSnapshot(); + indexInfo->ii_SnapshotRotation.snapshot = NULL; + } /* Roll back any GUC changes executed by index functions */ AtEOXact_GUC(false, save_nestlevel); diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat index e0a4906b278..4821d980595 100644 --- a/src/backend/utils/misc/guc_parameters.dat +++ b/src/backend/utils/misc/guc_parameters.dat @@ -511,7 +511,7 @@ { name => 'concurrent_index_reset_snapshot_interval', type => 'int', context => 'PGC_SUSET', group => 'DEVELOPER_OPTIONS', short_desc => 'Sets the minimum time between snapshot resets during concurrent index builds.', - long_desc => 'The first heap scan resets its snapshot at the next page boundary after this interval. 0 resets on every page. -1 disables resets.', + long_desc => 'Resets occur at page boundaries during the first heap scan and between tuples in later build phases. 0 resets at every opportunity. -1 disables resets.', flags => 'GUC_NOT_IN_SAMPLE | GUC_UNIT_MS', variable => 'concurrent_index_reset_snapshot_interval', boot_val => '100', diff --git a/src/include/access/hash.h b/src/include/access/hash.h index a8702f0e5ea..16c5c33833c 100644 --- a/src/include/access/hash.h +++ b/src/include/access/hash.h @@ -455,7 +455,8 @@ extern HSpool *_h_spoolinit(Relation heap, Relation index, uint32 num_buckets); extern void _h_spooldestroy(HSpool *hspool); extern void _h_spool(HSpool *hspool, const ItemPointerData *self, const Datum *values, const bool *isnull); -extern void _h_indexbuild(HSpool *hspool, Relation heapRel); +extern void _h_indexbuild(HSpool *hspool, Relation heapRel, + IndexInfo *indexInfo); /* hashutil.c */ extern bool _hash_checkqual(IndexScanDesc scan, IndexTuple itup); diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h index 5add66817cf..1fc39a75844 100644 --- a/src/include/catalog/index.h +++ b/src/include/catalog/index.h @@ -149,6 +149,7 @@ extern void FormIndexDatum(IndexInfo *indexInfo, Datum *values, bool *isnull); +extern void index_build_rotate_snapshot(IndexInfo *indexInfo); extern void index_build(Relation heapRelation, Relation indexRelation, IndexInfo *indexInfo, diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index bd8a99ada85..f3f8b8a789d 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -39,6 +39,7 @@ #include "partitioning/partdefs.h" #include "storage/buf.h" #include "utils/reltrigger.h" +#include "utils/snapmgr.h" /* @@ -168,8 +169,8 @@ typedef struct ExprState * entries for a particular index. Used for both index_build and * retail creation of index entries. * - * ii_Concurrent, ii_ResetSnapshot, ii_BrokenHotChain, and - * ii_ParallelWorkers are used only during index build; they're + * ii_Concurrent, ii_ResetSnapshot, ii_BrokenHotChain, ii_ParallelWorkers + * and ii_SnapshotRotation are used only during index build; they're * conventionally zeroed otherwise. * ---------------- */ @@ -240,6 +241,9 @@ typedef struct IndexInfo /* memory context holding this IndexInfo */ MemoryContext ii_Context; + + /* snapshot resets after the heap scan */ + SnapshotRotation ii_SnapshotRotation; } IndexInfo; /* ---------------- 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 63bdee5e757..a331dbc8c0d 100644 --- a/src/test/modules/injection_points/expected/cic_reset_snapshots.out +++ b/src/test/modules/injection_points/expected/cic_reset_snapshots.out @@ -69,6 +69,29 @@ SELECT injection_points_detach('heap_reset_scan_snapshot_swap'); (1 row) DROP INDEX CONCURRENTLY cic_reset_snap.rotation_probe; +-- After the scan, reset between tuples written to the index. A zero +-- interval makes the count deterministic. +CREATE TABLE cic_reset_snap.phase_probe(i int); +ALTER TABLE cic_reset_snap.phase_probe SET (parallel_workers=0); +INSERT INTO cic_reset_snap.phase_probe SELECT g FROM generate_series(1, 3) g; +SELECT injection_points_attach('index_build_rotate_snapshot', 'notice'); + injection_points_attach +------------------------- + +(1 row) + +CREATE INDEX CONCURRENTLY phase_probe_idx ON cic_reset_snap.phase_probe(i); +NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot +NOTICE: notice triggered for injection point index_build_rotate_snapshot +NOTICE: notice triggered for injection point index_build_rotate_snapshot +NOTICE: notice triggered for injection point index_build_rotate_snapshot +SELECT injection_points_detach('index_build_rotate_snapshot'); + injection_points_detach +------------------------- + +(1 row) + +DROP TABLE cic_reset_snap.phase_probe; CREATE FUNCTION cic_reset_snap.predicate_stable(integer) RETURNS bool IMMUTABLE LANGUAGE plpgsql AS $$ BEGIN @@ -162,6 +185,12 @@ NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot DROP INDEX CONCURRENTLY cic_reset_snap.idx; +-- Check resets while GiST empties node buffers after the scan. A low +-- fill factor gives the index enough levels to use buffering. +CREATE TABLE cic_reset_snap.gist_buf AS SELECT g AS i FROM generate_series(1, 20000) g; +CREATE INDEX CONCURRENTLY gist_buf_idx ON cic_reset_snap.gist_buf USING GIST(point(i, i)) WITH (buffering = on, fillfactor = 10); +NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot +DROP TABLE cic_reset_snap.gist_buf; CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl USING GIN((ARRAY[i, j])); NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; @@ -396,6 +425,85 @@ CREATE INDEX CONCURRENTLY enum_idx ON cic_reset_snap.enum_tbl(c); NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot RESET debug_discard_caches; DROP TABLE cic_reset_snap.enum_tbl; +-- Support functions may execute SQL after the heap scan. Verify that an +-- active snapshot is available, including in parallel workers. +CREATE FUNCTION cic_reset_snap.cmp_sql(int, int) RETURNS int IMMUTABLE +LANGUAGE plpgsql AS $$ +BEGIN + EXECUTE 'SELECT 1'; + RETURN CASE WHEN $1 < $2 THEN -1 WHEN $1 > $2 THEN 1 ELSE 0 END; +END; $$; +CREATE OPERATOR CLASS cic_reset_snap.int4_sql_ops FOR TYPE int4 USING btree AS + OPERATOR 1 <, OPERATOR 2 <=, OPERATOR 3 =, OPERATOR 4 >=, OPERATOR 5 >, + FUNCTION 1 cic_reset_snap.cmp_sql(int, int); +CREATE TABLE cic_reset_snap.opclass_tbl(i int); +ALTER TABLE cic_reset_snap.opclass_tbl SET (parallel_workers=0); +INSERT INTO cic_reset_snap.opclass_tbl SELECT g FROM generate_series(1, 200) g; +CREATE INDEX CONCURRENTLY opclass_idx + ON cic_reset_snap.opclass_tbl(i cic_reset_snap.int4_sql_ops); +NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot +REINDEX INDEX CONCURRENTLY cic_reset_snap.opclass_idx; +NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot +DROP INDEX CONCURRENTLY cic_reset_snap.opclass_idx; +ALTER TABLE cic_reset_snap.opclass_tbl SET (parallel_workers=2); +CREATE INDEX CONCURRENTLY opclass_idx + ON cic_reset_snap.opclass_tbl(i cic_reset_snap.int4_sql_ops); +NOTICE: notice triggered for injection point table_parallelscan_initialize_reset_snapshot +DROP INDEX CONCURRENTLY cic_reset_snap.opclass_idx; +ALTER TABLE cic_reset_snap.opclass_tbl SET (parallel_workers=0); +-- A support function in SQL does not go through SPI. +CREATE FUNCTION cic_reset_snap.cmp_sqllang(int, int) RETURNS int IMMUTABLE +LANGUAGE sql AS $$ + SELECT CASE WHEN $1 < $2 THEN -1 WHEN $1 > $2 THEN 1 ELSE 0 END +$$; +CREATE OPERATOR CLASS cic_reset_snap.int4_sqllang_ops FOR TYPE int4 USING btree AS + OPERATOR 1 <, OPERATOR 2 <=, OPERATOR 3 =, OPERATOR 4 >=, OPERATOR 5 >, + FUNCTION 1 cic_reset_snap.cmp_sqllang(int, int); +CREATE INDEX CONCURRENTLY opclass_idx + ON cic_reset_snap.opclass_tbl(i cic_reset_snap.int4_sqllang_ops); +NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot +DROP INDEX CONCURRENTLY cic_reset_snap.opclass_idx; +-- Committing or aborting a subtransaction in a support function must +-- preserve the build's snapshot. +CREATE FUNCTION cic_reset_snap.cmp_catch(int, int) RETURNS int IMMUTABLE +LANGUAGE plpgsql AS $$ +BEGIN + BEGIN + EXECUTE 'SELECT 1'; + EXCEPTION WHEN OTHERS THEN + NULL; + END; + BEGIN + EXECUTE 'SELECT 1'; + RAISE EXCEPTION 'caught'; + EXCEPTION WHEN OTHERS THEN + NULL; + END; + RETURN CASE WHEN $1 < $2 THEN -1 WHEN $1 > $2 THEN 1 ELSE 0 END; +END; $$; +CREATE OPERATOR CLASS cic_reset_snap.int4_catch_ops FOR TYPE int4 USING btree AS + OPERATOR 1 <, OPERATOR 2 <=, OPERATOR 3 =, OPERATOR 4 >=, OPERATOR 5 >, + FUNCTION 1 cic_reset_snap.cmp_catch(int, int); +CREATE INDEX CONCURRENTLY opclass_idx + ON cic_reset_snap.opclass_tbl(i cic_reset_snap.int4_catch_ops); +NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot +DROP INDEX CONCURRENTLY cic_reset_snap.opclass_idx; +-- Closing a cursor in a support function must preserve the build's snapshot. +CREATE FUNCTION cic_reset_snap.cmp_cursor(int, int) RETURNS int IMMUTABLE +LANGUAGE plpgsql AS $$ +DECLARE c refcursor; +BEGIN + OPEN c FOR SELECT 1; + CLOSE c; + RETURN CASE WHEN $1 < $2 THEN -1 WHEN $1 > $2 THEN 1 ELSE 0 END; +END; $$; +CREATE OPERATOR CLASS cic_reset_snap.int4_cursor_ops FOR TYPE int4 USING btree AS + OPERATOR 1 <, OPERATOR 2 <=, OPERATOR 3 =, OPERATOR 4 >=, OPERATOR 5 >, + FUNCTION 1 cic_reset_snap.cmp_cursor(int, int); +CREATE INDEX CONCURRENTLY opclass_idx + ON cic_reset_snap.opclass_tbl(i cic_reset_snap.int4_cursor_ops); +NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot +DROP TABLE cic_reset_snap.opclass_tbl; -- A cursor left open by a support function retains the scan snapshot. -- The build must tolerate this, even though it keeps xmin pinned. CREATE FUNCTION cic_reset_snap.leak_cursor(i int, at int) RETURNS bool IMMUTABLE @@ -442,12 +550,20 @@ CREATE INDEX CONCURRENTLY cur_idx ON cic_reset_snap.cur_tbl(i) NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot DROP TABLE cic_reset_snap.cur_tbl; DROP SCHEMA cic_reset_snap CASCADE; -NOTICE: drop cascades to 6 other objects +NOTICE: drop cascades to 14 other objects DETAIL: drop cascades to table cic_reset_snap.tbl drop cascades to function cic_reset_snap.predicate_stable(integer) drop cascades to function cic_reset_snap.predicate_stable_no_param() drop cascades to operator family cic_reset_snap.int4_bloom_storage_ops for access method brin drop cascades to type cic_reset_snap.rainbow +drop cascades to function cic_reset_snap.cmp_sql(integer,integer) +drop cascades to operator family cic_reset_snap.int4_sql_ops for access method btree +drop cascades to function cic_reset_snap.cmp_sqllang(integer,integer) +drop cascades to operator family cic_reset_snap.int4_sqllang_ops for access method btree +drop cascades to function cic_reset_snap.cmp_catch(integer,integer) +drop cascades to operator family cic_reset_snap.int4_catch_ops for access method btree +drop cascades to function cic_reset_snap.cmp_cursor(integer,integer) +drop cascades to operator family cic_reset_snap.int4_cursor_ops for access method btree drop cascades to function cic_reset_snap.leak_cursor(integer,integer) SELECT injection_points_detach('heap_beginscan_reset_snapshot'); injection_points_detach 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 3a83b202648..810b5e85471 100644 --- a/src/test/modules/injection_points/sql/cic_reset_snapshots.sql +++ b/src/test/modules/injection_points/sql/cic_reset_snapshots.sql @@ -27,6 +27,16 @@ CREATE INDEX CONCURRENTLY rotation_probe ON cic_reset_snap.tbl(i); SELECT injection_points_detach('heap_reset_scan_snapshot_swap'); DROP INDEX CONCURRENTLY cic_reset_snap.rotation_probe; +-- After the scan, reset between tuples written to the index. A zero +-- interval makes the count deterministic. +CREATE TABLE cic_reset_snap.phase_probe(i int); +ALTER TABLE cic_reset_snap.phase_probe SET (parallel_workers=0); +INSERT INTO cic_reset_snap.phase_probe SELECT g FROM generate_series(1, 3) g; +SELECT injection_points_attach('index_build_rotate_snapshot', 'notice'); +CREATE INDEX CONCURRENTLY phase_probe_idx ON cic_reset_snap.phase_probe(i); +SELECT injection_points_detach('index_build_rotate_snapshot'); +DROP TABLE cic_reset_snap.phase_probe; + CREATE FUNCTION cic_reset_snap.predicate_stable(integer) RETURNS bool IMMUTABLE LANGUAGE plpgsql AS $$ BEGIN @@ -96,6 +106,12 @@ CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl USING SPGIST(point(i, j)); REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; DROP INDEX CONCURRENTLY cic_reset_snap.idx; +-- Check resets while GiST empties node buffers after the scan. A low +-- fill factor gives the index enough levels to use buffering. +CREATE TABLE cic_reset_snap.gist_buf AS SELECT g AS i FROM generate_series(1, 20000) g; +CREATE INDEX CONCURRENTLY gist_buf_idx ON cic_reset_snap.gist_buf USING GIST(point(i, i)) WITH (buffering = on, fillfactor = 10); +DROP TABLE cic_reset_snap.gist_buf; + CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl USING GIN((ARRAY[i, j])); REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; DROP INDEX CONCURRENTLY cic_reset_snap.idx; @@ -253,6 +269,85 @@ CREATE INDEX CONCURRENTLY enum_idx ON cic_reset_snap.enum_tbl(c); RESET debug_discard_caches; DROP TABLE cic_reset_snap.enum_tbl; +-- Support functions may execute SQL after the heap scan. Verify that an +-- active snapshot is available, including in parallel workers. +CREATE FUNCTION cic_reset_snap.cmp_sql(int, int) RETURNS int IMMUTABLE +LANGUAGE plpgsql AS $$ +BEGIN + EXECUTE 'SELECT 1'; + RETURN CASE WHEN $1 < $2 THEN -1 WHEN $1 > $2 THEN 1 ELSE 0 END; +END; $$; +CREATE OPERATOR CLASS cic_reset_snap.int4_sql_ops FOR TYPE int4 USING btree AS + OPERATOR 1 <, OPERATOR 2 <=, OPERATOR 3 =, OPERATOR 4 >=, OPERATOR 5 >, + FUNCTION 1 cic_reset_snap.cmp_sql(int, int); + +CREATE TABLE cic_reset_snap.opclass_tbl(i int); +ALTER TABLE cic_reset_snap.opclass_tbl SET (parallel_workers=0); +INSERT INTO cic_reset_snap.opclass_tbl SELECT g FROM generate_series(1, 200) g; +CREATE INDEX CONCURRENTLY opclass_idx + ON cic_reset_snap.opclass_tbl(i cic_reset_snap.int4_sql_ops); +REINDEX INDEX CONCURRENTLY cic_reset_snap.opclass_idx; +DROP INDEX CONCURRENTLY cic_reset_snap.opclass_idx; + +ALTER TABLE cic_reset_snap.opclass_tbl SET (parallel_workers=2); +CREATE INDEX CONCURRENTLY opclass_idx + ON cic_reset_snap.opclass_tbl(i cic_reset_snap.int4_sql_ops); +DROP INDEX CONCURRENTLY cic_reset_snap.opclass_idx; +ALTER TABLE cic_reset_snap.opclass_tbl SET (parallel_workers=0); + +-- A support function in SQL does not go through SPI. +CREATE FUNCTION cic_reset_snap.cmp_sqllang(int, int) RETURNS int IMMUTABLE +LANGUAGE sql AS $$ + SELECT CASE WHEN $1 < $2 THEN -1 WHEN $1 > $2 THEN 1 ELSE 0 END +$$; +CREATE OPERATOR CLASS cic_reset_snap.int4_sqllang_ops FOR TYPE int4 USING btree AS + OPERATOR 1 <, OPERATOR 2 <=, OPERATOR 3 =, OPERATOR 4 >=, OPERATOR 5 >, + FUNCTION 1 cic_reset_snap.cmp_sqllang(int, int); +CREATE INDEX CONCURRENTLY opclass_idx + ON cic_reset_snap.opclass_tbl(i cic_reset_snap.int4_sqllang_ops); +DROP INDEX CONCURRENTLY cic_reset_snap.opclass_idx; + +-- Committing or aborting a subtransaction in a support function must +-- preserve the build's snapshot. +CREATE FUNCTION cic_reset_snap.cmp_catch(int, int) RETURNS int IMMUTABLE +LANGUAGE plpgsql AS $$ +BEGIN + BEGIN + EXECUTE 'SELECT 1'; + EXCEPTION WHEN OTHERS THEN + NULL; + END; + BEGIN + EXECUTE 'SELECT 1'; + RAISE EXCEPTION 'caught'; + EXCEPTION WHEN OTHERS THEN + NULL; + END; + RETURN CASE WHEN $1 < $2 THEN -1 WHEN $1 > $2 THEN 1 ELSE 0 END; +END; $$; +CREATE OPERATOR CLASS cic_reset_snap.int4_catch_ops FOR TYPE int4 USING btree AS + OPERATOR 1 <, OPERATOR 2 <=, OPERATOR 3 =, OPERATOR 4 >=, OPERATOR 5 >, + FUNCTION 1 cic_reset_snap.cmp_catch(int, int); +CREATE INDEX CONCURRENTLY opclass_idx + ON cic_reset_snap.opclass_tbl(i cic_reset_snap.int4_catch_ops); +DROP INDEX CONCURRENTLY cic_reset_snap.opclass_idx; + +-- Closing a cursor in a support function must preserve the build's snapshot. +CREATE FUNCTION cic_reset_snap.cmp_cursor(int, int) RETURNS int IMMUTABLE +LANGUAGE plpgsql AS $$ +DECLARE c refcursor; +BEGIN + OPEN c FOR SELECT 1; + CLOSE c; + RETURN CASE WHEN $1 < $2 THEN -1 WHEN $1 > $2 THEN 1 ELSE 0 END; +END; $$; +CREATE OPERATOR CLASS cic_reset_snap.int4_cursor_ops FOR TYPE int4 USING btree AS + OPERATOR 1 <, OPERATOR 2 <=, OPERATOR 3 =, OPERATOR 4 >=, OPERATOR 5 >, + FUNCTION 1 cic_reset_snap.cmp_cursor(int, int); +CREATE INDEX CONCURRENTLY opclass_idx + ON cic_reset_snap.opclass_tbl(i cic_reset_snap.int4_cursor_ops); +DROP TABLE cic_reset_snap.opclass_tbl; + -- A cursor left open by a support function retains the scan snapshot. -- The build must tolerate this, even though it keeps xmin pinned. CREATE FUNCTION cic_reset_snap.leak_cursor(i int, at int) RETURNS bool IMMUTABLE -- 2.43.0