From b1ab73ebb6668a8e41d0e5d796ebc97dc691946b Mon Sep 17 00:00:00 2001 From: Bingshuai Li Date: Sat, 15 Aug 2026 15:41:56 +0800 Subject: [PATCH v4] 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 in ReorderBufferAbort() when the sub->top association is known. The association can be unknown in a decoding pass that started after the subtransaction's first (toplevel-xid-bearing) WAL record, e.g. when decoding restarted in the middle of the transaction: tuplecid records are dispatched on the writing subtransaction without establishing the association, and abort records never carry the toplevel xid. Skipping the cleanup in that case 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 restart shape: the transaction stays open across checkpoints, the aborted subtransaction's catalog insert is vacuumed away by another session so that the toplevel's later insert deterministically reuses the same line pointer, and the final get_changes both replays the transaction across a restart and output-decodes its commit. On unfixed assert builds the test dies deterministically with the cmin assertion in ReorderBufferBuildTupleCidHash(); with the fix it passes. The existing regression test keeps covering the continuous (no restart) shape. Bug: #19555 Reported-by: Alexander Kozhemyakin Based-on-patch-by: Mark Dilger Discussion: https://postgr.es/m/CAHgHdKu5e3XY5e90Tuaxq_R4WrKxSV734Q%2BLwo5y39Omp2A-Gg@mail.gmail.com --- contrib/test_decoding/Makefile | 6 +- contrib/test_decoding/expected/tuplecid.out | 199 ++++++++++++++++++ .../expected/tuplecid_restart.out | 68 ++++++ contrib/test_decoding/meson.build | 2 + .../test_decoding/specs/tuplecid_restart.spec | 75 +++++++ contrib/test_decoding/sql/tuplecid.sql | 76 +++++++ .../replication/logical/reorderbuffer.c | 55 ++++- src/backend/replication/logical/snapbuild.c | 2 +- src/include/replication/reorderbuffer.h | 9 +- 9 files changed, 487 insertions(+), 5 deletions(-) create mode 100644 contrib/test_decoding/expected/tuplecid.out create mode 100644 contrib/test_decoding/expected/tuplecid_restart.out create mode 100644 contrib/test_decoding/specs/tuplecid_restart.spec create mode 100644 contrib/test_decoding/sql/tuplecid.sql diff --git a/contrib/test_decoding/Makefile b/contrib/test_decoding/Makefile index 0111124399a..5876d8d739e 100644 --- a/contrib/test_decoding/Makefile +++ b/contrib/test_decoding/Makefile @@ -5,11 +5,13 @@ PGFILEDESC = "test_decoding - example of a logical decoding output plugin" REGRESS = ddl xact rewrite toast permissions decoding_in_xact \ decoding_into_rel binary prepared replorigin time messages \ - repack spill slot truncate stream stats twophase twophase_stream + repack spill slot truncate stream stats twophase twophase_stream \ + tuplecid ISOLATION = mxact delayed_startup ondisk_startup concurrent_ddl_dml \ oldest_xmin snapshot_transfer subxact_without_top concurrent_stream \ twophase_snapshot slot_creation_error catalog_change_snapshot \ - skip_snapshot_restore invalidation_distribution parallel_session_origin + skip_snapshot_restore invalidation_distribution parallel_session_origin \ + tuplecid_restart 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.out b/contrib/test_decoding/expected/tuplecid.out new file mode 100644 index 00000000000..b559dd21bd4 --- /dev/null +++ b/contrib/test_decoding/expected/tuplecid.out @@ -0,0 +1,199 @@ +-- Tests for decoding of transactions in which a catalog-modifying +-- subtransaction was rolled back (BUG #19555). +-- +-- When a subtransaction modifies catalog tuples and is then rolled back, +-- its dead heap-only line pointers can be marked LP_UNUSED by on-access +-- pruning and reused by later catalog inserts of the same top-level +-- transaction. The aborted subtransaction's xl_heap_new_cid records must +-- not be consulted when building the historic snapshot used to decode the +-- transaction; on unfixed builds they collide with the records of the +-- reused line pointer and decoding dies with +-- TRAP: failed Assert("ent->cmin == change->data.tuplecid.cmin") +-- (or, on non-assert builds, silently uses wrong cmin/cmax mappings). +-- +-- Whether the collision triggers depends on the physical layout of the +-- catalog pages, so run in a fresh database where prior tests cannot have +-- changed it, and loop enough times to leave headroom for layout +-- differences across versions. +CREATE DATABASE regression_tuplecid; +-- snapshot the original database name (make and meson use different +-- names for it) so we can switch back before dropping the test database +\set prevdb :DBNAME +\c regression_tuplecid +-- predictability +SET synchronous_commit = on; +SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding'); + ?column? +---------- + init +(1 row) + +-- Catalog churn to push the catalog pages into a state where on-access +-- pruning frees the aborted subtransaction's line pointer and the second +-- ALTER's catalog insert reuses it. +CREATE TABLE tpc_filler1(f1 char(4)); +CREATE TABLE tpc_filler2(data text); +INSERT INTO tpc_filler2 VALUES ('before-test'); +CREATE TYPE tpc_complex AS (r float8, i float8); +CREATE TABLE tpc_filler3 (a int PRIMARY KEY, b text DEFAULT 'Unspecified'); +CREATE TABLE tpc_filler4 (id serial, t text); +CREATE TABLE tpc_filler5(data text); +INSERT INTO tpc_filler5 SELECT repeat('a', 2000) || g.i FROM generate_series(1, 1) g(i); +TRUNCATE table tpc_filler5; +CHECKPOINT; +SELECT count(*) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + count +------- + 9 +(1 row) + +-- Each iteration runs the trigger shape: catalog DDL inside a rolled-back +-- subtransaction, then the same DDL again, then decode. Unfixed builds +-- typically die within a handful of iterations. +-- (temporarily hide the queries, to avoid flooding the expected output) +\set ECHO none + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + + count +------- + 0 +(1 row) + +DROP TABLE tpc_filler1; +DROP TABLE tpc_filler2; +DROP TYPE tpc_complex; +DROP TABLE tpc_filler3; +DROP TABLE tpc_filler4; +DROP TABLE tpc_filler5; +SELECT count(*) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + count +------- + 0 +(1 row) + +SELECT pg_drop_replication_slot('regression_slot'); + pg_drop_replication_slot +-------------------------- + +(1 row) + +\c :prevdb +DROP DATABASE regression_tuplecid; 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 ac655853d26..a78024587dc 100644 --- a/contrib/test_decoding/meson.build +++ b/contrib/test_decoding/meson.build @@ -42,6 +42,7 @@ tests += { 'stats', 'twophase', 'twophase_stream', + 'tuplecid', ], 'regress_args': [ '--temp-config', files('logical.conf'), @@ -66,6 +67,7 @@ tests += { 'skip_snapshot_restore', 'invalidation_distribution', 'parallel_session_origin', + 'tuplecid_restart', ], 'regress_args': [ '--temp-config', files('logical.conf'), 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/contrib/test_decoding/sql/tuplecid.sql b/contrib/test_decoding/sql/tuplecid.sql new file mode 100644 index 00000000000..335fb6d1591 --- /dev/null +++ b/contrib/test_decoding/sql/tuplecid.sql @@ -0,0 +1,76 @@ +-- Tests for decoding of transactions in which a catalog-modifying +-- subtransaction was rolled back (BUG #19555). +-- +-- When a subtransaction modifies catalog tuples and is then rolled back, +-- its dead heap-only line pointers can be marked LP_UNUSED by on-access +-- pruning and reused by later catalog inserts of the same top-level +-- transaction. The aborted subtransaction's xl_heap_new_cid records must +-- not be consulted when building the historic snapshot used to decode the +-- transaction; on unfixed builds they collide with the records of the +-- reused line pointer and decoding dies with +-- TRAP: failed Assert("ent->cmin == change->data.tuplecid.cmin") +-- (or, on non-assert builds, silently uses wrong cmin/cmax mappings). +-- +-- Whether the collision triggers depends on the physical layout of the +-- catalog pages, so run in a fresh database where prior tests cannot have +-- changed it, and loop enough times to leave headroom for layout +-- differences across versions. +CREATE DATABASE regression_tuplecid; +-- snapshot the original database name (make and meson use different +-- names for it) so we can switch back before dropping the test database +\set prevdb :DBNAME +\c regression_tuplecid + +-- predictability +SET synchronous_commit = on; + +SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding'); + +-- Catalog churn to push the catalog pages into a state where on-access +-- pruning frees the aborted subtransaction's line pointer and the second +-- ALTER's catalog insert reuses it. +CREATE TABLE tpc_filler1(f1 char(4)); +CREATE TABLE tpc_filler2(data text); +INSERT INTO tpc_filler2 VALUES ('before-test'); +CREATE TYPE tpc_complex AS (r float8, i float8); +CREATE TABLE tpc_filler3 (a int PRIMARY KEY, b text DEFAULT 'Unspecified'); +CREATE TABLE tpc_filler4 (id serial, t text); +CREATE TABLE tpc_filler5(data text); +INSERT INTO tpc_filler5 SELECT repeat('a', 2000) || g.i FROM generate_series(1, 1) g(i); +TRUNCATE table tpc_filler5; +CHECKPOINT; + +SELECT count(*) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + +-- Each iteration runs the trigger shape: catalog DDL inside a rolled-back +-- subtransaction, then the same DDL again, then decode. Unfixed builds +-- typically die within a handful of iterations. +-- (temporarily hide the queries, to avoid flooding the expected output) +\set ECHO none +SELECT format($fmt$ +CREATE TABLE tpc_cmin(data int, pad1 text, pad2 int, pad3 text); +BEGIN; +SAVEPOINT a; +ALTER TABLE tpc_cmin ALTER COLUMN data TYPE text; +ROLLBACK TO SAVEPOINT a; +ALTER TABLE tpc_cmin ALTER COLUMN data TYPE bigint; +COMMIT; +SELECT count(*) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +DROP TABLE tpc_cmin; +$fmt$) +FROM generate_series(1, 25) \gexec +\set ECHO all + +DROP TABLE tpc_filler1; +DROP TABLE tpc_filler2; +DROP TYPE tpc_complex; +DROP TABLE tpc_filler3; +DROP TABLE tpc_filler4; +DROP TABLE tpc_filler5; + +SELECT count(*) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + +SELECT pg_drop_replication_slot('regression_slot'); + +\c :prevdb +DROP DATABASE regression_tuplecid; diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 6aed6346366..27dc6a40efc 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -3140,6 +3140,57 @@ ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, /* cosmetic... */ txn->final_lsn = lsn; + /* + * If this is a subtransaction, remove any tuplecid changes it added to + * the toplevel transaction's list. Unlike regular changes, tuplecid + * changes are always queued on the toplevel transaction (see + * ReorderBufferAddNewTupleCids), so they would otherwise survive the + * abort of the subtransaction that wrote them. That's wrong: they + * describe catalog tuple versions created or killed by the aborted + * subtransaction, which never became visible, and keeping them can + * corrupt ReorderBufferBuildTupleCidHash() at commit time -- both when + * the toplevel transaction later reuses the same tid (the stale entry + * collides with the fresh one, breaking the cmin equality assumption) + * and when nothing touches the tid again (a stale cmax would make a + * still-live tuple look deleted on historic snapshots). + * + * 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(). + */ + if (rbtxn_is_known_subxact(txn)) + { + ReorderBufferTXN *toptxn = rbtxn_get_toptxn(txn); + dlist_mutable_iter it; + + dlist_foreach_modify(it, &toptxn->tuplecids) + { + ReorderBufferChange *change; + + change = dlist_container(ReorderBufferChange, node, it.cur); + + Assert(change->action == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID); + + if (change->data.tuplecid.subxid == xid) + { + dlist_delete(&change->node); + ReorderBufferFreeChange(rb, change, false); + toptxn->ntuplecids--; + } + } + } + /* remove potential on-disk data, and deallocate */ ReorderBufferCleanupTXN(rb, txn); } @@ -3477,7 +3528,8 @@ void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, RelFileLocator locator, ItemPointerData tid, CommandId cmin, - CommandId cmax, CommandId combocid) + CommandId cmax, CommandId combocid, + TransactionId subxid) { ReorderBufferChange *change = ReorderBufferAllocChange(rb); ReorderBufferTXN *txn; @@ -3489,6 +3541,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 f60bcf09605..11cc3c08ffd 100644 --- a/src/backend/replication/logical/snapbuild.c +++ b/src/backend/replication/logical/snapbuild.c @@ -703,7 +703,7 @@ SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid, ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, lsn, xlrec->target_locator, xlrec->target_tid, xlrec->cmin, xlrec->cmax, - xlrec->combocid); + xlrec->combocid, xid); /* figure out new command id */ if (xlrec->cmin != InvalidCommandId && diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index ff825e4b7b2..9d4f01e1d77 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -146,6 +146,11 @@ typedef struct ReorderBufferChange CommandId cmin; CommandId cmax; CommandId combocid; + TransactionId subxid; /* xid that wrote the tuplecid record; + * differs from the xid of the txn this + * change is queued on (always the + * toplevel xid) when a subtransaction + * modified the catalog */ } tuplecid; /* Invalidation. */ @@ -751,7 +756,9 @@ extern void ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid, extern void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, RelFileLocator locator, ItemPointerData tid, - CommandId cmin, CommandId cmax, CommandId combocid); + CommandId cmin, CommandId cmax, + CommandId combocid, + TransactionId subxid); extern void ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs); extern void ReorderBufferAddDistributedInvalidations(ReorderBuffer *rb, TransactionId xid, -- 2.43.0