diff --git a/contrib/test_decoding/Makefile b/contrib/test_decoding/Makefile index a7f7b7ae789..a33e6295d8c 100644 --- a/contrib/test_decoding/Makefile +++ b/contrib/test_decoding/Makefile @@ -15,7 +15,8 @@ SLOT_CREATION_TST = $(if $(filter win32,$(PORTNAME)),,slot_creation_error) ISOLATION = mxact delayed_startup ondisk_startup concurrent_ddl_dml \ oldest_xmin snapshot_transfer subxact_without_top concurrent_stream \ twophase_snapshot $(SLOT_CREATION_TST) catalog_change_snapshot \ - skip_snapshot_restore invalidation_distribution parallel_session_origin + skip_snapshot_restore invalidation_distribution parallel_session_origin \ + tuplecid_restart tuplecid_nested REGRESS_OPTS = --temp-config $(top_srcdir)/contrib/test_decoding/logical.conf ISOLATION_OPTS = --temp-config $(top_srcdir)/contrib/test_decoding/logical.conf diff --git a/contrib/test_decoding/expected/tuplecid_nested.out b/contrib/test_decoding/expected/tuplecid_nested.out new file mode 100644 index 00000000000..5d816374fbc --- /dev/null +++ b/contrib/test_decoding/expected/tuplecid_nested.out @@ -0,0 +1,32 @@ +Parsed test spec with 3 sessions + +starting permutation: s1_init s0_begin s0_topwrite s0_outer s0_inner s0_catinsert1 s0_release_inner s0_rollback_outer s2_vacuum s0_catinsert2 s0_commit s1_get_changes +step s1_init: SELECT 'init' FROM pg_create_logical_replication_slot('tuplecid_nested_slot', 'test_decoding'); +?column? +-------- +init +(1 row) + +step s0_begin: BEGIN; +step s0_topwrite: INSERT INTO tbl1 VALUES (0); +step s0_outer: SAVEPOINT outer_sp; +step s0_inner: SAVEPOINT inner_sp; +step s0_catinsert1: INSERT INTO user_cat VALUES (5, 'e'); +step s0_release_inner: RELEASE SAVEPOINT inner_sp; +step s0_rollback_outer: ROLLBACK TO SAVEPOINT outer_sp; +step s2_vacuum: VACUUM user_cat; +step s0_catinsert2: INSERT INTO user_cat VALUES (6, 'f'); +step s0_commit: COMMIT; +step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('tuplecid_nested_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0') WHERE data NOT LIKE '%user_cat%'; +data +------------------------------------------ +BEGIN +table public.tbl1: INSERT: val1[integer]:0 +COMMIT +(3 rows) + +?column? +-------- +stop +(1 row) + diff --git a/contrib/test_decoding/expected/tuplecid_restart.out b/contrib/test_decoding/expected/tuplecid_restart.out new file mode 100644 index 00000000000..e5b4bc2543d --- /dev/null +++ b/contrib/test_decoding/expected/tuplecid_restart.out @@ -0,0 +1,68 @@ +Parsed test spec with 3 sessions + +starting permutation: s0_init s0_begin s0_savepoint s0_insert s1_checkpoint s1_get_changes s0_catinsert1 s0_rollback s2_vacuum s0_catinsert2 s1_checkpoint s1_get_changes s0_commit s1_get_changes +step s0_init: SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); +?column? +-------- +init +(1 row) + +step s0_begin: BEGIN; +step s0_savepoint: SAVEPOINT sp1; +step s0_insert: INSERT INTO tbl1 VALUES (1); +step s1_checkpoint: CHECKPOINT; +step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0') WHERE data NOT LIKE '%user_cat%'; +data +---- +(0 rows) + +step s0_catinsert1: INSERT INTO user_cat VALUES (5, 'e'); +step s0_rollback: ROLLBACK TO SAVEPOINT sp1; +step s2_vacuum: VACUUM user_cat; +step s0_catinsert2: INSERT INTO user_cat VALUES (6, 'f'); +step s1_checkpoint: CHECKPOINT; +step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0') WHERE data NOT LIKE '%user_cat%'; +data +---- +(0 rows) + +step s0_commit: COMMIT; +step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0') WHERE data NOT LIKE '%user_cat%'; +data +------ +BEGIN +COMMIT +(2 rows) + +?column? +-------- +stop +(1 row) + + +starting permutation: s0_begin s0_savepoint s0_insert s1_init s0_catinsert1 s0_rollback s2_vacuum s0_catinsert2 s0_commit s1_get_changes +step s0_begin: BEGIN; +step s0_savepoint: SAVEPOINT sp1; +step s0_insert: INSERT INTO tbl1 VALUES (1); +step s1_init: SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); +step s0_catinsert1: INSERT INTO user_cat VALUES (5, 'e'); +step s0_rollback: ROLLBACK TO SAVEPOINT sp1; +step s2_vacuum: VACUUM user_cat; +step s0_catinsert2: INSERT INTO user_cat VALUES (6, 'f'); +step s0_commit: COMMIT; +step s1_init: <... completed> +?column? +-------- +init +(1 row) + +step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0') WHERE data NOT LIKE '%user_cat%'; +data +---- +(0 rows) + +?column? +-------- +stop +(1 row) + diff --git a/contrib/test_decoding/meson.build b/contrib/test_decoding/meson.build index dbc0ef0f63e..667ea76518f 100644 --- a/contrib/test_decoding/meson.build +++ b/contrib/test_decoding/meson.build @@ -30,6 +30,8 @@ test_decoding_isolation = [ 'skip_snapshot_restore', 'invalidation_distribution', 'parallel_session_origin', + 'tuplecid_nested', + 'tuplecid_restart', ] # slot_creation_error terminates another backend, whose FATAL message can be diff --git a/contrib/test_decoding/specs/tuplecid_nested.spec b/contrib/test_decoding/specs/tuplecid_nested.spec new file mode 100644 index 00000000000..8dd5298432a --- /dev/null +++ b/contrib/test_decoding/specs/tuplecid_nested.spec @@ -0,0 +1,77 @@ +# Test for BUG #19555. +# +# Test decoding of a partially rolled back transaction where the aborted part +# is nested: the aborting outer subtransaction never produced any WAL of its +# own, so decoding cannot know its toplevel transaction, while an inner +# subtransaction it rolls back did produce WAL and its tuplecid records still +# have to be removed +# +# The tuplecid records written by the aborted inner subtransaction are queued +# on the toplevel transaction and must be removed when the outer +# subtransaction's abort is decoded; otherwise they collide with the records +# of the line pointer that gets reused by the toplevel transaction's later +# catalog insert. +# +# - s0_topwrite makes the toplevel transaction produce its first WAL record +# before any subtransaction exists. +# - s0_outer establishes the outer savepoint, which never gets a WAL record +# of its own. +# - s0_catinsert1 generates an xl_heap_new_cid record in the inner +# subtransaction, which is then released. +# - s0_rollback_outer rolls the outer savepoint back. Its abort record is +# written once the backend is already in TRANS_ABORT, so it carries no +# toplevel xid in its header and the outer subtransaction's association +# with the toplevel can never be established. The released inner +# subtransaction is listed in the same abort record, though, and its +# tuplecid records still must be removed from the toplevel's list. +# - s2_vacuum reclaims the aborted row's line pointer (dead inserts are +# removed regardless of the xid horizon, so this is deterministic even +# while the toplevel transaction is still open); s0_catinsert2 then reuses +# the same tid with a different cmin. +# - s1_get_changes then decodes the toplevel commit and builds the tuplecid +# hash. +setup +{ + DROP TABLE IF EXISTS tbl1; + DROP TABLE IF EXISTS user_cat; + CREATE TABLE tbl1 (val1 integer); + -- Four rows of ~1.5kB fill the first heap page to within a few hundred + -- bytes, so that vacuuming away the aborted fifth row leaves exactly one + -- line pointer for the next insert to reuse. + CREATE TABLE user_cat (c1 int, filler char(1500)) WITH (user_catalog_table = true); + INSERT INTO user_cat VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd'); +} + +teardown +{ + DROP TABLE tbl1; + DROP TABLE user_cat; + SELECT 'stop' FROM pg_drop_replication_slot('tuplecid_nested_slot'); +} + +session "s0" +setup { SET synchronous_commit=on; } +step "s0_begin" { BEGIN; } +step "s0_topwrite" { INSERT INTO tbl1 VALUES (0); } +step "s0_outer" { SAVEPOINT outer_sp; } +step "s0_inner" { SAVEPOINT inner_sp; } +step "s0_catinsert1" { INSERT INTO user_cat VALUES (5, 'e'); } +step "s0_release_inner" { RELEASE SAVEPOINT inner_sp; } +step "s0_rollback_outer" { ROLLBACK TO SAVEPOINT outer_sp; } +step "s0_catinsert2" { INSERT INTO user_cat VALUES (6, 'f'); } +step "s0_commit" { COMMIT; } + +session "s1" +setup { SET synchronous_commit=on; } +step "s1_init" { SELECT 'init' FROM pg_create_logical_replication_slot('tuplecid_nested_slot', 'test_decoding'); } +# The user_cat rows carry ~1.5kB filler values which would flood the expected +# output; filter them out (decoding still processes them server-side). +step "s1_get_changes" { SELECT data FROM pg_logical_slot_get_changes('tuplecid_nested_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0') WHERE data NOT LIKE '%user_cat%'; } + +session "s2" +setup { SET synchronous_commit=on; } +step "s2_vacuum" { VACUUM user_cat; } + +# The slot is created before the transaction begins, so the final +# s1_get_changes decodes the whole transaction through its commit. +permutation "s1_init" "s0_begin" "s0_topwrite" "s0_outer" "s0_inner" "s0_catinsert1" "s0_release_inner" "s0_rollback_outer" "s2_vacuum" "s0_catinsert2" "s0_commit" "s1_get_changes" diff --git a/contrib/test_decoding/specs/tuplecid_restart.spec b/contrib/test_decoding/specs/tuplecid_restart.spec new file mode 100644 index 00000000000..e3a4e6f7255 --- /dev/null +++ b/contrib/test_decoding/specs/tuplecid_restart.spec @@ -0,0 +1,74 @@ +# Test for BUG #19555. +# +# Test decoding of a transaction in which a catalog-modifying subtransaction +# was rolled back, where the transaction spans a logical decoding restart +# point. +# +# The tuplecid records written by the aborted subtransaction are queued on the +# toplevel transaction and must be removed when the subtransaction's abort is +# decoded; otherwise they collide with the records of the line pointer that +# gets reused by the toplevel transaction's later catalog insert. +# +# - s0_insert runs in a subtransaction and produces its first WAL record. +# - s0_catinsert1 generates an xl_heap_new_cid record in that subtransaction. +# - s0_rollback aborts the subtransaction. +# - s2_vacuum reclaims the aborted row's line pointer (dead inserts are +# removed regardless of the xid horizon, so this is deterministic even +# while the toplevel transaction is still open); s0_catinsert2 then reuses +# the same tid with a different cmin. +# - The transaction stays open across both checkpoints, so the final +# s1_get_changes replays the transaction's records from the first +# checkpoint and decodes its commit for output, building the tuplecid +# hash. +setup +{ + DROP TABLE IF EXISTS tbl1; + DROP TABLE IF EXISTS user_cat; + CREATE TABLE tbl1 (val1 integer); + -- Four rows of ~1.5kB fill the first heap page to within a few hundred + -- bytes, so that vacuuming away the aborted fifth row leaves exactly one + -- line pointer for the next insert to reuse. + CREATE TABLE user_cat (c1 int, filler char(1500)) WITH (user_catalog_table = true); + INSERT INTO user_cat VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd'); +} + +teardown +{ + DROP TABLE tbl1; + DROP TABLE user_cat; + SELECT 'stop' FROM pg_drop_replication_slot('isolation_slot'); +} + +session "s0" +setup { SET synchronous_commit=on; } +step "s0_init" { SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); } +step "s0_begin" { BEGIN; } +step "s0_savepoint" { SAVEPOINT sp1; } +step "s0_insert" { INSERT INTO tbl1 VALUES (1); } +step "s0_catinsert1" { INSERT INTO user_cat VALUES (5, 'e'); } +step "s0_rollback" { ROLLBACK TO SAVEPOINT sp1; } +step "s0_catinsert2" { INSERT INTO user_cat VALUES (6, 'f'); } +step "s0_commit" { COMMIT; } + +session "s1" +setup { SET synchronous_commit=on; } +step "s1_init" { SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); } +step "s1_checkpoint" { CHECKPOINT; } +# The user_cat rows carry ~1.5kB filler values which would flood the expected +# output; filter them out (decoding still processes them server-side). +step "s1_get_changes" { SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0') WHERE data NOT LIKE '%user_cat%'; } + +session "s2" +setup { SET synchronous_commit=on; } +step "s2_vacuum" { VACUUM user_cat; } + +# The transaction commits only after the second s1_get_changes, so the last +# s1_get_changes both replays the transaction from the first checkpoint and +# decodes its commit for output. +permutation "s0_init" "s0_begin" "s0_savepoint" "s0_insert" "s1_checkpoint" "s1_get_changes" "s0_catinsert1" "s0_rollback" "s2_vacuum" "s0_catinsert2" "s1_checkpoint" "s1_get_changes" "s0_commit" "s1_get_changes" + +# Variant with the slot created while the transaction is already open: slot +# creation waits for a consistent point past the open transaction, so the +# transaction is never decoded by this slot at all (no output). This locks in +# the behavior that decoding cannot start in the middle of a transaction. +permutation "s0_begin" "s0_savepoint" "s0_insert" "s1_init" "s0_catinsert1" "s0_rollback" "s2_vacuum" "s0_catinsert2" "s0_commit" "s1_get_changes" diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index 4a739230264..f44f94d9f07 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -860,7 +860,6 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid, bool two_phase) { - int i; XLogRecPtr origin_lsn = InvalidXLogRecPtr; TimestampTz abort_time = parsed->xact_time; ReplOriginId origin_id = XLogRecGetOrigin(buf->record); @@ -892,7 +891,15 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, } else { - for (i = 0; i < parsed->nsubxacts; i++) + /* + * Remove tuplecid changes queued by the aborted subtransactions from + * the toplevel's list. + */ + ReorderBufferCleanupAbortedTupleCids(ctx->reorder, xid, + parsed->nsubxacts, + parsed->subxacts); + + for (int i = 0; i < parsed->nsubxacts; i++) { ReorderBufferAbort(ctx->reorder, parsed->subxacts[i], buf->record->EndRecPtr, abort_time); diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 94112c30501..6c4eadc9972 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -242,6 +242,8 @@ static void ReorderBufferTransferSnapToParent(ReorderBufferTXN *txn, ReorderBufferTXN *subtxn); static void AssertTXNLsnOrder(ReorderBuffer *rb); +static bool TransactionIdInSubxactArray(TransactionId xid, + TransactionId *subxacts, int nsubxacts); /* --------------------------------------- * support functions for lsn-order iterating over the ->changes of a @@ -3245,6 +3247,102 @@ ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn) ReorderBufferCleanupTXN(rb, txn); } +/* + * Remove TUPLECID changes from a top-level transaction's tuplecid list. + * + * 'xid' is a subtransaction being aborted, and 'subxacts' is the list of + * its children -- which are, obviously, also being aborted. + * + * Unlike other types of changes, tuplecid ones are always queued on the + * toplevel transaction (see ReorderBufferAddNewTupleCids), so we must search + * for these changes by subxid. Because TIDs used by a catalog tuple written + * by a subtransaction that aborts can be reused later by a future command + * of the same transaction, we must remove these TUPLECIDs from the array, + * otherwise the tuplecid hash of the historical snapshot would be corrupt. + * + * We return without cleanup if the aborting transaction is not known to this + * reorderbuffer to be a subxact. It may still have queued TUPLECIDs onto the + * toplevel xact, if this snapbuild missed the WAL record(s) carrying the + * association between the two, but that should not be a problem because in + * that case we would "skip" the transaction at DecodeCommit. + * + * (This cleanup is pointless when xid is a top-level transaction. That case + * is handled instead by ReorderBufferCleanupTXN() clearing the whole list.) + */ +void +ReorderBufferCleanupAbortedTupleCids(ReorderBuffer *rb, + TransactionId xid, + int nsubxacts, + TransactionId *subxacts) +{ + ReorderBufferTXN *txn; + ReorderBufferTXN *toptxn; + dlist_mutable_iter it; + + /* + * Search the passed-in xids for the first known subtransaction, and use + * it to find the toplevel transaction. + */ + txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr, false); + for (int i = 0; txn == NULL && i < nsubxacts; i++) + txn = ReorderBufferTXNByXid(rb, subxacts[i], false, NULL, + InvalidXLogRecPtr, false); + + /* + * Skip when nothing usable was found. No entry at all means no NEW_CID + * record from these xids was decoded by this snapbuild, so nothing was + * queued. In the latter case, we won't find the TUPLECIDs, but we don't + * care. If this is a top-level xact, this cleanup is not necessary. + */ + if (txn == NULL || !rbtxn_is_known_subxact(txn)) + return; + + /* Get the top-level transaction whose tuplecid list we're to clean */ + toptxn = rbtxn_get_toptxn(txn); + + dlist_foreach_modify(it, &toptxn->tuplecids) + { + ReorderBufferChange *change; + + change = dlist_container(ReorderBufferChange, node, it.cur); + + Assert(change->action == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID); + + /* + * Remove the entry if it was recorded on behalf of the aborting + * subtransaction or one of its subcommitted children. The subxact + * array is known sorted in logical order because it comes directly + * from the xl_xact_abort WAL record, which has them sorted that way + * by construction (Xid assign order.) + */ + if (change->data.tuplecid.subxid == xid || + TransactionIdInSubxactArray(change->data.tuplecid.subxid, + subxacts, nsubxacts)) + { + dlist_delete(&change->node); + ReorderBufferFreeChange(rb, change, false); + Assert(toptxn->ntuplecids > 0); + toptxn->ntuplecids--; + } + } +} + +/* + * Check whether the given xid exists in the subxacts array, which must be + * sorted in logical order. + */ +static bool +TransactionIdInSubxactArray(TransactionId xid, TransactionId *subxacts, + int nsubxacts) +{ + /* if there are no subxacts, it's not there */ + if (nsubxacts == 0) + return false; + /* otherwise bsearch the (logically ordered) given subxact array */ + return bsearch(&xid, subxacts, nsubxacts, + sizeof(TransactionId), xidLogicalComparator) != NULL; +} + /* * Invalidate cache for those transactions that need to be skipped just in case * catalogs were manipulated as part of the transaction. @@ -3482,15 +3580,21 @@ ReorderBufferChangeMemoryUpdate(ReorderBuffer *rb, /* * Add new (relfilelocator, tid) -> (cmin, cmax) mappings. * + * The xid is always the top-level xid, for easy hash construction in + * ReorderBufferBuildTupleCidHash. We need the subxid separately for + * cleanup during subxact aborts; see ReorderBufferCleanupAbortedTupleCids. + * * We do not include this change type in memory accounting, because we * keep CIDs in a separate list and do not evict them when reaching * the memory limit. */ void -ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, +ReorderBufferAddNewTupleCids(ReorderBuffer *rb, + TransactionId xid, TransactionId subxid, XLogRecPtr lsn, RelFileLocator locator, - ItemPointerData tid, CommandId cmin, - CommandId cmax, CommandId combocid) + ItemPointerData tid, + CommandId cmin, CommandId cmax, + CommandId combocid) { ReorderBufferChange *change = ReorderBufferAllocChange(rb); ReorderBufferTXN *txn; @@ -3502,6 +3606,7 @@ ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, change->data.tuplecid.cmin = cmin; change->data.tuplecid.cmax = cmax; change->data.tuplecid.combocid = combocid; + change->data.tuplecid.subxid = subxid; change->lsn = lsn; change->txn = txn; change->action = REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID; diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c index 69caf5b02e7..765e25b03f1 100644 --- a/src/backend/replication/logical/snapbuild.c +++ b/src/backend/replication/logical/snapbuild.c @@ -740,7 +740,7 @@ SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid, */ ReorderBufferXidSetCatalogChanges(builder->reorder, xid, lsn); - ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, lsn, + ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, xid, lsn, xlrec->target_locator, xlrec->target_tid, xlrec->cmin, xlrec->cmax, xlrec->combocid); diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c index bbb6cb87203..048450da2d3 100644 --- a/src/backend/utils/time/snapmgr.c +++ b/src/backend/utils/time/snapmgr.c @@ -160,7 +160,7 @@ TransactionId TransactionXmin = FirstNormalTransactionId; TransactionId RecentXmin = FirstNormalTransactionId; /* (table, ctid) => (cmin, cmax) mapping during timetravel */ -static HTAB *tuplecid_data = NULL; +static HTAB *HistoricTupleCidHash = NULL; /* * Elements of the active snapshot stack. @@ -1708,7 +1708,7 @@ SetupHistoricSnapshot(Snapshot historic_snapshot, HTAB *tuplecids) HistoricSnapshot = historic_snapshot; /* setup (cmin, cmax) lookup hash */ - tuplecid_data = tuplecids; + HistoricTupleCidHash = tuplecids; } @@ -1719,7 +1719,7 @@ void TeardownHistoricSnapshot(bool is_error) { HistoricSnapshot = NULL; - tuplecid_data = NULL; + HistoricTupleCidHash = NULL; } bool @@ -1732,7 +1732,7 @@ HTAB * HistoricSnapshotGetTupleCids(void) { Assert(HistoricSnapshotActive()); - return tuplecid_data; + return HistoricTupleCidHash; } /* diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index ff825e4b7b2..cc47b472db3 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -136,8 +136,12 @@ typedef struct ReorderBufferChange CommandId command_id; /* - * New cid mapping for catalog changing transaction, set when action - * == *_INTERNAL_TUPLECID. + * tuplecid changes are used to follow catalog tuple movements (i.e., + * DDL changes); we need them to create the HistoricTupleCidHash table + * used with historical snapshots. For historical reasons we store + * these with the toplevel txn, not each subtxn; and yet we need to + * remove them on subxact aborts, so we record the subxid separately. + * Set when action == *_INTERNAL_TUPLECID. */ struct { @@ -146,6 +150,7 @@ typedef struct ReorderBufferChange CommandId cmin; CommandId cmax; CommandId combocid; + TransactionId subxid; } tuplecid; /* Invalidation. */ @@ -740,6 +745,10 @@ extern void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr TimestampTz abort_time); extern void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid); extern void ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn); +extern void ReorderBufferCleanupAbortedTupleCids(ReorderBuffer *rb, + TransactionId xid, + int nsubxacts, + TransactionId *subxacts); extern void ReorderBufferInvalidate(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn); extern void ReorderBufferSetBaseSnapshot(ReorderBuffer *rb, TransactionId xid, @@ -749,6 +758,7 @@ extern void ReorderBufferAddSnapshot(ReorderBuffer *rb, TransactionId xid, extern void ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, CommandId cid); extern void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, + TransactionId subxid, XLogRecPtr lsn, RelFileLocator locator, ItemPointerData tid, CommandId cmin, CommandId cmax, CommandId combocid);