Re: Add a hook for handling logical decoding messages on subscribers.

On Tue, Aug 4, 2026 at 9:42 AM Masahiko Sawada <sawada(dot)mshk(at)gmail(dot)com> wrote:
> I've rebased and updated the patch. Please review it.

Thanks for the patch!

I tried running the tests with it, and t/002_pg_dump.pl failed as follows:

t/002_pg_dump.pl .. 12151/? # Looks like you failed 26 tests of 12881.
t/002_pg_dump.pl .. Dubious, test returned 26 (wstat 6656, 0x1a00)
Failed 26/12881 subtests

Test Summary Report
-------------------
t/002_pg_dump.pl (Wstat: 6656 Tests: 12881 Failed: 26)
Failed tests: 201, 468, 735, 1269, 1803, 2074, 2348, 2620
2891, 3158, 3425, 3692, 3959, 4226, 5027
5294, 5561, 5829, 6096, 6630, 7432, 7699
8767, 9034, 10637, 11171
Non-zero exit status: 26
Files=1, Tests=12881, 31 wallclock secs ( 0.56 usr 0.10 sys + 4.81
cusr 1.72 csys = 7.19 CPU)
Result: FAIL

options->proto.logical.origin = pstrdup(MySubscription->origin);
+ options->proto.logical.messages = MySubscription->message;

Does this mean the tablesync worker also requests messages = true, even though
it always discards logical decoding messages in apply_handle_message()? If so,
that seems like unnecessary overhead. Shouldn't the tablesync worker always
request messages = false?

- subsynccommit, subwalrcvtimeout, subpublications, suborigin)
+ subsynccommit, subwalrcvtimeout, subpublications, suborigin,
+ submessage)

In pg_subscription, submessage is defined just before submaxretention, but here
it's added at the end of the column list. It's not a bug, but it would be better
to keep the ordering consistent.

+ /*
+ * A transactional message is applied as a step of the remote transaction
+ * that emitted it, and is committed together with it when applying the
+ * commit message. A non-transactional message belongs to no remote
+ * transaction, so commit it here.
+ */
+ if (!msg.transactional)
+ CommitTransactionCommand();

Shouldn't store_flush_position() also be called in the non-transactional case,
so that the remote/local LSN pair is tracked just as it is during normal commit
processing? Also pgstat_report_stat() etc should be called?

+typedef void (*LogicalRepMessageHandle_hook_type) (LogicalRepMessageData *msg);
+extern PGDLLIMPORT LogicalRepMessageHandle_hook_type
LogicalRepMessageHandle_hook;

These are added to worker_internal.h. Since that header is for worker internals,
would it be better to put them in a separate header such as logicalworker.h?

+ Oid argtypes[5] = {LSNOID, BOOLOID, TEXTOID, INT4OID, TEXTOID};
+ Datum values[5];
+ int ret;
+
+ ereport(LOG,
+ (errmsg("received message: LSN %X/%08X, prefix: %s, message: %s,
transactional: %d",
+ LSN_FORMAT_ARGS(msg->lsn),
+ msg->prefix, msg->message, msg->transactional)));

pg_logical_emit_message() can emit binary messages, but test_logicalmsg_hook.c
doesn't seem to handle them correctly. Is that intentional?

Should psql tab completion also be updated for the new message option
in CREATE/ALTER SUBSCRIPTION ... WITH/SET?

Regards,

--
Fujii Masao