From 46702d12a67acd82fe6656a7d985a767f31eec20 Mon Sep 17 00:00:00 2001
From: Zhijie Hou <houzj.fnst@fujitsu.com>
Date: Tue, 11 Aug 2026 13:53:51 +0800
Subject: [PATCH v23] Support remote relation synchronization for parallel
 apply workers

Today, a subscription with streaming disabled (or with small transactions)
applies every transaction serially in the leader apply worker: read a change,
apply it, commit, read the next. The leader quickly becomes the bottleneck,
while the subscriber machine sits mostly idle.

This patch series introduces parallel apply for non-stream transactions,
teaching the leader to act as a dispatcher: it reads the replication stream and
hands each non-streamed transaction to a pool of parallel apply workers (one
transaction per worker), so that independent transactions are applied
concurrently. Workers are reused across transactions to avoid process startup
costs.

Parallel apply workers are separate processes with their own relation mapping
caches, so the leader distributes remote relation information to keep them in
sync with the publisher's view of each table. A newly started worker receives
all known remote relation information. When a RELATION message arrives later
(e.g., due to a schema change or first touch of a table), the leader builds a
new relation message in a uniform internal format and distributes it to all
active workers, rather than forwarding the publisher's message as-is. This is
because the original format varies depending on whether the message is
streamed. A RELATION message also records a table-wide dependency (see below),
ensuring that transactions dispatched against the old definition are finished
before any worker applies changes using the new one.

To benefit from real parallelism, the leader does not wait for a worker to
commit an assigned transaction. When a COMMIT is dispatched, the leader records
the transaction in the flush-position list, bound to a shared-memory slot that
the worker fills with its local commit LSN. The leader collects the LSN lazily,
when reporting feedback to the walsender or reusing the worker, and never
advances the reported flush position past an uncommitted transaction.

Applying transactions concurrently is only safe if the result matches serial
application. To guarantee this, the patch set also adds dependency tracking and
commit order preservation:

Before dispatching changes to a parallel worker, the leader checks whether the
current modification affects the same row (identified by its replica identity
key) as another ongoing transaction. The replica identity keys of tuples
modified by parallelized transactions are recorded in a leader-local hash table.
If a change hits a key touched by an ongoing transaction, the leader sends the
list of dependent transaction IDs to the parallel worker, instructing it to wait
for those transactions to commit. If the leader applies the transaction itself,
it waits directly. Table-wide operations (TRUNCATE, RELATION messages) depend on
all ongoing transactions that modified the same table. This prevents conflicts
such as updating a row that a parallel worker has not finished inserting.

We preserve publisher commit order for all transactions for two reasons:
- User-visible consistency: out-of-order commits can expose states on the
  subscriber that were never visible on the publisher.
- Replication progress tracking: progress is tracked using the last
  transaction's commit LSN, which becomes ambiguous if commits happen out of
  order.

Each parallelized transaction waits for the last parallelized transaction before
committing; waiting for the immediate predecessor transitively waits for all
preceding transactions.

Beyond replica identity, dependencies are also tracked for local unique keys
and foreign keys on the subscriber in a similar way. This prevents conflicts
such as inserting a row that conflicts with a unique key that a parallel worker
has not finished deleting, or inserting a foreign key referencing a row that
another worker has not yet inserted.

--

This patch adds support for synchronizing remote relation information between
the leader and parallel apply workers. This is necessary when applying
non-streaming transactions, as parallel workers need to map local replication
target relations to their remote counterparts during change application. Since
the walsender does not send remote relation information with every transaction,
parallel workers may not have up-to-date relation info unless synchronized by
the leader.

A new internal worker message type (PAWorkerMsgType.PA_MSG_RELMAP) is introduced
to carry a list of relations known to the leader. This message is generated and
sent to parallel workers when the leader receives remote relation information
or allocates a new parallel worker to a transaction.

In the logical replication protocol, PAWorkerMsgType messages are encapsulated
within LOGICAL_REP_MSG_INTERNAL_MESSAGE using the uniform format:
LOGICAL_REP_MSG_INTERNAL_MESSAGE + PAWorkerMsgType + internal data. This format
is used both for sending messages to parallel workers and for serializing to
disk.

Later patches will add additional PAWorkerMsgType message types for dependency
waiting.

Author: Zhijie Hou <houzj.fnst@fujitsu.com>
Author: Hayato Kuroda <kuroda.hayato@fujitsu.com>
---
 .../replication/logical/applyparallelworker.c | 169 ++++++++++++++++--
 src/backend/replication/logical/proto.c       |  44 +++++
 src/backend/replication/logical/relation.c    |  43 +++++
 src/backend/replication/logical/worker.c      |  14 ++
 src/include/replication/logicalproto.h        |   7 +
 src/include/replication/logicalrelation.h     |   2 +
 src/include/replication/worker_internal.h     |  15 ++
 src/tools/pgindent/typedefs.list              |   1 +
 8 files changed, 277 insertions(+), 18 deletions(-)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 012d55e9d3d..cad437fd06e 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -410,6 +410,8 @@ pa_launch_parallel_worker(void)
 	bool		launched;
 	ParallelApplyWorkerInfo *winfo;
 	ListCell   *lc;
+	StringInfoData out;
+	int			num_rels;
 
 	/* Try to get an available parallel apply worker from the worker pool. */
 	foreach(lc, ParallelApplyWorkerPool)
@@ -459,6 +461,26 @@ pa_launch_parallel_worker(void)
 
 	MemoryContextSwitchTo(oldcontext);
 
+	initStringInfo(&out);
+
+	/*
+	 * Send all existing remote relation information to the parallel apply
+	 * worker. This allows the parallel worker to initialize the
+	 * LogicalRepRelMapEntry locally before applying remote changes. This is
+	 * needed since the walsender does not send remote relation information
+	 * with every transaction.
+	 */
+	logicalrep_write_all_internal_rels(&out, &num_rels);
+
+	/*
+	 * Timeout is unlikely here because the worker doesn't hold any locks while
+	 * processing relation information, so it's safe from deadlocks.
+	 */
+	if (num_rels && !pa_send_data(winfo, out.len, out.data))
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("could not send remote relation information to the logical replication parallel apply worker")));
+
 	return winfo;
 }
 
@@ -733,6 +755,52 @@ ProcessParallelApplyInterrupts(void)
 	}
 }
 
+/*
+ * Handle internal relation information.
+ *
+ * Update all relation details in the relation map cache.
+ */
+static void
+apply_handle_internal_relation(StringInfo s)
+{
+	int			nrels = pq_getmsgint(s, 4);
+
+	for (int i = 0; i < nrels; i++)
+	{
+		LogicalRepRelation *rel = logicalrep_read_rel(s);
+
+		logicalrep_relmap_update(rel);
+
+		/* Also reset all entries in the partition map that refer to remoterel. */
+		logicalrep_partmap_reset_relmap(rel);
+
+		elog(DEBUG1, "parallel apply worker init relmap for %s",
+			 rel->relname);
+	}
+}
+
+/*
+ * Handle an internal message generated by the leader apply worker.
+ */
+void
+apply_handle_internal_message(StringInfo s)
+{
+	PAWorkerMsgType action = pq_getmsgbyte(s);
+
+	Assert(am_parallel_apply_worker());
+
+	switch (action)
+	{
+		case PA_MSG_RELMAP:
+			apply_handle_internal_relation(s);
+			break;
+		default:
+			ereport(ERROR,
+					(errcode(ERRCODE_PROTOCOL_VIOLATION),
+					 errmsg("invalid worker internal message type \"??? (%d)\"", action)));
+	}
+}
+
 /* Parallel apply worker main loop. */
 static void
 LogicalParallelApplyLoop(shm_mq_handle *mqh)
@@ -741,6 +809,14 @@ LogicalParallelApplyLoop(shm_mq_handle *mqh)
 	ErrorContextCallback errcallback;
 	MemoryContext oldcxt = CurrentMemoryContext;
 
+	/*
+	 * Ensure LOGICAL_REP_MSG_INTERNAL_MESSAGE does not conflict with
+	 * PqReplMsg_WALData ('d'), as parallel apply workers may receive both types
+	 * of messages.
+	 */
+	StaticAssertDecl(LOGICAL_REP_MSG_INTERNAL_MESSAGE != PqReplMsg_WALData,
+					 "LOGICAL_REP_MSG_INTERNAL_MESSAGE conflicts with PqReplMsg_WALData");
+
 	/*
 	 * Init the ApplyMessageContext which we clean up after each replication
 	 * protocol message.
@@ -779,26 +855,30 @@ LogicalParallelApplyLoop(shm_mq_handle *mqh)
 
 			initReadOnlyStringInfo(&s, data, len);
 
-			/*
-			 * The first byte of messages sent from leader apply worker to
-			 * parallel apply workers can only be PqReplMsg_WALData.
-			 */
 			c = pq_getmsgbyte(&s);
-			if (c != PqReplMsg_WALData)
-				elog(ERROR, "unexpected message \"%c\"", c);
-
-			/*
-			 * Ignore statistics fields that have been updated by the leader
-			 * apply worker.
-			 *
-			 * XXX We can avoid sending the statistics fields from the leader
-			 * apply worker but for that, it needs to rebuild the entire
-			 * message by removing these fields which could be more work than
-			 * simply ignoring these fields in the parallel apply worker.
-			 */
-			s.cursor += SIZE_STATS_MESSAGE;
+			if (c == PqReplMsg_WALData)
+			{
+				/*
+				 * Ignore statistics fields that have been updated by the leader
+				 * apply worker.
+				 *
+				 * XXX We can avoid sending the statistics fields from the leader
+				 * apply worker but for that, it needs to rebuild the entire
+				 * message by removing these fields which could be more work than
+				 * simply ignoring these fields in the parallel apply worker.
+				 */
+				s.cursor += SIZE_STATS_MESSAGE;
 
-			apply_dispatch(&s);
+				apply_dispatch(&s);
+			}
+			else if (c == LOGICAL_REP_MSG_INTERNAL_MESSAGE)
+			{
+				/* Rewind so apply_dispatch can re-read the message type. */
+				s.cursor--;
+				apply_dispatch(&s);
+			}
+			else
+				elog(ERROR, "unexpected message \"%c\"", c);
 		}
 		else if (shmq_res == SHM_MQ_WOULD_BLOCK)
 		{
@@ -1656,3 +1736,56 @@ pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
 
 	pa_free_worker(winfo);
 }
+
+/*
+ * Distribute remote relation information to all active parallel apply workers.
+ */
+void
+pa_distribute_remote_rel_to_workers(LogicalRepRelation *rel)
+{
+	StringInfoData out;
+
+	if (!am_leader_apply_worker())
+		return;
+
+	if (!ParallelApplyWorkerPool)
+		return;
+
+	/*
+	 * Build a new relation message instead of reusing the one received from the
+	 * publisher. The original format may differ depending on whether the
+	 * message is streamed or not (e.g., streamed messages include an additional
+	 * XID). Since parallel workers may receive this message outside of any
+	 * transaction context, it would be difficult for them to interpret the
+	 * differences. To keep it simple and consistent, we construct a new
+	 * relation message in the uniform internal message format.
+	 */
+	initStringInfo(&out);
+	logicalrep_write_one_internal_rel(&out, rel);
+
+	foreach_ptr(ParallelApplyWorkerInfo, winfo, ParallelApplyWorkerPool)
+	{
+		/*
+		 * Skip the worker responsible for the current transaction, as the
+		 * relation information has already been sent to it.
+		 */
+		if (winfo == stream_apply_worker)
+			continue;
+
+		/*
+		 * Skip the worker that is in serialize mode, as they will soon stop
+		 * once they finish applying the transaction.
+		 */
+		if (winfo->serialize_changes)
+			continue;
+
+		/*
+		 * TODO: Support switching to PARTIAL_SERIALIZE mode when the send
+		 * buffer becomes full.
+		 */
+		if (!pa_send_data(winfo, out.len, out.data))
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("could not send remote relation information to the logical replication parallel apply worker")));
+	}
+}
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index 86ad97cd937..ae093aad7b1 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -691,6 +691,48 @@ logicalrep_write_rel(StringInfo out, TransactionId xid, Relation rel,
 	logicalrep_write_attrs(out, rel, columns, include_gencols_type);
 }
 
+/*
+ * Write internal relation description to the output stream.
+ *
+ * This is similar to logicalrep_write_rel, but takes a LogicalRepRelation and
+ * deparses it into the same message format. The output can be used in contexts
+ * where a relation description is needed without a live relation object.
+ */
+void
+logicalrep_write_internal_rel(StringInfo out, LogicalRepRelation *rel)
+{
+	pq_sendint32(out, rel->remoteid);
+
+	/* Write relation name. */
+	pq_sendstring(out, rel->nspname);
+	pq_sendstring(out, rel->relname);
+
+	/* Write the replica identity. */
+	pq_sendbyte(out, rel->replident);
+
+	/* Write attribute description. */
+	pq_sendint16(out, rel->natts);
+
+	for (int i = 0; i < rel->natts; i++)
+	{
+		uint8		flags = 0;
+
+		if (bms_is_member(i, rel->attkeys))
+			flags |= LOGICALREP_IS_REPLICA_IDENTITY;
+
+		pq_sendbyte(out, flags);
+
+		/* Attribute name. */
+		pq_sendstring(out, rel->attnames[i]);
+
+		/* Attribute type ID. */
+		pq_sendint32(out, rel->atttyps[i]);
+
+		/* Ignore attribute mode for now. */
+		pq_sendint32(out, 0);
+	}
+}
+
 /*
  * Read the relation info from stream and return as LogicalRepRelation.
  */
@@ -1253,6 +1295,8 @@ logicalrep_message_type(LogicalRepMsgType action)
 			return "STREAM ABORT";
 		case LOGICAL_REP_MSG_STREAM_PREPARE:
 			return "STREAM PREPARE";
+		case LOGICAL_REP_MSG_INTERNAL_MESSAGE:
+			return "INTERNAL MESSAGE";
 	}
 
 	/*
diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c
index 87498264256..d12b12c421a 100644
--- a/src/backend/replication/logical/relation.c
+++ b/src/backend/replication/logical/relation.c
@@ -23,6 +23,7 @@
 #include "catalog/namespace.h"
 #include "catalog/pg_subscription_rel.h"
 #include "executor/executor.h"
+#include "libpq/pqformat.h"
 #include "nodes/makefuncs.h"
 #include "replication/logicalrelation.h"
 #include "replication/worker_internal.h"
@@ -58,6 +59,48 @@ typedef struct LogicalRepPartMapEntry
 static Oid	FindLogicalRepLocalIndex(Relation localrel, LogicalRepRelation *remoterel,
 									 AttrMap *attrMap);
 
+/*
+ * Write all the remote relation information from the LogicalRepRelMapEntry to
+ * the output stream. The number of relations is stored in the num_rels.
+ */
+void
+logicalrep_write_all_internal_rels(StringInfo out, int *num_rels)
+{
+	LogicalRepRelMapEntry *entry;
+	HASH_SEQ_STATUS status;
+
+	if (LogicalRepRelMap)
+		*num_rels = hash_get_num_entries(LogicalRepRelMap);
+	else
+		*num_rels = 0;
+
+	if (*num_rels == 0)
+		return;
+
+	pq_sendbyte(out, LOGICAL_REP_MSG_INTERNAL_MESSAGE);
+	pq_sendbyte(out, PA_MSG_RELMAP);
+	pq_sendint(out, *num_rels, 4);
+
+	hash_seq_init(&status, LogicalRepRelMap);
+
+	while ((entry = (LogicalRepRelMapEntry *) hash_seq_search(&status)) != NULL)
+		logicalrep_write_internal_rel(out, &entry->remoterel);
+}
+
+/*
+ * Similar to logicalrep_write_all_internal_rels but writes only the given
+ * relation.
+ */
+void
+logicalrep_write_one_internal_rel(StringInfo out, LogicalRepRelation *rel)
+{
+	pq_sendbyte(out, LOGICAL_REP_MSG_INTERNAL_MESSAGE);
+	pq_sendbyte(out, PA_MSG_RELMAP);
+	pq_sendint(out, 1, 4);
+
+	logicalrep_write_internal_rel(out, rel);
+}
+
 /*
  * Relcache invalidation callback for our relation map cache.
  */
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 92ea1d0df24..5b3cd6cbeab 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -789,6 +789,14 @@ handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)
 	TransApplyAction apply_action;
 	StringInfoData original_msg;
 
+	/*
+	 * Return early if not in a streamed transaction. The leader distributes
+	 * relation messages to all workers, but if the worker is not handling any
+	 * transaction, there is nothing to do here.
+	 */
+	if (!in_streamed_transaction)
+		return false;
+
 	apply_action = get_transaction_apply_action(stream_xid, &winfo);
 
 	/* not in streaming mode */
@@ -2590,6 +2598,8 @@ apply_handle_relation(StringInfo s)
 
 	/* Also reset all entries in the partition map that refer to remoterel. */
 	logicalrep_partmap_reset_relmap(rel);
+
+	pa_distribute_remote_rel_to_workers(rel);
 }
 
 /*
@@ -3891,6 +3901,10 @@ apply_dispatch(StringInfo s)
 			apply_handle_stream_prepare(s);
 			break;
 
+		case LOGICAL_REP_MSG_INTERNAL_MESSAGE:
+			apply_handle_internal_message(s);
+			break;
+
 		default:
 			ereport(ERROR,
 					(errcode(ERRCODE_PROTOCOL_VIOLATION),
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index 058a955e20c..3a212d24c4a 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -53,6 +53,10 @@
  * in logical replication protocol, which uses a single byte to identify a
  * message type. Hence the values should be single-byte wide and preferably
  * human-readable characters.
+ *
+ * LOGICAL_REP_MSG_INTERNAL_MESSAGE ('i') is reserved for internal messages
+ * sent from the leader apply worker to parallel apply workers. The
+ * PAWorkerMsgType enum identifies the internal sub-message payload.
  */
 typedef enum LogicalRepMsgType
 {
@@ -75,6 +79,7 @@ typedef enum LogicalRepMsgType
 	LOGICAL_REP_MSG_STREAM_COMMIT = 'c',
 	LOGICAL_REP_MSG_STREAM_ABORT = 'A',
 	LOGICAL_REP_MSG_STREAM_PREPARE = 'p',
+	LOGICAL_REP_MSG_INTERNAL_MESSAGE = 'i',
 } LogicalRepMsgType;
 
 /*
@@ -251,6 +256,8 @@ extern void logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecP
 extern void logicalrep_write_rel(StringInfo out, TransactionId xid,
 								 Relation rel, Bitmapset *columns,
 								 PublishGencolsType include_gencols_type);
+extern void logicalrep_write_internal_rel(StringInfo out,
+							 LogicalRepRelation *rel);
 extern LogicalRepRelation *logicalrep_read_rel(StringInfo in);
 extern void logicalrep_write_typ(StringInfo out, TransactionId xid,
 								 Oid typoid);
diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h
index efe0f9d6031..8fe6ec8ed78 100644
--- a/src/include/replication/logicalrelation.h
+++ b/src/include/replication/logicalrelation.h
@@ -50,5 +50,7 @@ extern void logicalrep_rel_close(LogicalRepRelMapEntry *rel,
 								 LOCKMODE lockmode);
 extern bool IsIndexUsableForReplicaIdentityFull(Relation idxrel, AttrMap *attrmap);
 extern Oid	GetRelationIdentityOrPK(Relation rel);
+extern void logicalrep_write_all_internal_rels(StringInfo out, int *num_rels);
+extern void logicalrep_write_one_internal_rel(StringInfo out, LogicalRepRelation *rel);
 
 #endif							/* LOGICALRELATION_H */
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 88cb7c1e252..f2c2f90c451 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -233,6 +233,18 @@ typedef struct ParallelApplyWorkerInfo
 	ParallelApplyWorkerShared *shared;
 } ParallelApplyWorkerInfo;
 
+/*
+ * Parallel apply worker internal message types.
+ *
+ * These types of messages are generated by the leader apply worker and sent to
+ * the parallel apply worker, encapsulated within the
+ * LOGICAL_REP_MSG_INTERNAL_MESSAGE type.
+ */
+typedef enum PAWorkerMsgType
+{
+	PA_MSG_RELMAP = 'r',
+} PAWorkerMsgType;
+
 /* Main memory context for apply worker. Permanent during worker lifetime. */
 extern PGDLLIMPORT MemoryContext ApplyContext;
 
@@ -333,6 +345,8 @@ extern void pa_allocate_worker(TransactionId xid);
 extern ParallelApplyWorkerInfo *pa_find_worker(TransactionId xid);
 extern void pa_detach_all_error_mq(void);
 
+extern void apply_handle_internal_message(StringInfo s);
+
 extern bool pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes,
 						 const void *data);
 extern void pa_switch_to_partial_serialize(ParallelApplyWorkerInfo *winfo,
@@ -359,6 +373,7 @@ extern void pa_decr_and_wait_stream_block(void);
 
 extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
 						   XLogRecPtr remote_lsn);
+extern void pa_distribute_remote_rel_to_workers(LogicalRepRelation *rel);
 
 #define isParallelApplyWorker(worker) ((worker)->in_use && \
 									   (worker)->type == WORKERTYPE_PARALLEL_APPLY)
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index d7e289b3867..226b6cdf1ae 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1932,6 +1932,7 @@ OverridingKind
 PACE_HEADER
 PACL
 PATH
+PAWorkerMsgType
 PCtxtHandle
 PERL_CONTEXT
 PERL_SI
-- 
2.43.0

