From 40e51baf4b73f2606a7c5f50088e8c9911107044 Mon Sep 17 00:00:00 2001
From: alterego655 <824662526@qq.com>
Date: Mon, 10 Nov 2025 10:31:38 +0800
Subject: [PATCH v1 2/2] Optimize SnapBuildPurgeOlderTxn with binary search

SnapBuildPurgeOlderTxn() removes XIDs that precede xmin from the
committed.xip and catchange.xip arrays. Previously, this used an
O(n) linear scan with workspace allocation. Since these arrays are
maintained in sorted order, we can use binary search for O(log n)
performance.

The optimization exploits the fact that XIDs "preceding xmin" form
the modular interval [xmin - 2^31, xmin) on the 32-bit XID ring.
In a numeric-sorted array, this interval appears as either one or
two contiguous segments depending on numeric positions:

- Case A (boundary <= xmin numerically): One block [idx_boundary, idx_xmin)
- Case B (boundary > xmin numerically): Two blocks [0, idx_xmin) and
  [idx_boundary, n)

We use two binary searches to locate these boundaries, then compact
the array with a single memmove. This eliminates workspace allocation
and reduces the operation from O(n) to O(log n).

The same wraparound-aware two-case logic is applied to both committed.xip
and catchange.xip arrays.
---
 src/backend/replication/logical/snapbuild.c | 185 +++++++++++++++-----
 1 file changed, 137 insertions(+), 48 deletions(-)

diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index 344eb4c2ade..5c3bb8f377f 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -152,6 +152,11 @@ static ResourceOwner SavedResourceOwnerDuringExport = NULL;
 static bool ExportInProgress = false;
 
 /* ->committed and ->catchange manipulation */
+static int xid_bsearch_lower_bound(const TransactionId *xip, int n, TransactionId key);
+
+static int xid_purge_with_boundary(TransactionId *xip, int xcnt,
+									TransactionId boundary, TransactionId xmin);
+
 static void SnapBuildPurgeOlderTxn(SnapBuild *builder);
 
 /* snapshot building/manipulation/distribution functions */
@@ -889,6 +894,106 @@ SnapBuildAddCommittedTxns(SnapBuild *builder,
 	builder->committed.xcnt = old_xcnt + batch_cnt;
 }
 
+/*
+ * Binary search helper: find the first index where xip[i] >= key.
+ * Returns n if all elements are less than key.
+ *
+ * This is a standard lower_bound operation using unsigned comparison to match
+ * xidComparator ordering.
+ */
+static int
+xid_bsearch_lower_bound(const TransactionId *xip, int n, TransactionId key)
+{
+	int			lo = 0;
+	int			hi = n;
+
+	while (lo < hi)
+	{
+		int			mid = (lo + hi) >> 1;
+
+		if (xip[mid] < key)
+			lo = mid + 1;
+		else
+			hi = mid;
+	}
+	return lo;
+}
+
+/*
+ * Remove XIDs in the modular interval [boundary, xmin) from a sorted array.
+ *
+ * The array must be sorted in numeric uint32 order. XIDs in [boundary, xmin)
+ * are those that satisfy NormalTransactionIdPrecedes(xid, xmin).
+ *
+ * Returns the new count after removal. The array is compacted in-place.
+ */
+static int
+xid_purge_with_boundary(TransactionId *xip, int xcnt,
+						TransactionId boundary, TransactionId xmin)
+{
+	int			idx_boundary;
+	int			idx_xmin;
+	int			new_xcnt;
+
+	if (xcnt == 0)
+		return 0;
+
+	/*
+	 * Find where [boundary, xmin) appears in the numeric-sorted array.
+	 * idx_boundary = first index with xip[i] >= boundary
+	 * idx_xmin     = first index with xip[i] >= xmin
+	 */
+	idx_boundary = xid_bsearch_lower_bound(xip, xcnt, boundary);
+	idx_xmin = xid_bsearch_lower_bound(xip, xcnt, xmin);
+
+	/*
+	 * Case split based on numeric comparison (array is uint32-sorted):
+	 *
+	 * Case A: boundary <= xmin numerically
+	 *   Interval forms one block: [idx_boundary, idx_xmin)
+	 *   Keep: [0, idx_boundary) and [idx_xmin, n)
+	 *
+	 * Case B: boundary > xmin numerically
+	 *   Interval straddles numeric zero as two blocks:
+	 *     [0, idx_xmin) and [idx_boundary, n)
+	 *   Keep: [idx_xmin, idx_boundary)
+	 *
+	 * Note: Case B occurs due to ring geometry when the modular interval
+	 * crosses zero in numeric order, not because XIDs have "wrapped"
+	 * operationally.
+	 */
+	if (boundary <= xmin)
+	{
+		/* Case A: interval is contiguous, keep prefix and suffix */
+		int			prefix_len = idx_boundary;
+		int			suffix_len = xcnt - idx_xmin;
+
+		if (suffix_len > 0 && idx_xmin > idx_boundary)
+			memmove(xip + prefix_len,
+					xip + idx_xmin,
+					suffix_len * sizeof(TransactionId));
+
+		new_xcnt = prefix_len + suffix_len;
+	}
+	else
+	{
+		/* Case B: interval straddles zero, keep middle block */
+		int			keep_len;
+
+		Assert(idx_boundary >= idx_xmin);
+		keep_len = idx_boundary - idx_xmin;
+
+		if (keep_len > 0 && idx_xmin > 0)
+			memmove(xip,
+					xip + idx_xmin,
+					keep_len * sizeof(TransactionId));
+
+		new_xcnt = keep_len;
+	}
+
+	return new_xcnt;
+}
+
 /*
  * Remove knowledge about transactions we treat as committed or containing catalog
  * changes that are smaller than ->xmin. Those won't ever get checked via
@@ -902,74 +1007,58 @@ SnapBuildAddCommittedTxns(SnapBuild *builder,
 static void
 SnapBuildPurgeOlderTxn(SnapBuild *builder)
 {
-	int			off;
-	TransactionId *workspace;
-	int			surviving_xids = 0;
+	TransactionId boundary;
 
 	/* not ready yet */
 	if (!TransactionIdIsNormal(builder->xmin))
 		return;
 
-	/* TODO: Neater algorithm than just copying and iterating? */
-	workspace =
-		MemoryContextAlloc(builder->context,
-						   builder->committed.xcnt * sizeof(TransactionId));
+	/*
+	 * Purge committed.xip and catchange.xip using binary search.
+	 *
+	 * XID precedence: NormalTransactionIdPrecedes(a, b) means
+	 * (int32)(a - b) < 0, which identifies XIDs in the modular interval
+	 * [b - 2^31, b) on the 32-bit ring as "preceding b".
+	 *
+	 * Both arrays are sorted in numeric uint32 order: xip[i] <= xip[i+1].
+	 * This is a total order, but the "old" interval [xmin - 2^31, xmin)
+	 * may appear as one or two segments depending on numeric positions.
+	 */
+	boundary = builder->xmin - 0x80000000U;
 
-	/* copy xids that still are interesting to workspace */
-	for (off = 0; off < builder->committed.xcnt; off++)
+	/* Purge committed.xip */
+	if (builder->committed.xcnt > 0)
 	{
-		if (NormalTransactionIdPrecedes(builder->committed.xip[off],
-										builder->xmin))
-			;					/* remove */
-		else
-			workspace[surviving_xids++] = builder->committed.xip[off];
-	}
-
-	/* copy workspace back to persistent state */
-	memcpy(builder->committed.xip, workspace,
-		   surviving_xids * sizeof(TransactionId));
-
-	elog(DEBUG3, "purged committed transactions from %u to %u, xmin: %u, xmax: %u",
-		 (uint32) builder->committed.xcnt, (uint32) surviving_xids,
-		 builder->xmin, builder->xmax);
-	builder->committed.xcnt = surviving_xids;
+		int			old_committed_xcnt = builder->committed.xcnt;
 
-	pfree(workspace);
+		builder->committed.xcnt = xid_purge_with_boundary(builder->committed.xip,
+														  old_committed_xcnt,
+														  boundary,
+														  builder->xmin);
 
-	/*
-	 * Purge xids in ->catchange as well. The purged array must also be sorted
-	 * in xidComparator order.
-	 */
+		elog(DEBUG3, "purged committed transactions from %u to %u, xmin: %u, xmax: %u",
+			 (uint32) old_committed_xcnt, (uint32) builder->committed.xcnt,
+			 builder->xmin, builder->xmax);
+	}
+	/* Purge catchange.xip */
 	if (builder->catchange.xcnt > 0)
 	{
-		/*
-		 * Since catchange.xip is sorted, we find the lower bound of xids that
-		 * are still interesting.
-		 */
-		for (off = 0; off < builder->catchange.xcnt; off++)
-		{
-			if (TransactionIdFollowsOrEquals(builder->catchange.xip[off],
-											 builder->xmin))
-				break;
-		}
+		int			old_catchange_xcnt = builder->catchange.xcnt;
 
-		surviving_xids = builder->catchange.xcnt - off;
+		builder->catchange.xcnt = xid_purge_with_boundary(builder->catchange.xip,
+														  old_catchange_xcnt,
+														  boundary,
+														  builder->xmin);
 
-		if (surviving_xids > 0)
-		{
-			memmove(builder->catchange.xip, &(builder->catchange.xip[off]),
-					surviving_xids * sizeof(TransactionId));
-		}
-		else
+		if (builder->catchange.xcnt == 0)
 		{
 			pfree(builder->catchange.xip);
 			builder->catchange.xip = NULL;
 		}
 
 		elog(DEBUG3, "purged catalog modifying transactions from %u to %u, xmin: %u, xmax: %u",
-			 (uint32) builder->catchange.xcnt, (uint32) surviving_xids,
+			 (uint32) old_catchange_xcnt, (uint32) builder->catchange.xcnt,
 			 builder->xmin, builder->xmax);
-		builder->catchange.xcnt = surviving_xids;
 	}
 }
 
-- 
2.51.0

