From 7e3c3eacf234cf38a57fd6171ebda50a25e35226 Mon Sep 17 00:00:00 2001
From: Zhijie Hou <houzj.fnst@fujitsu.com>
Date: Tue, 11 Aug 2026 14:11:45 +0800
Subject: [PATCH v23 06/12] Parallel apply non-streaming transactions

This patch allows the leader apply worker to dispatch non-streaming
transactions to parallel apply workers, leveraging the commit order
preservation and row change dependency tracking mechanisms from preceding
patches.

On receiving a BEGIN message, the leader finds a free worker and assigns the
current transaction and its subsequent changes to it. If no workers are
available, the leader applies the transaction itself.

Before dispatching changes to a parallel worker, the leader checks whether the
current modification affects the same row (identified by the replica identity
key) as another ongoing transaction. If so, the leader sends a list of
dependent transaction IDs to the parallel worker, instructing it to wait for
those transactions to commit before proceeding.

Before sending the final COMMIT message of a transaction to a parallel worker,
the leader sends a PA_MSG_XACT_DEPENDENCY message containing the last
parallelized transaction ID. The parallel worker waits for that transaction to
commit, ensuring commit order is preserved.

The leader does not wait for the parallel worker to finish applying a
transaction, nor does it stop workers in the pool, enabling greater parallelism
in transaction application. The flush position is updated lazily, after the
parallel worker completes the transaction.

Currently, streamed transactions cannot be applied in parallel with non-streamed
ones, and PREPARED transactions are not dispatched to parallel workers. Later
patches will lift these restrictions and add support for dependency tracking
for streamed transactions and parallel apply of PREPARED transactions.

Author: Zhijie Hou <houzj.fnst@fujitsu.com>
Author: Hayato Kuroda <kuroda.hayato@fujitsu.com>
---
 .../replication/logical/applyparallelworker.c | 137 +++++-
 src/backend/replication/logical/worker.c      | 231 +++++++++-
 src/include/replication/worker_internal.h     |   2 +
 src/test/subscription/meson.build             |   1 +
 src/test/subscription/t/001_rep_changes.pl    |   2 +
 src/test/subscription/t/010_truncate.pl       |   2 +-
 src/test/subscription/t/015_stream.pl         |  10 +-
 src/test/subscription/t/026_stats.pl          |   1 +
 src/test/subscription/t/027_nosuperuser.pl    |   1 +
 src/test/subscription/t/050_parallel_apply.pl | 424 ++++++++++++++++++
 src/tools/pgindent/typedefs.list              |   2 +-
 11 files changed, 792 insertions(+), 21 deletions(-)
 create mode 100644 src/test/subscription/t/050_parallel_apply.pl

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index a2f4ae48a82..4649c662c6f 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -14,6 +14,9 @@
  * ParallelApplyWorkerInfo which is required so the leader worker and parallel
  * apply workers can communicate with each other.
  *
+ * Streaming transactions
+ * ======================
+ *
  * The parallel apply workers are assigned (if available) as soon as xact's
  * first stream is received for subscriptions that have set their 'streaming'
  * option as parallel. The leader apply worker will send changes to this new
@@ -152,6 +155,80 @@
  * session-level locks because both locks could be acquired outside the
  * transaction, and the stream lock in the leader needs to persist across
  * transaction boundaries i.e. until the end of the streaming transaction.
+ *
+ * Non-streaming transactions
+ * ======================
+ * The handling is similar to streaming transactions, but including few
+ * differences:
+ *
+ * Transaction dependency
+ * ----------------------
+ * Before dispatching changes to a parallel worker, the leader verifies if the
+ * current modification affects the same row (identitied by replica identity
+ * key) as another ongoing transaction (see handle_dependency_on_change for
+ * details). If so, the leader sends a list of dependent transaction IDs to the
+ * parallel worker, indicating that the parallel apply worker must wait for
+ * these transactions to commit before proceeding.
+ *
+ * Tracking dependencies is necessary even when commit order is preserved.
+ * Consider two transactions: TX-1 (INSERT row 1) and TX-2 (DELETE row 1). If
+ * both are allowed to apply in parallel, TX-2's DELETE could be applied before
+ * TX-1's INSERT, resulting in a delete_missing conflict. Simiarly, if TX-1
+ * (DELETE row 1) and TX-2 (INSERT row 1) are applied in parallel, TX-2's INSERT
+ * could be applied before TX-1's DELETE, resulting in a insert_conflict.
+ *
+ * Commit order
+ * ------------
+ * We preserve publisher commit order for all transactions for two reasons:
+ *
+ * 1) User-visible consistency
+ *
+ * Out-of-order commits can expose states on the subscriber that were never
+ * visible on the publisher.
+ *
+ * For example, suppose a user updates table A and then updates table B on the
+ * publisher. If the subscriber commits those transactions out of order, a
+ * query that sees the latest row in B might still see stale data in A.
+ * Although eventual consistency would still be reached, that behavior may be
+ * unacceptable for some users. In the future, we could provide a subscription
+ * option to allow out-of-order commits for users who prefer higher parallelism.
+ *
+ * 2) Replication progress tracking
+ *
+ * We currently track replication progress using the last transaction's commit
+ * LSNs. With out-of-order commits, this becomes ambiguous after failures.
+ *
+ * For example, if TX-2 is applied before TX-1 and replication stops due to an
+ * error, we cannot reliably determine whether TX-1 was applied before restart.
+ * As a result, transactions that were already committed on the subscriber may
+ * be replayed.
+ *
+ * Worker interaction
+ * ------------
+ * After sending the COMMIT message for a transaction, the leader apply worker
+ * does not wait for the parallel apply worker to finish applying that
+ * transaction. Instead, it sends a PA_MSG_XACT_DEPENDENCY message to
+ * the parallel apply worker, instructing it to wait for the last transaction to
+ * commit. This allows the leader to remain busy receiving and dispatching
+ * changes to more parallel apply workers, enabling greater parallelism in
+ * transaction application.
+ *
+ * To maximize parallelism, we do not stop workers in the pool. This is
+ * important because non-streaming transactions can occur frequently.
+ *
+ * Locking considerations
+ * ----------------------
+ * When handling a PA_MSG_XACT_DEPENDENCY message, the worker attempts
+ * to acquire the transaction lock of the depended transaction and releases it
+ * immediately after acquisition (see pa_wait_for_depended_transaction). This
+ * allows deadlock detection when one worker (either leader or parallel apply
+ * worker) is waiting for a dependency on a transaction being applied by another
+ * worker, while that other worker is also blocked by a lock held by the first
+ * worker.
+ *
+ * The lock graph for the above example will look as follows: Worker_1 (waiting
+ * for depended transaction to finish) -> Worker_2 (waiting to acquire a
+ * relation lock) -> Worker_1
  *-------------------------------------------------------------------------
  */
 
@@ -469,6 +546,7 @@ pa_setup_dsm(ParallelApplyWorkerInfo *winfo)
 	shared = shm_toc_allocate(toc, sizeof(ParallelApplyWorkerShared));
 	SpinLockInit(&shared->mutex);
 
+	shared->xid = InvalidTransactionId;
 	shared->xact_state = PARALLEL_TRANS_UNKNOWN;
 	pg_atomic_init_u32(&(shared->pending_stream_count), 0);
 	shared->last_commit_end = InvalidXLogRecPtr;
@@ -525,6 +603,15 @@ pa_launch_parallel_worker(void)
 			return winfo;
 	}
 
+	/*
+	 * Quick check to avoid allocating shared memory (pa_setup_dsm) and the
+	 * LWLock overhead and worker array scanning in logicalrep_worker_launch
+	 * when the worker pool is already full.
+	 */
+	if (list_length(ParallelApplyWorkerPool) ==
+		max_parallel_apply_workers_per_subscription)
+		return NULL;
+
 	/*
 	 * Start a new parallel apply worker.
 	 *
@@ -552,16 +639,15 @@ pa_launch_parallel_worker(void)
 										dsm_segment_handle(winfo->dsm_seg),
 										false);
 
-	if (launched)
-	{
-		ParallelApplyWorkerPool = lappend(ParallelApplyWorkerPool, winfo);
-	}
-	else
+	if (!launched)
 	{
+		MemoryContextSwitchTo(oldcontext);
 		pa_free_worker_info(winfo);
-		winfo = NULL;
+		return NULL;
 	}
 
+	ParallelApplyWorkerPool = lappend(ParallelApplyWorkerPool, winfo);
+
 	MemoryContextSwitchTo(oldcontext);
 
 	initStringInfo(&out);
@@ -1377,7 +1463,6 @@ pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
 	shm_mq_result result;
 	TimestampTz startTime = 0;
 
-	Assert(!IsTransactionState());
 	Assert(!winfo->serialize_changes);
 
 	/*
@@ -2123,3 +2208,41 @@ pa_transaction_committed(TransactionId xid)
 	return !entry ||
 		   pa_get_xact_state(entry->winfo->shared) == PARALLEL_TRANS_FINISHED;
 }
+
+/*
+ * Mark the transaction state as finished and remove the shared hash entry.
+ */
+void
+pa_commit_transaction(void)
+{
+	TransactionId xid = MyParallelShared->xid;
+
+	SpinLockAcquire(&MyParallelShared->mutex);
+	MyParallelShared->xact_state = PARALLEL_TRANS_FINISHED;
+	SpinLockRelease(&MyParallelShared->mutex);
+
+	dshash_delete_key(parallelized_txns, &xid);
+	elog(DEBUG1, "xid %u committed", xid);
+}
+
+/*
+ * Register a transaction to the shared hash table.
+ *
+ * This function is called by the leader during the commit phase of non-streamed
+ * transactions. The parallel apply worker that applies the transaction will
+ * remove it from the hash table upon completion.
+ */
+void
+pa_add_parallelized_transaction(TransactionId xid)
+{
+	bool		found;
+	ParallelizedTxnEntry *txn_entry;
+
+	Assert(parallelized_txns);
+	Assert(TransactionIdIsValid(xid));
+	Assert(am_leader_apply_worker());
+
+	txn_entry = dshash_find_or_insert(parallelized_txns, &xid, &found);
+
+	dshash_release_lock(parallelized_txns, txn_entry);
+}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 5a93f5428a8..230f8be6d4e 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -290,6 +290,7 @@
 #include "tcop/tcopprot.h"
 #include "utils/acl.h"
 #include "utils/guc.h"
+#include "utils/injection_point.h"
 #include "utils/inval.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
@@ -517,6 +518,7 @@ static List *on_commit_wakeup_workers_subids = NIL;
 
 bool		in_remote_transaction = false;
 static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr;
+static TransactionId remote_xid = InvalidTransactionId;
 
 /* fields valid only when processing streamed transaction */
 static bool in_streamed_transaction = false;
@@ -949,8 +951,6 @@ check_and_record_ri_dependency(Oid relid, LogicalRepTupleData *original_data,
 
 	for (int i_original = 0, i_ri = 0; i_original < original_data->ncols; i_original++)
 	{
-		StringInfo	original_colvalue = &original_data->colvalues[i_original];
-
 		if (!bms_is_member(i_original, relentry->remoterel.attkeys))
 			continue;
 
@@ -974,6 +974,8 @@ check_and_record_ri_dependency(Oid relid, LogicalRepTupleData *original_data,
 		 */
 		if (original_data->colstatus[i_original] != LOGICALREP_COLUMN_NULL)
 		{
+			StringInfo	original_colvalue = &original_data->colvalues[i_original];
+
 			initStringInfoExt(&ridata->colvalues[i_ri],
 							  original_colvalue->len + 1);
 			appendBinaryStringInfo(&ridata->colvalues[i_ri],
@@ -1453,10 +1455,7 @@ handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)
 
 	/* not in streaming mode */
 	if (apply_action == TRANS_LEADER_APPLY)
-	{
-		handle_dependency_on_change(action, s, InvalidTransactionId, winfo);
 		return false;
-	}
 
 	Assert(TransactionIdIsValid(stream_xid));
 
@@ -1531,6 +1530,73 @@ handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)
 	}
 }
 
+/*
+ * Handle non-streaming transactions when parallel apply is in use.
+ *
+ * This function runs only in the leader apply worker while processing a remote
+ * transaction. It checks whether the current change has dependencies on
+ * preceding parallelized transactions and decides whether to send the change to
+ * a parallel apply worker.
+ *
+ * Returns true if the change has been dispatched to a parallel worker,
+ * indicating the leader does not need to apply it directly. Returns false
+ * otherwise.
+ *
+ * Exception: RELATION and TYPE messages are also sent to the parallel apply
+ * worker, but false is still returned so that the leader updates its own
+ * relation and type caches as well (see apply_handle_relation()).
+ */
+static bool
+handle_parallelized_transaction(LogicalRepMsgType action, StringInfo s)
+{
+	ParallelApplyWorkerInfo *winfo;
+	TransApplyAction apply_action;
+
+	/*
+	 * Dependency checking for non-streaming transactions is only required in
+	 * the leader apply worker during a remote transaction.
+	 */
+	if (!in_remote_transaction || !am_leader_apply_worker())
+		return false;
+
+	apply_action = get_transaction_apply_action(remote_xid, &winfo);
+
+	/* not assigned to parallel apply worker, apply in leader */
+	if (apply_action == TRANS_LEADER_APPLY)
+	{
+		handle_dependency_on_change(action, s, InvalidTransactionId, winfo);
+		return false;
+	}
+
+	Assert(TransactionIdIsValid(remote_xid));
+
+	handle_dependency_on_change(action, s, remote_xid, winfo);
+
+	switch (apply_action)
+	{
+		case TRANS_LEADER_SEND_TO_PARALLEL:
+			Assert(winfo);
+
+			/* Always update relation and type cache in leader apply worker */
+			if (pa_send_data(winfo, s->len, s->data))
+				return (action != LOGICAL_REP_MSG_RELATION &&
+						action != LOGICAL_REP_MSG_TYPE);
+
+			/*
+			 * TODO: Support switching to PARTIAL_SERIALIZE mode when the send
+			 * buffer becomes full.
+			 */
+			ereport(ERROR,
+					errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					errmsg("could not send data to the logical replication parallel apply worker"));
+			return false;	/* silence compiler warning */
+
+		default:
+			elog(ERROR, "unexpected apply action: %d", (int) apply_action);
+			return false;		/* silence compiler warning */
+	}
+}
+
 /*
  * Executor state preparation for evaluation of constraint expressions,
  * indexes and triggers for the specified relation.
@@ -1892,17 +1958,61 @@ static void
 apply_handle_begin(StringInfo s)
 {
 	LogicalRepBeginData begin_data;
+	ParallelApplyWorkerInfo *winfo;
+	TransApplyAction apply_action;
 
 	/* There must not be an active streaming transaction. */
 	Assert(!TransactionIdIsValid(stream_xid));
 
 	logicalrep_read_begin(s, &begin_data);
-	set_apply_error_context_xact(begin_data.xid, begin_data.final_lsn);
+
+	remote_xid = begin_data.xid;
+
+	set_apply_error_context_xact(remote_xid, begin_data.final_lsn);
 
 	remote_final_lsn = begin_data.final_lsn;
 
 	maybe_start_skipping_changes(begin_data.final_lsn);
 
+	pa_allocate_worker(remote_xid);
+
+	apply_action = get_transaction_apply_action(remote_xid, &winfo);
+
+	elog(DEBUG1, "new remote_xid %u", remote_xid);
+	switch (apply_action)
+	{
+		case TRANS_LEADER_APPLY:
+			break;
+
+		case TRANS_LEADER_SEND_TO_PARALLEL:
+			Assert(winfo);
+
+			if (pa_send_data(winfo, s->len, s->data))
+			{
+				pa_set_stream_apply_worker(winfo);
+				break;
+			}
+
+			/*
+			 * TODO: Support switching to PARTIAL_SERIALIZE mode when the send
+			 * buffer becomes full.
+			 */
+			ereport(ERROR,
+					errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					errmsg("could not send data to the logical replication parallel apply worker"));
+			break;
+
+		case TRANS_PARALLEL_APPLY:
+			/* Hold the lock until the end of the transaction. */
+			pa_lock_transaction(MyParallelShared->xid, AccessExclusiveLock);
+			pa_set_xact_state(MyParallelShared, PARALLEL_TRANS_STARTED);
+			break;
+
+		default:
+			elog(ERROR, "unexpected apply action: %d", (int) apply_action);
+			break;
+	}
+
 	in_remote_transaction = true;
 
 	pgstat_report_activity(STATE_RUNNING, NULL);
@@ -1917,6 +2027,8 @@ static void
 apply_handle_commit(StringInfo s)
 {
 	LogicalRepCommitData commit_data;
+	ParallelApplyWorkerInfo *winfo;
+	TransApplyAction apply_action;
 
 	logicalrep_read_commit(s, &commit_data);
 
@@ -1927,7 +2039,86 @@ apply_handle_commit(StringInfo s)
 								 LSN_FORMAT_ARGS(commit_data.commit_lsn),
 								 LSN_FORMAT_ARGS(remote_final_lsn))));
 
-	apply_handle_commit_internal(&commit_data);
+	apply_action = get_transaction_apply_action(remote_xid, &winfo);
+
+	switch (apply_action)
+	{
+		case TRANS_LEADER_APPLY:
+
+			/*
+			 * Apart from parallelized transactions, we do not have to
+			 * register this transaction to parallelized_txns. The commit
+			 * ordering is always preserved.
+			 */
+
+			/* Wait until the last transaction finishes */
+			maintain_commit_order_dependency(NULL);
+
+			apply_handle_commit_internal(&commit_data);
+
+			break;
+
+		case TRANS_LEADER_SEND_TO_PARALLEL:
+			Assert(winfo);
+
+			/*
+			 * Mark this transaction as parallelized. This ensures that
+			 * upcoming transactions wait until this transaction is committed.
+			 */
+			pa_add_parallelized_transaction(remote_xid);
+
+			/*
+			 * Build a dependency between this transaction and the lastly
+			 * committed transaction to preserve the commit order.
+			 */
+			maintain_commit_order_dependency(winfo);
+
+			if (pa_send_data(winfo, s->len, s->data))
+			{
+				store_flush_position(commit_data.end_lsn, InvalidXLogRecPtr,
+									 remote_xid);
+				pa_set_stream_apply_worker(NULL);
+				break;
+			}
+
+			/*
+			 * TODO: Support switching to PARTIAL_SERIALIZE mode when the send
+			 * buffer becomes full.
+			 */
+			ereport(ERROR,
+					errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					errmsg("could not send data to the logical replication parallel apply worker"));
+			break;
+
+		case TRANS_PARALLEL_APPLY:
+
+			/*
+			 * If the parallel apply worker is applying spooled messages then
+			 * close the file before committing.
+			 */
+			if (stream_fd)
+				stream_close_file();
+
+			INJECTION_POINT("parallel-worker-before-commit", NULL);
+
+			apply_handle_commit_internal(&commit_data);
+
+			MyParallelShared->last_commit_end = XactLastCommitEnd;
+
+			pa_commit_transaction();
+
+			pa_unlock_transaction(remote_xid, AccessExclusiveLock);
+			break;
+
+		default:
+			elog(ERROR, "unexpected apply action: %d", (int) apply_action);
+			break;
+	}
+
+	elog(DEBUG1, "finished processing remote_xid %u", remote_xid);
+
+	remote_xid = InvalidTransactionId;
+	in_remote_transaction = false;
 
 	/*
 	 * Process any tables that are being synchronized in parallel, as well as
@@ -1935,8 +2126,6 @@ apply_handle_commit(StringInfo s)
 	 */
 	ProcessSyncingRelations(commit_data.end_lsn);
 
-	maintain_commit_order_dependency(NULL);
-
 	pgstat_report_activity(STATE_IDLE, NULL);
 	reset_apply_error_context_info();
 }
@@ -2463,6 +2652,14 @@ apply_handle_stream_start(StringInfo s)
 		case TRANS_LEADER_SEND_TO_PARALLEL:
 			Assert(winfo);
 
+			/*
+			 * TODO: Support dependency tracking for streamed transactions so
+			 * they can be applied in parallel with preceding non-streamed
+			 * transactions.
+			 */
+			if (first_segment)
+				maintain_commit_order_dependency(winfo);
+
 			/*
 			 * Once we start serializing the changes, the parallel apply
 			 * worker will wait for the leader to release the stream lock
@@ -3097,6 +3294,12 @@ apply_handle_stream_commit(StringInfo s)
 	switch (apply_action)
 	{
 		case TRANS_LEADER_APPLY:
+			/*
+			 * TODO: Support dependency tracking for streamed transactions so
+			 * they can be applied in parallel with preceding non-streamed
+			 * transactions.
+			 */
+			maintain_commit_order_dependency(winfo);
 
 			/*
 			 * The transaction has been serialized to file, so replay all the
@@ -3252,7 +3455,8 @@ apply_handle_relation(StringInfo s)
 {
 	LogicalRepRelation *rel;
 
-	if (handle_streamed_transaction(LOGICAL_REP_MSG_RELATION, s))
+	if (handle_parallelized_transaction(LOGICAL_REP_MSG_RELATION, s) ||
+		handle_streamed_transaction(LOGICAL_REP_MSG_RELATION, s))
 		return;
 
 	rel = logicalrep_read_rel(s);
@@ -3277,7 +3481,8 @@ apply_handle_type(StringInfo s)
 {
 	LogicalRepTyp typ;
 
-	if (handle_streamed_transaction(LOGICAL_REP_MSG_TYPE, s))
+	if (handle_parallelized_transaction(LOGICAL_REP_MSG_TYPE, s) ||
+		handle_streamed_transaction(LOGICAL_REP_MSG_TYPE, s))
 		return;
 
 	logicalrep_read_typ(s, &typ);
@@ -3337,6 +3542,7 @@ apply_handle_insert(StringInfo s)
 	 * streamed transactions.
 	 */
 	if (is_skipping_changes() ||
+		handle_parallelized_transaction(LOGICAL_REP_MSG_INSERT, s) ||
 		handle_streamed_transaction(LOGICAL_REP_MSG_INSERT, s))
 		return;
 
@@ -3497,6 +3703,7 @@ apply_handle_update(StringInfo s)
 	 * streamed transactions.
 	 */
 	if (is_skipping_changes() ||
+		handle_parallelized_transaction(LOGICAL_REP_MSG_UPDATE, s) ||
 		handle_streamed_transaction(LOGICAL_REP_MSG_UPDATE, s))
 		return;
 
@@ -3721,6 +3928,7 @@ apply_handle_delete(StringInfo s)
 	 * streamed transactions.
 	 */
 	if (is_skipping_changes() ||
+		handle_parallelized_transaction(LOGICAL_REP_MSG_DELETE, s) ||
 		handle_streamed_transaction(LOGICAL_REP_MSG_DELETE, s))
 		return;
 
@@ -4357,6 +4565,7 @@ apply_handle_truncate(StringInfo s)
 	 * streamed transactions.
 	 */
 	if (is_skipping_changes() ||
+		handle_parallelized_transaction(LOGICAL_REP_MSG_TRUNCATE, s) ||
 		handle_streamed_transaction(LOGICAL_REP_MSG_TRUNCATE, s))
 		return;
 
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index d895fc0284d..aab6c04cd14 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -386,6 +386,8 @@ extern void pa_bind_flush_position(TransactionId xid, XLogRecPtr *local_end);
 extern bool pa_get_last_commit_end(TransactionId xid, XLogRecPtr *local_end);
 extern void pa_wait_for_depended_transaction(TransactionId xid);
 extern bool pa_transaction_committed(TransactionId xid);
+extern void pa_commit_transaction(void);
+extern void pa_add_parallelized_transaction(TransactionId xid);
 
 #define isParallelApplyWorker(worker) ((worker)->in_use && \
 									   (worker)->type == WORKERTYPE_PARALLEL_APPLY)
diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index e71e95c6297..48ae698e786 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -48,6 +48,7 @@ tests += {
       't/036_sequences.pl',
       't/037_except.pl',
       't/038_walsnd_shutdown_timeout.pl',
+      't/050_parallel_apply.pl',
       't/100_bugs.pl',
     ],
   },
diff --git a/src/test/subscription/t/001_rep_changes.pl b/src/test/subscription/t/001_rep_changes.pl
index 7d41715ed81..c863b430bec 100644
--- a/src/test/subscription/t/001_rep_changes.pl
+++ b/src/test/subscription/t/001_rep_changes.pl
@@ -16,6 +16,8 @@ $node_publisher->start;
 # Create subscriber node
 my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
 $node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf',
+	"max_logical_replication_workers = 10");
 $node_subscriber->start;
 
 # Create some preexisting content on publisher
diff --git a/src/test/subscription/t/010_truncate.pl b/src/test/subscription/t/010_truncate.pl
index 945505d0239..e15a6bb2a03 100644
--- a/src/test/subscription/t/010_truncate.pl
+++ b/src/test/subscription/t/010_truncate.pl
@@ -17,7 +17,7 @@ $node_publisher->start;
 my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
 $node_subscriber->init;
 $node_subscriber->append_conf('postgresql.conf',
-	qq(max_logical_replication_workers = 6));
+	qq(max_logical_replication_workers = 7));
 $node_subscriber->start;
 
 my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
diff --git a/src/test/subscription/t/015_stream.pl b/src/test/subscription/t/015_stream.pl
index ac96bc3f009..4f1560b0725 100644
--- a/src/test/subscription/t/015_stream.pl
+++ b/src/test/subscription/t/015_stream.pl
@@ -230,6 +230,14 @@ $node_subscriber->wait_for_log(
 	qr/DEBUG: ( [A-Z0-9]+:)? applied [0-9]+ changes in the streaming chunk/,
 	$offset);
 
+# Non-streaming transactions are now also assigned to parallel apply
+# workers, so the first single-row transaction is handled by a parallel
+# apply worker (where it blocks on the unique index against the
+# uncommitted streamed changes) rather than by the leader.  Issue another
+# one so that it is applied by the leader itself (the worker pool is busy)
+# and blocks on the same conflict, recreating the leader-vs-parallel-worker
+# deadlock this test intends to exercise.
+$node_publisher->safe_psql('postgres', "INSERT INTO test_tab_2 values(1)");
 $node_publisher->safe_psql('postgres', "INSERT INTO test_tab_2 values(1)");
 
 $h->query_safe('COMMIT');
@@ -247,7 +255,7 @@ $node_publisher->wait_for_catchup($appname);
 
 $result =
   $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab_2");
-is($result, qq(5001), 'data replicated to subscriber after dropping index');
+is($result, qq(5002), 'data replicated to subscriber after dropping index');
 
 # Clean up test data from the environment.
 $node_publisher->safe_psql('postgres', "TRUNCATE TABLE test_tab_2");
diff --git a/src/test/subscription/t/026_stats.pl b/src/test/subscription/t/026_stats.pl
index 5d457060a02..911005bde20 100644
--- a/src/test/subscription/t/026_stats.pl
+++ b/src/test/subscription/t/026_stats.pl
@@ -16,6 +16,7 @@ $node_publisher->start;
 # Create subscriber node.
 my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
 $node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf', "max_logical_replication_workers = 10");
 $node_subscriber->start;
 
 
diff --git a/src/test/subscription/t/027_nosuperuser.pl b/src/test/subscription/t/027_nosuperuser.pl
index 322f5b4cc6a..fdfc44ac729 100644
--- a/src/test/subscription/t/027_nosuperuser.pl
+++ b/src/test/subscription/t/027_nosuperuser.pl
@@ -86,6 +86,7 @@ $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
 $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
 $node_publisher->init(allows_streaming => 'logical');
 $node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf', "max_logical_replication_workers = 10");
 $node_publisher->start;
 $node_subscriber->start;
 $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
diff --git a/src/test/subscription/t/050_parallel_apply.pl b/src/test/subscription/t/050_parallel_apply.pl
new file mode 100644
index 00000000000..3f8b523a769
--- /dev/null
+++ b/src/test/subscription/t/050_parallel_apply.pl
@@ -0,0 +1,424 @@
+
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+# This test verifies that a non-streamed transaction can launch a parallel apply
+# worker, and that dependency tracking and commit order preservation work
+# correctly during parallel apply.
+
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+if ($ENV{enable_injection_points} ne 'yes')
+{
+	plan skip_all => 'Injection points not supported by this build';
+}
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->start;
+
+# Create tables and insert initial data
+$node_publisher->safe_psql(
+    'postgres', qq(
+    CREATE TABLE regress_tab (id int PRIMARY KEY, value text);
+
+    CREATE TABLE tab_bin (k bytea PRIMARY KEY, v int);
+
+    CREATE TABLE tab_ri_full (id int, value text);
+    ALTER TABLE tab_ri_full REPLICA IDENTITY FULL;
+    INSERT INTO tab_ri_full VALUES (1, 'test');
+
+    CREATE TABLE tab_toast (a text NOT NULL, b text NOT NULL);
+    ALTER TABLE tab_toast ALTER COLUMN a SET STORAGE EXTERNAL;
+    CREATE UNIQUE INDEX tab_toast_ri_index on tab_toast (a, b);
+    ALTER TABLE tab_toast REPLICA IDENTITY USING INDEX tab_toast_ri_index;
+    INSERT INTO tab_toast(a, b) VALUES(repeat('1234567890', 200), '1234567890');
+));
+$node_publisher->safe_psql('postgres',
+    "INSERT INTO regress_tab VALUES (generate_series(1, 10), 'test');");
+
+# Create a publication
+$node_publisher->safe_psql('postgres',
+    "CREATE PUBLICATION regress_pub FOR ALL TABLES;");
+
+# Initialize subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf', "log_min_messages = debug1");
+$node_subscriber->append_conf('postgresql.conf',
+	"max_logical_replication_workers = 10");
+$node_subscriber->start;
+
+# Check if the extension injection_points is available, as it may be
+# possible that this script is run with installcheck, where the module
+# would not be installed by default.
+if (!$node_subscriber->check_extension('injection_points'))
+{
+	plan skip_all => 'Extension injection_points not installed';
+}
+
+$node_subscriber->safe_psql('postgres', 'CREATE EXTENSION injection_points;');
+
+# Create a subscription
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+
+$node_subscriber->safe_psql(
+    'postgres', qq(
+    CREATE TABLE regress_tab (id int PRIMARY KEY, value text);
+
+    CREATE TABLE tab_bin (k bytea PRIMARY KEY, v int);
+
+    CREATE TABLE tab_ri_full (id int, value text);
+    ALTER TABLE tab_ri_full REPLICA IDENTITY FULL;
+
+    CREATE TABLE tab_toast (a text NOT NULL, b text NOT NULL);
+    ALTER TABLE tab_toast ALTER COLUMN a SET STORAGE EXTERNAL;
+    CREATE UNIQUE INDEX tab_toast_ri_index on tab_toast (a, b);
+    ALTER TABLE tab_toast REPLICA IDENTITY USING INDEX tab_toast_ri_index;
+));
+$node_subscriber->safe_psql('postgres',
+    "CREATE SUBSCRIPTION regress_sub CONNECTION '$publisher_connstr' PUBLICATION regress_pub;");
+
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'regress_sub');
+
+##################################################
+# Test that a non-streamed transaction can be applied in a parallel apply worker
+##################################################
+
+# Start a transaction to ensure the leader worker has seen the latest table sync
+# READY state, ensuring parallel apply workers can be launched for later
+# non-streamed transactions.
+$node_publisher->safe_psql('postgres',
+    "INSERT INTO regress_tab VALUES (generate_series(11, 20), 'test');");
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Insert tuples again
+$node_publisher->safe_psql('postgres',
+    "INSERT INTO regress_tab VALUES (generate_series(21, 30), 'test');");
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Verify the parallel apply worker is launched
+my $result = $node_subscriber->safe_psql('postgres',
+    "SELECT count(1) FROM pg_stat_activity WHERE backend_type = 'logical replication parallel worker'");
+is($result, '1', "parallel apply worker is launched by a non-streamed transaction");
+
+##################################################
+# Test that the basic replica identity dependency tracking and commit order
+# preservation work correctly during parallel apply.
+##################################################
+
+# Attach an injection_point. Parallel workers would wait before the commit
+$node_subscriber->safe_psql('postgres',
+	"SELECT injection_points_attach('parallel-worker-before-commit','wait');"
+);
+
+# Insert tuples on publisher
+$node_publisher->safe_psql('postgres',
+    "INSERT INTO regress_tab VALUES (generate_series(31, 40), 'test');");
+
+# Wait until the parallel worker enters the injection point.
+$node_subscriber->wait_for_event('logical replication parallel worker',
+	'parallel-worker-before-commit');
+
+my $offset = -s $node_subscriber->logfile;
+
+# Insert tuples on publisher again. This transaction is independent from the
+# previous one, but the parallel worker would wait till it finishes
+$node_publisher->safe_psql('postgres',
+    "INSERT INTO regress_tab VALUES (generate_series(41, 50), 'test');");
+
+# Verify the parallel worker waits for the transaction
+my $str = $node_subscriber->wait_for_log(qr/wait for depended xid ([1-9][0-9]+)/, $offset);
+my $xid = $str =~ /wait for depended xid ([1-9][0-9]+)/;
+
+ok(1, "commit order dependency detected for parallel apply");
+
+$offset = -s $node_subscriber->logfile;
+
+# Update tuples which have not been applied yet on subscriber because the
+# parallel worker stops at the injection point. Newly assigned worker also
+# waits for the same transactions as above.
+$node_publisher->safe_psql('postgres',
+    "UPDATE regress_tab SET value = 'updated' WHERE id BETWEEN 31 AND 35;");
+
+# Verify the dependency is detected for the update
+$node_subscriber->wait_for_log(qr/found conflicting replica identity change on table [1-9][0-9]+ from $xid/, $offset);
+
+# Verify the parallel worker waits for the same transaction
+$node_subscriber->wait_for_log(qr/wait for depended xid $xid/, $offset);
+
+ok(1, "replica identity dependency detected for parallel apply");
+
+# Wakeup the parallel worker. We detach first no to stop other parallel workers
+$node_subscriber->safe_psql('postgres', qq[
+    SELECT injection_points_detach('parallel-worker-before-commit');
+    SELECT injection_points_wakeup('parallel-worker-before-commit');
+]);
+
+# Verify the parallel worker wakes up
+$node_subscriber->wait_for_log(qr/finish waiting for depended xid $xid/, $offset);
+
+$node_publisher->wait_for_catchup('regress_sub');
+
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM regress_tab");
+is ($result, 50, 'inserts are replicated to subscriber');
+
+$result =
+  $node_subscriber->safe_psql('postgres',
+    "SELECT count(1) FROM regress_tab WHERE value = 'updated'");
+is ($result, 5, 'updates are also replicated to subscriber');
+
+##################################################
+# Test that dependency hash key comparison handles values containing zero
+# bytes correctly when the subscription uses the binary option.
+##################################################
+
+$node_subscriber->safe_psql('postgres',
+    "ALTER SUBSCRIPTION regress_sub DISABLE;");
+$node_subscriber->poll_query_until('postgres',
+       "SELECT count(*) = 0 FROM pg_stat_activity WHERE backend_type = 'logical replication apply worker'"
+);
+$node_subscriber->safe_psql(
+    'postgres', "
+    ALTER SUBSCRIPTION regress_sub SET (binary = true);
+    ALTER SUBSCRIPTION regress_sub ENABLE;");
+
+# Insert the test row and send a couple of warm-up transactions so that
+# subsequent non-streamed transactions are assigned to parallel apply
+# workers (see AllTablesyncsReady).
+$node_publisher->safe_psql('postgres',
+    "INSERT INTO tab_bin VALUES ('\\x6162006364', 0);");
+$node_publisher->wait_for_catchup('regress_sub');
+$node_publisher->safe_psql('postgres',
+    "UPDATE tab_bin SET v = 0 WHERE k = '\\x6162006364';");
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Attach an injection_point. Parallel workers would wait before the commit
+$node_subscriber->safe_psql('postgres',
+       "SELECT injection_points_attach('parallel-worker-before-commit','wait');"
+);
+
+# TX-1: update the row whose key contains a zero byte ('ab\0cd'). The
+# parallel worker pauses before commit, keeping the key in the dependency
+# hash table.
+$node_publisher->safe_psql('postgres',
+    "UPDATE tab_bin SET v = 1 WHERE k = '\\x6162006364';");
+
+# Wait until the parallel worker enters the injection point.
+$node_subscriber->wait_for_event('logical replication parallel worker',
+       'parallel-worker-before-commit');
+
+$offset = -s $node_subscriber->logfile;
+
+# TX-2: update the same row (same key bytes). This must be detected as a
+# conflict with TX-1.
+$node_publisher->safe_psql('postgres',
+    "UPDATE tab_bin SET v = 2 WHERE k = '\\x6162006364';");
+
+# Verify the dependency is detected for the update
+$str = $node_subscriber->wait_for_log(qr/found conflicting replica identity change on table [1-9][0-9]+ from ([1-9][0-9]+)/, $offset);
+$xid = $str =~ /found conflicting replica identity change on table [1-9][0-9]+ from ([1-9][0-9]+)/;
+
+# Verify the parallel worker waits for the same transaction
+$node_subscriber->wait_for_log(qr/wait for depended xid $xid/, $offset);
+
+ok(1, "replica identity dependency detected for binary key with zero bytes");
+
+$offset = -s $node_subscriber->logfile;
+
+# TX-3: insert a row whose key differs from TX-1's key only after the zero
+# byte ('ab\0ef' vs 'ab\0cd'). This must NOT be treated as conflicting with
+# TX-1: the keys are distinct when compared as raw bytes.
+$node_publisher->safe_psql('postgres',
+    "INSERT INTO tab_bin VALUES ('\\x6162006566', 3);");
+
+# Wakeup the parallel workers and let everything apply
+$node_subscriber->safe_psql('postgres', qq[
+    SELECT injection_points_detach('parallel-worker-before-commit');
+    SELECT injection_points_wakeup('parallel-worker-before-commit');
+]);
+
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Verify that no replica identity conflict was reported for TX-3
+my $newlog = substr(slurp_file($node_subscriber->logfile), $offset);
+unlike($newlog, qr/found conflicting replica identity change/,
+       "no false dependency for keys differing after a zero byte");
+
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM tab_bin");
+is ($result, 2, 'changes are replicated to subscriber');
+
+$result =
+  $node_subscriber->safe_psql('postgres',
+    "SELECT v FROM tab_bin WHERE k = '\\x6162006364'");
+is ($result, 2, 'updates applied in commit order on subscriber');
+
+# Disable binary mode for the subscription
+$node_subscriber->safe_psql('postgres',
+    "ALTER SUBSCRIPTION regress_sub DISABLE;");
+$node_subscriber->poll_query_until('postgres',
+       "SELECT count(*) = 0 FROM pg_stat_activity WHERE backend_type = 'logical replication apply worker'"
+);
+$node_subscriber->safe_psql(
+    'postgres', "
+    ALTER SUBSCRIPTION regress_sub SET (binary = false);
+    ALTER SUBSCRIPTION regress_sub ENABLE;");
+
+##################################################
+# Test that the dependency tracking works correctly for unchanged toasted RI
+# columns.
+##################################################
+
+# Attach an injection_point. Parallel workers would wait before the commit
+$node_subscriber->safe_psql('postgres',
+	"SELECT injection_points_attach('parallel-worker-before-commit','wait');"
+);
+
+# Update one replica identity column but keep toasted column unchanged
+$node_publisher->safe_psql('postgres',
+    "UPDATE tab_toast SET b = '1';");
+
+# Wait until the parallel worker enters the injection point.
+$node_subscriber->wait_for_event('logical replication parallel worker',
+	'parallel-worker-before-commit');
+
+$offset = -s $node_subscriber->logfile;
+
+# Delete the updated row.
+$node_publisher->safe_psql('postgres',
+    "DELETE FROM tab_toast WHERE b = '1';");
+
+# Verify the dependency is detected for the delete
+$str = $node_subscriber->wait_for_log(qr/found conflicting replica identity change on table [1-9][0-9]+ from ([1-9][0-9]+)/, $offset);
+$xid = $str =~ /found conflicting replica identity change on table [1-9][0-9]+ from ([1-9][0-9]+)/;
+
+# Verify the parallel worker waits for the same transaction
+$node_subscriber->wait_for_log(qr/wait for depended xid $xid/, $offset);
+
+ok(1, "replica identity dependency from unchanged toasted column detected for parallel apply");
+
+# Wakeup the parallel worker. We detach first no to stop other parallel workers
+$node_subscriber->safe_psql('postgres', qq[
+    SELECT injection_points_detach('parallel-worker-before-commit');
+    SELECT injection_points_wakeup('parallel-worker-before-commit');
+]);
+
+# Verify the parallel worker wakes up
+$node_subscriber->wait_for_log(qr/finish waiting for depended xid $xid/, $offset);
+
+$node_publisher->wait_for_catchup('regress_sub');
+
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM tab_toast");
+is ($result, 0, 'changes are replicated to subscriber');
+
+##################################################
+# Test that dependency tracking still works for REPLICA IDENTITY FULL when
+# new tuple includes NULL key values.
+##################################################
+
+# Attach an injection_point. Parallel workers would wait before the commit
+$node_subscriber->safe_psql('postgres',
+	"SELECT injection_points_attach('parallel-worker-before-commit','wait');"
+);
+
+# Update one row to a non-NULL value and block commit in parallel worker.
+$node_publisher->safe_psql('postgres',
+        "UPDATE tab_ri_full SET value = NULL WHERE id = 1;");
+
+# Wait until the parallel worker enters the injection point.
+$node_subscriber->wait_for_event('logical replication parallel worker',
+	'parallel-worker-before-commit');
+
+$offset = -s $node_subscriber->logfile;
+
+# This update sets a NULL value under REPLICA IDENTITY FULL. We must still
+# detect dependency on the preceding update of the same row.
+$node_publisher->safe_psql('postgres',
+        "UPDATE tab_ri_full SET value = 'test' WHERE id = 1;");
+
+# Verify the dependency is detected for the update with NULL key value.
+$str = $node_subscriber->wait_for_log(qr/found conflicting replica identity change on table [1-9][0-9]+ from ([1-9][0-9]+)/, $offset);
+$xid = $str =~ /found conflicting replica identity change on table [1-9][0-9]+ from ([1-9][0-9]+)/;
+
+# Verify the parallel worker waits for the same transaction.
+$node_subscriber->wait_for_log(qr/wait for depended xid $xid/, $offset);
+
+ok(1, "replica identity FULL dependency with NULL values detected for parallel apply");
+
+# Wakeup the parallel worker. We detach first no to stop other parallel workers
+$node_subscriber->safe_psql('postgres', qq[
+        SELECT injection_points_detach('parallel-worker-before-commit');
+        SELECT injection_points_wakeup('parallel-worker-before-commit');
+]);
+
+# Verify the parallel worker wakes up.
+$node_subscriber->wait_for_log(qr/finish waiting for depended xid $xid/, $offset);
+
+$node_publisher->wait_for_catchup('regress_sub');
+
+$result =
+    $node_subscriber->safe_psql('postgres',
+        "SELECT count(1) FROM tab_ri_full WHERE id = 1 AND value = 'test'");
+is ($result, 1, 'update is replicated for REPLICA IDENTITY FULL table');
+
+##################################################
+# Test that table level dependency tracking by TRUNCATE work correctly during
+# parallel apply.
+##################################################
+
+# Truncate the data for upcoming tests
+$node_publisher->safe_psql('postgres', "TRUNCATE TABLE regress_tab;");
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Attach an injection_point. Parallel workers would wait before the commit
+$node_subscriber->safe_psql('postgres',
+	"SELECT injection_points_attach('parallel-worker-before-commit','wait');"
+);
+
+# Insert tuples on publisher
+$node_publisher->safe_psql('postgres',
+    "TRUNCATE regress_tab;");
+
+# Wait until the parallel worker enters the injection point.
+$node_subscriber->wait_for_event('logical replication parallel worker',
+	'parallel-worker-before-commit');
+
+$offset = -s $node_subscriber->logfile;
+
+# Insert a tuple that conflicts with the TRUNCATE operation.
+$node_publisher->safe_psql('postgres',
+    "INSERT INTO regress_tab VALUES (1, 'test');");
+
+# Verify the dependency is detected for the insert
+$str = $node_subscriber->wait_for_log(qr/found table-wide change affecting [1-9][0-9]+ from ([1-9][0-9]+)/, $offset);
+$xid = $str =~ /found table-wide change affecting [1-9][0-9]+ from ([1-9][0-9]+)/;
+
+ok(1, "table-wide dependency from TRUNCATE detected for parallel apply");
+
+# Wakeup the parallel worker. We detach first no to stop other parallel workers
+$node_subscriber->safe_psql('postgres', qq[
+    SELECT injection_points_detach('parallel-worker-before-commit');
+    SELECT injection_points_wakeup('parallel-worker-before-commit');
+]);
+
+# Verify the parallel worker waits for the same transaction
+$node_subscriber->wait_for_log(qr/wait for depended xid $xid/, $offset);
+
+# Verify the parallel worker wakes up
+$node_subscriber->wait_for_log(qr/finish waiting for depended xid $xid/, $offset);
+
+$node_publisher->wait_for_catchup('regress_sub');
+
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM regress_tab");
+is ($result, 1, 'inserts are replicated to subscriber');
+
+done_testing();
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 0461e9b51b1..e7849777f9d 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2692,7 +2692,6 @@ ReplicationSlotPersistentData
 ReplicationState
 ReplicationStateCtl
 ReplicationStateOnDisk
-replica_identity_hash
 ResTarget
 ReservoirState
 ReservoirStateData
@@ -4273,6 +4272,7 @@ rendezvousHashEntry
 rep
 replace_rte_variables_callback
 replace_rte_variables_context
+replica_identity_hash
 report_error_fn
 ret_type
 rewind_source
-- 
2.43.0

