From c2f7b1d3e5a69b0c8f4d2e1a3b7c9e5d0f8a2b46 Mon Sep 17 00:00:00 2001 From: Andrey Rachitskiy Date: Fri, 14 Aug 2026 09:27:00 +0500 Subject: [PATCH v3] 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. Mark a subxact as streamed only when it has changes and its top-level transaction is already marked as streamed. The streaming call sites mark the top-level xact first, so this keeps their behavior. The abort-discard path never marks the top-level xact, so its subxacts are left unmarked. BUG #19616 Reported-by: Tyler Smart Author: Andrey Rachitskiy Reviewed-by: Hayato Kuroda (Fujitsu) 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 | 36 +++++++++++-------- 3 files changed, 75 insertions(+), 14 deletions(-) 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..107c2db68c1 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -2132,26 +2132,34 @@ ReorderBufferSaveTXNSnapshot(ReorderBuffer *rb, ReorderBufferTXN *txn, } /* - * Mark the given transaction as streamed if it's a top-level transaction - * or has changes. + * Mark the given transaction as streamed, if appropriate. + * + * A top-level transaction is always marked. A subtransaction is marked + * only when it has changes and its top-level transaction is already + * marked as streamed. */ static void ReorderBufferMaybeMarkTXNStreamed(ReorderBuffer *rb, ReorderBufferTXN *txn) { /* - * The top-level transaction, is marked as streamed always, even if it - * does not contain any changes (that is, when all the changes are in - * subtransactions). - * - * For subtransactions, we only mark them as streamed when there are - * changes in them. - * - * We do it this way because of aborts - we don't want to send aborts for - * XIDs the downstream is not aware of. And of course, it always knows - * about the top-level xact (we send the XID in all messages), but we - * never stream XIDs of empty subxacts. + * The top-level transaction is marked as streamed always, even if it does + * not contain any changes (that is, when all the changes are in + * subtransactions). The downstream always knows about it, since we send + * its XID in every message. + */ + if (rbtxn_is_toptxn(txn)) + { + txn->txn_flags |= RBTXN_IS_STREAMED; + return; + } + + /* + * A subtransaction is marked only when it has changes, and only when its + * top-level transaction has already been marked as streamed. We never + * stream XIDs of empty subxacts, and we must not send an abort for an XID + * the downstream has never heard of. */ - if (rbtxn_is_toptxn(txn) || (txn->nentries_mem != 0)) + if (txn->nentries_mem != 0 && rbtxn_is_streamed(rbtxn_get_toptxn(txn))) txn->txn_flags |= RBTXN_IS_STREAMED; } -- 2.53.0