diff --git a/contrib/test_decoding/expected/stream.out b/contrib/test_decoding/expected/stream.out index 9879e02ca84..0ec5c933610 100644 --- a/contrib/test_decoding/expected/stream.out +++ b/contrib/test_decoding/expected/stream.out @@ -134,6 +134,29 @@ SELECT count(*) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, (1 row) RESET debug_logical_replication_streaming; +-- bug #19616 +-- +-- An aborted top-level transaction that is discarded at eviction must not +-- leave its subtransactions marked as streamed. Otherwise, decoding its +-- abort record invokes the stream abort callback for a subtransaction the +-- output plugin has never seen, even though streaming was never requested. +-- The trailing committed transaction is required to flush the ROLLBACK +-- record; without it decoding would stop before reaching the abort. +BEGIN; +SAVEPOINT s; +INSERT INTO stream_test VALUES ('subxact-change'); +RELEASE SAVEPOINT s; +INSERT INTO stream_test SELECT 'toplevel-change' || g.i FROM generate_series(1, 5000) g(i); +ROLLBACK; +INSERT INTO stream_test VALUES ('after-abort'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + data +---------------------------------------------------------------------------- + BEGIN + table public.stream_test: INSERT: data[text]:'after-abort' i[integer]:null + COMMIT +(3 rows) + DROP TABLE stream_test; SELECT pg_drop_replication_slot('regression_slot'); pg_drop_replication_slot diff --git a/contrib/test_decoding/sql/stream.sql b/contrib/test_decoding/sql/stream.sql index f1269403e0a..5e45a8e9b64 100644 --- a/contrib/test_decoding/sql/stream.sql +++ b/contrib/test_decoding/sql/stream.sql @@ -65,5 +65,22 @@ COMMIT; SELECT count(*) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'stream-changes', '1'); RESET debug_logical_replication_streaming; +-- bug #19616 +-- +-- An aborted top-level transaction that is discarded at eviction must not +-- leave its subtransactions marked as streamed. Otherwise, decoding its +-- abort record invokes the stream abort callback for a subtransaction the +-- output plugin has never seen, even though streaming was never requested. +-- The trailing committed transaction is required to flush the ROLLBACK +-- record; without it decoding would stop before reaching the abort. +BEGIN; +SAVEPOINT s; +INSERT INTO stream_test VALUES ('subxact-change'); +RELEASE SAVEPOINT s; +INSERT INTO stream_test SELECT 'toplevel-change' || g.i FROM generate_series(1, 5000) g(i); +ROLLBACK; +INSERT INTO stream_test VALUES ('after-abort'); +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); + DROP TABLE stream_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..6567dd52aed 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -2132,26 +2132,36 @@ 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->nentries_mem != 0)) + if (rbtxn_is_toptxn(txn)) + { + /* We only reach here when streaming is supported. */ + Assert(ReorderBufferCanStream(rb)); + 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 (txn->nentries_mem != 0 && rbtxn_is_streamed(rbtxn_get_toptxn(txn))) txn->txn_flags |= RBTXN_IS_STREAMED; }