Fix several DistributedTransaction related issues (#13810)

* Avoid enlarging inProgressXidArray when CreateDistributedSnapshot().
* latestCompletedXid should be uint64.
* DistributedLog_AdvanceOldestXmin() returns more accurate oldestXmin.
* Remove unnecessary check in DistributedSnapshotWithLocalMapping_CommittedTest().
* Remove unnecessary check in HeapTupleSatisfiesMVCC().

There are several issues/enhancements related to DistributedTransactionId or DistributedSnapshot.
Since they are closely related, we put them all together here.

1, latestCompletedGxid is uint32

	gxid is unit64 now, latestCompletedGxid should also be unit64.

2, Avoid enlarging inProgressXidArray during CreateDistributedSnapshot()

	DistributedSnapshot's xmax is set to latestCompletedGxid + 1 after ProcArrayLock is held.

	if (gxid > xmax)
	{
		xmax = gxid;
	}

	There is no need to update xmax because latestCompletedGxid can't be increased during
	CreateDistributedSnapshot() and any gxid bigger than latestCompletedGxid should be seen
	as a future gxid.
	And we should avoid enlarging inProgressXidArray for efficiency.

3, DistributedLog_AdvanceOldestXmin() should return more accurate oldestXmin

	while (1)
	{
		if (pg_atomic_compare_exchange_u32((pg_atomic_uint32 *)&DistributedLogShared->oldestXmin,
										&expected, (uint32)oldestXmin))
			break;
	
		if (TransactionIdPrecedesOrEquals(oldestXmin, expected))
			break;
	}

	When failed to update oldestXmin in a CAS operation and the expected value
	is bigger than the value we want to set.
	Some other processes must have updated oldestXmi to a bigger value.
	And we should return that more accurate value.

4, Enhancements in DistributedSnapshotWithLocalMapping_CommittedTest()

	redundant codes:

		/*
		 * We found it in the distributed log.
		 */
		Assert(distribXid != InvalidDistributedTransactionId);
	
		/*
		 * We have a distributed committed xid that corresponds to the
		 * local xid.
		 */
		Assert(distribXid != InvalidDistributedTransactionId);
	
	unnecessary check:
		if (LocalDistribXactCache_CommittedFind(localXid,&distribXid))
		{
			/*
			 * We cache local-only committed transactions for better performance,
			 * too.
			 */
			if (distribXid == InvalidDistributedTransactionId)
				return DISTRIBUTEDSNAPSHOT_COMMITTED_IGNORE;
	
			/*
			 * Fall below and evaluate the committed distributed transaction
			 * against the distributed snapshot.
			 */
		}
	
	It can't happen that distribXid == InvalidDistributedTransactionId.
	If we found it in LocalDistribXactCache_CommittedFind(), the distribXid must be valid.
	We add it to cache in LocalDistribXactCache_AddCommitted() and there is nowhere to update
	distribXid to InvalidDistributedTransactionId.
	distribXid in cache must be valid or we don't find it in cache(never added or removed by cache LRU).

5, Redundant checks in HeapTupleSatisfiesMVCC()

	if (snapshotCheckResult == XID_NOT_IN_SNAPSHOT)
	{
		if (snapshotCheckResult == XID_SURELY_COMMITTED || TransactionIdDidCommit(xvac))
		{
			SetHintBits(tuple, buffer, relation, HEAP_XMIN_INVALID,
						InvalidTransactionId);
			return false;
		}
		SetHintBits(tuple, buffer, relation, HEAP_XMIN_COMMITTED,
					InvalidTransactionId);
	}

	snapshotCheckResult == XID_SURELY_COMMITTED check is unnecessary.

Author: avamingli

src/backend/access/heap/heapam_visibility.c

@@ -1069,7 +1069,7 @@ HeapTupleSatisfiesMVCC(Relation relation, HeapTuple htup, Snapshot snapshot,
 													&setDistributedSnapshotIgnore);
 			if (snapshotCheckResult == XID_NOT_IN_SNAPSHOT)
 			{
-				if (snapshotCheckResult == XID_SURELY_COMMITTED || TransactionIdDidCommit(xvac))
+				if (TransactionIdDidCommit(xvac))
 				{
 					SetHintBits(tuple, buffer, relation, HEAP_XMIN_INVALID,
 								InvalidTransactionId);

src/backend/access/transam/distributedlog.c

@@ -273,7 +273,14 @@ DistributedLog_AdvanceOldestXmin(TransactionId oldestLocalXmin,
 				break;
 
 			if (TransactionIdPrecedesOrEquals(oldestXmin, expected))
+			{
+				/*
+				 * If we got here, other processes must have updated the oldestXmin.
+				 * Return the more accurate value.
+				 */
+				oldestXmin = expected;
 				break;
+			}
 		}
 	}
 

src/backend/cdb/cdbdistributedsnapshot.c

@@ -86,23 +86,15 @@ DistributedSnapshotWithLocalMapping_CommittedTest(
 
 	/*
 	 * Is this local xid in a process-local cache we maintain?
+	 *
+	 * If we found xact in cache, the distribXid must be a valid gxid.
+	 * We added it to cache in LocalDistribXactCache_AddCommitted()
+	 * and there is nowhere to update distribXid to InvalidDistributedTransactionId.
+	 * So that distribXid in cache must be valid or we don't find it
+	 * in cache (never added or removed by cache LRU).
 	 */
-	if (LocalDistribXactCache_CommittedFind(localXid,
+	if (!LocalDistribXactCache_CommittedFind(localXid,
 											&distribXid))
-	{
-		/*
-		 * We cache local-only committed transactions for better performance,
-		 * too.
-		 */
-		if (distribXid == InvalidDistributedTransactionId)
-			return DISTRIBUTEDSNAPSHOT_COMMITTED_IGNORE;
-
-		/*
-		 * Fall below and evaluate the committed distributed transaction
-		 * against the distributed snapshot.
-		 */
-	}
-	else
 	{
 		/*
 		 * Ok, now we must consult the distributed log.
@@ -114,12 +106,6 @@ DistributedSnapshotWithLocalMapping_CommittedTest(
 			 */
 			Assert(distribXid != InvalidDistributedTransactionId);
 
-			/*
-			 * We have a distributed committed xid that corresponds to the
-			 * local xid.
-			 */
-			Assert(distribXid != InvalidDistributedTransactionId);
-
 			/*
 			 * Since we did not find it in our process local cache, add it.
 			 */
@@ -137,6 +123,7 @@ DistributedSnapshotWithLocalMapping_CommittedTest(
 		}
 	}
 
+	Assert(distribXid != InvalidDistributedTransactionId);
 	Assert(ds->xminAllDistributedSnapshots != InvalidDistributedTransactionId);
 
 	/*

src/backend/storage/ipc/procarray.c

@@ -2000,9 +2000,20 @@ CreateDistributedSnapshot(DistributedSnapshot *ds)
 		if (dxid != InvalidDistributedTransactionId && dxid < globalXminDistributedSnapshots)
 			globalXminDistributedSnapshots = dxid;
 
-		/* just fetch once */
+		/*
+		 * Just fetch once
+		 *
+		 * Partial reading is possible on a 32 bit system, however we decided
+		 * not to take it serious currently.
+		 */
 		gxid = gxact_candidate->gxid;
-		if (gxid == InvalidDistributedTransactionId)
+
+		/*
+		* Skip further gxid to avoid enlarging inProgressXidArray
+		* as we already have held ProcArrayLock and latestCompletedGxid
+		* can not be changed.
+		*/
+		if (gxid == InvalidDistributedTransactionId || gxid >= xmax)
 			continue;
 
 		/*
@@ -2013,10 +2024,6 @@ CreateDistributedSnapshot(DistributedSnapshot *ds)
 		{
 			xmin = gxid;
 		}
-		if (gxid > xmax)
-		{
-			xmax = gxid;
-		}
 
 		if (gxact_candidate == MyTmGxact)
 			continue;

src/backend/storage/ipc/test/procarray_test.c

@@ -94,10 +94,10 @@ test__CreateDistributedSnapshot(void **state)
 	/* perform all the validations */
 	assert_true(ds.xminAllDistributedSnapshots == 5);
 	assert_true(ds.xmin == 10);
-	assert_true(ds.xmax == 30);
-	assert_true(ds.count == 2);
+	/* xmax should be latestCompletedGxid+1 */
+	assert_true(ds.xmax == 25);
+	assert_true(ds.count == 1);
 	assert_true(ds.inProgressXidArray[0] == 10);
-	assert_true(ds.inProgressXidArray[1] == 30);
 	assert_true(MyTmGxact->xminDistributedSnapshot == 10);
 
 	/*************************************************************************
@@ -123,13 +123,12 @@ test__CreateDistributedSnapshot(void **state)
 	/* perform all the validations */
 	assert_true(ds.xminAllDistributedSnapshots == 5);
 	assert_true(ds.xmin == 7);
-	assert_true(ds.xmax == 30);
-	assert_true(ds.count == 4);
+	assert_true(ds.xmax == 25);
+	assert_true(ds.count == 3);
 	assert_true(MyTmGxact->xminDistributedSnapshot == 7);
 	assert_true(ds.inProgressXidArray[0] == 7);
 	assert_true(ds.inProgressXidArray[1] == 10);
 	assert_true(ds.inProgressXidArray[2] == 15);
-	assert_true(ds.inProgressXidArray[3] == 30);
 
 	free(ds.inProgressXidArray);
 	free(allTmGxact);

src/include/access/transam.h

@@ -152,8 +152,8 @@ typedef struct VariableCacheData
 	 */
 	TransactionId latestCompletedXid;	/* newest XID that has committed or
 										 * aborted */
-	TransactionId latestCompletedGxid;	/* newest distributed XID that has
-										   committed or aborted */
+	DistributedTransactionId latestCompletedGxid;	/* newest distributed XID that has
+													   committed or aborted */
 
 	/*
 	 * The two variables are protected by shmGxidGenLock.  Note nextGxid won't