From d58b5f87a9211832eb339da2a4ce732f428e2cb7 Mon Sep 17 00:00:00 2001
From: Zhijie Hou <houzj.fnst@fujitsu.com>
Date: Tue, 11 Aug 2026 13:57:47 +0800
Subject: [PATCH v23 02/12] Prepare flush position tracking for parallel apply

This patch adds infrastructure for asynchronously tracking transaction flush
positions for transactions assigned to parallel apply workers. The goal is to
eliminate the need for the leader to wait for them to commit before handling the
next transaction.

After sending the COMMIT message to a parallel apply worker, the leader
records the transaction ID in the FlushPosition list and links it to the
corresponding ParallelApplyTxnHash entry, rather than waiting for the worker
to finish. This allows the leader to continue dispatching transactions to
parallel workers without blocking, achieving maximum parallelism. The leader
later fetches the local commit LSN from the corresponding ParallelApplyTxnHash
entry asynchronously when needed — either when collecting flush positions to
report feedback to the walsender, or when reusing the worker for a new
transaction.

To make the progress tracking design clearer, this patch also adds more comments
for the existing structures related to the tracking system and the hash table
entry.

Author: Zhijie Hou <houzj.fnst@fujitsu.com>
Author: Hayato Kuroda <kuroda.hayato@fujitsu.com>
---
 .../replication/logical/applyparallelworker.c | 139 +++++++++++++++++-
 src/backend/replication/logical/worker.c      |  82 ++++++++---
 src/include/replication/worker_internal.h     |   6 +-
 3 files changed, 201 insertions(+), 26 deletions(-)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index cad437fd06e..e5750c63db0 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -213,17 +213,36 @@
 #define PARALLEL_APPLY_LOCK_XACT	1
 
 /*
- * Hash table entry to map xid to the parallel apply worker state.
+ * Hash table entry of ParallelApplyTxnHash to map xid to the parallel apply
+ * worker state.
  */
 typedef struct ParallelApplyWorkerEntry
 {
-	TransactionId xid;			/* Hash key -- must be first */
-	ParallelApplyWorkerInfo *winfo;
+	TransactionId xid;			/* Remote transaction ID (hash key) ---
+								 * must be first */
+	ParallelApplyWorkerInfo *winfo;	/* The parallel apply worker assigned for
+									 * applying the transaction */
+
+	/*
+	 * The Local end LSN pointer from the node in the lsn_mapping list that is
+	 * bound to this transaction. The leader uses it to update the local end LSN
+	 * in the mapping before reusing the worker for a new transaction.
+	 *
+	 * This is NULL if the leader chose to wait for the transaction to finish,
+	 * in which case the local end LSN can be collected directly from the
+	 * worker's shared memory.
+	 */
+	XLogRecPtr *local_end;
 } ParallelApplyWorkerEntry;
 
 /*
  * A hash table used to cache the state of streaming transactions being applied
  * by the parallel apply workers.
+ *
+ * The leader apply worker adds an entry when assigning a transaction to a
+ * parallel apply worker, and removes it after collecting the transaction's
+ * local end LSN from that worker (see pa_maybe_reuse_worker and
+ * pa_get_last_commit_end).
  */
 static HTAB *ParallelApplyTxnHash = NULL;
 
@@ -318,6 +337,51 @@ pa_can_start(void)
 	return true;
 }
 
+/*
+ * Check if the given parallel apply worker can be reused for a new transaction,
+ * and perform any necessary cleanup if it can.
+ *
+ * Returns true if the worker is reusable, false otherwise.
+ */
+static bool
+pa_maybe_reuse_worker(ParallelApplyWorkerInfo *winfo)
+{
+	ParallelApplyWorkerEntry *entry;
+
+	/* Can reuse if explicitly marked as not in use */
+	if (!winfo->in_use)
+		return true;
+
+	/*
+	 * Cannot reuse the worker while it's still applying its current
+	 * transaction.
+	 */
+	if (pa_get_xact_state(winfo->shared) != PARALLEL_TRANS_FINISHED)
+		return false;
+
+	/*
+	 * No lock is needed because shared memory is not modified after the
+	 * transaction completes.
+	 */
+	entry = hash_search(ParallelApplyTxnHash, &winfo->shared->xid,
+						HASH_FIND, NULL);
+
+	/*
+	 * Update the flush position of the transaction being applied by the worker
+	 * before reusing it for a new transaction and remove the finished
+	 * transaction entry from the hash table.
+	 */
+	if (entry)
+	{
+		*entry->local_end = winfo->shared->last_commit_end;
+
+		if (!hash_search(ParallelApplyTxnHash, &winfo->shared->xid, HASH_REMOVE, NULL))
+			elog(ERROR, "hash table corrupted");
+	}
+
+	return true;
+}
+
 /*
  * Set up a dynamic shared memory segment.
  *
@@ -418,7 +482,7 @@ pa_launch_parallel_worker(void)
 	{
 		winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
 
-		if (!winfo->in_use)
+		if (pa_maybe_reuse_worker(winfo))
 			return winfo;
 	}
 
@@ -535,6 +599,7 @@ pa_allocate_worker(TransactionId xid)
 	winfo->in_use = true;
 	winfo->serialize_changes = false;
 	entry->winfo = winfo;
+	entry->local_end = NULL;
 }
 
 /*
@@ -1732,7 +1797,8 @@ pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
 	pa_wait_for_xact_finish(winfo);
 
 	if (XLogRecPtrIsValid(remote_lsn))
-		store_flush_position(remote_lsn, winfo->shared->last_commit_end);
+		store_flush_position(remote_lsn, winfo->shared->last_commit_end,
+							 InvalidTransactionId);
 
 	pa_free_worker(winfo);
 }
@@ -1789,3 +1855,66 @@ pa_distribute_remote_rel_to_workers(LogicalRepRelation *rel)
 					 errmsg("could not send remote relation information to the logical replication parallel apply worker")));
 	}
 }
+
+/*
+ * Bind a flush position node to a transaction being applied by a parallel apply
+ * worker.
+ */
+void
+pa_bind_flush_position(TransactionId xid, XLogRecPtr *local_end)
+{
+	ParallelApplyWorkerEntry *entry;
+
+	entry = hash_search(ParallelApplyTxnHash, &xid, HASH_FIND, NULL);
+	Assert(entry);
+
+	entry->local_end = local_end;
+}
+
+/*
+ * Get the local end LSN for a transaction applied by a parallel worker and
+ * store it *local_end.
+ *
+ * Returns true if the transaction has finished (the LSN may be
+ * InvalidXLogRecPtr if the transaction wrote no changes), false if it is still
+ * in progress.
+ *
+ * The transaction entry is removed from ParallelApplyTxnHash after the LSN is
+ * retrieved. Subsequent calls for the same transaction return true and leave
+ * *local_end unchanged.
+ */
+bool
+pa_get_last_commit_end(TransactionId xid, XLogRecPtr *local_end)
+{
+	ParallelApplyWorkerEntry *entry;
+	ParallelApplyWorkerInfo *winfo;
+
+	Assert(TransactionIdIsValid(xid));
+	Assert(ParallelApplyTxnHash);
+
+	entry = hash_search(ParallelApplyTxnHash, &xid, HASH_FIND, NULL);
+
+	/* Already collected, entry no longer exists */
+	if (!entry)
+		return true;
+
+	winfo = entry->winfo;
+
+	/*
+	 * Return InvalidXLogRecPtr if the transaction is still in progress in the
+	 * parallel apply worker.
+	 */
+	if (pa_get_xact_state(winfo->shared) != PARALLEL_TRANS_FINISHED)
+		return false;
+
+	/*
+	 * No lock is needed because shared memory is not modified after the
+	 * transaction completes.
+	 */
+	*local_end = winfo->shared->last_commit_end;
+
+	if (!hash_search(ParallelApplyTxnHash, &xid, HASH_REMOVE, NULL))
+		elog(ERROR, "hash table corrupted");
+
+	return true;
+}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 5b3cd6cbeab..45bdd881a69 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -303,11 +303,37 @@
 
 #define NAPTIME_PER_CYCLE 1000	/* max sleep time between cycles (1s) */
 
+/*
+ * Struct for tracking the progress of flushing the transaction received from
+ * the publisher.
+ *
+ * A list of these structs (lsn_mapping) is maintained in the apply worker, each
+ * representing a transaction that is being flushed. The entries are removed
+ * from the list when the transaction is fully flushed (see get_flush_position).
+ */
 typedef struct FlushPosition
 {
+	/*
+	 * The end of commit record applied by the worker, and the corresponding end
+	 * of commit record on the publisher.
+	 *
+	 * For transactions assigned to a parallel apply worker, the local_end is
+	 * not immediately available. The leader later fetches the local commit LSN
+	 * asynchronously when needed, either when collecting flush positions to
+	 * report feedback to the walsender, or when reusing the worker for a new
+	 * transaction (see pa_maybe_reuse_worker).
+	 */
 	dlist_node	node;
 	XLogRecPtr	local_end;
 	XLogRecPtr	remote_end;
+
+	/*
+	 * Remote transaction ID. This is valid only when the transaction is being
+	 * applied by a parallel apply worker. The leader uses this xid to look up
+	 * the ParallelApplyTxnHash table and retrieve the local_end from the
+	 * worker.
+	 */
+	TransactionId pa_remote_xid;
 } FlushPosition;
 
 static dlist_head lsn_mapping = DLIST_STATIC_INIT(lsn_mapping);
@@ -1395,7 +1421,8 @@ apply_handle_prepare(StringInfo s)
 	 * XactLastCommitEnd, and adding it for this purpose doesn't seems worth
 	 * it.
 	 */
-	store_flush_position(prepare_data.end_lsn, InvalidXLogRecPtr);
+	store_flush_position(prepare_data.end_lsn, InvalidXLogRecPtr,
+						 InvalidTransactionId);
 
 	in_remote_transaction = false;
 
@@ -1455,7 +1482,8 @@ apply_handle_commit_prepared(StringInfo s)
 	CommitTransactionCommand();
 	pgstat_report_stat(false);
 
-	store_flush_position(prepare_data.end_lsn, XactLastCommitEnd);
+	store_flush_position(prepare_data.end_lsn, XactLastCommitEnd,
+						 InvalidTransactionId);
 	in_remote_transaction = false;
 
 	/*
@@ -1524,7 +1552,8 @@ apply_handle_rollback_prepared(StringInfo s)
 	 * transaction because we always flush the WAL record for it. See
 	 * apply_handle_prepare.
 	 */
-	store_flush_position(rollback_data.rollback_end_lsn, InvalidXLogRecPtr);
+	store_flush_position(rollback_data.rollback_end_lsn, InvalidXLogRecPtr,
+						 InvalidTransactionId);
 	in_remote_transaction = false;
 
 	/*
@@ -1586,7 +1615,8 @@ apply_handle_stream_prepare(StringInfo s)
 			 * It is okay not to set the local_end LSN for the prepare because
 			 * we always flush the prepare record. See apply_handle_prepare.
 			 */
-			store_flush_position(prepare_data.end_lsn, InvalidXLogRecPtr);
+			store_flush_position(prepare_data.end_lsn, InvalidXLogRecPtr,
+								 InvalidTransactionId);
 
 			in_remote_transaction = false;
 
@@ -2565,7 +2595,8 @@ apply_handle_commit_internal(LogicalRepCommitData *commit_data)
 
 		pgstat_report_stat(false);
 
-		store_flush_position(commit_data->end_lsn, XactLastCommitEnd);
+		store_flush_position(commit_data->end_lsn, XactLastCommitEnd,
+							 InvalidTransactionId);
 	}
 	else
 	{
@@ -3943,6 +3974,15 @@ get_flush_position(XLogRecPtr *write, XLogRecPtr *flush,
 		FlushPosition *pos =
 			dlist_container(FlushPosition, node, iter.cur);
 
+		/*
+		 * Stop if the transaction is still being applied in a parallel apply
+		 * worker. Transactions are committed in the publisher's commit order,
+		 * so no later transaction can have been committed or flushed either.
+		 */
+		if (TransactionIdIsValid(pos->pa_remote_xid) &&
+			!pa_get_last_commit_end(pos->pa_remote_xid, &pos->local_end))
+			break;
+
 		*write = pos->remote_end;
 
 		if (pos->local_end <= local_flush)
@@ -3951,19 +3991,12 @@ get_flush_position(XLogRecPtr *write, XLogRecPtr *flush,
 			dlist_delete(iter.cur);
 			pfree(pos);
 		}
-		else
-		{
-			/*
-			 * Don't want to uselessly iterate over the rest of the list which
-			 * could potentially be long. Instead get the last element and
-			 * grab the write position from there.
-			 */
-			pos = dlist_tail_element(FlushPosition, node,
-									 &lsn_mapping);
-			*write = pos->remote_end;
-			*have_pending_txes = true;
-			return;
-		}
+
+		/*
+		 * We cannot stop here because later transactions may have been
+		 * committed by parallel workers, in which case we need to collect their
+		 * local_end and update the write position accordingly.
+		 */
 	}
 
 	*have_pending_txes = !dlist_is_empty(&lsn_mapping);
@@ -3971,9 +4004,15 @@ get_flush_position(XLogRecPtr *write, XLogRecPtr *flush,
 
 /*
  * Store current remote/local lsn pair in the tracking list.
+ *
+ * pa_remote_xid should be a valid transaction ID if the transaction was
+ * assigned to a parallel apply worker and the leader did not wait for it to
+ * commit. In that case, local_lsn will be collected from the worker later (see
+ * get_flush_position). Otherwise, pass InvalidTransactionId.
  */
 void
-store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn)
+store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn,
+					 TransactionId pa_remote_xid)
 {
 	FlushPosition *flushpos;
 
@@ -3991,12 +4030,15 @@ store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn)
 	flushpos = palloc_object(FlushPosition);
 	flushpos->local_end = local_lsn;
 	flushpos->remote_end = remote_lsn;
+	flushpos->pa_remote_xid = pa_remote_xid;
+
+	if (TransactionIdIsValid(pa_remote_xid))
+		pa_bind_flush_position(pa_remote_xid, &flushpos->local_end);
 
 	dlist_push_tail(&lsn_mapping, &flushpos->node);
 	MemoryContextSwitchTo(ApplyMessageContext);
 }
 
-
 /* Update statistics of the worker. */
 static void
 UpdateWorkerStats(XLogRecPtr last_lsn, TimestampTz send_time, bool reply)
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index f65e5b18d8c..ca1c2661f25 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -330,7 +330,9 @@ extern void SetupApplyOrSyncWorker(int worker_slot);
 
 extern void DisableSubscriptionAndExit(void);
 
-extern void store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn);
+extern void store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn,
+								 TransactionId pa_remote_xid);
+extern void update_flush_position(dlist_node *node, XLogRecPtr local_lsn);
 
 /* Function for apply error callback */
 extern void apply_error_callback(void *arg);
@@ -370,6 +372,8 @@ 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);
+extern void pa_bind_flush_position(TransactionId xid, XLogRecPtr *local_end);
+extern bool pa_get_last_commit_end(TransactionId xid, XLogRecPtr *local_end);
 
 #define isParallelApplyWorker(worker) ((worker)->in_use && \
 									   (worker)->type == WORKERTYPE_PARALLEL_APPLY)
-- 
2.43.0

