Add a hook for handling logical decoding messages on subscribers.
Hackorum builds and tests every patch posted to the lists, not only commitfest submissions. This is Hackorum's own CI rather than the PostgreSQL project's, and it is still under testing - please report anything that looks wrong.
You can run a PostgreSQL built from this patch straight from Docker, with no checkout and no build:
docker run --rm -p 5432:5432 ghcr.io/hackorum-dev/postgres-patch:t248589psql -h localhost -U postgresBuilt from patchset v19 (message #19), August 06, 2026 at 10:37 PM.
Every patchset is also pushed to a branch of our PostgreSQL fork, so you can check out the same tree CI built. Without a PostgreSQL checkout:
git clone --branch t248589_19 https://github.com/hackorum-dev/postgres.gitIn a checkout you already have, add the fork once:
git remote add hackorum https://github.com/hackorum-dev/postgres.gitthen, for this patchset and every later one:
git fetch hackorum t248589_19 && git checkout t248589_19Patchset v19 (message #19) is on t248589_19
Hi all,
Commit ac4645c015 allows pgoutput to send logical decoding messages,
but it's limited to applications that use the pgoutput plugin -- the
built-in logical replication doesn't use it. I'd like to propose
introducing a hook to the logical replication message handling so that
extensions can plug in their own handling routine. This feature can be
used for extensions to implement DDL replication, function
replication, or trigger user-specific routines on the subscriber side.
I've attached the PoC patch; it adds a hook function, and adds a new
'message' subscription option that allows the user to request the
publisher to send logical decoding messages. Therefore, users need to
enable the 'message' option and set up the hook function at server
startup in order to receive the messages and trigger the hook
function.
I I went with a hook function in the patch. While it lets you chain
the multiple hook functions, providing the registration API might be
better, or other types of registry can also be considered.
Feedback is very welcome.
Regards,
--
Masahiko Sawada
Amazon Web Services: https://aws.amazon.com
Hi,
On Fri, Jun 19, 2026 at 3:34 PM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
Hi all,
Commit ac4645c015 allows pgoutput to send logical decoding messages,
but it's limited to applications that use the pgoutput plugin -- the
built-in logical replication doesn't use it. I'd like to propose
introducing a hook to the logical replication message handling so that
extensions can plug in their own handling routine. This feature can be
used for extensions to implement DDL replication, function
replication, or trigger user-specific routines on the subscriber side.
Thanks for working on this!
I've attached the PoC patch; it adds a hook function, and adds a new
'message' subscription option that allows the user to request the
publisher to send logical decoding messages. Therefore, users need to
enable the 'message' option and set up the hook function at server
startup in order to receive the messages and trigger the hook
function.
I understand the intent of the proposal, but I'd like to get the
bigger picture first.
Do we have any external modules that actually implement DDL
replication (or any of the listed use-cases) with a similar hook? Or
any existing discussion? I could be missing something because I
haven't looked at all the DDL replication related threads.
Another thing I'm curious about - why a hook? Is the plan to implement
DDL replication as an external module rather than in core? If DDL
replication eventually gets into core, I'd expect it to be apply-side
logic executing the decoded DDL messages directly, not something going
through a hook.
Why not a hook at apply_dispatch to give external modules more freedom
with the pgoutput plugin?
I I went with a hook function in the patch. While it lets you chain
the multiple hook functions, providing the registration API might be
better, or other types of registry can also be considered.
It's hard to tell how many external modules would make use of this
hook (rather, how many external modules implementing this hook one
would allow to be installed in a production database requiring
chaining), but my first thought is that a registration-based API along
the lines of RegisterXactCallback would be cleaner and work better.
Feedback is very welcome.
A few comments on the patch:
1/
+ bool message; /* True if the subscription wants to receive
+ * logical messages */
} Subscription;Nit: I'd call these logical decoding messages or generic logical
messages - something to match the docs and pg_logical_emit_message.
2/
+void
+test_logical_message_handler(LogicalRepMessageData *msg)
+{
+ ereport(LOG,
+ (errmsg("received message: LSN %X/%08X, prefix: %s, message: %s",
+ LSN_FORMAT_ARGS(msg->lsn),
+ msg->prefix,
+ msg->message)));
+}Why not have the test extension do a simple DDL/function/event-trigger
based replication? It doesn't need to be a full-blown implementation,
but to show the usefulness of this hook, it's better to have one
demonstrating the listed use-cases.
3/
+ if (subinfo->submessage)
+ appendPQExpBufferStr(query, ", message = true");Why a subscription-level option? Why not leave the decision of whether
or not to act on the message to the external module implementers?
4/
+ /*
+ * Logical messages are handled only the (parallel) apply workers
+ */
+ if (am_tablesync_worker() || am_sequencesync_worker())
+ return;Why these restrictions? Why not leave the decision to external module
implementers? Isn't this limiting - what if someone wants to use this
hook for the schema sync during the initial table sync phase?
5/ With this change, pg_logical_emit_message does affect the logical
replication apply if the subscriber has defined this hook. I think
it's worth mentioning in the docs for pg_logical_emit_message.
--
Bharath Rupireddy
Amazon Web Services: https://aws.amazon.com
On Tue, Jun 23, 2026 at 4:09 AM Bharath Rupireddy
<bharath.rupireddyforpostgres@gmail.com> wrote:
On Fri, Jun 19, 2026 at 3:34 PM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
Hi all,
Commit ac4645c015 allows pgoutput to send logical decoding messages,
but it's limited to applications that use the pgoutput plugin -- the
built-in logical replication doesn't use it. I'd like to propose
introducing a hook to the logical replication message handling so that
extensions can plug in their own handling routine. This feature can be
used for extensions to implement DDL replication, function
replication, or trigger user-specific routines on the subscriber side.Thanks for working on this!
I've attached the PoC patch; it adds a hook function, and adds a new
'message' subscription option that allows the user to request the
publisher to send logical decoding messages. Therefore, users need to
enable the 'message' option and set up the hook function at server
startup in order to receive the messages and trigger the hook
function.I understand the intent of the proposal, but I'd like to get the
bigger picture first.Do we have any external modules that actually implement DDL
replication (or any of the listed use-cases) with a similar hook? Or
any existing discussion? I could be missing something because I
haven't looked at all the DDL replication related threads.Another thing I'm curious about - why a hook? Is the plan to implement
DDL replication as an external module rather than in core? If DDL
replication eventually gets into core, I'd expect it to be apply-side
logic executing the decoded DDL messages directly, not something going
through a hook.
I think it is important to have some example extension implementation
to see how the hook could be utilized. One more use of such a hook
could be to use for audit of DDLs replayed on subscribers. BTW, can we
also consider it as a solution implementing basic DDL replication for
tables? The key question is what if someday we have in-core DDL
replication. I think extensions can still be used to implement
filtering or transformation of DDL. We can implement capture of DDL
using JSON format [1]The JSON format for WAL could be of form to keep it extendable: so that it is forward compatible with in-core
DDL replication. So considering that, the extension handlers will look
like:
_PG_init(void)
{
/* Publisher side */
prev_ProcessUtility = ProcessUtility_hook;
ProcessUtility_hook = ddlrep_ProcessUtility;
.....
/* Subscriber side */
prev_message_handler = logical_message_handler;
logical_message_handler = ddlrep_message_handler;
static void
ddlrep_ProcessUtility(PlannedStmt *pstmt,
const char *queryString,
bool readOnlyTree,
ProcessUtilityContext context,
ParamListInfo params,
QueryEnvironment *queryEnv,
DestReceiver *dest,
QueryCompletion *qc)
{
bool replicate = should_replicate(pstmt);
/*
* Execute the DDL first.
*/
if (prev_ProcessUtility)
prev_ProcessUtility(pstmt, queryString, readOnlyTree, context,
params, queryEnv, dest, qc);
else
standard_ProcessUtility(pstmt, queryString, readOnlyTree, context,
params, queryEnv, dest, qc);
if (replicate)
{
const char *tag =
GetCommandTagName(CreateCommandTag(pstmt->utilityStmt));
char *msg = build_ddl_message(tag, queryString);
/*
* transactional = true → message held in ReorderBuffer, emitted
* to subscribers at COMMIT in WAL order relative to any DML in
* the same transaction.
*
* omit_lsn = false → include the LSN so the subscriber can log
* exactly which WAL position a given DDL came from.
*/
LogLogicalMessage("pg_ddl", msg, strlen(msg),
true /* transactional */,
false /* omit_lsn */);
...
static void
ddlrep_message_handler(const char *prefix,
Size sz,
const char *message,
bool transactional,
XLogRecPtr lsn)
{
char *payload;
char *ddl;
char *search_path;
StringInfoData cmd;
int spi_rc;
/*
* Always pass through to the previous handler first. This ensures
* correct behaviour when chained with other extensions.
*/
if (prev_message_handler)
prev_message_handler(prefix, sz, message, transactional, lsn);
if (strcmp(prefix, "pg_ddl") != 0)
return;
/* Write code to execute/perform DDL. */
When we have a built-in handler then the apply worker carefully
registers the same and gives an ERROR if the extension one is already
registered.
void
ApplyWorkerMain(Datum main_arg)
{
/* ... existing initialisation ... */
/*
* If the subscription requests built-in DDL replication and an
* extension has also registered a logical message hook, both would
* process the same "pg_ddl" messages and execute DDL twice.
* Refuse to start rather than silently corrupt.
*/
if (MySubscription->ddloption != DDL_OPTION_NONE &&
logical_message_hook != NULL)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
....
/*
* No conflict — register the built-in DDL handler when the
* subscription requests DDL replication and no extension owns
* the hook.
*/
if (MySubscription->ddloption != DDL_OPTION_NONE)
logical_message_hook = builtin_ddl_message_handler;
[1]: The JSON format for WAL could be of form to keep it extendable:
The JSON format for WAL could be of form to keep it extendable:
{
"version": 1,
"command_tag": "CREATE TABLE",
"object_type": "table",
"schema": "public",
"identity": "public.foo",
"ddl_text": "CREATE TABLE public.foo (id int PRIMARY KEY)",
"search_path": "public"
}
Why not a hook at apply_dispatch to give external modules more freedom
with the pgoutput plugin?
What advantage do you see with the same?
--
With Regards,
Amit Kapila.
On Tue, Jun 23, 2026 at 1:52 AM Amit Kapila <amit.kapila16@gmail.com> wrote:
On Tue, Jun 23, 2026 at 4:09 AM Bharath Rupireddy
<bharath.rupireddyforpostgres@gmail.com> wrote:On Fri, Jun 19, 2026 at 3:34 PM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
Hi all,
Commit ac4645c015 allows pgoutput to send logical decoding messages,
but it's limited to applications that use the pgoutput plugin -- the
built-in logical replication doesn't use it. I'd like to propose
introducing a hook to the logical replication message handling so that
extensions can plug in their own handling routine. This feature can be
used for extensions to implement DDL replication, function
replication, or trigger user-specific routines on the subscriber side.Thanks for working on this!
I've attached the PoC patch; it adds a hook function, and adds a new
'message' subscription option that allows the user to request the
publisher to send logical decoding messages. Therefore, users need to
enable the 'message' option and set up the hook function at server
startup in order to receive the messages and trigger the hook
function.I understand the intent of the proposal, but I'd like to get the
bigger picture first.Do we have any external modules that actually implement DDL
replication (or any of the listed use-cases) with a similar hook? Or
any existing discussion? I could be missing something because I
haven't looked at all the DDL replication related threads.Another thing I'm curious about - why a hook? Is the plan to implement
DDL replication as an external module rather than in core? If DDL
replication eventually gets into core, I'd expect it to be apply-side
logic executing the decoded DDL messages directly, not something going
through a hook.I think it is important to have some example extension implementation
to see how the hook could be utilized. One more use of such a hook
could be to use for audit of DDLs replayed on subscribers.
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.
BTW, can we
also consider it as a solution implementing basic DDL replication for
tables? The key question is what if someday we have in-core DDL
replication. I think extensions can still be used to implement
filtering or transformation of DDL. We can implement capture of DDL
using JSON format [1] so that it is forward compatible with in-core
DDL replication. So considering that, the extension handlers will look
like:
While a common JSON format might be helpful for forward compatibility
with in-core DDL replication, I think that we need to keep the generic
logical decoding messages used by this feature distinct from the
messages that would be used for DDL replication. A generic logical
decoding message can be written via pg_logical_emit_message() SQL
function, which is granted to PUBLIC, so any user could directly emit
a DDL command in the JSON format with the 'pg_ddl' prefix and create
arbitrary tables on the subscriber, which could cause a privilege
escalation problem. I don't think we should build something as
powerful as DDL execution on top of a message channel that any user
can write to.
Regards,
--
Masahiko Sawada
Amazon Web Services: https://aws.amazon.com
On Mon, Jun 22, 2026 at 3:39 PM Bharath Rupireddy
<bharath.rupireddyforpostgres@gmail.com> wrote:
Hi,
On Fri, Jun 19, 2026 at 3:34 PM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
Hi all,
Commit ac4645c015 allows pgoutput to send logical decoding messages,
but it's limited to applications that use the pgoutput plugin -- the
built-in logical replication doesn't use it. I'd like to propose
introducing a hook to the logical replication message handling so that
extensions can plug in their own handling routine. This feature can be
used for extensions to implement DDL replication, function
replication, or trigger user-specific routines on the subscriber side.Thanks for working on this!
I've attached the PoC patch; it adds a hook function, and adds a new
'message' subscription option that allows the user to request the
publisher to send logical decoding messages. Therefore, users need to
enable the 'message' option and set up the hook function at server
startup in order to receive the messages and trigger the hook
function.I understand the intent of the proposal, but I'd like to get the
bigger picture first.Do we have any external modules that actually implement DDL
replication (or any of the listed use-cases) with a similar hook? Or
any existing discussion? I could be missing something because I
haven't looked at all the DDL replication related threads.
Not that I'm aware of. That said, in past DDL replication discussions
several approaches were proposed (e.g. sending the DDL query text vs.
sending schema diffs in some form of intermediate representation),
each with its own pros and cons. So even once we have an in-core DDL
replication implementation, I think it's still plausible that an
extension would want to implement a different approach, and this hook
would let it do so without patching core.
Another thing I'm curious about - why a hook? Is the plan to implement
DDL replication as an external module rather than in core? If DDL
replication eventually gets into core, I'd expect it to be apply-side
logic executing the decoded DDL messages directly, not something going
through a hook.
I'm working on in-core DDL replication too, and I agree that the
in-core version would be apply-side logic executing the decoded DDL
directly, not something going through this hook. The hook isn't meant
to be to implement the in-core DDL replication. It's there for things
core wouldn't cover: auditing replayed DDLs, propagating
context-specific information to the subscriber, or an extension
implementing a different DDL replication approach than the in-core
one.
Why not a hook at apply_dispatch to give external modules more freedom
with the pgoutput plugin?
It would let an extension intercept every message type
(INSERT/UPDATE/DELETE, COMMIT, etc.) and could easily break apply
consistency. At the same time, the incoming data is just the standard
logical replication protocol, so there isn't much an extension could
usefully do with the non-MESSAGE types beyond what core already does.
I I went with a hook function in the patch. While it lets you chain
the multiple hook functions, providing the registration API might be
better, or other types of registry can also be considered.It's hard to tell how many external modules would make use of this
hook (rather, how many external modules implementing this hook one
would allow to be installed in a production database requiring
chaining), but my first thought is that a registration-based API along
the lines of RegisterXactCallback would be cleaner and work better.
Yeah, it would be better.
Feedback is very welcome.
A few comments on the patch:
1/ + bool message; /* True if the subscription wants to receive + * logical messages */ } Subscription;Nit: I'd call these logical decoding messages or generic logical
messages - something to match the docs and pg_logical_emit_message.
Will fix.
2/ +void +test_logical_message_handler(LogicalRepMessageData *msg) +{ + ereport(LOG, + (errmsg("received message: LSN %X/%08X, prefix: %s, message: %s", + LSN_FORMAT_ARGS(msg->lsn), + msg->prefix, + msg->message))); +}Why not have the test extension do a simple DDL/function/event-trigger
based replication? It doesn't need to be a full-blown implementation,
but to show the usefulness of this hook, it's better to have one
demonstrating the listed use-cases.
I'd prefer to keep this test module minimal, since its purpose is to
improve coverage of the newly added code. A worked example that
demonstrates the usefulness of the hook is valuable, but I think it
belongs in a separate contrib module rather than in the test module,
and that's probably a separate discussion.
3/ + if (subinfo->submessage) + appendPQExpBufferStr(query, ", message = true");Why a subscription-level option? Why not leave the decision of whether
or not to act on the message to the external module implementers?
This is because not to affect the existing logical replication users.
Since the logical decoding messages are not sent today, I think that
it should be an explcit opt-in feature for users who don't want to
allow the publisher to send logical decoding messages.
4/ + /* + * Logical messages are handled only the (parallel) apply workers + */ + if (am_tablesync_worker() || am_sequencesync_worker()) + return;Why these restrictions? Why not leave the decision to external module
implementers? Isn't this limiting - what if someone wants to use this
hook for the schema sync during the initial table sync phase?
Logical decoding messages are relation-agnostic, so they don't map
cleanly onto the table sync phase. If every tablesync worker processed
the messages they'd be handled multiple times, and if only one did,
there'd be no well-defined ordering between a message and the initial
copy progress. So I think letting tablesync workers run the hook seems
more confusing than useful.
5/ With this change, pg_logical_emit_message does affect the logical
replication apply if the subscriber has defined this hook. I think
it's worth mentioning in the docs for pg_logical_emit_message.
Agreed.
Regards,
--
Masahiko Sawada
Amazon Web Services: https://aws.amazon.com
On Tue, Jun 23, 2026 at 11:01 AM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
On Tue, Jun 23, 2026 at 1:52 AM Amit Kapila <amit.kapila16@gmail.com> wrote:
On Tue, Jun 23, 2026 at 4:09 AM Bharath Rupireddy
<bharath.rupireddyforpostgres@gmail.com> wrote:On Fri, Jun 19, 2026 at 3:34 PM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
Hi all,
Commit ac4645c015 allows pgoutput to send logical decoding messages,
but it's limited to applications that use the pgoutput plugin -- the
built-in logical replication doesn't use it. I'd like to propose
introducing a hook to the logical replication message handling so that
extensions can plug in their own handling routine. This feature can be
used for extensions to implement DDL replication, function
replication, or trigger user-specific routines on the subscriber side.Thanks for working on this!
I've attached the PoC patch; it adds a hook function, and adds a new
'message' subscription option that allows the user to request the
publisher to send logical decoding messages. Therefore, users need to
enable the 'message' option and set up the hook function at server
startup in order to receive the messages and trigger the hook
function.I understand the intent of the proposal, but I'd like to get the
bigger picture first.Do we have any external modules that actually implement DDL
replication (or any of the listed use-cases) with a similar hook? Or
any existing discussion? I could be missing something because I
haven't looked at all the DDL replication related threads.Another thing I'm curious about - why a hook? Is the plan to implement
DDL replication as an external module rather than in core? If DDL
replication eventually gets into core, I'd expect it to be apply-side
logic executing the decoded DDL messages directly, not something going
through a hook.I think it is important to have some example extension implementation
to see how the hook could be utilized. One more use of such a hook
could be to use for audit of DDLs replayed on subscribers.Right. I'm implementing a basic DDL replication solution using this
hook as a sample implementation.
FYI, I've implemented a test extension, pg_logical_ddl[1]https://github.com/MasahikoSawada/pg_logical_ddl, which can
be used with the proposed patch.
The extension uses both hooks ProcessUtility_hook to capture DDL
events and LogicalRepMessageHandle_hook to apply the DDL message. It
sends a JSON blob containing the DDL command, command tag, user name,
and search_path. The implementation is straightforward and I find
that the proposed hook is useful to implement DDL replication.
One thing that would simplify the implementation, and that I'd like to
have, is an easy way to capture the OID of the affected relation (or
its object address) without event triggers. That information is useful
for figuring out which object a DDL affects: for instance, to filter
DDLs by table, or to map the command to the right object on the
subscriber.
If we use an event trigger to capture DDL events, it would be easy to
get the object address, but it's not easy for extensions like
pg_logical_ddl that don't rely on event triggers. The OID of the
affected table can be resolved by the name and search_path but it's
not a reliable solution as its schema can be renamed without taking a
heavy lock on the table. If they have to use event triggers, it could
be cumbersome to manage additional objects (e.g., users can drop them
etc.). It would be good to have a mechanism to easily capture the
information that is collected by event triggers today without event
triggers. FYI I proposed a similar idea before in the DDL replication
thread[2]/messages/by-id/CAD21AoCEVO+zpLmQqKwZbJ5+rvqsJ1e0wnTNBH437p8tDw7B=g@mail.gmail.com since the same is true for the in-core DDL replication and I
have a PoC patch for that.
Also, since this is an extension, it cannot necessarily work well with
PUBLICATION; it doesn't respect the table filtering setting and some
options like publish_via_partition_root. In that sense, working with
the existing publication feature is the advantage of the in-core DDL
replication.
Regards,
[1]: https://github.com/MasahikoSawada/pg_logical_ddl
[2]: /messages/by-id/CAD21AoCEVO+zpLmQqKwZbJ5+rvqsJ1e0wnTNBH437p8tDw7B=g@mail.gmail.com
--
Masahiko Sawada
Amazon Web Services: https://aws.amazon.com
On Wed, Jun 24, 2026 at 3:02 AM Masahiko Sawada <sawada.mshk@gmail.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.
Regards,
--
Fujii Masao
On Thu, Jul 9, 2026 at 5:14 AM Fujii Masao <masao.fujii@gmail.com> wrote:
On Wed, Jun 24, 2026 at 3:02 AM Masahiko Sawada <sawada.mshk@gmail.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
On Tue, Aug 4, 2026 at 9:42 AM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
I've rebased and updated the patch. Please review it.
Thanks for the patch!
I tried running the tests with it, and t/002_pg_dump.pl failed as follows:
t/002_pg_dump.pl .. 12151/? # Looks like you failed 26 tests of 12881.
t/002_pg_dump.pl .. Dubious, test returned 26 (wstat 6656, 0x1a00)
Failed 26/12881 subtests
Test Summary Report
-------------------
t/002_pg_dump.pl (Wstat: 6656 Tests: 12881 Failed: 26)
Failed tests: 201, 468, 735, 1269, 1803, 2074, 2348, 2620
2891, 3158, 3425, 3692, 3959, 4226, 5027
5294, 5561, 5829, 6096, 6630, 7432, 7699
8767, 9034, 10637, 11171
Non-zero exit status: 26
Files=1, Tests=12881, 31 wallclock secs ( 0.56 usr 0.10 sys + 4.81
cusr 1.72 csys = 7.19 CPU)
Result: FAIL
options->proto.logical.origin = pstrdup(MySubscription->origin);
+ options->proto.logical.messages = MySubscription->message;
Does this mean the tablesync worker also requests messages = true, even though
it always discards logical decoding messages in apply_handle_message()? If so,
that seems like unnecessary overhead. Shouldn't the tablesync worker always
request messages = false?
- subsynccommit, subwalrcvtimeout, subpublications, suborigin)
+ subsynccommit, subwalrcvtimeout, subpublications, suborigin,
+ submessage)In pg_subscription, submessage is defined just before submaxretention, but here
it's added at the end of the column list. It's not a bug, but it would be better
to keep the ordering consistent.
+ /*
+ * 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();Shouldn't store_flush_position() also be called in the non-transactional case,
so that the remote/local LSN pair is tracked just as it is during normal commit
processing? Also pgstat_report_stat() etc should be called?
+typedef void (*LogicalRepMessageHandle_hook_type) (LogicalRepMessageData *msg);
+extern PGDLLIMPORT LogicalRepMessageHandle_hook_type
LogicalRepMessageHandle_hook;These are added to worker_internal.h. Since that header is for worker internals,
would it be better to put them in a separate header such as logicalworker.h?
+ Oid argtypes[5] = {LSNOID, BOOLOID, TEXTOID, INT4OID, TEXTOID};
+ Datum values[5];
+ int ret;
+
+ ereport(LOG,
+ (errmsg("received message: LSN %X/%08X, prefix: %s, message: %s,
transactional: %d",
+ LSN_FORMAT_ARGS(msg->lsn),
+ msg->prefix, msg->message, msg->transactional)));pg_logical_emit_message() can emit binary messages, but test_logicalmsg_hook.c
doesn't seem to handle them correctly. Is that intentional?
Should psql tab completion also be updated for the new message option
in CREATE/ALTER SUBSCRIPTION ... WITH/SET?
Regards,
--
Fujii Masao
Hi,
On Mon, Aug 3, 2026 at 5:42 PM Masahiko Sawada <sawada.mshk@gmail.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 workersCould 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 wayThis 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
On Aug 4, 2026, at 08:41, Masahiko Sawada <sawada.mshk@gmail.com> wrote:
On Thu, Jul 9, 2026 at 5:14 AM Fujii Masao <masao.fujii@gmail.com> wrote:
On Wed, Jun 24, 2026 at 3:02 AM Masahiko Sawada <sawada.mshk@gmail.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/
On Mon, Aug 3, 2026 at 8:03 PM Fujii Masao <masao.fujii@gmail.com> wrote:
On Tue, Aug 4, 2026 at 9:42 AM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
I've rebased and updated the patch. Please review it.
Thanks for the patch!
Thank you for reviewing the patch!
I tried running the tests with it, and t/002_pg_dump.pl failed as follows:
t/002_pg_dump.pl .. 12151/? # Looks like you failed 26 tests of 12881.
t/002_pg_dump.pl .. Dubious, test returned 26 (wstat 6656, 0x1a00)
Failed 26/12881 subtestsTest Summary Report
-------------------
t/002_pg_dump.pl (Wstat: 6656 Tests: 12881 Failed: 26)
Failed tests: 201, 468, 735, 1269, 1803, 2074, 2348, 2620
2891, 3158, 3425, 3692, 3959, 4226, 5027
5294, 5561, 5829, 6096, 6630, 7432, 7699
8767, 9034, 10637, 11171
Non-zero exit status: 26
Files=1, Tests=12881, 31 wallclock secs ( 0.56 usr 0.10 sys + 4.81
cusr 1.72 csys = 7.19 CPU)
Result: FAIL
Fixed.
options->proto.logical.origin = pstrdup(MySubscription->origin);
+ options->proto.logical.messages = MySubscription->message;Does this mean the tablesync worker also requests messages = true, even though
it always discards logical decoding messages in apply_handle_message()? If so,
that seems like unnecessary overhead. Shouldn't the tablesync worker always
request messages = false?
Good catch. We can always disable messages for tablesync workers.
- subsynccommit, subwalrcvtimeout, subpublications, suborigin) + subsynccommit, subwalrcvtimeout, subpublications, suborigin, + submessage)In pg_subscription, submessage is defined just before submaxretention, but here
it's added at the end of the column list. It's not a bug, but it would be better
to keep the ordering consistent.
Fixed.
+ /* + * 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();Shouldn't store_flush_position() also be called in the non-transactional case,
so that the remote/local LSN pair is tracked just as it is during normal commit
processing? Also pgstat_report_stat() etc should be called?
True. In addition to that, I think we should call
replorigin_xact_clear(false) before the commit. Will fix it.
+typedef void (*LogicalRepMessageHandle_hook_type) (LogicalRepMessageData *msg); +extern PGDLLIMPORT LogicalRepMessageHandle_hook_type LogicalRepMessageHandle_hook;These are added to worker_internal.h. Since that header is for worker internals,
would it be better to put them in a separate header such as logicalworker.h?
Agreed.
+ Oid argtypes[5] = {LSNOID, BOOLOID, TEXTOID, INT4OID, TEXTOID}; + Datum values[5]; + int ret; + + ereport(LOG, + (errmsg("received message: LSN %X/%08X, prefix: %s, message: %s, transactional: %d", + LSN_FORMAT_ARGS(msg->lsn), + msg->prefix, msg->message, msg->transactional)));pg_logical_emit_message() can emit binary messages, but test_logicalmsg_hook.c
doesn't seem to handle them correctly. Is that intentional?
Yes. It supports only text-format messages, which seems fine to me as
we can ensure that only text messages are emitted in this test module.
Do you think it's better to make it work in more general cases?
Should psql tab completion also be updated for the new message option
in CREATE/ALTER SUBSCRIPTION ... WITH/SET?
Fixed.
Regards,
--
Masahiko Sawada
Amazon Web Services: https://aws.amazon.com
On Mon, Aug 3, 2026 at 8:04 PM Bharath Rupireddy
<bharath.rupireddyforpostgres@gmail.com> wrote:
Hi,
On Mon, Aug 3, 2026 at 5:42 PM Masahiko Sawada <sawada.mshk@gmail.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 workersCould 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 wayThis 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
Hi,
On Tue, Aug 4, 2026 at 1:26 PM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
Nice call-out about the more-than-once delivery and the idempotency
requirement. It's true for logical replication in general, but would....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?
WFM. I missed that, sorry for the noise!
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.
Ah, I see it. Also, it's better to send a null-terminated string to
the hooks than sending everything across from the received input
buffer.
6/
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.
Agreed to keep it simple the way you have it in the v2 patch.
7/
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.
Thanks for pointing me to the code. I understand it now. So, there can
never be a case where a non-transactional message is part of an
in-progress transaction on the apply worker, even in streaming mode,
because logical replication can't interleave transactions for apply.
If so, can we assert this or enhance the comment in
apply_handle_message() a bit?
--
Bharath Rupireddy
Amazon Web Services: https://aws.amazon.com
On Tue, Aug 4, 2026 at 1:05 AM Chao Li <li.evan.chao@gmail.com> wrote:
On Aug 4, 2026, at 08:41, Masahiko Sawada <sawada.mshk@gmail.com> wrote:
On Thu, Jul 9, 2026 at 5:14 AM Fujii Masao <masao.fujii@gmail.com> wrote:
On Wed, Jun 24, 2026 at 3:02 AM Masahiko Sawada <sawada.mshk@gmail.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.
Regards,
--
Masahiko Sawada
Amazon Web Services: https://aws.amazon.com
On Wed, Aug 5, 2026 at 9:32 AM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
On Tue, Aug 4, 2026 at 1:05 AM Chao Li <li.evan.chao@gmail.com> wrote:
On Aug 4, 2026, at 08:41, Masahiko Sawada <sawada.mshk@gmail.com> wrote:
On Thu, Jul 9, 2026 at 5:14 AM Fujii Masao <masao.fujii@gmail.com> wrote:
On Wed, Jun 24, 2026 at 3:02 AM Masahiko Sawada <sawada.mshk@gmail.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
Attachments:
t248589_16v3-0001-Add-a-hook-for-handling-logical-messages-on-subsc.patchtext/x-patch; charset=US-ASCII; name=v3-0001-Add-a-hook-for-handling-logical-messages-on-subsc.patchDownload+818-86
On Aug 6, 2026, at 02:55, Masahiko Sawada <sawada.mshk@gmail.com> wrote:
On Wed, Aug 5, 2026 at 9:32 AM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
On Tue, Aug 4, 2026 at 1:05 AM Chao Li <li.evan.chao@gmail.com> wrote:
On Aug 4, 2026, at 08:41, Masahiko Sawada <sawada.mshk@gmail.com> wrote:
On Thu, Jul 9, 2026 at 5:14 AM Fujii Masao <masao.fujii@gmail.com> wrote:
On Wed, Jun 24, 2026 at 3:02 AM Masahiko Sawada <sawada.mshk@gmail.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/
Hi,
On Wed, Aug 5, 2026 at 11:56 AM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
I've addressed all comments I got so far unless I'm missing anything,
and attached the updated patch. Feedback is very welcome.
Thanks for sending the v3 patch. It looks good to me as-is. I verified
the patch with different cases such as emitting a bunch of ~900MB
messages to see the memory growth, enabling/disabling the subscription
option, pg_dump with the option enabled/disabled, regression tests,
and pgindent.
Here are some comments (may not need any code changes, feel free to ignore):
1/
+ if (options->proto.logical.messages &&
+ PQserverVersion(conn->streamConn) >= 140000)
+ appendStringInfo(&cmd, ", messages 'on'");+ <term><literal>message</literal> (<type>boolean</type>)</term>
The publisher option is "messages". Should we have the subscription
option also use the same plural form to keep it consistent, and for
the reason that we receive a stream of messages, not just one?
2/
+ /*
+ * The message doesn't belong to any remote transaction, so there is
+ * no remote commit LSN nor timestamp to record. Clear the state left
+ * over by the previously applied transaction so that this commit
+ * doesn't inherit it.
+ */
+ replorigin_xact_clear(false);Why is this a problem if we let the non-transactional message inherit it?
3/
+ logicalrep_read_message(s, &msg);
+
+ begin_replication_step();
+
+ (*LogicalRepMessageHandle_hook) (&msg);
+
+ end_replication_step();Wrapping the hook with begin and end replication step is nice. This
lets the hook see the correct command ID, snapshot, and memory
context. There are callers that do the begin first and read message
next (insert), but it seems okay this way because read message doesn't
do any catalog or table accesses, so it should be fine.
--
Bharath Rupireddy
Amazon Web Services: https://aws.amazon.com
On Thu, Aug 6, 2026 at 12:03 AM Bharath Rupireddy
<bharath.rupireddyforpostgres@gmail.com> wrote:
Hi,
On Wed, Aug 5, 2026 at 11:56 AM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
I've addressed all comments I got so far unless I'm missing anything,
and attached the updated patch. Feedback is very welcome.Thanks for sending the v3 patch. It looks good to me as-is. I verified
the patch with different cases such as emitting a bunch of ~900MB
messages to see the memory growth, enabling/disabling the subscription
option, pg_dump with the option enabled/disabled, regression tests,
and pgindent.Here are some comments (may not need any code changes, feel free to ignore):
1/ + if (options->proto.logical.messages && + PQserverVersion(conn->streamConn) >= 140000) + appendStringInfo(&cmd, ", messages 'on'");+ <term><literal>message</literal> (<type>boolean</type>)</term>
The publisher option is "messages". Should we have the subscription
option also use the same plural form to keep it consistent, and for
the reason that we receive a stream of messages, not just one?
Agreed.
2/ + /* + * The message doesn't belong to any remote transaction, so there is + * no remote commit LSN nor timestamp to record. Clear the state left + * over by the previously applied transaction so that this commit + * doesn't inherit it. + */ + replorigin_xact_clear(false);Why is this a problem if we let the non-transactional message inherit it?
IIUC non-transactional messages would have the same commit timestamp
as the previously applied transaction, which is wrong to me.
3/ + logicalrep_read_message(s, &msg); + + begin_replication_step(); + + (*LogicalRepMessageHandle_hook) (&msg); + + end_replication_step();Wrapping the hook with begin and end replication step is nice. This
lets the hook see the correct command ID, snapshot, and memory
context. There are callers that do the begin first and read message
next (insert), but it seems okay this way because read message doesn't
do any catalog or table accesses, so it should be fine.
begin_replication_step() switches the memory context to
ApplyMessageContext. Given logicalrep_read_message() palloc's for
messages, it should be called after begin_replication_step(). Fixed it.
I've attached the updated patch.
Regards,
--
Masahiko Sawada
Amazon Web Services: https://aws.amazon.com
Attachments:
t248589_19v4-0001-Add-a-hook-for-handling-logical-messages-on-subsc.patchtext/x-patch; charset=US-ASCII; name=v4-0001-Add-a-hook-for-handling-logical-messages-on-subsc.patchDownload+822-86
Hi,
On Thu, Aug 6, 2026 at 9:59 AM Masahiko Sawada <sawada.mshk@gmail.com> wrote:
2/ + /* + * The message doesn't belong to any remote transaction, so there is + * no remote commit LSN nor timestamp to record. Clear the state left + * over by the previously applied transaction so that this commit + * doesn't inherit it. + */ + replorigin_xact_clear(false);Why is this a problem if we let the non-transactional message inherit it?
IIUC non-transactional messages would have the same commit timestamp
as the previously applied transaction, which is wrong to me.
Having replorigin_xact_clear there looks fine to me. The next
transaction commit would anyway set the origin LSN and timestamp.
Wrapping the hook with begin and end replication step is nice. This
lets the hook see the correct command ID, snapshot, and memory
context. There are callers that do the begin first and read message
next (insert), but it seems okay this way because read message doesn't
do any catalog or table accesses, so it should be fine.begin_replication_step() switches the memory context to
ApplyMessageContext. Given logicalrep_read_message() palloc's for
messages, it should be called after begin_replication_step(). Fixed it.
Right. I verified other places and wherever the read does a palloc, it
is wrapped within begin and end replication step.
I've attached the updated patch.
Thanks. The v4 patch looks good to me. pgindent and tests are happy. I
have no further comments. I marked the CF entry RfC
(https://commitfest.postgresql.org/patch/7092/). FWIW, the CF bot
complains with "needs rebase":
https://cfbot.cputube.org/patch_7092.log.
--
Bharath Rupireddy
Amazon Web Services: https://aws.amazon.com