Hi,
On Sat, Oct 31, 2026 at 06:34 AM Masahiko Sawada wrote:
It seems to me worth adding a check for tablesync workers since both
the apply workers and the tablesync workers call
walrcv_startstreaming() and it would prevent mistakes by future
changes.
I've attached the updated patch.
While reviewing how transactional logical messages interact with
subtransaction aborts, I noticed an issue with streamed transactions. In
non-streaming mode, it works as expected (i.e., the message is not
applied if the subtransaction is aborted). However, in streaming mode,
if a transactional message is emitted in a subtransaction that is later
aborted, the STREAM ABORT message does not discard or roll back that
logical message, so it still gets applied in both streaming=on and
streaming=parallel modes.
I think the reason is that ReorderBufferApplyMessage() passes the
top-level txn rather than change->txn to rb->stream_message(), so
logical decoding always sends the top-level XID instead of the
subtransaction XID when outputting a streamed message. Thus the apply
worker cannot associate the message with the subtransaction and cannot
roll it back properly.
Strictly speaking, this is an existing issue, but it becomes more
visible once subscribers can handle logical messages via this hook. So
it may be worth considering whether to fix it along with this patch or
separately.
Attached 0002 adds a test diff in 001_basic.pl to show the issue when
streaming a message in an aborted subtransaction. 0001 is a draft patch
showing that passing the subtransaction of the message when streaming
allows the message to be rolled back properly (we may not necessarily
fix it in this exact style, just sharing it to show the behavior change).
--
Best Regards,
Zhijie Hou
From 01d4606950996fd8d5d7e2af2b5a19a752b561c2 Mon Sep 17 00:00:00 2001
From: Zhijie Hou <[email protected]>
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)
From 4b377025db31174f8f2a43123f0b2c4205b240ed Mon Sep 17 00:00:00 2001
From: Zhijie Hou <[email protected]>
Date: Sat, 31 Oct 2026 16:28:45 +0800
Subject: [PATCH] Top-up for v6-0001: Test streamed subtransaction rollback
with logical messages.
Verify in 001_basic.pl that a transactional logical message emitted in a
streamed subtransaction that later rolls back to a savepoint is not
applied under both streaming = on and streaming = parallel.
---
src/test/modules/test_logicalmsg_hooks/t/001_basic.pl | 7 +++++++
1 file changed, 7 insertions(+)
diff --git a/src/test/modules/test_logicalmsg_hooks/t/001_basic.pl
b/src/test/modules/test_logicalmsg_hooks/t/001_basic.pl
index a9ff6171ad6..59d38769c03 100644
--- a/src/test/modules/test_logicalmsg_hooks/t/001_basic.pl
+++ b/src/test/modules/test_logicalmsg_hooks/t/001_basic.pl
@@ -124,7 +124,10 @@ foreach my $mode ('on', 'parallel')
$pub->safe_psql(
'postgres', qq{
BEGIN;
+SAVEPOINT sp;
+SELECT pg_logical_emit_message(true, 'streaming_subxact_aborted_$mode', 'not
delivered');
INSERT INTO test SELECT generate_series(1000, 6000);
+ROLLBACK TO SAVEPOINT sp;
SELECT pg_logical_emit_message(true, 'streaming_$mode', 'streamed message');
COMMIT;
});
@@ -136,6 +139,10 @@ COMMIT;
q{SELECT stream_txns > 0 FROM pg_stat_replication_slots WHERE
slot_name = 'sub'}
) or die "transaction was not streamed";
+ is(logged_count("streaming_subxact_aborted_$mode"),
+ 0,
+ "message of an aborted streamed subtransaction is not applied
(streaming = $mode)"
+ );
is(logged_count("streaming_$mode"),
1,
"message in a streamed transaction is applied (streaming =
$mode)");
--
2.54.0 (Apple Git-157)
