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

> On Aug 4, 2026, at 08:41, Masahiko Sawada <sawada(dot)mshk(at)gmail(dot)com> wrote:
>
> On Thu, Jul 9, 2026 at 5:14 AM Fujii Masao <masao(dot)fujii(at)gmail(dot)com> wrote:
>>
>> On Wed, Jun 24, 2026 at 3:02 AM Masahiko Sawada <sawada(dot)mshk(at)gmail(dot)com> wrote:
>>> Right. I'm implementing a basic DDL replication solution using this
>>> hook as a sample implementation. Other than that, this feature can be
>>> used to add additional information for the replicated transaction that
>>> is not replicated via logical replication protocol, such as the
>>> executed user or context-specific information, which can then be
>>> dispatched to a CDC system connected to the subscriber.
>>
>> +1 for the proposed hook, although I haven't read the patch yet.
>>
>> I know of a system that uses logical decoding messages to propagate data
>> to external systems. In that system, the application writes the data to
>> be propagated as logical decoding messages, and a CDC pipeline consumes
>> them via logical decoding and sends them to other systems, for example
>> through Kafka.
>>
>> In that system, when the remote site is maintained by physical replication,
>> the same logical decoding messages can be decoded on the standby
>> at the remote site to deliver them to external systems there. However,
>> if the remote site uses logical replication instead, those messages are
>> currently neither delivered to nor processed on the subscriber at
>> the remote site. As a result, the CDC pipeline at the remote site cannot
>> consume the data they carry.
>>
>> The proposed hook might be useful for this use case. An extension could
>> process the incoming logical decoding messages on the subscriber and
>> forward them to the local CDC pipeline, or store or re-emit them in a form
>> that local consumers can process.
>
> 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.
>
> Regards,
>
> --
> Masahiko Sawada
> Amazon Web Services: https://aws.amazon.com
> <v2-0001-Add-a-hook-for-handling-logical-messages-on-subsc.patch>

I played with this feature today. I have a few comments from a design perspective:

1. Would it make sense to pass the receiving subscription’s identity explicitly to the hook? Otherwise, a hook that needs to distinguish messages received by different subscriptions has to obtain it from a global variable such as MySubscription->oid.

2. A hook is invoked by an apply worker, and since a message has no target relation, the hook runs as the subscription owner regardless of run_as_owner. I think it would be useful to document this explicitly.

3. I’m thinking out loud here. Would it be useful to provide an opt-in default hook that re-emits received messages, so that they are written to the subscriber-side WAL and can be consumed by a local logical-decoding client? I understand that a re-emitted message would have a different LSN, just as logically replicated row changes generate new local WAL records. For the use case Fujii-san described, this might allow an existing decoder to continue working after the remote site switches from physical to logical replication.

And a few comment for the code changes:

1 - worker_internal.h
```
+typedef void (*LogicalRepMessageHandle_hook_type) (LogicalRepMessageData *msg);
+extern PGDLLIMPORT LogicalRepMessageHandle_hook_type LogicalRepMessageHandle_hook;
```

I guess we don’t expect a hook function to mutate the message, so maybe make the pointer const.

2 - worker.c
```
+ /*
+ * Logical messages are relation-agnostic, so they don't map cleanly onto
+ * the tablesync worker of a particular relation, and there would be no
+ * 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
+ */
```

The last sentence looks duplicate and incomplete.

3 - proto.c
```
+ /* read message length */
+ len = pq_getmsgint(in, 4);
+ msg_data->message_size = len;
+
+ /* and data */
+ msg = palloc(len + 1);
+ pq_copymsgbytes(in, msg, len);
+
+ msg[len] = '\0';
+ msg_data->message = msg;
```

logicalrep_read_message() allocates one extra byte for NULL terminator, which is unnecessary, as the contract is to use msg_data->message_size to decide the message length, and a message contain contains 0 in the middle. But I agree it may be useful for debugging and logging, so maybe add a comment to explain why using this extra byte.

Best regards,
--
Chao Li (Evan)
HighGo Software Co., Ltd.
https://www.highgo.com/