On 2026-Oct-09, Álvaro Herrera wrote: > On 2026-10-09, Zhijie Hou wrote: > > > Thanks for the explanation. I re-read the code and comments and have some > > suggestions to simplify the code and improve some comments. I think we can > > reduce the code in Kuroda-san's patch with a simpler loop: > > > > 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); > > > > followed by a single early return covering all the skip cases: > > > > if (txn == NULL || !rbtxn_is_known_subxact(txn)) > > return; > > > > I think this makes the code cleaner and shorter, and also fixes the BF > > failure > > IIUC. > > > > I also rewrote some of the comments above > > ReorderBufferCleanupAbortedSubTxnTupleCids(), using wording that's more > > consistent with the rest of the codebase and simplifying the explanation of > > logic that isn't directly related, hoping it reads better. > > Hmm I had already made essentially the same changes locally plus a few > others. I'll share my patches over the weekend.
Thanks. It turned out that the changes I had done were not the same at all, so I ended up the coding of the new routine essentially the same as what's in your patch. But I didn't have time to fully merge my comment updates with yours, so I kept mine in the function header comment and mostly yours (but also partly mine) within the function. You can see I added an #ifdef block around around the part you found objectionable (which I also do). I will give these another read tomorrow to merge everything in a sane manner. The new comment in reorderbuffer.h was rewritten. I think I would like to add an XXX at that spot, to point out that this is a somewhat silly arrangement, and that we should instead make tuplecid changes attached to the subtransaction rather than the top-level transaction, to make the work at abort time more reasonble. (That would be a master-only patch, of course.) I think XXX comments are effective to motivate people to do that kind of work. I renamed the new function to ReorderBufferCleanupAbortedTupleCids, and moved the new subxid argument to another position. This patch also renames the tuplecid_data global in snapmgr.c to HistoricTupleCidHash. I intend to make this change in master only. I think the other name is confusing considering that reorderbuffer.c has a function parameter using the same name. -- Álvaro Herrera Breisgau, Deutschland — https://www.EnterpriseDB.com/ "Los dioses no protegen a los insensatos. Éstos reciben protección de otros insensatos mejor dotados" (Luis Wu, Mundo Anillo)
>From 6e192c8b9fbbe8469eb8ca0bd1499e3ba2f7fe77 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=81lvaro=20Herrera?= <[email protected]> Date: Sun, 11 Oct 2026 23:22:26 +0200 Subject: [PATCH v8] Fix stale tuplecid records left behind by aborted subtransactions Tuplecid changes are always queued on the toplevel transaction, so unlike regular changes they are not dropped when the subtransaction that wrote them aborts. The stale mappings corrupt ReorderBufferBuildTupleCidHash() at commit time: if a later catalog insert reuses the aborted subtransaction's line pointer (made LP_UNUSED by on-access pruning or by vacuum), the fresh record collides with the stale one with a different cmin, tripping the cmin equality assertion; where nothing reuses the tid, a stale cmax makes a still-live catalog tuple look deleted on historic snapshots. On non-assert builds there is no crash, but the stale mappings risk misjudging historic catalog visibility, which could produce incorrect decoding output. Remove the aborted subtransaction's tuplecid entries when its abort is decoded. The cleanup is not needed when the whole toplevel transaction is being aborted, because ReorderBufferCleanupTXN() frees the whole list anyway; to avoid scanning the list once per aborted subxid in that case, DecodeAbort() passes the abort record's primary xid to the new ReorderBufferCleanupSubTxnTupleCids(), which skips the scan when that xid is the toplevel of the xid being cleaned. The decision cannot be based on whether the primary xid's own association with its toplevel is known: abort records are written after the subtransaction has entered TRANS_ABORT, so they never carry the toplevel xid in their header, and an outer subtransaction that never wrote WAL of its own -- for example a savepoint that only wraps other savepoints -- can never get its association established at all. When such an outer subtransaction is rolled back, the released inner subtransactions listed in its abort record do have known associations, and their tuplecids must be removed from the surviving toplevel's list. The existing tuplecid and tuplecid_restart tests only cover aborts of subtransactions that wrote WAL themselves, which is why this shape needs its own test. Skipping the cleanup when the association of the xid being cleaned is unknown is safe because of how a slot's restart point advances: SnapBuildProcessRunningXacts() cannot move the restart point past the oldest in-progress transaction, so any pass that output-decodes the toplevel commit must have replayed the subtransaction's first record and therefore knows the association. Conversely, once the restart point has advanced past that record, the commit has already been consumed by an earlier pass and is skipped via SnapBuildXactNeedsSkip(), so any stale tuplecid entries left behind are dropped along with the transaction state and never reach ReorderBufferBuildTupleCidHash(). Add an isolation test for the nested shape: the toplevel writes WAL first, an inner subtransaction modifies a user catalog table and is released, and the outer subtransaction -- which produces no WAL of its own -- is then rolled back; with the page-fill and vacuum recipe of tuplecid_restart, the toplevel's later insert deterministically reuses the aborted subtransaction's line pointer. On unfixed assert builds the final get_changes dies with the original cmin assertion in ReorderBufferBuildTupleCidHash(); with the fix it passes. The existing tests keep covering the simple savepoint and the restart shapes. Author: Bingshuai Li <[email protected]> Author: Hayato Kuroda <[email protected]> Author: Zhijie Hou <[email protected]> Author: Mark Dilger <[email protected]> Reported-by: Alexander Kozhemyakin <[email protected]> Backpatch-through: 14 Bug: #19555 Discussion: https://postgr.es/m/cahghdku5e3xy5e90tuaxq_r4wrkxsv734q%[email protected] --- 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 | 75 ++++++++++++ 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 | 18 ++- 11 files changed, 393 insertions(+), 14 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'); <waiting ...> +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..52c80eb55eb --- /dev/null +++ b/contrib/test_decoding/specs/tuplecid_nested.spec @@ -0,0 +1,77 @@ +# 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 (BUG #19555 nested shape). +# +# 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, failing Assert(ent->cmin == change->data.tuplecid.cmin) in +# ReorderBufferBuildTupleCidHash() when the commit is decoded. +# +# - 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; on unfixed builds the stale tuplecid of the aborted inner +# subtransaction collides there with the fresh one. +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..58da414e8b7 --- /dev/null +++ b/contrib/test_decoding/specs/tuplecid_restart.spec @@ -0,0 +1,75 @@ +# Test decoding of a transaction in which a catalog-modifying subtransaction +# was rolled back, where the transaction spans a logical decoding restart +# point (BUG #19555 restart shape). +# +# 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, failing +# Assert(ent->cmin == change->data.tuplecid.cmin) in +# ReorderBufferBuildTupleCidHash() when the commit is decoded. +# +# - 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. On unfixed builds the stale tuplecid of the aborted +# subtransaction collides there with the fresh one. +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..a4e77728854 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 @@ -3157,6 +3159,109 @@ ReorderBufferAbort(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. + * +#ifdef INCLUDE_LLM_WRITTEN_COMMENTS + * If this pass does not know the association between the aborting + * subtransaction and its toplevel, there is nothing we can clean up here, and + * that's fine: such a pass must have started after the subtransaction's first + * (toplevel-xid-bearing) WAL record. A pass that will output-decode the + * toplevel commit cannot have started that late: a slot's restart point + * cannot advance past the oldest in-progress transaction + * (SnapBuildProcessRunningXacts()), so the commit-decoding pass has replayed + * the record that established the association. Conversely, once the restart + * point has advanced past that record, the commit has already been consumed + * by an earlier pass and is skipped here via SnapBuildXactNeedsSkip(), so + * stale tuplecid entries left behind in that case are dropped along with the + * transaction state without ever reaching ReorderBufferBuildTupleCidHash(). +#endif + * + * (This is pointless when xid is a top-level transaction. That case is + * handled instead by ReorderBufferCleanupTXN() removing 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 a 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 in this pass, so nothing was queued. + * An entry that is not a known subtransaction means either the toplevel + * is being aborted or this pass started after the assigning WAL record, + * so the commit is skipped here anyway. + */ + 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's part 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) +{ + /* otherwise bsearch the (logically ordered) given subxact array */ + return bsearch(&xid, subxacts, nsubxacts, + sizeof(TransactionId), xidLogicalComparator) != NULL; +} + /* * Abort all transactions that aren't actually running anymore because the * server restarted. @@ -3489,8 +3594,9 @@ ReorderBufferChangeMemoryUpdate(ReorderBuffer *rb, void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, RelFileLocator locator, - ItemPointerData tid, CommandId cmin, - CommandId cmax, CommandId combocid) + ItemPointerData tid, TransactionId subxid, + CommandId cmin, CommandId cmax, + CommandId combocid) { ReorderBufferChange *change = ReorderBufferAllocChange(rb); ReorderBufferTXN *txn; @@ -3502,6 +3608,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..0d31724746d 100644 --- a/src/backend/replication/logical/snapbuild.c +++ b/src/backend/replication/logical/snapbuild.c @@ -742,7 +742,7 @@ SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid, ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, lsn, xlrec->target_locator, xlrec->target_tid, - xlrec->cmin, xlrec->cmax, + xid, xlrec->cmin, xlrec->cmax, xlrec->combocid); /* figure out new command id */ 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..c8ec007c9f5 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. */ @@ -738,6 +743,10 @@ extern void ReorderBufferCommitChild(ReorderBuffer *rb, TransactionId xid, XLogRecPtr end_lsn); extern void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, TimestampTz abort_time); +extern void ReorderBufferCleanupAbortedTupleCids(ReorderBuffer *rb, + TransactionId xid, + int nsubxacts, + TransactionId *subxacts); extern void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid); extern void ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn); extern void ReorderBufferInvalidate(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn); @@ -750,8 +759,9 @@ extern void ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, CommandId cid); extern void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, RelFileLocator locator, - ItemPointerData tid, - CommandId cmin, CommandId cmax, CommandId combocid); + ItemPointerData tid, TransactionId subxid, + CommandId cmin, CommandId cmax, + CommandId combocid); extern void ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs); extern void ReorderBufferAddDistributedInvalidations(ReorderBuffer *rb, TransactionId xid, -- 2.47.3
