v2-0002-Maintain-the-oldest-non-removeable-tranasction-id.patch

From cc780560d69268cad6ec465cb12eb2f709938ff2 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Thu, 26 Sep 2024 14:11:40 +0800
Subject: [PATCH v2 2/5] Maintain the oldest non removeable tranasction id by
 apply worker

This patch allows each apply worker to maintain the oldest_nonremovable_xid in
the shared memory. The xid is advanced based on the following process:

1) Calling GetRunningTransactionData() to take oldestRunningXid as the
candidate xid and send a new message to request the remote WAL position from
the walsender.
2) It then waits (non-blocking) to receive the WAL position from the walsender.
3) After receiving the WAL position, the non-removable transaction ID is
advanced if the current flush location has reached or surpassed the received
WAL position.

These steps are repeated at intervals defined by wal_receiver_status_interval
to minimize performance impact.
---
 doc/src/sgml/protocol.sgml                |  52 +++++++
 src/backend/replication/logical/worker.c  | 182 +++++++++++++++++++++-
 src/backend/replication/walsender.c       |  23 +++
 src/include/replication/worker_internal.h |  18 +++
 src/tools/pgindent/typedefs.list          |   1 +
 5 files changed, 275 insertions(+), 1 deletion(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 11b6456779..4896086220 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2440,6 +2440,41 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
          </variablelist>
         </listitem>
        </varlistentry>
+
+       <varlistentry id="protocol-replication-primary-wal-status-update">
+        <term>Primary WAL status update (B)</term>
+        <listitem>
+         <variablelist>
+          <varlistentry>
+           <term>Byte1('s')</term>
+           <listitem>
+            <para>
+             Identifies the message as a primary WAL status update.
+            </para>
+           </listitem>
+          </varlistentry>
+
+          <varlistentry>
+           <term>Int64</term>
+           <listitem>
+            <para>
+             The latest WAL write position on the server.
+            </para>
+           </listitem>
+          </varlistentry>
+
+          <varlistentry>
+           <term>Int64</term>
+           <listitem>
+            <para>
+             The server's system clock at the time of transmission, as
+             microseconds since midnight on 2000-01-01.
+            </para>
+           </listitem>
+          </varlistentry>
+         </variablelist>
+        </listitem>
+       </varlistentry>
       </variablelist>
 
       <para>
@@ -2584,6 +2619,23 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
          </variablelist>
         </listitem>
        </varlistentry>
+
+       <varlistentry id="protocol-replication-standby-wal-status-request">
+        <term>Standby WAL status request (F)</term>
+        <listitem>
+         <variablelist>
+          <varlistentry>
+           <term>Byte1('x')</term>
+           <listitem>
+            <para>
+             Identifies the message as a request for the WAL status on the primary.
+            </para>
+           </listitem>
+          </varlistentry>
+         </variablelist>
+        </listitem>
+       </varlistentry>
+
       </variablelist>
      </listitem>
     </varlistentry>
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 925dff9cc4..d74d8fe2de 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -173,12 +173,14 @@
 #include "replication/logicalrelation.h"
 #include "replication/logicalworker.h"
 #include "replication/origin.h"
+#include "replication/slot.h"
 #include "replication/walreceiver.h"
 #include "replication/worker_internal.h"
 #include "rewrite/rewriteHandler.h"
 #include "storage/buffile.h"
 #include "storage/ipc.h"
 #include "storage/lmgr.h"
+#include "storage/procarray.h"
 #include "tcop/tcopprot.h"
 #include "utils/acl.h"
 #include "utils/dynahash.h"
@@ -275,6 +277,19 @@ typedef enum
 	TRANS_PARALLEL_APPLY,
 } TransApplyAction;
 
+/*
+ * The phases involved in advancing the non-removable transaction ID.
+ *
+ * Refer to maybe_advance_nonremovable_xid() for details on how the function
+ * transitions between these phases.
+ */
+typedef enum
+{
+	DTR_REQUEST_WALSENDER_WAL_POS,
+	DTR_WAIT_FOR_WALSENDER_WAL_POS,
+	DTR_WAIT_FOR_LOCAL_FLUSH
+} DeadTupleRetainPhase;
+
 /* errcontext tracker */
 static ApplyErrorCallbackArg apply_error_callback_arg =
 {
@@ -339,6 +354,8 @@ static XLogRecPtr skip_xact_finish_lsn = InvalidXLogRecPtr;
 /* BufFile handle of the current streaming file */
 static BufFile *stream_fd = NULL;
 
+static XLogRecPtr last_flushpos = InvalidXLogRecPtr;
+
 typedef struct SubXactInfo
 {
 	TransactionId xid;			/* XID of the subxact */
@@ -378,6 +395,8 @@ static void stream_open_and_write_change(TransactionId xid, char action, StringI
 static void stream_close_file(void);
 
 static void send_feedback(XLogRecPtr recvpos, bool force, bool requestReply);
+static void maybe_advance_nonremovable_xid(XLogRecPtr *remote_wal_pos,
+										   DeadTupleRetainPhase *phase);
 
 static void apply_handle_commit_internal(LogicalRepCommitData *commit_data);
 static void apply_handle_insert_internal(ApplyExecutionData *edata,
@@ -3573,6 +3592,8 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 	bool		ping_sent = false;
 	TimeLineID	tli;
 	ErrorContextCallback errcallback;
+	XLogRecPtr	remote_lsn = InvalidXLogRecPtr;
+	DeadTupleRetainPhase phase = DTR_REQUEST_WALSENDER_WAL_POS;
 
 	/*
 	 * Init the ApplyMessageContext which we clean up after each replication
@@ -3694,6 +3715,16 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 						send_feedback(last_received, reply_requested, false);
 						UpdateWorkerStats(last_received, timestamp, true);
 					}
+					else if (c == 's')
+					{
+						TimestampTz timestamp;
+
+						remote_lsn = pq_getmsgint64(&s);
+						timestamp = pq_getmsgint64(&s);
+
+						maybe_advance_nonremovable_xid(&remote_lsn, &phase);
+						UpdateWorkerStats(last_received, timestamp, false);
+					}
 					/* other message types are purposefully ignored */
 
 					MemoryContextReset(ApplyMessageContext);
@@ -3706,6 +3737,8 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 		/* confirm all writes so far */
 		send_feedback(last_received, false, false);
 
+		maybe_advance_nonremovable_xid(&remote_lsn, &phase);
+
 		if (!in_remote_transaction && !in_streamed_transaction)
 		{
 			/*
@@ -3838,7 +3871,6 @@ send_feedback(XLogRecPtr recvpos, bool force, bool requestReply)
 
 	static XLogRecPtr last_recvpos = InvalidXLogRecPtr;
 	static XLogRecPtr last_writepos = InvalidXLogRecPtr;
-	static XLogRecPtr last_flushpos = InvalidXLogRecPtr;
 
 	XLogRecPtr	writepos;
 	XLogRecPtr	flushpos;
@@ -3916,6 +3948,154 @@ send_feedback(XLogRecPtr recvpos, bool force, bool requestReply)
 		last_flushpos = flushpos;
 }
 
+/*
+ * Attempt to advance the non-removable transaction id.
+ *
+ * The oldest_nonremovable_xid is maintained in shared memory to prevent dead
+ * rows from being removed prematurely when the apply worker still needs them
+ * to detect update-delete conflicts.
+ *
+ * The non-removable transaction id is advanced to the oldest running
+ * transaction id once all concurrent transactions on the publisher have been
+ * applied and flushed locally. The process involves:
+ *
+ * DTR_REQUEST_WALSENDER_WAL_POS - Call GetRunningTransactionData() to get the
+ * candidate xmin and send a message to request the remote WAL position from
+ * the walsender.
+ *
+ * DTR_WAIT_FOR_WALSENDER_WAL_POS - Wait for receiving the WAL position from
+ * the walsender.
+ *
+ * DTR_WAIT_FOR_LOCAL_FLUSH - Advance the non-removable transaction ID if the
+ * current flush location has reached or surpassed the received WAL position.
+ *
+ * Retaining the dead tuples for this period is sufficient because any
+ * subsequent transaction from the publisher will have a later timestamp.
+ * Therefore, it is acceptable if dead tuples are removed by vacuum and an
+ * update_missing conflict is detected, as the correct resolution for the
+ * last-update-wins strategy in this case is to convert the UPDATE to an INSERT
+ * and apply it anyway.
+ *
+ * The 'remote_wal_pos' will be reset after sending a new request to walsender.
+ */
+static void
+maybe_advance_nonremovable_xid(XLogRecPtr *remote_wal_pos,
+							   DeadTupleRetainPhase *phase)
+{
+	static TimestampTz xid_advance_attemp_time = 0;
+	static FullTransactionId candidate_xid;
+
+	Assert(remote_wal_pos);
+
+	/* Exit early if retaining dead tuples is not required */
+	if (!MySubscription->retaindeadtuples)
+		return;
+
+	if (*phase == DTR_WAIT_FOR_WALSENDER_WAL_POS)
+	{
+		Assert(xid_advance_attemp_time);
+
+		/*
+		 * Return if we have requested but not yet received the remote wal
+		 * position.
+		 */
+		if (XLogRecPtrIsInvalid(*remote_wal_pos))
+			return;
+
+		*phase = DTR_WAIT_FOR_LOCAL_FLUSH;
+	}
+
+	if (*phase == DTR_WAIT_FOR_LOCAL_FLUSH)
+	{
+		Assert(!XLogRecPtrIsInvalid(*remote_wal_pos) &&
+			   FullTransactionIdIsValid(candidate_xid));
+
+		/*
+		 * Do not attempt to advance the non-removable transaction id when
+		 * table sync is in progress. During this time, changes from a single
+		 * transaction may be applied by multiple table sync workers
+		 * corresponding to the target tables. In this case, confirming the
+		 * apply and flush progress across all table sync workers is complex
+		 * and not worth the effort.
+		 */
+		if (!AllTablesyncsReady())
+			return;
+
+		/* Return to wait for the changes to be applied */
+		if (last_flushpos < *remote_wal_pos)
+			return;
+
+		/*
+		 * Advance the non-removable transaction id if the remote wal position
+		 * has been received, and all transactions up to that position on the
+		 * publisher have been applied and flushed locally.
+		 */
+		SpinLockAcquire(&MyLogicalRepWorker->relmutex);
+		MyLogicalRepWorker->oldest_nonremovable_xid = candidate_xid;
+		SpinLockRelease(&MyLogicalRepWorker->relmutex);
+
+		*phase = DTR_REQUEST_WALSENDER_WAL_POS;
+	}
+
+	if (*phase == DTR_REQUEST_WALSENDER_WAL_POS)
+	{
+		TimestampTz now;
+		RunningTransactions running_transactions;
+		TransactionId oldest_running_xid;
+		FullTransactionId full_xid;
+		uint32		epoch;
+
+		/*
+		 * Exit early if the user has disabled sending messages to the
+		 * publisher.
+		 */
+		if (wal_receiver_status_interval <= 0)
+			return;
+
+		now = GetCurrentTimestamp();
+
+		/*
+		 * Compute the candidate_xid and send a message at most once per
+		 * wal_receiver_status_interval.
+		 */
+		if (!TimestampDifferenceExceeds(xid_advance_attemp_time, now,
+										wal_receiver_status_interval * 1000))
+			return;
+
+		xid_advance_attemp_time = now;
+
+		running_transactions = GetRunningTransactionData();
+		oldest_running_xid = running_transactions->oldestRunningXid;
+		epoch = EpochFromFullTransactionId(TransamVariables->nextXid);
+
+		/* Release the locks acquired in GetRunningTransactionData() */
+		LWLockRelease(ProcArrayLock);
+		LWLockRelease(XidGenLock);
+
+		/* Compute the epoch of the oldestRunningXid */
+		if (oldest_running_xid > running_transactions->nextXid)
+			epoch--;
+
+		full_xid = FullTransactionIdFromEpochAndXid(epoch,
+													oldest_running_xid);
+
+		/* Return if the new transaction ID is unchanged */
+		if (FullTransactionIdFollowsOrEquals(MyLogicalRepWorker->oldest_nonremovable_xid,
+											 full_xid))
+			return;
+
+		candidate_xid = full_xid;
+
+		elog(DEBUG2, "sending wal position request message");
+
+		/* Send a wal position request message to the server */
+		walrcv_send(LogRepWorkerWalRcvConn, "x", sizeof(uint8));
+
+		*remote_wal_pos = InvalidXLogRecPtr;
+		*phase = DTR_WAIT_FOR_WALSENDER_WAL_POS;
+	}
+}
+
 /*
  * Exit routine for apply workers due to subscription parameter changes.
  */
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index c5f1009f37..5371711488 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -253,6 +253,7 @@ static void StartLogicalReplication(StartReplicationCmd *cmd);
 static void ProcessStandbyMessage(void);
 static void ProcessStandbyReplyMessage(void);
 static void ProcessStandbyHSFeedbackMessage(void);
+static void ProcessStandbyWalPosRequestMessage(void);
 static void ProcessRepliesIfAny(void);
 static void ProcessPendingWrites(void);
 static void WalSndKeepalive(bool requestReply, XLogRecPtr writePtr);
@@ -2342,6 +2343,10 @@ ProcessStandbyMessage(void)
 			ProcessStandbyHSFeedbackMessage();
 			break;
 
+		case 'x':
+			ProcessStandbyWalPosRequestMessage();
+			break;
+
 		default:
 			ereport(COMMERROR,
 					(errcode(ERRCODE_PROTOCOL_VIOLATION),
@@ -2688,6 +2693,24 @@ ProcessStandbyHSFeedbackMessage(void)
 	}
 }
 
+/*
+ * Process the standby message requesting the latest WAL write position.
+ */
+static void
+ProcessStandbyWalPosRequestMessage(void)
+{
+	elog(DEBUG2, "sending wal write position");
+
+	/* construct the message... */
+	resetStringInfo(&output_message);
+	pq_sendbyte(&output_message, 's');
+	pq_sendint64(&output_message, GetXLogWriteRecPtr());
+	pq_sendint64(&output_message, GetCurrentTimestamp());
+
+	/* ... and send it wrapped in CopyData */
+	pq_putmessage_noblock('d', output_message.data, output_message.len);
+}
+
 /*
  * Compute how long send/receive loops should sleep.
  *
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 9646261d7e..370c71c93e 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -86,6 +86,24 @@ typedef struct LogicalRepWorker
 	/* Indicates whether apply can be performed in parallel. */
 	bool		parallel_apply;
 
+	/*
+	 * The changes made by this and later transactions are still non-removable
+	 * to allow for the detection of update_delete conflicts when applying
+	 * changes in this logical replication worker.
+	 *
+	 * Note that this info cannot directly protect dead tuples from being
+	 * prematurely frozen or removed. The logical replication launcher
+	 * asynchronously collects this info to determine whether to advance the
+	 * xmin value of the replication slot.
+	 *
+	 * Therefore, FullTransactionId that includes both the transaction ID and
+	 * its epoch is used here instead of a single Transaction ID. This is
+	 * critical because without considering the epoch, the transaction ID
+	 * alone may appear as if it is in the future due to transaction ID
+	 * wraparound.
+	 */
+	FullTransactionId oldest_nonremovable_xid;
+
 	/* Stats. */
 	XLogRecPtr	last_lsn;
 	TimestampTz last_send_time;
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 5fabb127d7..66d7f16a73 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -591,6 +591,7 @@ DbInfoArr
 DbLocaleInfo
 DeClonePtrType
 DeadLockState
+DeadTupleRetainPhase
 DeallocateStmt
 DeclareCursorStmt
 DecodedBkpBlock
-- 
2.30.0.windows.2