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

On Mon, Aug 3, 2026 at 8:04 PM Bharath Rupireddy
<bharath(dot)rupireddyforpostgres(at)gmail(dot)com> wrote:
>
> Hi,
>
> On Mon, Aug 3, 2026 at 5:42 PM Masahiko Sawada <sawada(dot)mshk(at)gmail(dot)com> wrote:
> >
> > Thank you for sharing the concrete use case. This is one of the use
> > cases I initially imagined.
> >
> > I've rebased and updated the patch. Please review it.
>
> Thanks, Sawada-san, for the v2 patch.
>
> I reviewed it and here are some comments:

Thank you for reviewing the patch!

>
> 1/
> + * The handler runs in the apply worker, inside a transaction, with an active
> + * snapshot pushed. For a transactional message the handler's work is part of
> + * the remote transaction and commits with it; a non-transactional message is
> + * committed as an independent unit. The same message may be delivered more
> + * than once, so the handler must be idempotent.
>
> Nice call-out about the more-than-once delivery and the idempotency
> requirement. It's true for logical replication in general, but would
> it be worth mentioning near the option description in
> create_subscription.sgml too?

The patch already mentions in the doc:

+ <para>
+ Received messages are only acted upon if an extension installed on
+ the subscriber has registered a handler for them; otherwise they are
+ received and discarded. A message may be passed to the handler more
+ than once, for example after an apply worker restart, so a handler
+ must be prepared to process the same message repeatedly.
+ </para>

Does it work for you?

>
> 2/
> + * well-defined ordering between a message and the initial copy. Leave
> + * them to the (parallel) apply workers. Logical messages are handled only
> + * the (parallel) apply workers
>
> Could you fix the last two sentences? They say the same thing.

Fixed.

>
> 3/
> + /*
> + * The log table is created by the test, not by this module, so it may not
>
> +#define LOG_SCHEMA "public"
> +#define LOG_TABLE "test_logicalmsg_log"
>
> Hard-coding is fine for a test module, but would it be better to have
> the module create the log table itself on first use (via SPI, or
> heap_create_with_catalog) instead of relying on the test to create it
> beforehand?

Agreed.

>
> 4/
> + msg_data->prefix = pstrdup(pq_getmsgstring(in));
> + msg = palloc(len + 1);
>
> I understand this isn't a memory leak, because ApplyMessageContext is
> reset after every message is handled at the end of the apply loop in
> LogicalRepApplyLoop(). But if the intention were to hand the caller
> the message in long-lived memory, this palloc in ApplyMessageContext
> doesn't achieve that either, since it gets reset after every message
> anyway. So why not hand the hook the message directly from the
> received StringInfo input buffer and let hook implementors copy it
> into whatever context they need?
>
> The point is just that the palloc here doesn't give us anything (if
> I'm not missing something), so I'm wondering whether we can avoid it.

It's just to make the message handling consistent with the rest of
proto.c does (see logicalrep_read_tuple()). I don't think it would be
common for the hook to keep the received message in a long-lived
memory context, so the current API seems fine to me.

> 5/
> +/*
> + * Module load callback
> + */
> +void
> +_PG_init(void)
> +{
> + LogicalRepMessageHandle_hook = &test_logical_message_handler;
> +}
>
> This overwrites LogicalRepMessageHandle_hook without saving or calling
> the previous value, so any previously registered hook is lost. Is it
> intentional? It's just a test module, but others may use it as a
> template.

Agreed.

>
> 6/
> + * test_logicalmsg_hooks.c
> + * Code for testing LogicalRepMessageHandle_hook
> + *
> + * The handler records every logical decoding message it receives into the
> + * table public.test_logicalmsg_log, so that tests can assert on the contents
> + * of a table rather than on the server log. Recording the messages this way
>
> This is fine, but I personally like the idea Amit posted above: a
> simple DDL replication example on the publisher and subscriber. Why
> not have that instead? I'm aware it would mean more LoC for the test
> extension, but that's fine IMO, since I see DDL replication as a good
> use case to demonstrate in the test module.

While query-shipping style DDL replication can be implemented using
this hook, it would not be a great solution to me as neither a test
module nor a contrib module. For a test module, it would need too much
things to be implemented as a test module and people would not use it
as it's not shipped in the package. As for implementing it as a
contrib module, I think the module would not be integrated with the
built-in logical replication well. For example, the extension cannot
respect the table filter and publication options like
publish_via_partition_root etc. Also, if we support DDL replication in
the built-in logical replication, the contrib module would no longer
be necessary but it would be hard to remove it as people might be
using it.

>
> 7/
> + /*
> + * 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();
>
> Is it safe to commit the transaction that the non-transactional
> message was emitted from, given that transaction may have already done
> some work?

I think that non-transactional messages are never interleaved with the
remote transaction. We emit a non-transactional message as soon as we
decode it (see ReorderBufferQueueMessage()) instead of buffering it.

> I haven't verified this in depth, but going by intuition:
>
> BEGIN;
> INSERT ...
> UPDATE ...
> DELETE ...
> emit non-transactional message --> apply worker commits here
> UPDATE ...
> DELETE ...
> COMMIT;
>
> Is something like this safe?

In this case, the non-transaction message is sent first and the
subscriber handles it as a single separate transaction, and then the
transaction without the message is sent to the subscriber. I think it
works fine.

Regards,

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