From 12a55f446cdb520e841320c910dd30e677d0f9e7 Mon Sep 17 00:00:00 2001 From: Mikhail Nikalayeu Date: Mon, 21 Apr 2025 14:11:53 +0200 Subject: [PATCH v39 7/7] Refresh snapshots during concurrent index validation Keep a snapshot active throughout validation and replace it between phases and heap pages, so long validations can advance xmin. Register the new snapshot before releasing the old one. Put snapshot rotation and interval checks in snapmgr.c. Use concurrent_index_reset_snapshot_interval, with a default of 100ms. Zero resets at every opportunity; -1 disables resets. Transaction snapshots keep xmin pinned, so skip resets at REPEATABLE READ and above. Return the newest xmin used for visibility checks, so the caller can wait for older snapshots before marking the index valid. Add tests for resets, disabled resets and cursors holding back xmin, and check reset notices in the SQL test output. --- doc/src/sgml/config.sgml | 27 +++ src/backend/access/heap/README.HOT | 4 +- src/backend/access/heap/heapam.c | 2 + src/backend/access/heap/heapam_handler.c | 27 ++- src/backend/catalog/index.c | 81 +++++-- src/backend/commands/indexcmds.c | 52 +---- src/backend/utils/misc/guc_parameters.dat | 10 + src/backend/utils/time/snapmgr.c | 82 +++++++ src/include/access/tableam.h | 31 ++- src/include/access/transam.h | 15 ++ src/include/catalog/index.h | 4 +- src/include/miscadmin.h | 1 + src/include/nodes/execnodes.h | 12 +- src/include/utils/snapmgr.h | 4 + src/include/utils/snapshot.h | 12 + .../expected/reindex_conc.out | 49 ++++ .../injection_points/sql/reindex_conc.sql | 21 ++ src/test/modules/test_misc/meson.build | 1 + .../test_misc/t/015_cic_validate_snapshot.pl | 218 ++++++++++++++++++ src/test/regress/expected/create_index.out | 3 + src/test/regress/sql/create_index.sql | 4 + 21 files changed, 572 insertions(+), 88 deletions(-) create mode 100644 src/test/modules/test_misc/t/015_cic_validate_snapshot.pl diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 236ee067f40..64065ce538e 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -12725,6 +12725,33 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' + + concurrent_index_reset_snapshot_interval (integer) + + concurrent_index_reset_snapshot_interval configuration parameter + + + + + Sets the minimum time between snapshot resets during validation of + CREATE INDEX CONCURRENTLY and + REINDEX CONCURRENTLY. Resets occur between validation + phases and heap pages, allowing the xmin horizon to advance. Each + index scan, sort, and merge of TID lists keeps its snapshot until it + finishes. + If no unit is specified, the value is in milliseconds. The default + is 100ms. A value of 0 resets at + every opportunity; -1 disables resets. + + + Resets are disabled at REPEATABLE READ and + SERIALIZABLE, where the transaction snapshot keeps + xmin from advancing. Other registered snapshots, such as those held + by cursors, can also hold back xmin. + + + + backtrace_functions (string) diff --git a/src/backend/access/heap/README.HOT b/src/backend/access/heap/README.HOT index b1c797517ee..382fe1723a5 100644 --- a/src/backend/access/heap/README.HOT +++ b/src/backend/access/heap/README.HOT @@ -401,12 +401,12 @@ live tuple. We mark the index open for inserts (but still not ready for reads) then we again wait for transactions which have the table open. Then validate the index. This searches for tuples missing from the index in auxiliary -index, and inserts any missing ones if they are visible to reference snapshot. +index, and inserts any missing ones if they are visible to a fresh snapshot. Again, the index entries have to have TIDs equal to HOT-chain root TIDs, but the value to be inserted is the one from the live tuple. Then we wait until every transaction that could have a snapshot older than -the second reference snapshot is finished. This ensures that nobody is +the latest used snapshot is finished. This ensures that nobody is alive any longer who could need to see any tuples that might be missing from the index, as well as ensuring that no one can see any inconsistent rows in a broken HOT chain (the first condition is stronger than the diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index 8b488cfd8f6..aa7f681beba 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -57,6 +57,8 @@ #include "utils/syscache.h" +int concurrent_index_reset_snapshot_interval = 100; + static HeapTuple heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid, CommandId cid, uint32 options); static XLogRecPtr log_heap_update(Relation reln, Buffer oldbuf, diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index 22e010aee1a..a0e3c6b2665 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -42,7 +42,6 @@ #include "storage/lmgr.h" #include "storage/lock.h" #include "storage/predicate.h" -#include "storage/proc.h" #include "storage/procarray.h" #include "storage/smgr.h" #include "utils/builtins.h" @@ -1965,17 +1964,19 @@ heapam_index_validate_scan_read_stream_next( return result; } -static void +static TransactionId heapam_index_validate_scan(Relation heapRelation, Relation indexRelation, IndexInfo *indexInfo, - Snapshot snapshot, ValidateIndexState *state, ValidateIndexState *auxState) { + TransactionId limitXmin; + Datum values[INDEX_MAX_KEYS]; bool isnull[INDEX_MAX_KEYS]; + SnapshotRotation *rotation = &indexInfo->ii_SnapshotRotation; TupleTableSlot *slot; EState *estate; ExprContext *econtext; @@ -1983,6 +1984,7 @@ heapam_index_validate_scan(Relation heapRelation, int64 num_to_check; Tuplestorestate *tuples_for_check; + ValidateIndexScanState callback_private_data; Buffer buf; @@ -1999,12 +2001,20 @@ heapam_index_validate_scan(Relation heapRelation, * pass-by-value on most platforms. */ tuples_for_check = tuplestore_begin_datum(INT8OID, false, false, store_work_mem_part); + validate_index_reset_snapshot(indexInfo); /* * sanity checks */ Assert(OidIsValid(indexRelation->rd_rel->relam)); + /* + * TODO: When adding _bt_load()'s heap-fetch budget, include skipped TIDs + * here too. Rows inserted before STIR was created may be absent from + * both indexes, so the difference below would miss them. + * + * https://commitfest.postgresql.org/patch/6401/ + */ num_to_check = heapam_index_validate_tuplesort_difference(state->tuplesort, auxState->tuplesort, tuples_for_check); @@ -2015,6 +2025,9 @@ heapam_index_validate_scan(Relation heapRelation, state->tuplesort = auxState->tuplesort = NULL; + /* The caller waits for snapshots older than any used to skip dead tuples. */ + limitXmin = validate_index_reset_snapshot(indexInfo); + estate = CreateExecutorState(); econtext = GetPerTupleExprContext(estate); slot = MakeSingleTupleTableSlot(RelationGetDescr(heapRelation), @@ -2046,6 +2059,9 @@ heapam_index_validate_scan(Relation heapRelation, CHECK_FOR_INTERRUPTS(); + limitXmin = TransactionIdNewer(limitXmin, + validate_index_reset_snapshot(indexInfo)); + LockBuffer(buf, BUFFER_LOCK_SHARE); block_number = BufferGetBlockNumber(buf); @@ -2056,7 +2072,7 @@ heapam_index_validate_scan(Relation heapRelation, bool all_dead, found; ItemPointerSet(&tid, block_number, off); - found = heap_hot_search_buffer(&tid, heapRelation, buf, snapshot, + found = heap_hot_search_buffer(&tid, heapRelation, buf, rotation->snapshot, &heap_tuple_data[i], &all_dead, true); if (!found) ItemPointerSetInvalid(&heap_tuple_data[i].t_self); @@ -2117,6 +2133,7 @@ heapam_index_validate_scan(Relation heapRelation, i++; } + ExecClearTuple(slot); ReleaseBuffer(buf); } @@ -2132,6 +2149,8 @@ heapam_index_validate_scan(Relation heapRelation, /* These may have been pointing to the now-gone estate */ indexInfo->ii_ExpressionsState = NIL; indexInfo->ii_PredicateState = NULL; + + return limitXmin; } /* diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c index cea1eb0207b..03bbd2dbbad 100644 --- a/src/backend/catalog/index.c +++ b/src/backend/catalog/index.c @@ -69,10 +69,12 @@ #include "storage/bufmgr.h" #include "storage/lmgr.h" #include "storage/predicate.h" +#include "storage/proc.h" #include "storage/smgr.h" #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" @@ -3562,8 +3564,9 @@ IndexCheckExclusion(Relation heapRelation, * insert their new tuples into it. At the same moment we clear "indisready" for * auxiliary index, since it is no more required to be updated. * - * We then take a new reference snapshot, any tuples that are valid according - * to this snap, but are not in the index, must be added to the index. + * We then take a new snapshot, any tuples that are valid according + * to this snap, but are not in the index, must be added to the index. In + * order to propagate xmin we reset that snapshot every so often. * (Any tuples committed live after the snap will be inserted into the * index by their originating transaction. Any tuples committed dead before * the snap need not be indexed, because we will wait out all transactions @@ -3576,7 +3579,7 @@ IndexCheckExclusion(Relation heapRelation, * TIDs of both auxiliary and target indexes, and doing a "merge join" against * the TID lists to see which tuples from auxiliary index are missing from the * target index. Thus we will ensure that all tuples valid according to the - * reference snapshot are in the index. Notice we need to do bulkdelete in the + * latest snapshot are in the index. Notice we need to do bulkdelete in the * particular order: auxiliary first, target last. * * Building a unique index this way is tricky: we might try to insert a @@ -3589,21 +3592,24 @@ IndexCheckExclusion(Relation heapRelation, * before it declares a uniqueness error. * * After completing validate_index(), we wait until all transactions that - * were alive at the time of the reference snapshot are gone; this is - * necessary to be sure there are none left with a transaction snapshot - * older than the reference (and hence possibly able to see tuples we did - * not index). Then we mark the index "indisvalid" and commit. Subsequent - * transactions will be able to use it for queries. + * were alive at the time of the latest snapshot used during validation are + * gone; this is necessary to be sure there are none left with a transaction + * snapshot older than that (and hence possibly able to see tuples we did + * not index). The snapshot is periodically refreshed during the heap scan + * to propagate the xmin horizon, so limitXmin tracks the most recent one. + * Then we mark the index "indisvalid" and commit. Subsequent transactions + * will be able to use it for queries. * * Also, some actions to concurrent drop the auxiliary index are performed. */ -void -validate_index(Oid heapId, Oid indexId, Oid auxIndexId, Snapshot snapshot) +TransactionId +validate_index(Oid heapId, Oid indexId, Oid auxIndexId) { Relation heapRelation, indexRelation, auxIndexRelation; IndexInfo *indexInfo; + TransactionId limitXmin; IndexVacuumInfo ivinfo, auxivinfo; ValidateIndexState state, auxState; Oid save_userid; @@ -3616,6 +3622,9 @@ validate_index(Oid heapId, Oid indexId, Oid auxIndexId, Snapshot snapshot) int main_work_mem_part = (int)((int64) maintenance_work_mem * 8 / 10); int aux_work_mem_part = maintenance_work_mem / 10; + /* Keep a snapshot active through index scans, sorting, and insertion. */ + PushActiveSnapshot(GetTransactionSnapshot()); + { const int progress_index[] = { PROGRESS_CREATEIDX_PHASE, @@ -3658,6 +3667,7 @@ validate_index(Oid heapId, Oid indexId, Oid auxIndexId, Snapshot snapshot) /* mark build is concurrent just for consistency */ indexInfo->ii_Concurrent = true; + validate_index_reset_snapshot(indexInfo); /* * Scan the index and gather up all the TIDs into a tuplesort object. @@ -3689,9 +3699,12 @@ validate_index(Oid heapId, Oid indexId, Oid auxIndexId, Snapshot snapshot) aux_work_mem_part, NULL, TUPLESORT_NONE); auxState.htups = auxState.itups = auxState.tups_inserted = 0; + validate_index_reset_snapshot(indexInfo); (void) index_bulk_delete(&auxivinfo, NULL, validate_index_callback, &auxState); + limitXmin = validate_index_reset_snapshot(indexInfo); + /* If aux index is empty, merge may be skipped */ if (auxState.itups == 0) { @@ -3709,7 +3722,10 @@ validate_index(Oid heapId, Oid indexId, Oid auxIndexId, Snapshot snapshot) index_close(indexRelation, NoLock); table_close(heapRelation, NoLock); - return; + PopActiveSnapshot(); + indexInfo->ii_SnapshotRotation.snapshot = NULL; + + return limitXmin; } state.tuplesort = tuplesort_begin_datum(INT8OID, Int8LessOperator, @@ -3717,10 +3733,12 @@ validate_index(Oid heapId, Oid indexId, Oid auxIndexId, Snapshot snapshot) (int) main_work_mem_part, NULL, TUPLESORT_NONE); state.htups = state.itups = state.tups_inserted = 0; + validate_index_reset_snapshot(indexInfo); /* ambulkdelete updates progress metrics */ (void) index_bulk_delete(&ivinfo, NULL, validate_index_callback, &state); + validate_index_reset_snapshot(indexInfo); /* Execute the sort */ { @@ -3737,19 +3755,20 @@ validate_index(Oid heapId, Oid indexId, Oid auxIndexId, Snapshot snapshot) pgstat_progress_update_multi_param(3, progress_index, progress_vals); } tuplesort_performsort(state.tuplesort); + validate_index_reset_snapshot(indexInfo); tuplesort_performsort(auxState.tuplesort); + validate_index_reset_snapshot(indexInfo); /* * Now merge both indexes */ pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE, PROGRESS_CREATEIDX_PHASE_VALIDATE_IDXMERGE); - table_index_validate_scan(heapRelation, - indexRelation, - indexInfo, - snapshot, - &state, - &auxState); + limitXmin = table_index_validate_scan(heapRelation, + indexRelation, + indexInfo, + &state, + &auxState); /* Tuple sort closed by table_index_validate_scan */ Assert(state.tuplesort == NULL && auxState.tuplesort == NULL); @@ -3772,6 +3791,34 @@ validate_index(Oid heapId, Oid indexId, Oid auxIndexId, Snapshot snapshot) index_close(auxIndexRelation, NoLock); index_close(indexRelation, NoLock); table_close(heapRelation, NoLock); + + PopActiveSnapshot(); + indexInfo->ii_SnapshotRotation.snapshot = NULL; + return limitXmin; +} + +/* + * Reset between validation phases or heap pages, once tuple data from the + * previous snapshot is no longer needed. Return the current snapshot's xmin. + */ +TransactionId +validate_index_reset_snapshot(IndexInfo *indexInfo) +{ + SnapshotRotation *rotation = &indexInfo->ii_SnapshotRotation; + + if (rotation->snapshot == NULL) + StartSnapshotRotation(rotation); + + if (SnapshotRotationDue(rotation)) + { + RotateSnapshot(rotation); + + INJECTION_POINT("validate-index-snapshot-reset", NULL); + if (unlikely(MyProc->xmin != rotation->snapshot->xmin)) + INJECTION_POINT("validate-index-xmin-not-advanced", NULL); + } + + return rotation->snapshot->xmin; } /* diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c index 96f34aadc44..0877a3fbceb 100644 --- a/src/backend/commands/indexcmds.c +++ b/src/backend/commands/indexcmds.c @@ -606,7 +606,6 @@ DefineIndex(ParseState *pstate, LockRelId heaprelid; LOCKTAG heaplocktag; LOCKMODE lockmode; - Snapshot snapshot; Oid root_save_userid; int root_save_sec_context; int root_save_nestlevel; @@ -1826,32 +1825,11 @@ DefineIndex(ParseState *pstate, /* Tell concurrent index builds to ignore us, if index qualifies */ if (safe_index) set_indexsafe_procflags(); - - /* - * Now take the "reference snapshot" that will be used by validate_index() - * to filter candidate tuples. Beware! There might still be snapshots in - * use that treat some transaction as in-progress that our reference - * snapshot treats as committed. If such a recently-committed transaction - * deleted tuples in the table, we will not include them in the index; yet - * those transactions which see the deleting one as still-in-progress will - * expect such tuples to be there once we mark the index as valid. - * - * We solve this by waiting for all endangered transactions to exit before - * we mark the index as valid. - * - * We also set ActiveSnapshot to this snap, since functions in indexes may - * need a snapshot. - */ - snapshot = RegisterSnapshot(GetTransactionSnapshot()); - PushActiveSnapshot(snapshot); /* * Merge content of auxiliary and target indexes - insert any missing index entries. */ - validate_index(tableId, indexRelationId, auxIndexRelationId, snapshot); - limitXmin = snapshot->xmin; + limitXmin = validate_index(tableId, indexRelationId, auxIndexRelationId); - PopActiveSnapshot(); - UnregisterSnapshot(snapshot); /* * The snapshot subsystem could still contain registered snapshots that * are holding back our process's advertised xmin; in particular, if @@ -1873,8 +1851,8 @@ DefineIndex(ParseState *pstate, /* * The index is now valid in the sense that it contains all currently * interesting tuples. But since it might not contain tuples deleted just - * before the reference snap was taken, we have to wait out any - * transactions that might have older snapshots. + * before the last snapshot during validating was taken, we have to wait + * out any transactions that might have older snapshots. */ INJECTION_POINT("define-index-before-set-valid", NULL); pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE, @@ -4573,7 +4551,6 @@ ReindexRelationConcurrently(const ReindexStmt *stmt, Oid relationOid, const Rein { ReindexIndexInfo *newidx = lfirst(lc); TransactionId limitXmin; - Snapshot snapshot; StartTransactionCommand(); @@ -4588,13 +4565,6 @@ ReindexRelationConcurrently(const ReindexStmt *stmt, Oid relationOid, const Rein if (newidx->safe) set_indexsafe_procflags(); - /* - * Take the "reference snapshot" that will be used by validate_index() - * to filter candidate tuples. - */ - snapshot = RegisterSnapshot(GetTransactionSnapshot()); - PushActiveSnapshot(snapshot); - /* * Update progress for the index to build, with the correct parent * table involved. @@ -4606,16 +4576,7 @@ ReindexRelationConcurrently(const ReindexStmt *stmt, Oid relationOid, const Rein progress_vals[3] = newidx->amId; pgstat_progress_update_multi_param(4, progress_index, progress_vals); - validate_index(newidx->tableId, newidx->indexId, newidx->auxIndexId, snapshot); - - /* - * We can now do away with our active snapshot, we still need to save - * the xmin limit to wait for older snapshots. - */ - limitXmin = snapshot->xmin; - - PopActiveSnapshot(); - UnregisterSnapshot(snapshot); + limitXmin = validate_index(newidx->tableId, newidx->indexId, newidx->auxIndexId); /* * To ensure no deadlocks, we must commit and start yet another @@ -4625,10 +4586,13 @@ ReindexRelationConcurrently(const ReindexStmt *stmt, Oid relationOid, const Rein CommitTransactionCommand(); StartTransactionCommand(); + /* We should now definitely not be advertising any xmin. */ + Assert(!TransactionIdIsValid(MyProc->xmin)); + /* * The index is now valid in the sense that it contains all currently * interesting tuples. But since it might not contain tuples deleted - * just before the reference snap was taken, we have to wait out any + * just before the latest snap was taken, we have to wait out any * transactions that might have older snapshots. * * Because we don't take a snapshot or Xid in this transaction, diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat index 1fb19b16078..453fcac00a3 100644 --- a/src/backend/utils/misc/guc_parameters.dat +++ b/src/backend/utils/misc/guc_parameters.dat @@ -516,6 +516,16 @@ options => 'compute_query_id_options', }, +{ 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 => 'Validation resets its snapshot between phases and heap pages. 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', + min => '-1', + max => 'INT_MAX', +}, + { name => 'config_file', type => 'string', context => 'PGC_POSTMASTER', group => 'FILE_LOCATIONS', short_desc => 'Sets the server\'s main configuration file.', flags => 'GUC_DISALLOW_IN_FILE | GUC_SUPERUSER_ONLY', diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c index bc98a4361bf..85010520ab9 100644 --- a/src/backend/utils/time/snapmgr.c +++ b/src/backend/utils/time/snapmgr.c @@ -376,6 +376,88 @@ GetLatestSnapshot(void) return SecondarySnapshot; } +/* + * StartSnapshotRotation + * Initialize rotation state with the active snapshot and current time. + * + * The operation must not hold another reference to the snapshot, so + * RotateSnapshot() can release it. The caller remains responsible for + * popping the active snapshot when the operation ends. + */ +void +StartSnapshotRotation(SnapshotRotation *rotation) +{ + rotation->snapshot = GetActiveSnapshot(); + INSTR_TIME_SET_CURRENT_FAST(rotation->last_reset); +} + +/* + * SnapshotRotationDue + * Check whether the reset interval has elapsed, and restart the timer. + * + * Resets are disabled at REPEATABLE READ and above, where the transaction + * snapshot pins xmin regardless of the active snapshot. + */ +bool +SnapshotRotationDue(SnapshotRotation *rotation) +{ + instr_time now, + elapsed; + + if (concurrent_index_reset_snapshot_interval < 0 || + IsolationUsesXactSnapshot()) + return false; + + INSTR_TIME_SET_CURRENT_FAST(now); + elapsed = now; + INSTR_TIME_SUBTRACT(elapsed, rotation->last_reset); + if (INSTR_TIME_GET_MILLISEC(elapsed) < concurrent_index_reset_snapshot_interval) + return false; + + rotation->last_reset = now; + return true; +} + +/* + * RotateSnapshot + * Replace the active snapshot while keeping xmin valid throughout. + * + * Register the new snapshot before popping the old one. With the active + * stack empty, SnapshotResetXmin() can advance xmin to the oldest registered + * snapshot. An older snapshot held by another caller, such as a cursor, can + * still hold back xmin. + * + * Push the new snapshot and unregister it, leaving the active-stack entry as + * the operation's only reference for the next reset. + */ +void +RotateSnapshot(SnapshotRotation *rotation) +{ + Snapshot newsnap; + + /* + * The snapshot must be the only active-stack entry. A nested snapshot + * cannot be replaced here, and an older entry would keep xmin pinned. + */ + Assert(ActiveSnapshot != NULL && + ActiveSnapshot->as_snap == rotation->snapshot && + ActiveSnapshot->as_next == NULL); + Assert(!HistoricSnapshotActive()); + /* The operation is running, so the transaction snapshot was taken. */ + Assert(FirstSnapshotSet); + + newsnap = RegisterSnapshot(GetSnapshotData(&SecondarySnapshotData)); + + /* The catalog snapshot is registered too and would hold back xmin. */ + InvalidateCatalogSnapshot(); + + PopActiveSnapshot(); + PushActiveSnapshot(newsnap); + UnregisterSnapshot(newsnap); + + rotation->snapshot = GetActiveSnapshot(); +} + /* * GetCatalogSnapshot * Get a snapshot that is sufficiently up-to-date for scan of the diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h index 27d4bca6afd..002e5d167ac 100644 --- a/src/include/access/tableam.h +++ b/src/include/access/tableam.h @@ -739,12 +739,11 @@ typedef struct TableAmRoutine TableScanDesc scan); /* see table_index_validate_scan for reference about parameters */ - void (*index_validate_scan) (Relation table_rel, - Relation index_rel, - IndexInfo *index_info, - Snapshot snapshot, - ValidateIndexState *state, - ValidateIndexState *aux_state); + TransactionId (*index_validate_scan) (Relation table_rel, + Relation index_rel, + IndexInfo *index_info, + ValidateIndexState *state, + ValidateIndexState *aux_state); /* ------------------------------------------------------------------------ @@ -1906,23 +1905,23 @@ table_index_build_range_scan(Relation table_rel, * * See validate_index() for an explanation. * - * Note: it is responsibility of that function to close sortstates in - * both `state` and `auxstate`. + * The caller supplies an active snapshot. The scan may replace it, but + * must leave it active on return. Return the newest xmin used to check + * visibility, so the caller can wait for older snapshots before marking + * the index valid. The scan must close both sortstates. */ -static inline void +static inline TransactionId table_index_validate_scan(Relation table_rel, Relation index_rel, IndexInfo *index_info, - Snapshot snapshot, ValidateIndexState *state, ValidateIndexState *auxstate) { - table_rel->rd_tableam->index_validate_scan(table_rel, - index_rel, - index_info, - snapshot, - state, - auxstate); + return table_rel->rd_tableam->index_validate_scan(table_rel, + index_rel, + index_info, + state, + auxstate); } diff --git a/src/include/access/transam.h b/src/include/access/transam.h index 55a4ab26b34..923aadbab43 100644 --- a/src/include/access/transam.h +++ b/src/include/access/transam.h @@ -415,6 +415,21 @@ NormalTransactionIdOlder(TransactionId a, TransactionId b) return b; } +/* return the newer of the two IDs */ +static inline TransactionId +TransactionIdNewer(TransactionId a, TransactionId b) +{ + if (!TransactionIdIsValid(a)) + return b; + + if (!TransactionIdIsValid(b)) + return a; + + if (TransactionIdFollows(a, b)) + return a; + return b; +} + /* return the newer of the two IDs */ static inline FullTransactionId FullTransactionIdNewer(FullTransactionId a, FullTransactionId b) diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h index bbdc82448ee..5e6873c959c 100644 --- a/src/include/catalog/index.h +++ b/src/include/catalog/index.h @@ -160,7 +160,9 @@ extern void index_build(Relation heapRelation, bool parallel, bool progress); -extern void validate_index(Oid heapId, Oid indexId, Oid auxIndexId, Snapshot snapshot); +extern TransactionId validate_index(Oid heapId, Oid indexId, Oid auxIndexId); + +extern TransactionId validate_index_reset_snapshot(IndexInfo *indexInfo); extern void index_set_state_flags(Oid indexId, IndexStateFlagsAction action); diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h index 6bf0dd63247..e673cc567f7 100644 --- a/src/include/miscadmin.h +++ b/src/include/miscadmin.h @@ -273,6 +273,7 @@ extern PGDLLIMPORT double hash_mem_multiplier; extern PGDLLIMPORT int maintenance_work_mem; extern PGDLLIMPORT int debug_cic_validate_store_mem_pct; extern PGDLLIMPORT int max_parallel_maintenance_workers; +extern PGDLLIMPORT int concurrent_index_reset_snapshot_interval; /* * Upper and lower hard limits for the buffer access strategy ring size diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 1b8d3b88f8a..07e21c1bb82 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" #include "utils/typcache.h" @@ -169,10 +170,10 @@ typedef struct ExprState * entries for a particular index. Used for both index_build and * retail creation of index entries. * - * ii_Concurrent, ii_BrokenHotChain, and ii_ParallelWorkers are used only - * during index build; they're conventionally zeroed otherwise. ii_Auxiliary - * is also used during retail inserts to skip datum formation for auxiliary - * indexes. + * ii_Concurrent, ii_BrokenHotChain, ii_ParallelWorkers, and ii_SnapshotRotation + * are used only during index build; they're conventionally zeroed otherwise. + * ii_Auxiliary is also used during retail inserts to skip datum formation for + * auxiliary indexes. * ---------------- */ typedef struct IndexInfo @@ -244,6 +245,9 @@ typedef struct IndexInfo /* memory context holding this IndexInfo */ MemoryContext ii_Context; + + /* snapshot resets during index validation */ + SnapshotRotation ii_SnapshotRotation; } IndexInfo; /* ---------------- diff --git a/src/include/utils/snapmgr.h b/src/include/utils/snapmgr.h index 1c550096393..a96e7e67364 100644 --- a/src/include/utils/snapmgr.h +++ b/src/include/utils/snapmgr.h @@ -76,6 +76,10 @@ extern PGDLLIMPORT SnapshotData SnapshotToastData; extern Snapshot GetTransactionSnapshot(void); extern Snapshot GetLatestSnapshot(void); + +extern void StartSnapshotRotation(SnapshotRotation *rotation); +extern bool SnapshotRotationDue(SnapshotRotation *rotation); +extern void RotateSnapshot(SnapshotRotation *rotation); extern void SnapshotSetCommandId(CommandId curcid); extern Snapshot GetCatalogSnapshot(Oid relid); diff --git a/src/include/utils/snapshot.h b/src/include/utils/snapshot.h index 9766aabcad4..9a4538d2aa6 100644 --- a/src/include/utils/snapshot.h +++ b/src/include/utils/snapshot.h @@ -14,6 +14,7 @@ #define SNAPSHOT_H #include "lib/pairingheap.h" +#include "portability/instr_time.h" /* @@ -209,4 +210,15 @@ typedef struct SnapshotData uint64 snapXactCompletionCount; } SnapshotData; + +/* + * State for periodically replacing an operation's active snapshot. See + * StartSnapshotRotation(). + */ +typedef struct SnapshotRotation +{ + Snapshot snapshot; /* current active snapshot */ + instr_time last_reset; /* when it was last replaced */ +} SnapshotRotation; + #endif /* SNAPSHOT_H */ diff --git a/src/test/modules/injection_points/expected/reindex_conc.out b/src/test/modules/injection_points/expected/reindex_conc.out index db8de4bbe85..d909116fe2c 100644 --- a/src/test/modules/injection_points/expected/reindex_conc.out +++ b/src/test/modules/injection_points/expected/reindex_conc.out @@ -33,7 +33,56 @@ REINDEX INDEX CONCURRENTLY reindex_inj.ind_pred; NOTICE: notice triggered for injection point reindex-conc-index-not-safe REINDEX INDEX CONCURRENTLY reindex_inj.ind_expr_pred; NOTICE: notice triggered for injection point reindex-conc-index-not-safe +-- Validation must reset snapshots even when STIR is empty. +SELECT injection_points_attach('validate-index-snapshot-reset', 'notice'); + injection_points_attach +------------------------- + +(1 row) + +SELECT injection_points_attach('validate-index-xmin-not-advanced', 'notice'); + injection_points_attach +------------------------- + +(1 row) + +SET concurrent_index_reset_snapshot_interval = 0; +INSERT INTO reindex_inj.tbl(i) VALUES (1), (2), (3); +CREATE INDEX CONCURRENTLY ind_reset ON reindex_inj.tbl(i); +NOTICE: notice triggered for injection point validate-index-snapshot-reset +NOTICE: notice triggered for injection point validate-index-snapshot-reset +NOTICE: notice triggered for injection point validate-index-snapshot-reset +REINDEX INDEX CONCURRENTLY reindex_inj.ind_reset; +NOTICE: notice triggered for injection point reindex-conc-index-safe +NOTICE: notice triggered for injection point validate-index-snapshot-reset +NOTICE: notice triggered for injection point validate-index-snapshot-reset +NOTICE: notice triggered for injection point validate-index-snapshot-reset +-- These cases keep a fixed snapshot. +SET concurrent_index_reset_snapshot_interval = -1; +REINDEX INDEX CONCURRENTLY reindex_inj.ind_reset; +NOTICE: notice triggered for injection point reindex-conc-index-safe +SET concurrent_index_reset_snapshot_interval = 0; +SET default_transaction_isolation = 'repeatable read'; +REINDEX INDEX CONCURRENTLY reindex_inj.ind_reset; +NOTICE: notice triggered for injection point reindex-conc-index-safe +SET default_transaction_isolation = 'serializable'; +REINDEX INDEX CONCURRENTLY reindex_inj.ind_reset; +NOTICE: notice triggered for injection point reindex-conc-index-safe +RESET default_transaction_isolation; +RESET concurrent_index_reset_snapshot_interval; -- Cleanup +SELECT injection_points_detach('validate-index-snapshot-reset'); + injection_points_detach +------------------------- + +(1 row) + +SELECT injection_points_detach('validate-index-xmin-not-advanced'); + injection_points_detach +------------------------- + +(1 row) + SELECT injection_points_detach('reindex-conc-index-safe'); injection_points_detach ------------------------- diff --git a/src/test/modules/injection_points/sql/reindex_conc.sql b/src/test/modules/injection_points/sql/reindex_conc.sql index 6cf211e6d5d..d8c54c79247 100644 --- a/src/test/modules/injection_points/sql/reindex_conc.sql +++ b/src/test/modules/injection_points/sql/reindex_conc.sql @@ -19,7 +19,28 @@ REINDEX INDEX CONCURRENTLY reindex_inj.ind_expr; REINDEX INDEX CONCURRENTLY reindex_inj.ind_pred; REINDEX INDEX CONCURRENTLY reindex_inj.ind_expr_pred; +-- Validation must reset snapshots even when STIR is empty. +SELECT injection_points_attach('validate-index-snapshot-reset', 'notice'); +SELECT injection_points_attach('validate-index-xmin-not-advanced', 'notice'); +SET concurrent_index_reset_snapshot_interval = 0; +INSERT INTO reindex_inj.tbl(i) VALUES (1), (2), (3); +CREATE INDEX CONCURRENTLY ind_reset ON reindex_inj.tbl(i); +REINDEX INDEX CONCURRENTLY reindex_inj.ind_reset; + +-- These cases keep a fixed snapshot. +SET concurrent_index_reset_snapshot_interval = -1; +REINDEX INDEX CONCURRENTLY reindex_inj.ind_reset; +SET concurrent_index_reset_snapshot_interval = 0; +SET default_transaction_isolation = 'repeatable read'; +REINDEX INDEX CONCURRENTLY reindex_inj.ind_reset; +SET default_transaction_isolation = 'serializable'; +REINDEX INDEX CONCURRENTLY reindex_inj.ind_reset; +RESET default_transaction_isolation; +RESET concurrent_index_reset_snapshot_interval; + -- Cleanup +SELECT injection_points_detach('validate-index-snapshot-reset'); +SELECT injection_points_detach('validate-index-xmin-not-advanced'); SELECT injection_points_detach('reindex-conc-index-safe'); SELECT injection_points_detach('reindex-conc-index-not-safe'); DROP TABLE reindex_inj.tbl; diff --git a/src/test/modules/test_misc/meson.build b/src/test/modules/test_misc/meson.build index ee290698b31..2de4ab61b4b 100644 --- a/src/test/modules/test_misc/meson.build +++ b/src/test/modules/test_misc/meson.build @@ -23,6 +23,7 @@ tests += { 't/012_ddlutils.pl', 't/013_temp_obj_multisession.pl', 't/014_log_statement_max_length.pl', + 't/015_cic_validate_snapshot.pl', ], # The injection points are cluster-wide, so disable installcheck 'runningcheck': false, diff --git a/src/test/modules/test_misc/t/015_cic_validate_snapshot.pl b/src/test/modules/test_misc/t/015_cic_validate_snapshot.pl new file mode 100644 index 00000000000..c75520961ad --- /dev/null +++ b/src/test/modules/test_misc/t/015_cic_validate_snapshot.pl @@ -0,0 +1,218 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Check validation snapshot resets with and without STIR candidates, disabled +# resets, transaction isolation, and snapshots held by cursors. + +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +plan skip_all => 'Injection points not supported by this build' + unless $ENV{enable_injection_points} eq 'yes'; + +my $node = PostgreSQL::Test::Cluster->new('node'); +$node->init(); +$node->append_conf('postgresql.conf', 'fsync = off'); +# Injection point notices must reach the server log. +$node->append_conf('postgresql.conf', 'log_min_messages = notice'); +$node->start(); + +plan skip_all => 'Extension injection_points not installed' + unless $node->check_extension('injection_points'); + +$node->safe_psql('postgres', 'CREATE EXTENSION injection_points;'); +$node->safe_psql('postgres', 'CREATE EXTENSION amcheck;'); + +$node->safe_psql( + 'postgres', q[ +CREATE UNLOGGED TABLE tbl (i int); +INSERT INTO tbl SELECT generate_series(1, 1000); +-- The function lies about immutability so that the build scan of CIC can +-- be paused on an advisory lock held by another session. +CREATE FUNCTION indexpr(int) RETURNS int IMMUTABLE LANGUAGE plpgsql +AS $$ BEGIN PERFORM pg_advisory_xact_lock(4971); RETURN $1; END $$; +]); + +my $blocker = $node->background_psql('postgres', on_error_stop => 0); +my $cic = $node->background_psql('postgres', on_error_stop => 0); + +# Pause the CIC build scan on the first evaluation of the index +# expression, insert rows that will only be reachable through the +# auxiliary index (making the validation phase do real work), then let +# CIC finish. +sub run_cic_with_candidates +{ + my ($marker, $first_new_row, $interval, $isolation) = @_; + $interval //= 0; + $isolation //= 'read committed'; + + $blocker->query_safe(q[SELECT pg_advisory_lock(4971);]); + + $cic->query_until( + qr/$marker/, qq[ +\\echo $marker +SET concurrent_index_reset_snapshot_interval = $interval; +SET default_transaction_isolation = '$isolation'; +CREATE INDEX CONCURRENTLY idx ON tbl (indexpr(i)); +]); + + # Wait for the build scan to block on the advisory lock. + $node->poll_query_until( + 'postgres', q[ +SELECT EXISTS (SELECT FROM pg_locks + WHERE locktype = 'advisory' AND objid = 4971 AND NOT granted)]) + or die "timed out waiting for CIC to block on the advisory lock"; + + # These rows are seen neither by the build scan snapshot nor by the + # target index (not ready yet), so the validation phase will have to + # fetch and insert all of them. + my $last_new_row = $first_new_row + 999; + $node->safe_psql('postgres', + "INSERT INTO tbl SELECT generate_series($first_new_row, $last_new_row);"); + + $blocker->query_safe(q[SELECT pg_advisory_unlock(4971);]); +} + +# An empty STIR skips the heap scan, but validation must still reset snapshots. +$node->safe_psql('postgres', + q[SELECT injection_points_attach('validate-index-snapshot-reset', 'error');]); + +my ($ret, $stdout, $stderr) = $node->psql('postgres', q[ +SET concurrent_index_reset_snapshot_interval = 0; +CREATE INDEX CONCURRENTLY idx ON tbl (indexpr(i)); +]); +like($stderr, + qr/error triggered for injection point validate-index-snapshot-reset/, + 'snapshot reset happens before the validation heap scan'); + +# The failed CIC left an invalid index behind; dropping it must take the +# auxiliary index with it. +$node->safe_psql('postgres', q[DROP INDEX idx;]); +is( $node->safe_psql( + 'postgres', + q[SELECT count(*) FROM pg_class WHERE relname IN ('idx', 'idx_ccaux')]), + '0', + 'failed CIC cleaned up after DROP INDEX'); + +$node->safe_psql('postgres', + q[SELECT injection_points_detach('validate-index-snapshot-reset');]); + +# Every reset must fully advance xmin. The reset point now only +# raises a notice, both to prove that resets do happen during this +# (successful) run and to let it complete. +$node->safe_psql('postgres', + q[SELECT injection_points_attach('validate-index-snapshot-reset', 'notice');]); +$node->safe_psql('postgres', + q[SELECT injection_points_attach('validate-index-xmin-not-advanced', 'error');]); + +my $log_offset = -s $node->logfile; + +# The early return for an empty STIR must also leave a complete index. +$node->safe_psql('postgres', q[ +SET concurrent_index_reset_snapshot_interval = 0; +CREATE INDEX CONCURRENTLY idx ON tbl (indexpr(i)); +SELECT bt_index_check('idx', heapallindexed => true); +]); +$node->wait_for_log( + qr/notice triggered for injection point validate-index-snapshot-reset/, + $log_offset); +ok(1, 'CIC with an empty STIR completes with snapshot resets'); +$node->safe_psql('postgres', q[DROP INDEX idx;]); + +$log_offset = -s $node->logfile; +run_cic_with_candidates('starting_cic2', 200001); + +$node->poll_query_until( + 'postgres', q[ +SELECT indisvalid FROM pg_index + WHERE indexrelid = (SELECT oid FROM pg_class WHERE relname = 'idx')]) + or die "timed out waiting for CIC to complete"; +ok(1, 'CIC completed with snapshot resets fully advancing xmin'); + +$node->wait_for_log( + qr/notice triggered for injection point validate-index-snapshot-reset/, + $log_offset); +ok(1, 'snapshot resets happened during the successful CIC'); + +# The index must contain everything despite the snapshot changes. +is( $node->safe_psql('postgres', + q[SELECT bt_index_check('idx', heapallindexed => true)]), + '', 'index is complete'); + +$node->safe_psql('postgres', q[DROP INDEX idx;]); +$node->safe_psql('postgres', + q[SELECT injection_points_detach('validate-index-snapshot-reset'); +SELECT injection_points_attach('validate-index-snapshot-reset', 'error');]); + +# Disabled resets and transaction snapshots must keep a fixed snapshot. +my @fixed_cases = ( + ['disabled', -1, 'read committed'], + ['repeatable_read', 0, 'repeatable read'], + ['serializable', 0, 'serializable']); +my $first_row = 300001; +for my $case (@fixed_cases) +{ + my ($name, $interval, $isolation) = @$case; + run_cic_with_candidates("starting_$name", $first_row, $interval, $isolation); + $node->poll_query_until('postgres', + q[SELECT indisvalid FROM pg_index WHERE indexrelid = to_regclass('idx')]) + or die "timed out waiting for $name CIC"; + is($node->safe_psql('postgres', + q[SELECT bt_index_check('idx', heapallindexed => true)]), + '', "$name: index is complete without resets"); + $node->safe_psql('postgres', q[DROP INDEX idx;]); + $first_row += 100000; +} + +# A cursor opened by an index expression keeps an extra snapshot reference. +$node->safe_psql('postgres', + q[SELECT injection_points_detach('validate-index-snapshot-reset'); +SELECT injection_points_attach('validate-index-snapshot-reset', 'notice'); +SELECT injection_points_detach('validate-index-xmin-not-advanced'); +SELECT injection_points_attach('validate-index-xmin-not-advanced', 'notice'); +SELECT injection_points_attach('validation-cursor-open', 'wait');]); +$node->safe_psql('postgres', qq[ +CREATE OR REPLACE FUNCTION indexpr(int) RETURNS int IMMUTABLE LANGUAGE plpgsql +AS \$\$ +DECLARE c refcursor := 'validation_pin'; +BEGIN + PERFORM pg_advisory_xact_lock(4971); + IF \$1 = $first_row THEN + OPEN c FOR SELECT 1; + PERFORM public.injection_points_run('validation-cursor-open'); + END IF; + RETURN \$1; +END \$\$; +]); +$log_offset = -s $node->logfile; +run_cic_with_candidates('starting_pinned', $first_row); +$node->poll_query_until('postgres', + q[SELECT EXISTS (SELECT FROM pg_stat_activity + WHERE wait_event = 'validation-cursor-open')]) + or die 'timed out waiting for the validation cursor'; +# Give the next snapshot a newer xmin than the cursor's snapshot. +$node->safe_psql('postgres', q[SELECT pg_current_xact_id();]); +$node->safe_psql('postgres', + q[SELECT injection_points_wakeup('validation-cursor-open');]); +$node->poll_query_until('postgres', + q[SELECT indisvalid FROM pg_index WHERE indexrelid = to_regclass('idx')]) + or die 'timed out waiting for CIC with a pinned snapshot'; +$node->wait_for_log( + qr/notice triggered for injection point validate-index-snapshot-reset/, + $log_offset); +$node->wait_for_log( + qr/notice triggered for injection point validate-index-xmin-not-advanced/, + $log_offset); +ok(1, 'cursor keeps xmin pinned across a snapshot reset'); +$node->safe_psql('postgres', + q[SELECT injection_points_detach('validation-cursor-open');]); +is($node->safe_psql('postgres', + q[SELECT bt_index_check('idx', heapallindexed => true)]), + '', 'index is complete with a snapshot held by a cursor'); + +eval { $blocker->quit }; +eval { $cic->quit }; + +done_testing(); diff --git a/src/test/regress/expected/create_index.out b/src/test/regress/expected/create_index.out index e4072033d7f..bf793d9b94d 100644 --- a/src/test/regress/expected/create_index.out +++ b/src/test/regress/expected/create_index.out @@ -3392,6 +3392,9 @@ DROP INDEX aux_index_ind6; --------+---------+-----------+----------+--------- c1 | integer | | | +SET default_transaction_isolation = 'repeatable read'; +CREATE INDEX CONCURRENTLY aux_index_ind6 ON aux_index_tab5 (c1); +SET default_transaction_isolation = 'read committed'; DROP TABLE aux_index_tab5; -- Check handling of indexes with expressions and predicates. The -- definitions of the rebuilt indexes should match the original diff --git a/src/test/regress/sql/create_index.sql b/src/test/regress/sql/create_index.sql index 5fe373b7b34..30f33eb898c 100644 --- a/src/test/regress/sql/create_index.sql +++ b/src/test/regress/sql/create_index.sql @@ -1404,6 +1404,10 @@ DROP INDEX aux_index_ind6; -- Make sure auxiliary index dropped too \d aux_index_tab5 +SET default_transaction_isolation = 'repeatable read'; +CREATE INDEX CONCURRENTLY aux_index_ind6 ON aux_index_tab5 (c1); +SET default_transaction_isolation = 'read committed'; + DROP TABLE aux_index_tab5; -- Check handling of indexes with expressions and predicates. The -- 2.43.0