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

On Mon, Aug 3, 2026 at 8:03 PM Fujii Masao <masao(dot)fujii(at)gmail(dot)com> wrote:
>
> 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!

Thank you for reviewing 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

Fixed.

> 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?

Good catch. We can always disable messages for tablesync workers.

>
>
> - 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.

Fixed.

>
>
> + /*
> + * 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?

True. In addition to that, I think we should call
replorigin_xact_clear(false) before the commit. Will fix it.

> +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?

Agreed.

>
> + 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?

Yes. It supports only text-format messages, which seems fine to me as
we can ensure that only text messages are emitted in this test module.
Do you think it's better to make it work in more general cases?

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

Fixed.

Regards,

--
Masahiko Sawada
Amazon Web Services: https://aws.amazon.com