diff --git a/src/backend/cdb/cdbtm.c b/src/backend/cdb/cdbtm.c index 5733b21cadb..1fc3ac44596 100644 --- a/src/backend/cdb/cdbtm.c +++ b/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; diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index 7a3078c5ca6..9f634383192 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/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); -} \ No newline at end of file +} diff --git a/src/include/cdb/cdbtm.h b/src/include/cdb/cdbtm.h index dc05e1a79dd..116c1da9f7a 100644 --- a/src/include/cdb/cdbtm.h +++ b/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; diff --git a/src/test/isolation2/expected/issue1465_distributed_snapshot.out b/src/test/isolation2/expected/issue1465_distributed_snapshot.out new file mode 100644 index 00000000000..d342bef2e5d --- /dev/null +++ b/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; +-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; + +-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; +-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; + +-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 diff --git a/src/test/isolation2/isolation2_schedule b/src/test/isolation2/isolation2_schedule index 82d23731fb6..f3989e0dde4 100644 --- a/src/test/isolation2/isolation2_schedule +++ b/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 diff --git a/src/test/isolation2/sql/issue1465_distributed_snapshot.sql b/src/test/isolation2/sql/issue1465_distributed_snapshot.sql new file mode 100644 index 00000000000..989d325623c --- /dev/null +++ b/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;