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

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