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

From: Masahiko Sawada <[email-masked]>
Date: Fri, 19 Jun 2026 15:33:21 -0700

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
From 546809304986e5cdcdee4ee41387c5edeb435fb2 Mon Sep 17 00:00:00 2001
From: Masahiko Sawada <[email-masked]>
Date: Mon, 8 Jun 2026 11:25:47 -0700
Subject: [PATCH v1] Add a hook for handling logical messages on subscribers.

Add a "message" subscription option (default off) that asks the
publisher's output plugin to stream messages, and dispatch each
received message to a new LogicalMessageHandle_hook.  When the option
is off or no hook is installed, behavior is unchanged.	This lets an
extension act on logical messages on the subscriber -- for instance to
carry deparsed DDL to a custom apply worker -- without modifying core
apply code.

Message handling honors transaction streaming and ALTER SUBSCRIPTION
... SKIP, as exercised by the new test_logicalmsg_hooks module.

XXX Bump catalog version for the new pg_subscription.submessage column.
---
 src/backend/catalog/pg_subscription.c         |   1 +
 src/backend/catalog/system_views.sql          |   2 +-
 src/backend/commands/subscriptioncmds.c       |  24 ++-
 .../libpqwalreceiver/libpqwalreceiver.c       |   4 +
 src/backend/replication/logical/proto.c       |  25 +++
 src/backend/replication/logical/worker.c      |  42 +++-
 src/bin/pg_dump/pg_dump.c                     |  16 +-
 src/bin/pg_dump/pg_dump.h                     |   1 +
 src/bin/psql/describe.c                       |   5 +-
 src/include/catalog/pg_subscription.h         |   5 +
 src/include/replication/logicalproto.h        |  13 ++
 src/include/replication/walreceiver.h         |   1 +
 src/include/replication/worker_internal.h     |   4 +
 src/test/modules/Makefile                     |   1 +
 src/test/modules/meson.build                  |   1 +
 .../modules/test_logicalmsg_hooks/.gitignore  |   4 +
 .../modules/test_logicalmsg_hooks/Makefile    |  20 ++
 .../modules/test_logicalmsg_hooks/meson.build |  28 +++
 .../test_logicalmsg_hooks/t/001_basic.pl      | 113 ++++++++++
 .../test_logicalmsg_hooks.c                   |  40 ++++
 src/test/regress/expected/subscription.out    | 202 ++++++++++--------
 src/test/regress/sql/subscription.sql         |  17 ++
 src/tools/pgindent/typedefs.list              |   1 +
 23 files changed, 469 insertions(+), 101 deletions(-)
 create mode 100644 src/test/modules/test_logicalmsg_hooks/.gitignore
 create mode 100644 src/test/modules/test_logicalmsg_hooks/Makefile
 create mode 100644 src/test/modules/test_logicalmsg_hooks/meson.build
 create mode 100644 src/test/modules/test_logicalmsg_hooks/t/001_basic.pl
 create mode 100644 src/test/modules/test_logicalmsg_hooks/test_logicalmsg_hooks.c

diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index b5cb301db88..f14649a245b 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -132,6 +132,7 @@ GetSubscription(Oid subid, bool missing_ok, bool conninfo_needed,
 	sub->retaindeadtuples = subform->subretaindeadtuples;
 	sub->maxretention = subform->submaxretention;
 	sub->retentionactive = subform->subretentionactive;
+	sub->message = subform->submessage;
 
 	if (conninfo_needed)
 	{
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 8f129baec90..6ebaa93b1ff 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1528,7 +1528,7 @@ GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
 			  subpasswordrequired, subrunasowner, subfailover,
               subretaindeadtuples, submaxretention, subretentionactive,
               subserver, subslotname, subsynccommit, subwalrcvtimeout,
-              subpublications, suborigin)
+              subpublications, suborigin, submessage)
     ON pg_subscription TO public;
 
 CREATE VIEW pg_stat_subscription_stats AS
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index ee06a726f42..5183ab01156 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -79,6 +79,7 @@
 #define SUBOPT_WAL_RECEIVER_TIMEOUT			0x00010000
 #define SUBOPT_LSN					0x00020000
 #define SUBOPT_ORIGIN				0x00040000
+#define SUBOPT_MESSAGE				0x00080000
 
 /* check if the 'val' has 'bits' set */
 #define IsSet(val, bits)  (((val) & (bits)) == (bits))
@@ -105,6 +106,7 @@ typedef struct SubOpts
 	bool		runasowner;
 	bool		failover;
 	bool		retaindeadtuples;
+	bool		message;
 	int32		maxretention;
 	char	   *origin;
 	XLogRecPtr	lsn;
@@ -192,6 +194,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 		opts->failover = false;
 	if (IsSet(supported_opts, SUBOPT_RETAIN_DEAD_TUPLES))
 		opts->retaindeadtuples = false;
+	if (IsSet(supported_opts, SUBOPT_MESSAGE))
+		opts->message = false;
 	if (IsSet(supported_opts, SUBOPT_MAX_RETENTION_DURATION))
 		opts->maxretention = 0;
 	if (IsSet(supported_opts, SUBOPT_ORIGIN))
@@ -348,6 +352,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 			opts->specified_opts |= SUBOPT_RETAIN_DEAD_TUPLES;
 			opts->retaindeadtuples = defGetBoolean(defel);
 		}
+		else if (IsSet(supported_opts, SUBOPT_MESSAGE) &&
+				 strcmp(defel->defname, "message") == 0)
+		{
+			if (IsSet(opts->specified_opts, SUBOPT_MESSAGE))
+				errorConflictingDefElem(defel, pstate);
+
+			opts->specified_opts |= SUBOPT_MESSAGE;
+			opts->message = defGetBoolean(defel);
+		}
 		else if (IsSet(supported_opts, SUBOPT_MAX_RETENTION_DURATION) &&
 				 strcmp(defel->defname, "max_retention_duration") == 0)
 		{
@@ -674,7 +687,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 					  SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
 					  SUBOPT_RETAIN_DEAD_TUPLES |
 					  SUBOPT_MAX_RETENTION_DURATION |
-					  SUBOPT_WAL_RECEIVER_TIMEOUT | SUBOPT_ORIGIN);
+					  SUBOPT_WAL_RECEIVER_TIMEOUT | SUBOPT_ORIGIN | SUBOPT_MESSAGE);
 	parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
 
 	/*
@@ -824,6 +837,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 	values[Anum_pg_subscription_subfailover - 1] = BoolGetDatum(opts.failover);
 	values[Anum_pg_subscription_subretaindeadtuples - 1] =
 		BoolGetDatum(opts.retaindeadtuples);
+	values[Anum_pg_subscription_submessage - 1] = BoolGetDatum(opts.message);
 	values[Anum_pg_subscription_submaxretention - 1] =
 		Int32GetDatum(opts.maxretention);
 	values[Anum_pg_subscription_subretentionactive - 1] =
@@ -1499,7 +1513,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 							  SUBOPT_RETAIN_DEAD_TUPLES |
 							  SUBOPT_MAX_RETENTION_DURATION |
 							  SUBOPT_WAL_RECEIVER_TIMEOUT |
-							  SUBOPT_ORIGIN);
+							  SUBOPT_ORIGIN | SUBOPT_MESSAGE);
 			break;
 
 		case ALTER_SUBSCRIPTION_ENABLED:
@@ -1858,6 +1872,12 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					replaces[Anum_pg_subscription_subwalrcvtimeout - 1] = true;
 				}
 
+				if (IsSet(opts.specified_opts, SUBOPT_MESSAGE))
+				{
+					values[Anum_pg_subscription_submessage - 1] = BoolGetDatum(opts.message);
+					replaces[Anum_pg_subscription_submessage - 1] = true;
+				}
+
 				update_tuple = true;
 				break;
 			}
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 5376519fea5..490e0590970 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -614,6 +614,10 @@ libpqrcv_startstreaming(WalReceiverConn *conn,
 			appendQuotedLiteral(&cmd, options->proto.logical.origin);
 		}
 
+		if (options->proto.logical.messages &&
+			PQserverVersion(conn->streamConn) >= 140000)
+			appendStringInfo(&cmd, ", messages 'on'");
+
 		pubnames = options->proto.logical.publication_names;
 		pubnames_str = stringlist_to_identifierstr(pubnames);
 		appendStringInfoString(&cmd, ", publication_names ");
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index 86ad97cd937..af8ee26ed0c 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -660,6 +660,31 @@ logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn,
 	pq_sendbytes(out, message, sz);
 }
 
+/*
+ * Read MESSAGE from stream.
+ */
+void
+logicalrep_read_message(StringInfo in, LogicalRepMessageData *msg_data)
+{
+	Size		len;
+	char	   *msg;
+
+	msg_data->flags = pq_getmsgint(in, 1);
+	msg_data->lsn = pq_getmsgint64(in);
+	msg_data->prefix = pstrdup(pq_getmsgstring(in));
+
+	/* 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;
+}
+
 /*
  * Write relation description to the output stream.
  */
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 7799266c614..3457ed7b379 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -549,6 +549,9 @@ typedef struct ApplySubXactData
 
 static ApplySubXactData subxact_data = {0, 0, InvalidTransactionId, NULL};
 
+/* Hook for plugins to get control in handling logical decoding messages */
+LogicalRepMessageHandle_hook_type LogicalRepMessageHandle_hook = NULL;
+
 static inline void subxact_filename(char *path, Oid subid, TransactionId xid);
 static inline void changes_filename(char *path, Oid subid, TransactionId xid);
 
@@ -1694,6 +1697,34 @@ apply_handle_origin(StringInfo s)
 				 errmsg_internal("ORIGIN message sent out of order")));
 }
 
+/*
+ * Handle MESSAGE message.
+ *
+ * Invoke a hook function if set.
+ */
+static void
+apply_handle_message(StringInfo s)
+{
+	LogicalRepMessageData msg;
+
+	if (!LogicalRepMessageHandle_hook)
+		return;
+
+	/*
+	 * Logical messages are handled only the (parallel) apply workers
+	 */
+	if (am_tablesync_worker() || am_sequencesync_worker())
+		return;
+
+	if (is_skipping_changes() ||
+		handle_streamed_transaction(LOGICAL_REP_MSG_MESSAGE, s))
+		return;
+
+	logicalrep_read_message(s, &msg);
+
+	(*LogicalRepMessageHandle_hook) (&msg);
+}
+
 /*
  * Initialize fileset (if not already done).
  *
@@ -3846,12 +3877,7 @@ apply_dispatch(StringInfo s)
 			break;
 
 		case LOGICAL_REP_MSG_MESSAGE:
-
-			/*
-			 * Logical replication does not use generic logical messages yet.
-			 * Although, it could be used by other applications that use this
-			 * output plugin.
-			 */
+			apply_handle_message(s);
 			break;
 
 		case LOGICAL_REP_MSG_STREAM_START:
@@ -5128,7 +5154,8 @@ maybe_reread_subscription(void)
 		newsub->passwordrequired != MySubscription->passwordrequired ||
 		strcmp(newsub->origin, MySubscription->origin) != 0 ||
 		newsub->owner != MySubscription->owner ||
-		!equal(newsub->publications, MySubscription->publications))
+		!equal(newsub->publications, MySubscription->publications) ||
+		newsub->message != MySubscription->message)
 	{
 		if (am_parallel_apply_worker())
 			ereport(LOG,
@@ -5614,6 +5641,7 @@ set_stream_options(WalRcvStreamOptions *options,
 
 	options->proto.logical.twophase = false;
 	options->proto.logical.origin = pstrdup(MySubscription->origin);
+	options->proto.logical.messages = MySubscription->message;
 }
 
 /*
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index c56437d6057..4441b14082c 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -5184,6 +5184,7 @@ getSubscriptions(Archive *fout)
 	int			i_subfailover;
 	int			i_subretaindeadtuples;
 	int			i_submaxretention;
+	int			i_submessage;
 	int			i,
 				ntups;
 
@@ -5282,9 +5283,14 @@ getSubscriptions(Archive *fout)
 							 " '-1' AS subwalrcvtimeout,\n");
 
 	if (fout->remoteVersion >= 190000)
-		appendPQExpBufferStr(query, " fs.srvname AS subservername\n");
+		appendPQExpBufferStr(query, " fs.srvname AS subservername,\n");
 	else
-		appendPQExpBufferStr(query, " NULL AS subservername\n");
+		appendPQExpBufferStr(query, " NULL AS subservername,\n");
+
+	if (fout->remoteVersion >= 190000)
+		appendPQExpBufferStr(query, " s.submessage\n");
+	else
+		appendPQExpBufferStr(query, "false AS submessage\n");
 
 	appendPQExpBufferStr(query,
 						 "FROM pg_subscription s\n");
@@ -5333,6 +5339,7 @@ getSubscriptions(Archive *fout)
 	i_subpublications = PQfnumber(res, "subpublications");
 	i_suborigin = PQfnumber(res, "suborigin");
 	i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn");
+	i_submessage = PQfnumber(res, "submessage");
 
 	subinfo = pg_malloc_array(SubscriptionInfo, ntups);
 
@@ -5390,6 +5397,8 @@ getSubscriptions(Archive *fout)
 		else
 			subinfo[i].suboriginremotelsn =
 				pg_strdup(PQgetvalue(res, i, i_suboriginremotelsn));
+		subinfo[i].submessage =
+			(strcmp(PQgetvalue(res, i, i_submessage), "t") == 0);
 
 		/* Decide whether we want to dump it */
 		selectDumpableObject(&(subinfo[i].dobj), fout);
@@ -5642,6 +5651,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
 	if (subinfo->subretaindeadtuples)
 		appendPQExpBufferStr(query, ", retain_dead_tuples = true");
 
+	if (subinfo->submessage)
+		appendPQExpBufferStr(query, ", message = true");
+
 	if (subinfo->submaxretention)
 		appendPQExpBuffer(query, ", max_retention_duration = %d", subinfo->submaxretention);
 
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 5a6726d8b12..30bd1575a26 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -721,6 +721,7 @@ typedef struct _SubscriptionInfo
 	bool		subrunasowner;
 	bool		subfailover;
 	bool		subretaindeadtuples;
+	bool		submessage;
 	int			submaxretention;
 	char	   *subservername;
 	char	   *subconninfo;
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index af3935b0078..ac5a2c1835b 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -7093,7 +7093,7 @@ describeSubscriptions(const char *pattern, bool verbose)
 	printQueryOpt myopt = pset.popt;
 	static const bool translate_columns[] = {false, false, false, false,
 		false, false, false, false, false, false, false, false, false, false,
-	false, false, false, false, false, false, false};
+	false, false, false, false, false, false, false, false};
 
 	if (pset.sversion < 100000)
 	{
@@ -7179,6 +7179,9 @@ describeSubscriptions(const char *pattern, bool verbose)
 			appendPQExpBuffer(&buf,
 							  ", subretentionactive AS \"%s\"\n",
 							  gettext_noop("Retention active"));
+			appendPQExpBuffer(&buf,
+							  ", submessage AS \"%s\"\n",
+							  gettext_noop("Message"));
 		}
 
 		appendPQExpBuffer(&buf,
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 48944201889..cdda8265573 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -83,6 +83,9 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
 	bool		subretaindeadtuples;	/* True if dead tuples useful for
 										 * conflict detection are retained */
 
+	bool		submessage;		/* True if the subscription wants to receive
+								 * logical messages. */
+
 	int32		submaxretention;	/* The maximum duration (in milliseconds)
 									 * for which information useful for
 									 * conflict detection can be retained */
@@ -171,6 +174,8 @@ typedef struct Subscription
 	List	   *publications;	/* List of publication names to subscribe to */
 	char	   *origin;			/* Only publish data originating from the
 								 * specified origin */
+	bool		message;		/* True if the subscription wants to receive
+								 * logical messages */
 } Subscription;
 
 #ifdef EXPOSE_TO_CLIENT_CODE
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index 058a955e20c..85add073202 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -191,6 +191,18 @@ typedef struct LogicalRepStreamAbortData
 	TimestampTz abort_time;
 } LogicalRepStreamAbortData;
 
+/*
+ * Logical decoding message information
+ */
+typedef struct LogicalRepMessageData
+{
+	uint8		flags;
+	XLogRecPtr	lsn;
+	const char *prefix;
+	Size		message_size;	/* length of message in bytes */
+	const char *message;		/* payload; may contain embedded nulls */
+} LogicalRepMessageData;
+
 extern void logicalrep_write_begin(StringInfo out, ReorderBufferTXN *txn);
 extern void logicalrep_read_begin(StringInfo in,
 								  LogicalRepBeginData *begin_data);
@@ -248,6 +260,7 @@ extern List *logicalrep_read_truncate(StringInfo in,
 									  bool *cascade, bool *restart_seqs);
 extern void logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn,
 									 bool transactional, const char *prefix, Size sz, const char *message);
+extern void logicalrep_read_message(StringInfo in, LogicalRepMessageData *msg_data);
 extern void logicalrep_write_rel(StringInfo out, TransactionId xid,
 								 Relation rel, Bitmapset *columns,
 								 PublishGencolsType include_gencols_type);
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 47c07574d4d..44a08cc2170 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -188,6 +188,7 @@ typedef struct
 									 * prepare time */
 			char	   *origin; /* Only publish data originating from the
 								 * specified origin */
+			bool		messages;	/* Logical message */
 		}			logical;
 	}			proto;
 } WalRcvStreamOptions;
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 745b7d9e969..b07042846cb 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -18,6 +18,7 @@
 #include "miscadmin.h"
 #include "replication/logicalrelation.h"
 #include "replication/walreceiver.h"
+#include "replication/logicalproto.h"
 #include "storage/buffile.h"
 #include "storage/fileset.h"
 #include "storage/shm_mq.h"
@@ -255,6 +256,9 @@ extern PGDLLIMPORT bool InitializingApplyWorker;
 
 extern PGDLLIMPORT List *table_states_not_ready;
 
+typedef void (*LogicalRepMessageHandle_hook_type) (LogicalRepMessageData *msg);
+extern PGDLLIMPORT LogicalRepMessageHandle_hook_type LogicalRepMessageHandle_hook;
+
 extern void logicalrep_worker_attach(int slot);
 extern LogicalRepWorker *logicalrep_worker_find(LogicalRepWorkerType wtype,
 												Oid subid, Oid relid,
diff --git a/src/test/modules/Makefile b/src/test/modules/Makefile
index 0a74ab5c86f..603fe1ff907 100644
--- a/src/test/modules/Makefile
+++ b/src/test/modules/Makefile
@@ -36,6 +36,7 @@ SUBDIRS = \
 		  test_integerset \
 		  test_json_parser \
 		  test_lfind \
+		  test_logicalmsg_hooks \
 		  test_lwlock_tranches \
 		  test_misc \
 		  test_oat_hooks \
diff --git a/src/test/modules/meson.build b/src/test/modules/meson.build
index 4bca42bb370..5f6c0cae77a 100644
--- a/src/test/modules/meson.build
+++ b/src/test/modules/meson.build
@@ -37,6 +37,7 @@ subdir('test_int128')
 subdir('test_integerset')
 subdir('test_json_parser')
 subdir('test_lfind')
+subdir('test_logicalmsg_hooks')
 subdir('test_lwlock_tranches')
 subdir('test_misc')
 subdir('test_oat_hooks')
diff --git a/src/test/modules/test_logicalmsg_hooks/.gitignore b/src/test/modules/test_logicalmsg_hooks/.gitignore
new file mode 100644
index 00000000000..5dcb3ff9723
--- /dev/null
+++ b/src/test/modules/test_logicalmsg_hooks/.gitignore
@@ -0,0 +1,4 @@
+# Generated subdirectories
+/log/
+/results/
+/tmp_check/
diff --git a/src/test/modules/test_logicalmsg_hooks/Makefile b/src/test/modules/test_logicalmsg_hooks/Makefile
new file mode 100644
index 00000000000..af129009b8a
--- /dev/null
+++ b/src/test/modules/test_logicalmsg_hooks/Makefile
@@ -0,0 +1,20 @@
+# src/test/modules/test_logicalmsg_hooks/Makefile
+
+MODULE_big = test_logicalmsg_hooks
+OBJS = \
+	$(WIN32RES) \
+	test_logicalmsg_hooks.o
+PGFILEDESC = "test_logicalmsg_hooks - test logical replication message hooks"
+
+TAP_TESTS = 1
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = src/test/modules/test_logicalmsg_hooks
+top_builddir = ../../../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/src/test/modules/test_logicalmsg_hooks/meson.build b/src/test/modules/test_logicalmsg_hooks/meson.build
new file mode 100644
index 00000000000..8084d82e47b
--- /dev/null
+++ b/src/test/modules/test_logicalmsg_hooks/meson.build
@@ -0,0 +1,28 @@
+# Copyright (c) 2022-2026, PostgreSQL Global Development Group
+
+test_logicalmsg_hooks_sources = files(
+  'test_logicalmsg_hooks.c',
+)
+
+if host_system == 'windows'
+  test_logicalmsg_hooks_sources += rc_lib_gen.process(win32ver_rc, extra_args: [
+    '--NAME', 'test_logicalmsg_hooks',
+    '--FILEDESC', 'test_logicalmsg_hooks - test logical message hooks',])
+endif
+
+test_logicalmsg_hooks = shared_module('test_logicalmsg_hooks',
+  test_logicalmsg_hooks_sources,
+  kwargs: pg_test_mod_args,
+)
+test_install_libs += test_logicalmsg_hooks
+
+tests += {
+  'name': 'test_logicalmsg_hooks',
+  'sd': meson.current_source_dir(),
+  'bd': meson.current_build_dir(),
+  'tap': {
+    'tests': [
+      't/001_basic.pl',
+    ],
+  },
+}
diff --git a/src/test/modules/test_logicalmsg_hooks/t/001_basic.pl b/src/test/modules/test_logicalmsg_hooks/t/001_basic.pl
new file mode 100644
index 00000000000..64ca51d1786
--- /dev/null
+++ b/src/test/modules/test_logicalmsg_hooks/t/001_basic.pl
@@ -0,0 +1,113 @@
+# Copyright (c) 2021-2026, PostgreSQL Global Development Group
+
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+my $pub = PostgreSQL::Test::Cluster->new('publisher');
+$pub->init(allows_streaming => 'logical');
+$pub->start;
+
+my $sub = PostgreSQL::Test::Cluster->new('subscriber');
+$sub->init(allows_streaming => 'logical');
+$sub->append_conf(
+	'postgresql.conf', q{
+shared_preload_libraries = 'test_logicalmsg_hooks'
+});
+$sub->start;
+
+$pub->safe_psql(
+	'postgres',
+	qq{
+CREATE TABLE test (a int);
+INSERT INTO test VALUES (1);
+CREATE PUBLICATION pub FOR ALL TABLES;
+});
+
+my $pub_connstr = $pub->connstr . ' dbname=postgres';
+
+$sub->safe_psql(
+	'postgres',
+	qq{
+CREATE TABLE test (a int primary key);
+CREATE SUBSCRIPTION sub CONNECTION '$pub_connstr' PUBLICATION pub WITH (message = true, disable_on_error = true)
+});
+
+$sub->wait_for_subscription_sync($pub, 'sub');
+
+my $log_location = -s $sub->logfile;
+
+# Emit a logical message and verify that the subscriber processed it in the
+# hook function.
+$pub->safe_psql(
+	'postgres',
+	qq{
+SELECT pg_logical_emit_message(true, 'test_prefix', 'test_mesasge');
+});
+$pub->wait_for_catchup('sub');
+$sub->wait_for_log(
+	qr/LOG[^\n]+received message: LSN [[:xdigit:]]+\/[[:xdigit:]]+, prefix: test_prefix, message: test_mesasge/,
+	$log_location);
+ok(1, "logical message is correctly handled on the subscriber");
+
+$log_location = -s $sub->logfile;
+
+# INSERT a tuple with a=1 conflicts on the subscriber. Test that ALTER SUBSCRIPTION ... SKIP
+# the transacitonal message as well.
+$pub->safe_psql(
+	'postgres',
+	q{
+BEGIN;
+INSERT INTO test VALUES (1);
+SELECT pg_logical_emit_message(true, 'conflict', 'should be skipped');
+COMMIT;
+});
+
+# Wait until a conflict occurs on the subscriber.
+$sub->poll_query_until('postgres',
+	q{SELECT subenabled = FALSE FROM pg_subscription WHERE subname = 'sub'});
+
+# Get the finish LSN of the error transaction.
+my $contents = slurp_file($sub->logfile, $log_location);
+$contents =~
+  qr/conflict detected on relation "public.test".*\n.*DETAIL:.*Could not apply remote change.*\n.*Key already exists in unique index "test_pkey", modified in transaction \d+: key .*, local row .*\n.*CONTEXT:.* for replication target relation "public.test" in transaction \d+, finished at ([[:xdigit:]]+\/[[:xdigit:]]+)/m
+  or die "could not get error-LSN";
+my $lsn = $1;
+
+$log_location = -s $sub->logfile;
+
+# Set skip LSN and re-enable the subscription.
+$sub->safe_psql('postgres', qq{ALTER SUBSCRIPTION sub SKIP (lsn = '$lsn');});
+$sub->safe_psql('postgres', "ALTER SUBSCRIPTION sub ENABLE");
+
+# Wait for the failed transaction to be skipped
+$sub->poll_query_until('postgres',
+	q{SELECT subskiplsn = '0/0' FROM pg_subscription WHERE subname = 'sub'});
+
+ok( !$sub->log_contains(
+		qr/LOG[^\n]+received message: LSN .*, prefix: conflict, message: should be skipped/
+	),
+	"logical message is skipped by ALTER SUBSCRIPTION SKIP");
+
+# Test that a non-transactional message is applied even if the transaction rolls back.
+$log_location = -s $sub->logfile;
+$pub->safe_psql(
+	'postgres',
+	q{
+BEGIN;
+SELECT pg_logical_emit_message(false, 'test_prefix', 'non-transactional message');
+ROLLBACK;
+});
+$pub->wait_for_catchup('sub');
+$sub->wait_for_log(
+	qr/LOG[^\n]+received message: LSN [[:xdigit:]]+\/[[:xdigit:]]+, prefix: test_prefix, message: non-transactional message/,
+	$log_location);
+ok(1, "logical message is correctly handled on the subscriber");
+
+
+$pub->stop;
+$sub->stop;
+done_testing();
diff --git a/src/test/modules/test_logicalmsg_hooks/test_logicalmsg_hooks.c b/src/test/modules/test_logicalmsg_hooks/test_logicalmsg_hooks.c
new file mode 100644
index 00000000000..9167043508e
--- /dev/null
+++ b/src/test/modules/test_logicalmsg_hooks/test_logicalmsg_hooks.c
@@ -0,0 +1,40 @@
+/*--------------------------------------------------------------------------
+ *
+ * test_logicalmsg_hooks.c
+ *		Code for testing LogicalMessageHandle_hook
+ *
+ * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ *		src/test/modules/test_logicalmsg_hooks/test_logicalmsg_hooks.c
+ *
+ * -------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "replication/worker_internal.h"
+
+PG_MODULE_MAGIC;
+
+static void test_logical_message_handler(LogicalRepMessageData *msg);
+
+/*
+ * Module load callback
+ */
+void
+_PG_init(void)
+{
+	LogicalRepMessageHandle_hook = &test_logical_message_handler;
+}
+
+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)));
+}
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 8dbfac66326..29c2fac6504 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -139,18 +139,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
 \dRs+ regress_testsub4
-                                                                                                                                                                       List of subscriptions
-       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | none   | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | none   | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
 \dRs+ regress_testsub4
-                                                                                                                                                                       List of subscriptions
-       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 DROP SUBSCRIPTION regress_testsub3;
@@ -237,10 +237,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
 ERROR:  invalid connection string syntax: missing "=" after "foobar" in connection info string
 
 \dRs+
-                                                                                                                                                                          List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  |    Description    
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | test subscription
+                                                                                                                                                                               List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  |    Description    
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | test subscription
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -249,10 +249,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname');
 ALTER SUBSCRIPTION regress_testsub SET (password_required = false);
 ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true);
 \dRs+
-                                                                                                                                                                              List of subscriptions
-      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |           Conninfo           | Receiver timeout |  Skip LSN  |    Description    
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------
- regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | parallel  | d                | f                | any    | f                 | t             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist2 | -1               | 0/00000000 | test subscription
+                                                                                                                                                                                   List of subscriptions
+      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |           Conninfo           | Receiver timeout |  Skip LSN  |    Description    
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+------------------------------+------------------+------------+-------------------
+ regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | parallel  | d                | f                | any    | f                 | t             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist2 | -1               | 0/00000000 | test subscription
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
@@ -268,10 +268,10 @@ ERROR:  unrecognized subscription parameter: "create_slot"
 -- ok
 ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
 \dRs+
-                                                                                                                                                                              List of subscriptions
-      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |           Conninfo           | Receiver timeout |  Skip LSN  |    Description    
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------
- regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist2 | -1               | 0/00012345 | test subscription
+                                                                                                                                                                                   List of subscriptions
+      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |           Conninfo           | Receiver timeout |  Skip LSN  |    Description    
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+------------------------------+------------------+------------+-------------------
+ regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist2 | -1               | 0/00012345 | test subscription
 (1 row)
 
 -- ok - with lsn = NONE
@@ -280,10 +280,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
 ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
 ERROR:  invalid WAL location (LSN): 0/0
 \dRs+
-                                                                                                                                                                              List of subscriptions
-      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |           Conninfo           | Receiver timeout |  Skip LSN  |    Description    
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------
- regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist2 | -1               | 0/00000000 | test subscription
+                                                                                                                                                                                   List of subscriptions
+      Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |           Conninfo           | Receiver timeout |  Skip LSN  |    Description    
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+------------------------------+------------------+------------+-------------------
+ regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist2 | -1               | 0/00000000 | test subscription
 (1 row)
 
 BEGIN;
@@ -319,10 +319,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (wal_receiver_timeout = '80s');
 ALTER SUBSCRIPTION regress_testsub_foo SET (wal_receiver_timeout = 'foobar');
 ERROR:  invalid value for parameter "wal_receiver_timeout": "foobar"
 \dRs+
-                                                                                                                                                                                List of subscriptions
-        Name         |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |           Conninfo           | Receiver timeout |  Skip LSN  |    Description    
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------
- regress_testsub_foo | regress_subscription_user | f       | {testpub2,testpub3} | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | local              | dbname=regress_doesnotexist2 | 80s              | 0/00000000 | test subscription
+                                                                                                                                                                                     List of subscriptions
+        Name         |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |           Conninfo           | Receiver timeout |  Skip LSN  |    Description    
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+------------------------------+------------------+------------+-------------------
+ regress_testsub_foo | regress_subscription_user | f       | {testpub2,testpub3} | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | local              | dbname=regress_doesnotexist2 | 80s              | 0/00000000 | test subscription
 (1 row)
 
 -- rename back to keep the rest simple
@@ -351,19 +351,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
 \dRs+
-                                                                                                                                                                       List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f       | {testpub}   | t      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | t      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (binary = false);
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 \dRs+
-                                                                                                                                                                       List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 DROP SUBSCRIPTION regress_testsub;
@@ -375,27 +375,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
 \dRs+
-                                                                                                                                                                       List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
 \dRs+
-                                                                                                                                                                       List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 \dRs+
-                                                                                                                                                                       List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 -- fail - publication already exists
@@ -410,10 +410,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
 ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
 ERROR:  publication "testpub1" is already in subscription "regress_testsub"
 \dRs+
-                                                                                                                                                                               List of subscriptions
-      Name       |           Owner           | Enabled |         Publication         | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f       | {testpub,testpub1,testpub2} | f      | off       | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                                    List of subscriptions
+      Name       |           Owner           | Enabled |         Publication         | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub,testpub1,testpub2} | f      | off       | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 -- fail - publication used more than once
@@ -428,10 +428,10 @@ ERROR:  publication "testpub3" is not in subscription "regress_testsub"
 -- ok - delete publications
 ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
 \dRs+
-                                                                                                                                                                       List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 DROP SUBSCRIPTION regress_testsub;
@@ -467,19 +467,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
 \dRs+
-                                                                                                                                                                       List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | p                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | p                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 -- we can alter streaming when two_phase enabled
 ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
 \dRs+
-                                                                                                                                                                       List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -489,10 +489,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
 \dRs+
-                                                                                                                                                                       List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -505,18 +505,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
 \dRs+
-                                                                                                                                                                       List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
 \dRs+
-                                                                                                                                                                       List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | t                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | t                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -529,10 +529,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
 \dRs+
-                                                                                                                                                                       List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -549,10 +549,10 @@ NOTICE:  max_retention_duration is ineffective when retain_dead_tuples is disabl
 WARNING:  subscription was created, but is not connected
 HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
 \dRs+
-                                                                                                                                                                       List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                   1000 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                   1000 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
 -- fail - max_retention_duration must be non-negative
@@ -561,12 +561,38 @@ ERROR:  max_retention_duration cannot be negative
 -- ok
 ALTER SUBSCRIPTION regress_testsub SET (max_retention_duration = 0);
 \dRs+
-                                                                                                                                                                       List of subscriptions
-      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+(1 row)
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+-- fail - message must be boolean
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, message = foo);
+ERROR:  message requires a Boolean value
+-- ok
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, message = true);
+WARNING:  subscription was created, but is not connected
+HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
+\dRs+
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | t       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
+(1 row)
+
+-- ok
+ALTER SUBSCRIPTION regress_testsub SET (message = false);
+\dRs+
+                                                                                                                                                                            List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit |          Conninfo           | Receiver timeout |  Skip LSN  | Description 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+-------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        |        | f                  |                      0 | f                | f       | off                | dbname=regress_doesnotexist | -1               | 0/00000000 | 
 (1 row)
 
+-- cleanup
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 DROP SUBSCRIPTION regress_testsub;
 -- let's do some tests with pg_create_subscription rather than superuser
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 05533d66675..7eef3721a9a 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -423,6 +423,23 @@ ALTER SUBSCRIPTION regress_testsub SET (max_retention_duration = 0);
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 DROP SUBSCRIPTION regress_testsub;
 
+-- fail - message must be boolean
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, message = foo);
+
+-- ok
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, message = true);
+
+\dRs+
+
+-- ok
+ALTER SUBSCRIPTION regress_testsub SET (message = false);
+
+\dRs+
+
+-- cleanup
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+
 -- let's do some tests with pg_create_subscription rather than superuser
 SET SESSION AUTHORIZATION regress_subscription_user3;
 
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 1969d467c1d..09c9b87d483 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1701,6 +1701,7 @@ LogicalRepBeginData
 LogicalRepCommitData
 LogicalRepCommitPreparedTxnData
 LogicalRepCtxStruct
+LogicalRepMessageData
 LogicalRepMsgType
 LogicalRepPartMapEntry
 LogicalRepPreparedTxnData
-- 
2.54.0

---

From: Bharath Rupireddy <[email-masked]>
Date: Mon, 22 Jun 2026 15:39:00 -0700

Hi,

> replication, or trigger user-specific routines on the subscriber side.

Thanks for working on this!

> startup in order to receive the messages and trigger the hook

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?

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

Nit: I'd call these logical decoding messages or generic logical
messages - something to match the docs and pg_logical_emit_message.

2/
+void

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/

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/
+ /*

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

---

From: Amit Kapila <[email-masked]>
Date: Tue, 23 Jun 2026 14:22:29 +0530

> logic executing the decoded DDL messages directly, not something going

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] 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:

{
    "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"
 }

> with the pgoutput plugin?

What advantage do you see with the same?

-- 
With Regards,
Amit Kapila.

---

From: Masahiko Sawada <[email-masked]>
Date: Tue, 23 Jun 2026 11:01:49 -0700

On Tue, Jun 23, 2026 at 1:52 AM Amit Kapila <[email-masked]> wrote:
> 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

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,

---

From: Masahiko Sawada <[email-masked]>
Date: Tue, 23 Jun 2026 11:48:28 -0700

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

> logic executing the decoded DDL messages directly, not something going

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.

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

> the lines of RegisterXactCallback would be cleaner and work better.

Yeah, it would be better.

> messages - something to match the docs and pg_logical_emit_message.

Will fix.

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

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

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

> it's worth mentioning in the docs for pg_logical_emit_message.

Agreed.

Regards,

--
Masahiko Sawada

---

From: Masahiko Sawada <[email-masked]>
Date: Tue, 23 Jun 2026 17:21:51 -0700

On Tue, Jun 23, 2026 at 11:01 AM Masahiko Sawada <[email-masked]> wrote:
> hook as a sample implementation.

FYI, I've implemented a test extension, pg_logical_ddl[1], 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] 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] https://www.postgresql.org/message-id/CAD21AoCEVO%2BzpLmQqKwZbJ5%2BrvqsJ1e0wnTNBH437p8tDw7B%3Dg%40mail.gmail.com

---

From: Fujii Masao <[email-masked]>
Date: Thu, 9 Jul 2026 21:14:46 +0900

> 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