From f903160de5ce67e514dd063a30ffb27faf5dcc3a Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Mon, 16 Dec 2024 11:32:25 +0800
Subject: [PATCH] wait for committing txns to finish in walsender

---
 src/backend/replication/logical/worker.c | 47 ++++++++++--------------
 src/backend/replication/walsender.c      | 27 +++++++++++---
 src/backend/storage/ipc/procarray.c      | 16 ++------
 3 files changed, 44 insertions(+), 46 deletions(-)

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 6406be4f17..4abf3ded95 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -3771,8 +3771,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 						 * to avoid accumulating dead rows when the worker is
 						 * busy.
 						 */
-						if (can_advance_nonremovable_xid(&data))
-							maybe_advance_nonremovable_xid(&data, false);
+						maybe_advance_nonremovable_xid(&data, false);
 					}
 					else if (c == 'k')
 					{
@@ -3796,9 +3795,6 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 					else if (c == 's')	/* Primary status update */
 					{
 						data.remote_lsn = pq_getmsgint64(&s);
-						data.remote_oldestxid = pq_getmsgint(&s, 4);
-						data.remote_nextxid = pq_getmsgint(&s, 4);
-						data.remote_epoch = pq_getmsgint(&s, 4);
 						data.reply_time = pq_getmsgint64(&s);
 
 						/*
@@ -4291,28 +4287,7 @@ wait_for_publisher_status(RetainConflictInfoData *data, bool status_received)
 	if (!status_received)
 		return;
 
-	if (!FullTransactionIdIsValid(data->last_phase_at))
-		data->last_phase_at = FullTransactionIdFromEpochAndXid(data->remote_epoch,
-															   data->remote_nextxid);
-
-	/* Compute the epoch of the remote oldest running transaction ID */
-	if (data->remote_oldestxid > data->remote_nextxid)
-		remote_epoch--;
-
-	remote_full_xid = FullTransactionIdFromEpochAndXid(remote_epoch,
-													   data->remote_oldestxid);
-
-	/*
-	 * Check if all remote concurrent transactions that were active at the
-	 * first status request have now completed. If completed, proceed to the
-	 * next phase; otherwise, continue checking the publisher status until
-	 * these transactions finish.
-	 */
-	if (FullTransactionIdPrecedesOrEquals(data->last_phase_at,
-										  remote_full_xid))
-		data->phase = RCI_WAIT_FOR_LOCAL_FLUSH;
-	else
-		data->phase = RCI_REQUEST_PUBLISHER_STATUS;
+	data->phase = RCI_WAIT_FOR_LOCAL_FLUSH;
 
 	/* process the next phase */
 	maybe_advance_nonremovable_xid(data, false);
@@ -4327,6 +4302,24 @@ wait_for_local_flush(RetainConflictInfoData *data)
 	Assert(!XLogRecPtrIsInvalid(data->remote_lsn) &&
 		   FullTransactionIdIsValid(data->candidate_xid));
 
+	/*
+	 * Update the remote flush position if we are applying changes in a loop
+	 * and. Otherwise, the position is only updated when the feedback
+	 * is sent to the server.
+	 */
+	if (data->last_recv_time)
+	{
+		XLogRecPtr	writepos;
+		XLogRecPtr	flushpos;
+		bool		have_pending_txes;
+
+		/* Fetch the latest remote flush position */
+		get_flush_position(&writepos, &flushpos, &have_pending_txes);
+
+		if (flushpos > last_flushpos)
+			last_flushpos = flushpos;
+	}
+
 	/*
 	 * We expect the publisher and subscriber clocks to be in sync using time
 	 * sync service like NTP. Otherwise, we will advance this worker's
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 8c18264f58..eafd658728 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -2676,7 +2676,7 @@ ProcessStandbyPSRequestMessage(void)
 {
 	XLogRecPtr	lsn = InvalidXLogRecPtr;
 	TransactionId oldestXidInCommit;
-	FullTransactionId nextFullXid;
+	TransactionId nextXid = InvalidTransactionId;
 	WalSnd	   *walsnd = MyWalSnd;
 	TimestampTz replyTime;
 
@@ -2693,8 +2693,26 @@ ProcessStandbyPSRequestMessage(void)
 	walsnd->replyTime = replyTime;
 	SpinLockRelease(&walsnd->mutex);
 
-	oldestXidInCommit = GetOldestTransactionIdInCommit();
-	nextFullXid = ReadNextFullTransactionId();
+	for (;;)
+	{
+		oldestXidInCommit = GetOldestTransactionIdInCommit();
+
+		if (!TransactionIdIsValid(oldestXidInCommit))
+			break;
+
+		if (!TransactionIdIsValid(nextXid))
+			nextXid = XidFromFullTransactionId(ReadNextFullTransactionId());
+		else if (TransactionIdPrecedesOrEquals(nextXid, oldestXidInCommit))
+			break;
+
+		(void) WaitLatch(MyLatch,
+						 WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+						 10L,
+						 WAIT_EVENT_LOGICAL_PARALLEL_APPLY_STATE_CHANGE);
+
+		CHECK_FOR_INTERRUPTS();
+	}
+	
 	lsn = GetXLogWriteRecPtr();
 
 	elog(DEBUG2, "sending primary status");
@@ -2703,9 +2721,6 @@ ProcessStandbyPSRequestMessage(void)
 	resetStringInfo(&output_message);
 	pq_sendbyte(&output_message, 's');
 	pq_sendint64(&output_message, lsn);
-	pq_sendint32(&output_message, oldestXidInCommit);
-	pq_sendint32(&output_message, XidFromFullTransactionId(nextFullXid));
-	pq_sendint32(&output_message, EpochFromFullTransactionId(nextFullXid));
 	pq_sendint64(&output_message, GetCurrentTimestamp());
 
 	/* ... and send it wrapped in CopyData */
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index ae6da9fd91..3bb7eaa93c 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -2937,22 +2937,11 @@ GetOldestTransactionIdInCommit(void)
 {
 	ProcArrayStruct *arrayP = procArray;
 	TransactionId *other_xids = ProcGlobal->xids;
-	TransactionId oldestXidInCommit;
+	TransactionId oldestXidInCommit = InvalidTransactionId;
 	int			index;
 
 	Assert(!RecoveryInProgress());
 
-	/*
-	 * Read nextXid, as the upper bound of what's still active.
-	 *
-	 * Reading a TransactionId is atomic, but we must grab the lock to make
-	 * sure that all XIDs < nextXid are already present in the proc array (or
-	 * have already completed), when we spin over it.
-	 */
-	LWLockAcquire(XidGenLock, LW_SHARED);
-	oldestXidInCommit = XidFromFullTransactionId(TransamVariables->nextXid);
-	LWLockRelease(XidGenLock);
-
 	/*
 	 * Spin over procArray collecting all xids and subxids.
 	 */
@@ -2970,7 +2959,8 @@ GetOldestTransactionIdInCommit(void)
 			continue;
 
 		if ((proc->delayChkptFlags & DELAY_CHKPT_IN_COMMIT) != 0 &&
-			TransactionIdPrecedes(xid, oldestXidInCommit))
+			(!TransactionIdIsValid(oldestXidInCommit) ||
+			 TransactionIdPrecedes(xid, oldestXidInCommit)))
 			oldestXidInCommit = xid;
 
 		/*
-- 
2.30.0.windows.2

