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

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:

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?

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.

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?

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.

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.

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.

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

--
Bharath Rupireddy
Amazon Web Services: https://aws.amazon.com