Fix the incorrect assertion introduced in commit 7f13ac8123.

It has been incorrectly assumed in commit 7f13ac8123 that we can either
purge all or none in the catalog modifying xids list retrieved from a
serialized snapshot. It is quite possible that some of the xids in that
array are old enough to be pruned but not others.

As per buildfarm

Author: Amit Kapila and Masahiko Sawada
Reviwed-by: Masahiko Sawada
Discussion: https://postgr.es/m/CAA4eK1LBtv6ayE+TvCcPmC-xse=DVg=SmbyQD1nv_AaqcpUJEg@mail.gmail.com

1 files changed, 31 insertions, 19 deletions

diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index 1ff2c12240d..bf72ad45ec7 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -969,28 +969,40 @@ SnapBuildPurgeOlderTxn(SnapBuild *builder)
 	pfree(workspace);
 
 	/*
-	 * Either all the xacts got purged or none. It is only possible to
-	 * partially remove the xids from this array if one or more of the xids
-	 * are still running but not all. That can happen if we start decoding
-	 * from a point (LSN where the snapshot state became consistent) where all
-	 * the xacts in this were running and then at least one of those got
-	 * committed and a few are still running. We will never start from such a
-	 * point because we won't move the slot's restart_lsn past the point where
-	 * the oldest running transaction's restart_decoding_lsn is.
+	 * Purge xids in ->catchange as well. The purged array must also be sorted
+	 * in xidComparator order.
 	 */
-	if (builder->catchange.xcnt == 0 ||
-		TransactionIdFollowsOrEquals(builder->catchange.xip[0],
-									 builder->xmin))
-		return;
+	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;
+		}
 
-	Assert(TransactionIdFollows(builder->xmin,
-								builder->catchange.xip[builder->catchange.xcnt - 1]));
-	pfree(builder->catchange.xip);
-	builder->catchange.xip = NULL;
-	builder->catchange.xcnt = 0;
+		surviving_xids = builder->catchange.xcnt - off;
 
-	elog(DEBUG3, "purged catalog modifying transactions, oldest running xid %u",
-		 builder->xmin);
+		if (surviving_xids > 0)
+		{
+			memmove(builder->catchange.xip, &(builder->catchange.xip[off]),
+					surviving_xids * sizeof(TransactionId));
+		}
+		else
+		{
+			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,
+			 builder->xmin, builder->xmax);
+		builder->catchange.xcnt = surviving_xids;
+	}
 }
 
 /*