Fix distributed snapshot race during commit

src/backend/cdb/cdbtm.c

@@ -129,6 +129,7 @@ static void retryAbortPrepared(void);
 static void doQEDistributedExplicitBegin();
 static void currentDtxActivate(void);
 static void setCurrentDtxState(DtxState state);
+static void markDtxCommitInProgress(void);
 
 static bool isDtxQueryDispatcher(void);
 static void performDtxProtocolCommitPrepared(const char *gid, bool raiseErrorIfNotFound);
@@ -176,6 +177,21 @@ getDistributedTransactionId(void)
 		return InvalidDistributedTransactionId;
 }
 
+/*
+ * Mark the DTX before sending the first commit notification to a QE. The
+ * snapshot builder uses this state to wait for the QD's normal transaction
+ * cleanup, which keeps the proc-array and QE commit views atomic.
+ */
+static void
+markDtxCommitInProgress(void)
+{
+	Assert(MyTmGxact->gxid != InvalidDistributedTransactionId);
+
+	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+	MyTmGxact->commitInProgress = true;
+	LWLockRelease(ProcArrayLock);
+}
+
 bool
 getDistributedTransactionIdentifier(char *id)
 {
@@ -560,6 +576,8 @@ doNotifyingOnePhaseCommit(void)
 	Assert(MyTmGxactLocal->state == DTX_STATE_ONE_PHASE_COMMIT);
 	setCurrentDtxState(DTX_STATE_NOTIFYING_ONE_PHASE_COMMIT);
 
+	markDtxCommitInProgress();
+
 	succeeded = currentDtxDispatchProtocolCommand(DTX_PROTOCOL_COMMAND_COMMIT_ONEPHASE, true);
 	if (!succeeded)
 	{
@@ -587,14 +605,15 @@ doNotifyingCommitPrepared(void)
 	Assert(MyTmGxactLocal->state == DTX_STATE_INSERTED_COMMITTED);
 	setCurrentDtxState(DTX_STATE_NOTIFYING_COMMIT_PREPARED);
 
-	SIMPLE_FAULT_INJECTOR("dtm_broadcast_commit_prepared");
-
 	/*
 	 * Acquire TwophaseCommitLock in shared mode to block any GPDB restore
 	 * points from being created while commit prepared messages are being
 	 * broadcasted.
 	 */
 	LWLockAcquire(TwophaseCommitLock, LW_SHARED);
+	markDtxCommitInProgress();
+
+	SIMPLE_FAULT_INJECTOR("dtm_broadcast_commit_prepared");
 
 	savedInterruptHoldoffCount = InterruptHoldoffCount;
 
@@ -1491,6 +1510,7 @@ resetTmGxact(void)
 	Assert(MyTmGxact->gxid == InvalidDistributedTransactionId);
 	MyTmGxact->xminDistributedSnapshot = InvalidDistributedTransactionId;
 	MyTmGxact->includeInCkpt = false;
+	MyTmGxact->commitInProgress = false;
 	MyTmGxact->sessionId = 0;
 
 	MyTmGxactLocal->explicitBeginRemembered = false;

src/backend/storage/ipc/procarray.c

@@ -61,6 +61,7 @@
 #include "port/atomics.h"
 #include "pgstat.h"
 #include "port/pg_lfind.h"
+#include "storage/lmgr.h"
 #include "storage/proc.h"
 #include "storage/procarray.h"
 #include "storage/spin.h"
@@ -698,6 +699,7 @@ ProcArrayEndGxact(TMGXACT *tmGxact)
 	pg_atomic_init_u64(&(tmGxact->atomic_gxid), InvalidDistributedTransactionId);
 	tmGxact->xminDistributedSnapshot = InvalidDistributedTransactionId;
 	tmGxact->includeInCkpt = false;
+	tmGxact->commitInProgress = false;
 	tmGxact->sessionId = 0;
 
 	/*
@@ -2625,6 +2627,48 @@ DistributedSnapshotMappedEntry_Compare(const void *p1, const void *p2)
 		return -1;
 }
 
+/*
+ * Wait for DTX commit notifications that have started before taking a QD
+ * distributed snapshot. The caller must not hold ProcArrayLock. On return the
+ * lock is held in shared mode, so the caller can build its snapshot from the
+ * same proc-array state that it checked here.
+ */
+static void
+WaitForDtxCommit(void)
+{
+	DistributedTransactionId waitGxid;
+
+	for (;;)
+	{
+		waitGxid = InvalidDistributedTransactionId;
+
+		LWLockAcquire(ProcArrayLock, LW_SHARED);
+
+		for (int i = 0; i < procArray->numProcs; i++)
+		{
+			int					 pgprocno = procArray->pgprocnos[i];
+			volatile TMGXACT *tmGxact = &allTmGxact[pgprocno];
+
+			DistributedTransactionId gxid =
+				pg_atomic_read_u64(&tmGxact->atomic_gxid);
+
+			if (tmGxact == MyTmGxact ||
+				gxid == InvalidDistributedTransactionId ||
+				!tmGxact->commitInProgress)
+				continue;
+
+			waitGxid = gxid;
+			break;
+		}
+
+		if (waitGxid == InvalidDistributedTransactionId)
+			break;
+
+		LWLockRelease(ProcArrayLock);
+		GxactLockTableWait(waitGxid);
+	}
+}
+
 /*
  * create distributed snapshot based on current visible distributed transaction
  */
@@ -3020,9 +3064,16 @@ GetSnapshotData(Snapshot snapshot, DtxContext distributedTransactionContext)
 
 	/*
 	 * It is sufficient to get shared lock on ProcArrayLock, even if we are
-	 * going to set MyProc->xmin.
+	 * going to set MyProc->xmin. A DTX whose QE commit notification has
+	 * started is different: wait until its QD cleanup completes before taking
+	 * the snapshot, otherwise some QEs can expose the commit while others do
+	 * not.
 	 */
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	if (distributedTransactionContext == DTX_CONTEXT_QD_DISTRIBUTED_CAPABLE &&
+		!Debug_disable_distributed_snapshot && needDistributedSnapshot)
+		WaitForDtxCommit();
+	else
+		LWLockAcquire(ProcArrayLock, LW_SHARED);
 
 	/*
 	* GPDB_14_MERGE_FIXME:
@@ -6671,4 +6722,4 @@ LoopBackendProc(BackendProcCallbackFunction func, void *args)
 		(*func)(proc, args);
 	}
 	LWLockRelease(ProcArrayLock);
-}
+}

src/include/cdb/cdbtm.h

@@ -229,6 +229,8 @@ typedef struct TMGXACT
 	DistributedTransactionId	xminDistributedSnapshot;
 
 	bool						includeInCkpt;
+	/* True after QE commit notification starts, until the DTX ends. */
+	bool						commitInProgress;
 	int							sessionId;
 }	TMGXACT;
 

src/test/isolation2/expected/issue1465_distributed_snapshot.out

@@ -0,0 +1,117 @@
+-- Verify that a distributed snapshot cannot retain a gxid after the QE has
+-- committed it.  Session 1 is opened before the fault is installed because
+-- the fault is scoped to the isolation2test database.
+CREATE TABLE issue1465_snapshot (k int, v int) DISTRIBUTED REPLICATED;
+CREATE
+INSERT INTO issue1465_snapshot VALUES (1, 400);
+INSERT 1
+
+1: SELECT 1;
+ ?column? 
+----------
+ 1        
+(1 row)
+1: BEGIN;
+BEGIN
+1: UPDATE issue1465_snapshot SET v = 500 WHERE k = 1;
+UPDATE 1
+-1U:@db_name postgres: SELECT gp_inject_fault('before_xact_end_procarray', 'suspend', '', 'isolation2test', '', 1, 1, 0, 1);
+ gp_inject_fault 
+-----------------
+ Success:        
+(1 row)
+1&: COMMIT;  <waiting ...>
+-1U:@db_name postgres: SELECT gp_wait_until_triggered_fault('before_xact_end_procarray', 1, 1);
+ gp_wait_until_triggered_fault 
+-------------------------------
+ Success:                      
+(1 row)
+
+2: BEGIN ISOLATION LEVEL REPEATABLE READ;
+BEGIN
+2&: SELECT k, v FROM issue1465_snapshot ORDER BY k;  <waiting ...>
+
+-1U:@db_name postgres: SELECT gp_inject_fault('before_xact_end_procarray', 'resume', 1);
+ gp_inject_fault 
+-----------------
+ Success:        
+(1 row)
+1<:  <... completed>
+COMMIT
+2<:  <... completed>
+ k | v   
+---+-----
+ 1 | 500 
+(1 row)
+-1U:@db_name postgres: SELECT gp_inject_fault('all', 'reset', 1);
+ gp_inject_fault 
+-----------------
+ Success:        
+(1 row)
+
+-- The same snapshot must retain the committed version after the writer exits.
+2: SELECT k, v FROM issue1465_snapshot ORDER BY k;
+ k | v   
+---+-----
+ 1 | 500 
+(1 row)
+2: ROLLBACK;
+ROLLBACK
+
+DROP TABLE issue1465_snapshot;
+DROP
+
+-- The one-phase path also notifies a QE before QD transaction cleanup.
+CREATE TABLE issue1465_onephase (k int, v int) DISTRIBUTED BY (k);
+CREATE
+INSERT INTO issue1465_onephase VALUES (1, 400);
+INSERT 1
+
+1: BEGIN;
+BEGIN
+1: UPDATE issue1465_onephase SET v = 500 WHERE k = 1;
+UPDATE 1
+-1U:@db_name postgres: SELECT gp_inject_fault('before_xact_end_procarray', 'suspend', '', 'isolation2test', '', 1, 1, 0, 1);
+ gp_inject_fault 
+-----------------
+ Success:        
+(1 row)
+1&: COMMIT;  <waiting ...>
+-1U:@db_name postgres: SELECT gp_wait_until_triggered_fault('before_xact_end_procarray', 1, 1);
+ gp_wait_until_triggered_fault 
+-------------------------------
+ Success:                      
+(1 row)
+
+2: BEGIN ISOLATION LEVEL REPEATABLE READ;
+BEGIN
+2&: SELECT k, v FROM issue1465_onephase WHERE k = 1;  <waiting ...>
+
+-1U:@db_name postgres: SELECT gp_inject_fault('before_xact_end_procarray', 'resume', 1);
+ gp_inject_fault 
+-----------------
+ Success:        
+(1 row)
+1<:  <... completed>
+COMMIT
+2<:  <... completed>
+ k | v   
+---+-----
+ 1 | 500 
+(1 row)
+-1U:@db_name postgres: SELECT gp_inject_fault('all', 'reset', 1);
+ gp_inject_fault 
+-----------------
+ Success:        
+(1 row)
+
+2: SELECT k, v FROM issue1465_onephase WHERE k = 1;
+ k | v   
+---+-----
+ 1 | 500 
+(1 row)
+2: ROLLBACK;
+ROLLBACK
+
+DROP TABLE issue1465_onephase;
+DROP

src/test/isolation2/isolation2_schedule

@@ -76,6 +76,7 @@ test: instr_in_shmem_terminate
 test: vacuum_recently_dead_tuple_due_to_distributed_snapshot
 test: vacuum_full_interrupt
 test: distributedlog-bug
+test: issue1465_distributed_snapshot
 test: invalidated_toast_index
 test: distributed_snapshot
 test: gp_collation

src/test/isolation2/sql/issue1465_distributed_snapshot.sql

@@ -0,0 +1,49 @@
+-- Verify that a distributed snapshot cannot retain a gxid after the QE has
+-- committed it.  Session 1 is opened before the fault is installed because
+-- the fault is scoped to the isolation2test database.
+CREATE TABLE issue1465_snapshot (k int, v int) DISTRIBUTED REPLICATED;
+INSERT INTO issue1465_snapshot VALUES (1, 400);
+
+1: SELECT 1;
+1: BEGIN;
+1: UPDATE issue1465_snapshot SET v = 500 WHERE k = 1;
+-1U:@db_name postgres: SELECT gp_inject_fault('before_xact_end_procarray', 'suspend', '', 'isolation2test', '', 1, 1, 0, 1);
+1&: COMMIT;
+-1U:@db_name postgres: SELECT gp_wait_until_triggered_fault('before_xact_end_procarray', 1, 1);
+
+2: BEGIN ISOLATION LEVEL REPEATABLE READ;
+2&: SELECT k, v FROM issue1465_snapshot ORDER BY k;
+
+-1U:@db_name postgres: SELECT gp_inject_fault('before_xact_end_procarray', 'resume', 1);
+1<:
+2<:
+-1U:@db_name postgres: SELECT gp_inject_fault('all', 'reset', 1);
+
+-- The same snapshot must retain the committed version after the writer exits.
+2: SELECT k, v FROM issue1465_snapshot ORDER BY k;
+2: ROLLBACK;
+
+DROP TABLE issue1465_snapshot;
+
+-- The one-phase path also notifies a QE before QD transaction cleanup.
+CREATE TABLE issue1465_onephase (k int, v int) DISTRIBUTED BY (k);
+INSERT INTO issue1465_onephase VALUES (1, 400);
+
+1: BEGIN;
+1: UPDATE issue1465_onephase SET v = 500 WHERE k = 1;
+-1U:@db_name postgres: SELECT gp_inject_fault('before_xact_end_procarray', 'suspend', '', 'isolation2test', '', 1, 1, 0, 1);
+1&: COMMIT;
+-1U:@db_name postgres: SELECT gp_wait_until_triggered_fault('before_xact_end_procarray', 1, 1);
+
+2: BEGIN ISOLATION LEVEL REPEATABLE READ;
+2&: SELECT k, v FROM issue1465_onephase WHERE k = 1;
+
+-1U:@db_name postgres: SELECT gp_inject_fault('before_xact_end_procarray', 'resume', 1);
+1<:
+2<:
+-1U:@db_name postgres: SELECT gp_inject_fault('all', 'reset', 1);
+
+2: SELECT k, v FROM issue1465_onephase WHERE k = 1;
+2: ROLLBACK;
+
+DROP TABLE issue1465_onephase;