From 40cb75de37aee10c2d9b9e8b767e3b269c5770d0 Mon Sep 17 00:00:00 2001 From: Andrey Rachitskiy Date: Thu, 13 Aug 2026 22:44:00 +0500 Subject: [PATCH] Don't mark discarded aborted subxacts as streamed. ReorderBufferTruncateTXN is also used when discarding already-aborted transactions at eviction. Marking every subxact with in-memory changes as streamed in that path made a later abort emit stream_abort to clients that never enabled streaming. BUG #19616 Reported-by: Tyler Smart Author: Andrey Rachitskiy Discussion: https://www.postgresql.org/message-id/19616-f6153af509910853%40postgresql.org --- contrib/test_decoding/expected/spill.out | 35 +++++++++++++++++++++++++ contrib/test_decoding/sql/spill.sql | 18 +++++++++++++ src/backend/replication/logical/reorderbuffer.c | 10 ++++++- 3 files changed, 62 insertions(+), 1 deletion(-) diff --git a/contrib/test_decoding/expected/spill.out b/contrib/test_decoding/expected/spill.out index 10734bdb6af..1b56615af01 100644 --- a/contrib/test_decoding/expected/spill.out +++ b/contrib/test_decoding/expected/spill.out @@ -247,6 +247,41 @@ GROUP BY 1 ORDER BY 1; 'serialize-nested-subbig-subbigabort-subbig-3 | 5000 | table public.spill_test: INSERT: data[text]:'serialize-nested-subbig-subbigabort-subbig-3:5001' | table public.spill_test: INSERT: data[text]:'serialize-nested-subbig-subbigabort-subbig-3:10000' (2 rows) +-- Aborted xact discarded at eviction, with a subxact still in memory. +-- proto_version 1 must not see Stream Abort ('A'). +CREATE PUBLICATION spill_pub FOR TABLE spill_test; +SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot_pgoutput', 'pgoutput'); + ?column? +---------- + init +(1 row) + +BEGIN; +SAVEPOINT s; +INSERT INTO spill_test VALUES ('subtransaction-change'); +RELEASE SAVEPOINT s; +INSERT INTO spill_test SELECT repeat('x', 1000) FROM generate_series(1, 5000) g(i); +ROLLBACK; +INSERT INTO spill_test VALUES ('after-abort'); +SELECT chr(get_byte(data, 0)) AS msgtype, count(*) +FROM pg_logical_slot_peek_binary_changes('regression_slot_pgoutput', NULL, NULL, + 'proto_version', '1', 'publication_names', 'spill_pub') +GROUP BY 1 ORDER BY 1; + msgtype | count +---------+------- + B | 1 + C | 1 + I | 1 + R | 1 +(4 rows) + +SELECT pg_drop_replication_slot('regression_slot_pgoutput'); + pg_drop_replication_slot +-------------------------- + +(1 row) + +DROP PUBLICATION spill_pub; DROP TABLE spill_test; SELECT pg_drop_replication_slot('regression_slot'); pg_drop_replication_slot diff --git a/contrib/test_decoding/sql/spill.sql b/contrib/test_decoding/sql/spill.sql index e638cacd3f9..3f0b77ac269 100644 --- a/contrib/test_decoding/sql/spill.sql +++ b/contrib/test_decoding/sql/spill.sql @@ -174,6 +174,24 @@ SELECT (regexp_split_to_array(data, ':'))[4] COLLATE "C", COUNT(*), (array_agg(d FROM pg_logical_slot_get_changes('regression_slot', NULL,NULL) WHERE data ~ 'INSERT' GROUP BY 1 ORDER BY 1; +-- Aborted xact discarded at eviction, with a subxact still in memory. +-- proto_version 1 must not see Stream Abort ('A'). +CREATE PUBLICATION spill_pub FOR TABLE spill_test; +SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot_pgoutput', 'pgoutput'); +BEGIN; +SAVEPOINT s; +INSERT INTO spill_test VALUES ('subtransaction-change'); +RELEASE SAVEPOINT s; +INSERT INTO spill_test SELECT repeat('x', 1000) FROM generate_series(1, 5000) g(i); +ROLLBACK; +INSERT INTO spill_test VALUES ('after-abort'); +SELECT chr(get_byte(data, 0)) AS msgtype, count(*) +FROM pg_logical_slot_peek_binary_changes('regression_slot_pgoutput', NULL, NULL, + 'proto_version', '1', 'publication_names', 'spill_pub') +GROUP BY 1 ORDER BY 1; +SELECT pg_drop_replication_slot('regression_slot_pgoutput'); +DROP PUBLICATION spill_pub; + DROP TABLE spill_test; SELECT pg_drop_replication_slot('regression_slot'); diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 6aed6346366..c1e22becd97 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -1674,7 +1674,15 @@ ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, bool txn_prep Assert(rbtxn_is_known_subxact(subtxn)); Assert(subtxn->nsubtxns == 0); - ReorderBufferMaybeMarkTXNStreamed(rb, subtxn); + /* + * Don't mark subxacts as streamed unless the top-level xact already + * is. Streaming callers set that flag before we get here. This + * routine is also used to throw away already-aborted xacts that were + * never sent. Marking those would make abort emit stream_abort for + * XIDs the downstream has never heard of. + */ + if (rbtxn_is_streamed(txn)) + ReorderBufferMaybeMarkTXNStreamed(rb, subtxn); ReorderBufferTruncateTXN(rb, subtxn, txn_prepared); } -- 2.53.0