From d5aca2292c7b45e5b398a9c9840d117016aa4d92 Mon Sep 17 00:00:00 2001 From: Bingshuai Li Date: Wed, 30 Sep 2026 12:09:23 +0800 Subject: [PATCH] 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. 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 | 5 +- contrib/test_decoding/expected/tuplecid.out | 74 ++++++++++++++++ .../expected/tuplecid_nested.out | 32 +++++++ .../expected/tuplecid_restart.out | 68 +++++++++++++++ .../test_decoding/specs/tuplecid_nested.spec | 77 +++++++++++++++++ .../test_decoding/specs/tuplecid_restart.spec | 75 ++++++++++++++++ contrib/test_decoding/sql/tuplecid.sql | 76 ++++++++++++++++ src/backend/replication/logical/decode.c | 21 +++++ .../replication/logical/reorderbuffer.c | 86 ++++++++++++++++++- src/backend/replication/logical/snapbuild.c | 2 +- src/include/replication/reorderbuffer.h | 10 ++- 11 files changed, 521 insertions(+), 5 deletions(-) create mode 100644 contrib/test_decoding/expected/tuplecid.out 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 create mode 100644 contrib/test_decoding/sql/tuplecid.sql diff --git a/contrib/test_decoding/Makefile b/contrib/test_decoding/Makefile index 1e5f4f6b1cd..9fb90b80299 100644 --- a/contrib/test_decoding/Makefile +++ b/contrib/test_decoding/Makefile @@ -5,11 +5,12 @@ 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 \ - spill slot truncate stream stats twophase twophase_stream + 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 catalog_change_snapshot skip_snapshot_restore \ - invalidation_distribution + invalidation_distribution 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.out b/contrib/test_decoding/expected/tuplecid.out new file mode 100644 index 00000000000..0105bbb674d --- /dev/null +++ b/contrib/test_decoding/expected/tuplecid.out @@ -0,0 +1,74 @@ +-- 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 +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_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/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/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/decode.c b/src/backend/replication/logical/decode.c index c6858da9b0b..e9f2fced3ff 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -912,12 +912,33 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, } else { + /* + * Remove tuplecid changes queued by the aborted (sub)transactions + * from the toplevel's list, before the transactions are torn down. + * The abort record's primary xid tells + * ReorderBufferCleanupSubTxnTupleCids() whether the whole toplevel + * is going away, in which case scanning the list is pointless. + * + * Note that we must not try to decide that from the primary xid's + * own association instead: an abort record is written once the + * subtransaction is already in TRANS_ABORT, so it never carries + * the toplevel xid in its header (IsSubxactTopXidLogPending() + * requires IsTransactionState()), and an outer subtransaction + * that never wrote WAL of its own -- e.g. a savepoint that only + * wraps other savepoints -- can therefore have no association at + * all, while the released inner subtransactions it rolls back do + * have one and their tuplecids still need to be removed. + */ for (i = 0; i < parsed->nsubxacts; i++) { + ReorderBufferCleanupSubTxnTupleCids(ctx->reorder, + parsed->subxacts[i], + xid); ReorderBufferAbort(ctx->reorder, parsed->subxacts[i], buf->record->EndRecPtr); } + ReorderBufferCleanupSubTxnTupleCids(ctx->reorder, xid, xid); ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr); } diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 0a84443208f..f08137b862f 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -2947,6 +2947,88 @@ ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn) ReorderBufferCleanupTXN(rb, txn); } +/* + * Remove tuplecid changes queued by subtransaction xid from its 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). + * + * The cleanup is pointless when the whole toplevel transaction is + * being aborted, since ReorderBufferCleanupTXN() frees the whole list + * anyway; the caller passes the abort record's primary xid and we skip + * the scan when it is xid's toplevel. Note that the opposite decision + * -- cleaning only when the primary xid's own association is known -- + * would be wrong: abort records are written once the subtransaction is + * already in TRANS_ABORT, so they carry no toplevel xid in their + * header, and an outer subtransaction that never wrote WAL of its own + * never gets an association at all. When such an outer subtransaction + * rolls back, released inner subtransactions listed in its abort record + * do have known associations and are cleaned here. + * + * 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(). + */ +void +ReorderBufferCleanupSubTxnTupleCids(ReorderBuffer *rb, TransactionId xid, + TransactionId primary_xid) +{ + ReorderBufferTXN *txn; + ReorderBufferTXN *toptxn; + dlist_mutable_iter it; + + txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr, + false); + + /* unknown transaction, or unknown association: nothing to remove */ + if (txn == NULL || !rbtxn_is_known_subxact(txn)) + return; + + toptxn = txn->toptxn; + + /* + * The whole toplevel transaction is being aborted (the abort record's + * primary xid is xid's toplevel): ReorderBufferCleanupTXN() will free + * the whole list shortly, don't scan it once per aborted subxid. + */ + if (toptxn->xid == primary_xid) + return; + + 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); + ReorderBufferReturnChange(rb, change, false); + toptxn->ntuplecids--; + } + } +} + /* * Abort all transactions that aren't actually running anymore because the * server restarted. @@ -3259,7 +3341,8 @@ void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, RelFileNode node, ItemPointerData tid, CommandId cmin, - CommandId cmax, CommandId combocid) + CommandId cmax, CommandId combocid, + TransactionId subxid) { ReorderBufferChange *change = ReorderBufferGetChange(rb); ReorderBufferTXN *txn; @@ -3271,6 +3354,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 bcee0896500..ddf7a50a088 100644 --- a/src/backend/replication/logical/snapbuild.c +++ b/src/backend/replication/logical/snapbuild.c @@ -867,7 +867,7 @@ SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid, ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, lsn, xlrec->target_node, 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 cb633315d04..d59ea121b4f 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -150,6 +150,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. */ @@ -657,6 +662,8 @@ void ReorderBufferAssignChild(ReorderBuffer *, TransactionId, TransactionId, XL void ReorderBufferCommitChild(ReorderBuffer *, TransactionId, TransactionId, XLogRecPtr commit_lsn, XLogRecPtr end_lsn); void ReorderBufferAbort(ReorderBuffer *, TransactionId, XLogRecPtr lsn); +extern void ReorderBufferCleanupSubTxnTupleCids(ReorderBuffer *rb, TransactionId xid, + TransactionId primary_xid); void ReorderBufferAbortOld(ReorderBuffer *, TransactionId xid); void ReorderBufferForget(ReorderBuffer *, TransactionId, XLogRecPtr lsn); void ReorderBufferInvalidate(ReorderBuffer *, TransactionId, XLogRecPtr lsn); @@ -667,7 +674,8 @@ void ReorderBufferAddNewCommandId(ReorderBuffer *, TransactionId, XLogRecPtr ls CommandId cid); void ReorderBufferAddNewTupleCids(ReorderBuffer *, TransactionId, XLogRecPtr lsn, RelFileNode node, ItemPointerData pt, - CommandId cmin, CommandId cmax, CommandId combocid); + CommandId cmin, CommandId cmax, + CommandId combocid, TransactionId subxid); void ReorderBufferAddInvalidations(ReorderBuffer *, TransactionId, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs); void ReorderBufferAddDistributedInvalidations(ReorderBuffer *rb, TransactionId xid, -- 2.43.0