From 587f0c596dbd392f78e2db433cd3c3fbb01793ef Mon Sep 17 00:00:00 2001 From: Zhijie Hou Date: Sat, 31 Oct 2026 19:55:25 +0800 Subject: [PATCH v8 1/2] Add an xid argument to the logical decoding message callbacks. Previously, when streaming an in-progress transaction, a transactional logical decoding message emitted in a subtransaction was passed to the stream_message callback with the top-level transaction only, and the callback had no way to know which subtransaction had emitted it. The change callbacks get that through ReorderBufferChange, but a message has none. pgoutput therefore sent the message with the top-level XID, and when the subtransaction was later rolled back and STREAM ABORT arrived for it, the subscriber could not discard the message with it. Oversight in commit 7259736a6e5. To fix this, we add the XID of the (sub)transaction that emitted the message as a new argument of both the message and stream_message callbacks, and have pgoutput send it. Output plugins can now see which subtransaction a message belongs to, as they already can for the other change types, while still getting the top-level transaction as before. Both callbacks get the argument so that a plugin can keep implementing them with one function, as pgoutput does. We considered fixing this in a backpatchable way, such as passing the subtransaction's ReorderBufferTXN to the callback instead of the top-level one, but that silently changes the meaning of an existing argument, and an output plugin that keeps its state on the top-level transaction would break. Adding an argument breaks the ABI, but it is the cleaner interface, so we chose it and don't backpatch. The bug exists in the back branches too, but there have been no reports, and the built-in logical replication doesn't use logical messages there. Author: Zhijie Hou Co-authored-by: Masahiko Sawada Reviewed-by: Amit Kapila 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 | 22 ++++++++------- doc/src/sgml/logicaldecoding.sgml | 14 +++++++++- src/backend/replication/logical/logical.c | 14 +++++----- .../replication/logical/reorderbuffer.c | 11 ++++---- src/backend/replication/pgoutput/pgoutput.c | 17 +++++------ src/include/replication/output_plugin.h | 10 +++++-- src/include/replication/reorderbuffer.h | 2 ++ 9 files changed, 105 insertions(+), 33 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..0f6e05b3bef 100644 --- a/contrib/test_decoding/test_decoding.c +++ b/contrib/test_decoding/test_decoding.c @@ -72,9 +72,9 @@ static void pg_decode_truncate(LogicalDecodingContext *ctx, static bool pg_decode_filter(LogicalDecodingContext *ctx, ReplOriginId origin_id); static void pg_decode_message(LogicalDecodingContext *ctx, - ReorderBufferTXN *txn, XLogRecPtr lsn, - bool transactional, const char *prefix, - Size sz, const char *message); + ReorderBufferTXN *txn, TransactionId xid, + XLogRecPtr lsn, bool transactional, + const char *prefix, Size sz, const char *message); static bool pg_decode_filter_prepare(LogicalDecodingContext *ctx, TransactionId xid, const char *gid); @@ -112,9 +112,10 @@ static void pg_decode_stream_change(LogicalDecodingContext *ctx, Relation relation, ReorderBufferChange *change); static void pg_decode_stream_message(LogicalDecodingContext *ctx, - ReorderBufferTXN *txn, XLogRecPtr lsn, - bool transactional, const char *prefix, - Size sz, const char *message); + ReorderBufferTXN *txn, TransactionId xid, + XLogRecPtr lsn, bool transactional, + const char *prefix, Size sz, + const char *message); static void pg_decode_stream_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, int nrelations, Relation relations[], @@ -753,8 +754,9 @@ pg_decode_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, static void pg_decode_message(LogicalDecodingContext *ctx, - ReorderBufferTXN *txn, XLogRecPtr lsn, bool transactional, - const char *prefix, Size sz, const char *message) + ReorderBufferTXN *txn, TransactionId xid, XLogRecPtr lsn, + bool transactional, const char *prefix, Size sz, + const char *message) { TestDecodingData *data = ctx->output_plugin_private; TestDecodingTxnData *txndata; @@ -954,8 +956,8 @@ pg_decode_stream_change(LogicalDecodingContext *ctx, */ static void pg_decode_stream_message(LogicalDecodingContext *ctx, - ReorderBufferTXN *txn, XLogRecPtr lsn, bool transactional, - const char *prefix, Size sz, const char *message) + ReorderBufferTXN *txn, TransactionId xid, XLogRecPtr lsn, + bool transactional, const char *prefix, Size sz, const char *message) { /* Output stream start if we haven't yet for transactional messages. */ if (transactional) diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgml index 6dc49108997..f21e1115a94 100644 --- a/doc/src/sgml/logicaldecoding.sgml +++ b/doc/src/sgml/logicaldecoding.sgml @@ -1128,6 +1128,7 @@ typedef bool (*LogicalDecodeFilterByOriginCB) (struct LogicalDecodingContext *ct typedef void (*LogicalDecodeMessageCB) (struct LogicalDecodingContext *ctx, ReorderBufferTXN *txn, + TransactionId xid, XLogRecPtr message_lsn, bool transactional, const char *prefix, @@ -1138,7 +1139,11 @@ typedef void (*LogicalDecodeMessageCB) (struct LogicalDecodingContext *ctx, the transaction, like the time stamp at which it has been committed and its XID. Note however that it can be NULL when the message is non-transactional and the XID was not assigned yet in the transaction - which logged the message. The lsn has WAL + which logged the message. The xid parameter is + the XID of the transaction that emitted the message; for a + transactional message emitted in a subtransaction, it is the + subtransaction's XID while txn is the top-level + transaction. The lsn has WAL location of the message. The transactional says if the message was sent as transactional or not. Similar to the change callback, in case of decoding a prepared (but yet uncommitted) @@ -1360,12 +1365,19 @@ typedef void (*LogicalDecodeStreamChangeCB) (struct LogicalDecodingContext *ctx, typedef void (*LogicalDecodeStreamMessageCB) (struct LogicalDecodingContext *ctx, ReorderBufferTXN *txn, + TransactionId xid, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size message_size, const char *message); + The txn parameter is the top-level transaction + being streamed, and xid is the XID of the + transaction that emitted the message, which is a subtransaction's XID + if it was emitted in one. For the other stream callbacks, the XID of + the subtransaction is available as + change->txn->xid. diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 8fe255fa960..44a27d0b653 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -81,7 +81,7 @@ static void change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, static void truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change); static void message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, - XLogRecPtr message_lsn, bool transactional, + TransactionId xid, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size message_size, const char *message); /* streaming callbacks */ @@ -98,7 +98,7 @@ static void stream_commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn static void stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change); static void stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, - XLogRecPtr message_lsn, bool transactional, + TransactionId xid, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size message_size, const char *message); static void stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change); @@ -1336,7 +1336,7 @@ filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, ReplOriginId origin_id) static void message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, - XLogRecPtr message_lsn, bool transactional, + TransactionId xid, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size message_size, const char *message) { LogicalDecodingContext *ctx = cache->private_data; @@ -1364,7 +1364,7 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, ctx->end_xact = false; /* do the actual work: call callback */ - ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix, + ctx->callbacks.message_cb(ctx, txn, xid, message_lsn, transactional, prefix, message_size, message); /* Pop the error context stack */ @@ -1647,7 +1647,7 @@ stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, static void stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, - XLogRecPtr message_lsn, bool transactional, + TransactionId xid, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size message_size, const char *message) { LogicalDecodingContext *ctx = cache->private_data; @@ -1674,12 +1674,12 @@ stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, /* set output state */ ctx->accept_writes = true; - ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId; + ctx->write_xid = xid; ctx->write_location = message_lsn; ctx->end_xact = false; /* do the actual work: call callback */ - ctx->callbacks.stream_message_cb(ctx, txn, message_lsn, transactional, prefix, + ctx->callbacks.stream_message_cb(ctx, txn, xid, message_lsn, transactional, prefix, message_size, message); /* Pop the error context stack */ diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 46843985317..2917d971f17 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -920,7 +920,8 @@ ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid, SetupHistoricSnapshot(snapshot_now, NULL); PG_TRY(); { - rb->message(rb, txn, lsn, false, prefix, message_size, message); + rb->message(rb, txn, xid, lsn, false, prefix, message_size, + message); TeardownHistoricSnapshot(false); } @@ -2104,13 +2105,13 @@ ReorderBufferApplyMessage(ReorderBuffer *rb, ReorderBufferTXN *txn, ReorderBufferChange *change, bool streaming) { if (streaming) - rb->stream_message(rb, txn, change->lsn, true, - change->data.msg.prefix, + rb->stream_message(rb, txn, change->txn->xid, 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, + rb->message(rb, txn, change->txn->xid, change->lsn, + true, change->data.msg.prefix, change->data.msg.message_size, change->data.msg.message); } diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index c8caa163c15..1c0a9cf09b9 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -55,9 +55,10 @@ static void pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change); static void pgoutput_message(LogicalDecodingContext *ctx, - ReorderBufferTXN *txn, XLogRecPtr message_lsn, - bool transactional, const char *prefix, - Size sz, const char *message); + ReorderBufferTXN *txn, TransactionId xid, + XLogRecPtr message_lsn, bool transactional, + const char *prefix, Size sz, + const char *message); static bool pgoutput_origin_filter(LogicalDecodingContext *ctx, ReplOriginId origin_id); static void pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx, @@ -1722,11 +1723,11 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, static void pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, - XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, - const char *message) + TransactionId xid, XLogRecPtr message_lsn, bool transactional, + const char *prefix, Size sz, const char *message) { PGOutputData *data = (PGOutputData *) ctx->output_plugin_private; - TransactionId xid = InvalidTransactionId; + TransactionId stream_xid = InvalidTransactionId; if (!data->messages) return; @@ -1736,7 +1737,7 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, * pgoutput_change. */ if (data->in_streaming) - xid = txn->xid; + stream_xid = xid; /* * Output BEGIN if we haven't yet. Avoid for non-transactional messages. @@ -1752,7 +1753,7 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, OutputPluginPrepareWrite(ctx, true); logicalrep_write_message(ctx->out, - xid, + stream_xid, message_lsn, transactional, prefix, diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h index 842fcde67f9..2f12fe09ea2 100644 --- a/src/include/replication/output_plugin.h +++ b/src/include/replication/output_plugin.h @@ -80,10 +80,14 @@ typedef void (*LogicalDecodeCommitCB) (struct LogicalDecodingContext *ctx, XLogRecPtr commit_lsn); /* - * Called for the generic logical decoding messages. + * Called for the generic logical decoding messages. "xid" is the XID of the + * (sub)transaction that emitted the message; for a transactional message + * emitted in a subtransaction it differs from txn->xid, as txn is the + * top-level transaction. */ typedef void (*LogicalDecodeMessageCB) (struct LogicalDecodingContext *ctx, ReorderBufferTXN *txn, + TransactionId xid, XLogRecPtr message_lsn, bool transactional, const char *prefix, @@ -191,10 +195,12 @@ typedef void (*LogicalDecodeStreamChangeCB) (struct LogicalDecodingContext *ctx, /* * Callback for streaming generic logical decoding messages from in-progress - * transactions. + * transactions. "txn" is the top-level transaction and "xid" the XID of the + * (sub)transaction that emitted the message. */ typedef void (*LogicalDecodeStreamMessageCB) (struct LogicalDecodingContext *ctx, ReorderBufferTXN *txn, + TransactionId xid, XLogRecPtr message_lsn, bool transactional, const char *prefix, diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index cc47b472db3..612a7a81f40 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -500,6 +500,7 @@ typedef void (*ReorderBufferCommitCB) (ReorderBuffer *rb, /* message callback signature */ typedef void (*ReorderBufferMessageCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, + TransactionId xid, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, @@ -559,6 +560,7 @@ typedef void (*ReorderBufferStreamChangeCB) (ReorderBuffer *rb, /* stream message callback signature */ typedef void (*ReorderBufferStreamMessageCB) (ReorderBuffer *rb, ReorderBufferTXN *txn, + TransactionId xid, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, -- 2.56.0