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

> On Aug 6, 2026, at 02:55, Masahiko Sawada <sawada(dot)mshk(at)gmail(dot)com> wrote:
>
> On Wed, Aug 5, 2026 at 9:32 AM Masahiko Sawada <sawada(dot)mshk(at)gmail(dot)com> wrote:
>>
>> On Tue, Aug 4, 2026 at 1:05 AM Chao Li <li(dot)evan(dot)chao(at)gmail(dot)com> wrote:
>>>
>>>
>>>
>>>> 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:
>>
>> Thank you for reviewing the patch!
>>
>>> 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.
>>
>> It's common for hook functions to access a global variable to get more
>> information such as MyDatabaseId and MyProcPort etc. Using
>> MySubscription to get a subscription's identity works for me.
>>
>>> 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.
>>
>> I think it depends on the hook function implementations. They can
>> switch the role as they want.
>>
>>>
>>> 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.
>>
>> Interesting idea, but I'm not sure we should have it in the core or in
>> the contrib. I think it's a good topic to discuss in a separate
>> thread.
>>
>>>
>>> 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.
>>
>> Agreed with the all above comments.
>>
>> I'll submit the updated patch shortly.
>
> I've addressed all comments I got so far unless I'm missing anything,
> and attached the updated patch. Feedback is very welcome.
>
> Regards,
>
> --
> Masahiko Sawada
> Amazon Web Services: https://aws.amazon.com
> <v3-0001-Add-a-hook-for-handling-logical-messages-on-subsc.patch>

V3 LGTM.

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