Fix distributed snapshot race during commit

contrib/pax_storage/src/test/isolation2/expected/commit_transaction_block_checkpoint.out

@@ -1,5 +1,8 @@
 -- TEST 1: block checkpoint on segments
 
+SET debug_disable_distributed_snapshot = on;
+SET
+
 -- pause the 2PC after setting inCommit flag
 select gp_inject_fault_infinite('before_xlog_xact_commit_prepared', 'suspend', 3);
  gp_inject_fault_infinite 

contrib/pax_storage/src/test/isolation2/expected/prepare_limit.out

@@ -32,6 +32,19 @@ CREATE
  Success:                 
 (1 row)
 
+-- This test uses a suspended commit as a test barrier.  Keep the concurrent
+-- writers from waiting for that barrier while obtaining their snapshots.
+1: SET debug_disable_distributed_snapshot = on;
+SET
+2: SET debug_disable_distributed_snapshot = on;
+SET
+3: SET debug_disable_distributed_snapshot = on;
+SET
+4: SET debug_disable_distributed_snapshot = on;
+SET
+5: SET debug_disable_distributed_snapshot = on;
+SET
+
 -- Note first insert after table create triggers auto_stats and leads to 2pc
 -- transaction.
 

contrib/pax_storage/src/test/isolation2/expected/segwalrep/cancel_commit_pending_replication.out

@@ -9,6 +9,10 @@ select gp_inject_fault('wal_sender_loop', 'reset', 2);
  Success:        
 (1 row)
 
+-- This session coordinates a suspended commit rather than testing snapshots.
+SET debug_disable_distributed_snapshot = on;
+SET
+
 create or replace function wait_for_replication(iterations int) returns bool as $$ begin /* in func */ for i in 1 .. iterations loop /* in func */ if exists (select wait_event from pg_stat_activity where sess_id in (select sess_id from store_session_id) and wait_event = 'SyncRep') then /* in func */ return true; /* in func */ end if; /* in func */ perform pg_sleep(0.1); /* in func */ perform pg_stat_clear_snapshot(); /* in func */ end loop; /* in func */ return false; /* in func */ end; /* in func */ $$ language plpgsql VOLATILE;
 CREATE
 

contrib/pax_storage/src/test/isolation2/expected/segwalrep/die_commit_pending_replication.out

@@ -15,6 +15,9 @@ CREATE
 
 -- Suspend to hit commit-prepared point on segment (as we are
 -- interested in testing Commit here and not really Prepare)
+-- The following waits are test coordination, not snapshot coverage.
+SET debug_disable_distributed_snapshot = on;
+SET
 select gp_inject_fault_infinite('finish_prepared_start_of_function', 'suspend', dbid) from gp_segment_configuration where role='p' and content = 0;
  gp_inject_fault_infinite 
 --------------------------

contrib/pax_storage/src/test/isolation2/expected/segwalrep/dtx_recovery_wait_lsn.out

@@ -8,6 +8,12 @@
 1: create table t_wait_lsn(a int);
 CREATE
 
+-- These sessions coordinate suspended commits rather than test snapshots.
+2: SET debug_disable_distributed_snapshot = on;
+SET
+3: SET debug_disable_distributed_snapshot = on;
+SET
+
 -- suspend segment 0 before performing 'COMMIT PREPARED'
 2: select gp_inject_fault_infinite('finish_prepared_start_of_function', 'suspend', dbid) from gp_segment_configuration where content=0 and role='p';
  gp_inject_fault_infinite 

contrib/pax_storage/src/test/isolation2/expected/segwalrep/twophase_tolerance_with_mirror_promotion.out

@@ -182,6 +182,8 @@ INSERT 1
 
 -- Scenario 3: Commit-Prepare received on primary but not acknowledged to master
 -- NOTICE: Don't use session 2 again because it's cached gang is invalid
+4: SET debug_disable_distributed_snapshot = on;
+SET
 1:SELECT gp_inject_fault_infinite('finish_prepared_start_of_function', 'infinite_loop', dbid) FROM gp_segment_configuration WHERE content = 2 AND role = 'p';
  gp_inject_fault_infinite 
 --------------------------

contrib/pax_storage/src/test/isolation2/sql/commit_transaction_block_checkpoint.sql

@@ -1,5 +1,7 @@
 -- TEST 1: block checkpoint on segments
 
+SET debug_disable_distributed_snapshot = on;
+
 -- pause the 2PC after setting inCommit flag
 select gp_inject_fault_infinite('before_xlog_xact_commit_prepared', 'suspend', 3);
 

contrib/pax_storage/src/test/isolation2/sql/prepare_limit.sql

@@ -17,6 +17,14 @@
 
 5: select gp_inject_fault_infinite('dtm_before_insert_forget_comitted', 'suspend', 1);
 
+-- This test uses a suspended commit as a test barrier.  Keep the concurrent
+-- writers from waiting for that barrier while obtaining their snapshots.
+1: SET debug_disable_distributed_snapshot = on;
+2: SET debug_disable_distributed_snapshot = on;
+3: SET debug_disable_distributed_snapshot = on;
+4: SET debug_disable_distributed_snapshot = on;
+5: SET debug_disable_distributed_snapshot = on;
+
 -- Note first insert after table create triggers auto_stats and leads to 2pc
 -- transaction.
 

contrib/pax_storage/src/test/isolation2/sql/segwalrep/cancel_commit_pending_replication.sql

@@ -1,6 +1,9 @@
 select gp_inject_fault('sync_rep_query_cancel', 'reset', 2);
 select gp_inject_fault('wal_sender_loop', 'reset', 2);
 
+-- This session coordinates a suspended commit rather than testing snapshots.
+SET debug_disable_distributed_snapshot = on;
+
 create or replace function wait_for_replication(iterations int)
 returns bool as $$
 begin /* in func */

contrib/pax_storage/src/test/isolation2/sql/segwalrep/die_commit_pending_replication.sql

@@ -11,6 +11,8 @@ create table store_session_id(a int, sess_id int);
 
 -- Suspend to hit commit-prepared point on segment (as we are
 -- interested in testing Commit here and not really Prepare)
+-- The following waits are test coordination, not snapshot coverage.
+SET debug_disable_distributed_snapshot = on;
 select gp_inject_fault_infinite('finish_prepared_start_of_function', 'suspend', dbid) from gp_segment_configuration where role='p' and content = 0;
 1&: insert into die_commit_pending_replication values(2),(1);
 select gp_wait_until_triggered_fault('finish_prepared_start_of_function', 1, dbid) from gp_segment_configuration where role='p' and content = 0;

contrib/pax_storage/src/test/isolation2/sql/segwalrep/dtx_recovery_wait_lsn.sql

@@ -7,6 +7,10 @@
 
 1: create table t_wait_lsn(a int);
 
+-- These sessions coordinate suspended commits rather than test snapshots.
+2: SET debug_disable_distributed_snapshot = on;
+3: SET debug_disable_distributed_snapshot = on;
+
 -- suspend segment 0 before performing 'COMMIT PREPARED'
 2: select gp_inject_fault_infinite('finish_prepared_start_of_function', 'suspend', dbid) from gp_segment_configuration where content=0 and role='p';
 1&: insert into t_wait_lsn values(2),(1);

contrib/pax_storage/src/test/isolation2/sql/segwalrep/twophase_tolerance_with_mirror_promotion.sql

@@ -81,6 +81,7 @@
 
 -- Scenario 3: Commit-Prepare received on primary but not acknowledged to master
 -- NOTICE: Don't use session 2 again because it's cached gang is invalid
+4: SET debug_disable_distributed_snapshot = on;
 1:SELECT gp_inject_fault_infinite('finish_prepared_start_of_function', 'infinite_loop', dbid)
   FROM gp_segment_configuration WHERE content = 2 AND role = 'p';
 1&:DROP TABLE tolerance_test_table;

gpMgmt/bin/gppylib/commands/gp.py

@@ -1310,9 +1310,14 @@ def _get_phase2_status(self):
 
         try:
             dburl = dbconn.DbURL(dbname=self.dbname)
-            with closing(dbconn.connect(dburl, encoding='UTF8')) as conn:
+            # This catalog-only existence check must remain available while a
+            # distributed transaction is completing.  The status table itself
+            # is distributed, so read it through the normal dispatcher below.
+            with closing(dbconn.connect(dburl, utility=True, encoding='UTF8')) as conn:
                 if not dbconn.querySingleton(conn, status_table_exists_sql):
                     return False
+
+            with closing(dbconn.connect(dburl, encoding='UTF8')) as conn:
                 status = dbconn.querySingleton(conn, sql)
         except Exception:
             # schema table not found

gpMgmt/bin/gppylib/test/unit/test_unit_gpexpand_status.py

@@ -6,6 +6,8 @@
 
 import os
 
+from mock import call, patch
+
 from gppylib.test.unit.gp_unittest import GpTestCase, run_tests
 from gppylib.commands import gp
 from gppylib.db import dbconn
@@ -19,6 +21,23 @@ class Context(object):
 
 ctx = Context()
 
+class GpExpandStatusConnectionMode(GpTestCase):
+
+    @patch('gppylib.commands.gp.dbconn.connect')
+    @patch('gppylib.commands.gp.dbconn.DbURL')
+    @patch('gppylib.commands.gp.dbconn.querySingleton', side_effect=[1, 'SETUP DONE'])
+    def test_phase2_catalog_check_uses_utility_connection(self, query_singleton, dburl_cls, connect):
+        status = gp._GpExpandStatus()
+        status.dbname = 'testdb'
+
+        self.assertTrue(status._get_phase2_status())
+        self.assertEqual(status.status, 'SETUP DONE')
+        dburl_cls.assert_called_once_with(dbname='testdb')
+        self.assertEqual(connect.call_args_list, [
+            call(dburl_cls.return_value, utility=True, encoding='UTF8'),
+            call(dburl_cls.return_value, encoding='UTF8'),
+        ])
+
 def get_gpexpand_status():
     st = gp.get_gpexpand_status()
     st.dbname = ctx.dbname

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,15 +605,16 @@ 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);
 
+	SIMPLE_FAULT_INJECTOR("dtm_broadcast_commit_prepared");
+	markDtxCommitInProgress();
+
 	savedInterruptHoldoffCount = InterruptHoldoffCount;
 
 	Assert(MyTmGxactLocal->dtxSegments != NIL);
@@ -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"
@@ -75,6 +76,7 @@
 #include "access/xact.h"		/* setting the shared xid */
 #include "cdb/cdbtm.h"
 #include "cdb/cdbvars.h"
+#include "postmaster/fts.h"
 #include "utils/faultinjector.h"
 #include "utils/sharedsnapshot.h"
 #include "libpq/libpq-be.h"
@@ -698,6 +700,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 +2628,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 +3065,22 @@ 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.
+	 *
+	 * FTS only reads coordinator configuration and probes segments through
+	 * libpq.  It must not wait for a user DTX here, because the probe can be
+	 * needed to resolve that DTX.
 	 */
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	if (distributedTransactionContext == DTX_CONTEXT_QD_DISTRIBUTED_CAPABLE &&
+		Gp_role != GP_ROLE_UTILITY &&
+		!Debug_disable_distributed_snapshot && !am_ftsprobe &&
+		needDistributedSnapshot)
+		WaitForDtxCommit();
+	else
+		LWLockAcquire(ProcArrayLock, LW_SHARED);
 
 	/*
 	* GPDB_14_MERGE_FIXME:
@@ -3283,9 +3341,14 @@ GetSnapshotData(Snapshot snapshot, DtxContext distributedTransactionContext)
 	if (!TransactionIdIsValid(MyProc->xmin))
 		MyProc->xmin = TransactionXmin = xmin;
 
-	/* GP: QD takes a distributed snapshot iff QD not in retry phase and the query needs distributed snapshot */
-	if (distributedTransactionContext == DTX_CONTEXT_QD_DISTRIBUTED_CAPABLE && !Debug_disable_distributed_snapshot 
-			&& needDistributedSnapshot)
+	/*
+	 * GP: A QD takes a distributed snapshot iff it is not in retry phase and
+	 * the query needs one.  FTS only reads coordinator configuration while it
+	 * probes segments, so it does not need a distributed snapshot.
+	 */
+	if (distributedTransactionContext == DTX_CONTEXT_QD_DISTRIBUTED_CAPABLE &&
+		!Debug_disable_distributed_snapshot && !am_ftsprobe &&
+		needDistributedSnapshot)
 	{
 		CreateDistributedSnapshot(ds);
 		snapshot->haveDistribSnapshot = true;
@@ -6671,4 +6734,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/Makefile

@@ -78,6 +78,9 @@ installcheck-expandshrink: install
 installcheck-isolation2: install
 	$(pg_isolation2_regress_installcheck) --init-file=$(top_builddir)/src/test/regress/init_file --init-file=./init_file_isolation2 --schedule=$(srcdir)/isolation2_schedule
 
+installcheck-distributed-snapshot: install
+	$(pg_isolation2_regress_installcheck) --init-file=$(top_builddir)/src/test/regress/init_file --init-file=./init_file_isolation2 issue1465_distributed_snapshot
+
 installcheck-isolation2-crash: install
 	$(pg_isolation2_regress_installcheck) --init-file=$(top_builddir)/src/test/regress/init_file --init-file=./init_file_isolation2 --schedule=$(srcdir)/isolation2_crash_schedule