From 01d4606950996fd8d5d7e2af2b5a19a752b561c2 Mon Sep 17 00:00:00 2001 From: Zhijie Hou Date: Sat, 31 Oct 2026 19:55:25 +0800 Subject: [PATCH] Fix subtransaction XID in streamed logical messages. When streaming a transactional logical decoding message emitted inside a subtransaction, ReorderBufferApplyMessage() passed the top-level ReorderBufferTXN to stream_message_cb instead of the subtransaction's ReorderBufferTXN (change->txn). As a result, stream_message_cb_wrapper() and pgoutput_message() recorded the top-level XID instead of the subtransaction XID for the streamed message, unlike pgoutput_change() and pgoutput_truncate(). Any downstream consumer of streamed logical messages therefore could not associate the message with its subtransaction or roll it back upon receiving STREAM ABORT for a rolled-back savepoint. Fix this by passing change->txn to rb->stream_message() in ReorderBufferApplyMessage(). Also update pg_decode_stream_message() in test_decoding to look up output_plugin_private on the top-level transaction via rbtxn_get_toptxn(txn), matching pg_decode_stream_abort(). --- 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.54.0 (Apple Git-157)