From eb8800526cad139b2507172e313a57f3a1fcff10 Mon Sep 17 00:00:00 2001 From: Mikhail Nikalayeu Date: Mon, 21 Apr 2025 14:11:53 +0200 Subject: [PATCH v40 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/016_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/016_cic_validate_snapshot.pl diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index f7db2995c3..ade2f59875 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -12780,6 +12780,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 b1c797517e..382fe1723a 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 6a09a1a151..a89e792e84 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 86c6b5b705..68d3338d68 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" @@ -1958,17 +1957,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; @@ -1976,6 +1977,7 @@ heapam_index_validate_scan(Relation heapRelation, int64 num_to_check; Tuplestorestate *tuples_for_check; + ValidateIndexScanState callback_private_data; Buffer buf; @@ -1992,12 +1994,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); @@ -2008,6 +2018,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), @@ -2039,6 +2052,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); @@ -2049,7 +2065,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); @@ -2110,6 +2126,7 @@ heapam_index_validate_scan(Relation heapRelation, i++; } + ExecClearTuple(slot); ReleaseBuffer(buf); } @@ -2125,6 +2142,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 457146c565..0f08b92ddf 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" @@ -3640,8 +3642,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 @@ -3654,7 +3657,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 @@ -3667,21 +3670,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; @@ -3694,6 +3700,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, @@ -3736,6 +3745,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. @@ -3768,9 +3778,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) { @@ -3788,7 +3801,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, @@ -3796,10 +3812,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 */ { @@ -3816,19 +3834,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); @@ -3851,6 +3870,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 f30e16b19f..931c506a39 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; @@ -1842,32 +1841,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 @@ -1889,8 +1867,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, @@ -4589,7 +4567,6 @@ ReindexRelationConcurrently(const ReindexStmt *stmt, Oid relationOid, const Rein { ReindexIndexInfo *newidx = lfirst(lc); TransactionId limitXmin; - Snapshot snapshot; StartTransactionCommand(); @@ -4604,13 +4581,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. @@ -4622,16 +4592,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 @@ -4641,10 +4602,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 54e0eebc90..a00dab9095 100644 --- a/src/backend/utils/misc/guc_parameters.dat +++ b/src/backend/utils/misc/guc_parameters.dat @@ -509,6 +509,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 048450da2d..69755dde14 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 ed9221eaf5..c0df062f6e 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); /* ------------------------------------------------------------------------ @@ -1928,23 +1927,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 dbe228527a..797cee09fe 100644 --- a/src/include/access/transam.h +++ b/src/include/access/transam.h @@ -417,6 +417,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 0e2e678783..b771997d98 100644 --- a/src/include/catalog/index.h +++ b/src/include/catalog/index.h @@ -163,7 +163,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 1cdb4f4a7e..98dc305a4b 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 c893932b9b..0e8d169438 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,10 +169,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 @@ -243,6 +244,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 1c55009639..a96e7e6736 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 9766aabcad..9a4538d2aa 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 db8de4bbe8..d909116fe2 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 6cf211e6d5..d8c54c7924 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 5d81f5b13b..450a1cd0a9 100644 --- a/src/test/modules/test_misc/meson.build +++ b/src/test/modules/test_misc/meson.build @@ -24,6 +24,7 @@ tests += { 't/013_temp_obj_multisession.pl', 't/014_log_statement_max_length.pl', 't/015_temp_schema_exit_deferrable.pl', + 't/016_cic_validate_snapshot.pl', ], # The injection points are cluster-wide, so disable installcheck 'runningcheck': false, diff --git a/src/test/modules/test_misc/t/016_cic_validate_snapshot.pl b/src/test/modules/test_misc/t/016_cic_validate_snapshot.pl new file mode 100644 index 0000000000..c75520961a --- /dev/null +++ b/src/test/modules/test_misc/t/016_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 e4072033d7..bf793d9b94 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 5fe373b7b3..30f33eb898 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