From 27712ed56a6ddcd6e05c36c201191061de690a6a Mon Sep 17 00:00:00 2001 From: Zhijie Hou Date: Sat, 31 Oct 2026 19:55:25 +0800 Subject: [PATCH v7 1/2] Fix subtransaction XID in streamed logical messages. Previously, when a transaction was streamed, a transactional logical decoding message emitted in a subtransaction was passed to the stream_message callback with the top-level transaction, whereas row changes and truncates in the same subtransaction are passed with the subtransaction. pgoutput therefore sent the message with the top-level XID, so a consumer could not tell which subtransaction it belonged to. When that subtransaction was rolled back and STREAM ABORT arrived for it, the message was not discarded with it. Fix this by passing the subtransaction, as for the other change types. Although this is an oversight in commit 7259736a6e5, we don't backpatch the fix. An output plugin written after test_decoding's example would crash the same way on a subtransaction, and no in-core consumer acts on streamed messages in the back branches, so we don't think the fix is worth that risk there. Author: Zhijie Hou Reviewed-by: Masahiko Sawada Discussion: https://postgr.es/m/b9b5bd27-6416-48b4-8f73-c5a60e2c97e2@gmail.com --- contrib/test_decoding/expected/stream.out | 28 +++++++++++++++++++ contrib/test_decoding/sql/stream.sql | 20 +++++++++++++ contrib/test_decoding/test_decoding.c | 11 ++++++-- .../replication/logical/reorderbuffer.c | 10 ++++++- 4 files changed, 66 insertions(+), 3 deletions(-) diff --git a/contrib/test_decoding/expected/stream.out b/contrib/test_decoding/expected/stream.out index 0ec5c933610..f34a07ba761 100644 --- a/contrib/test_decoding/expected/stream.out +++ b/contrib/test_decoding/expected/stream.out @@ -157,6 +157,34 @@ SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'inc COMMIT (3 rows) +-- Verify that a transactional message emitted in a streamed subtransaction +-- is associated with the subtransaction's XID. +SET debug_logical_replication_streaming = immediate; +BEGIN; +SAVEPOINT sp; +SELECT 'msg_subxact' FROM pg_logical_emit_message(true, 'test', 'msg'); + ?column? +------------- + msg_subxact +(1 row) + +ROLLBACK TO SAVEPOINT sp; +COMMIT; +WITH changes AS ( + SELECT xid, data + FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, + 'include-xids', '0', 'skip-empty-xacts', '1', 'stream-changes', '1') +) +SELECT m.xid = a.xid AS subxid_matches +FROM changes m, changes a +WHERE m.data LIKE 'streaming message:%' + AND a.data = 'aborting streamed (sub)transaction'; + subxid_matches +---------------- + t +(1 row) + +RESET debug_logical_replication_streaming; 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 5e45a8e9b64..ec68efed421 100644 --- a/contrib/test_decoding/sql/stream.sql +++ b/contrib/test_decoding/sql/stream.sql @@ -82,5 +82,25 @@ 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'); +-- Verify that a transactional message emitted in a streamed subtransaction +-- is associated with the subtransaction's XID. +SET debug_logical_replication_streaming = immediate; +BEGIN; +SAVEPOINT sp; +SELECT 'msg_subxact' FROM pg_logical_emit_message(true, 'test', 'msg'); +ROLLBACK TO SAVEPOINT sp; +COMMIT; + +WITH changes AS ( + SELECT xid, data + FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, + 'include-xids', '0', 'skip-empty-xacts', '1', 'stream-changes', '1') +) +SELECT m.xid = a.xid AS subxid_matches +FROM changes m, changes a +WHERE m.data LIKE 'streaming message:%' + AND a.data = 'aborting streamed (sub)transaction'; +RESET debug_logical_replication_streaming; + DROP TABLE stream_test; SELECT pg_drop_replication_slot('regression_slot'); diff --git a/contrib/test_decoding/test_decoding.c b/contrib/test_decoding/test_decoding.c index 78185837aef..961dc475a12 100644 --- a/contrib/test_decoding/test_decoding.c +++ b/contrib/test_decoding/test_decoding.c @@ -961,11 +961,18 @@ pg_decode_stream_message(LogicalDecodingContext *ctx, if (transactional) { TestDecodingData *data = ctx->output_plugin_private; - TestDecodingTxnData *txndata = txn->output_plugin_private; + + /* + * stream_message_cb can be called for an individual subtransaction + * but we maintain the output_plugin_private only under the toptxn so + * if this is not the toptxn then fetch the toptxn. + */ + ReorderBufferTXN *toptxn = rbtxn_get_toptxn(txn); + TestDecodingTxnData *txndata = toptxn->output_plugin_private; if (data->skip_empty_xacts && !txndata->stream_wrote_changes) { - pg_output_stream_start(ctx, data, txn, false); + pg_output_stream_start(ctx, data, toptxn, false); } txndata->xact_wrote_changes = txndata->stream_wrote_changes = true; } diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 46843985317..7d53879d18e 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -2104,10 +2104,18 @@ ReorderBufferApplyMessage(ReorderBuffer *rb, ReorderBufferTXN *txn, ReorderBufferChange *change, bool streaming) { if (streaming) - rb->stream_message(rb, txn, change->lsn, true, + { + /* + * Pass the change's transaction entry (which could be a + * subtransaction) so that the output plugin can associate the message + * with it and discard the message if the subtransaction is aborted + * later. + */ + rb->stream_message(rb, change->txn, change->lsn, true, change->data.msg.prefix, change->data.msg.message_size, change->data.msg.message); + } else rb->message(rb, txn, change->lsn, true, change->data.msg.prefix, -- 2.56.0