From beff3cd00989e2d532cde376da05be42eeb98577 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=81lvaro=20Herrera?= Date: Wed, 14 Oct 2026 19:42:04 +0200 Subject: [PATCH] Reorderbuffer: handle catalog-change-bearing aborted subxacts better MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When constructing a historic snapshot for logical decoding, reorderbuffer needs to keep track of movement of tuples in catalog tables. For the most part this works correctly, except that when a subtransaction that wrote a catalog tuple aborts, we failed to "forget" about that change. There's no known reports of anything actually breaking except assertion failures, but it is likely possible to construct cases where historical snapshots produce bogus visibility answers, potentially causing incorrect decoding output. To fix, have the decode of a subtransaction abort go over the TUPLECID list that has been accumulated on the ReorderBufferTXN for its topmost parent transaction, and remove all the entries that were marked as originating from the aborted subtransaction or its children. There are some little details in which various devils lie, which the added code comments explain (we hope). Add a couple of isolation test cases that trip the assertions with the original coding. They're not ideal test design, because they pass in non-assert builds without the fix. But we'd rather have the fix be verified in that limited scenario than nowhere at all, and they are cheap enough. This bug is sibling to the one fixed in commit 8c67d29fd51c and, like that one, it goes all the way back to the introduction of reorderbuffers, so backpatch to all live branches. While reviewing this, I (Álvaro) was bothered by the snapmgr.c tuplecid_data file-scoped variable, as it's easily confused with a function parameter used widely in reorderbuffer.c. In branches 19 and master, rename that to avoid confusion. Reported-by: Arseny Sher Reported-by: Mark Dilger Reported-by: Alexander Kozhemyakin Author: Bingshuai Li Author: Hayato Kuroda Author: Zhijie Hou Backpatch-through: 14 Bug: #19555 Discussion: https://postgr.es/m/87r2c8xocq.fsf@ars-thinkpad Discussion: https://postgr.es/m/CAHgHdKu5e3XY5e90Tuaxq_R4WrKxSV734Q%2BLwo5y39Omp2A-Gg@mail.gmail.com Discussion: https://postgr.es/m/19555-3c700f40a13045cd@postgresql.org --- contrib/test_decoding/Makefile | 3 +- .../expected/tuplecid_nested.out | 32 +++++ .../expected/tuplecid_restart.out | 68 +++++++++++ contrib/test_decoding/meson.build | 2 + .../test_decoding/specs/tuplecid_nested.spec | 77 ++++++++++++ .../test_decoding/specs/tuplecid_restart.spec | 74 ++++++++++++ src/backend/replication/logical/decode.c | 11 +- .../replication/logical/reorderbuffer.c | 111 +++++++++++++++++- src/backend/replication/logical/snapbuild.c | 2 +- src/backend/utils/time/snapmgr.c | 8 +- src/include/replication/reorderbuffer.h | 14 ++- 11 files changed, 389 insertions(+), 13 deletions(-) create mode 100644 contrib/test_decoding/expected/tuplecid_nested.out create mode 100644 contrib/test_decoding/expected/tuplecid_restart.out create mode 100644 contrib/test_decoding/specs/tuplecid_nested.spec create mode 100644 contrib/test_decoding/specs/tuplecid_restart.spec 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);