From 8967aa442acc84cfd28848329c3ad8ea4ec3b792 Mon Sep 17 00:00:00 2001 From: zyf-gpu1 <205618093@qq.com> Date: Wed, 5 Aug 2026 16:37:04 +0000 Subject: [PATCH 1/2] Fix distributed snapshot race during DTX commit Keep the QD GXID in the proc array through normal transaction cleanup and mark DTX commits while QE notifications are in progress. QD snapshot creation waits on the existing GXID lock until the coordinator and all QEs have a coherent commit view. Retain the existing checkpoint and FORGET COMMITTED lifecycle, and add deterministic isolation coverage for two-phase and one-phase commit paths. Fixes #1465 --- src/backend/cdb/cdbtm.c | 24 +++- src/backend/storage/ipc/procarray.c | 52 +++++++- src/include/cdb/cdbtm.h | 2 + .../issue1465_distributed_snapshot.out | 117 ++++++++++++++++++ src/test/isolation2/isolation2_schedule | 1 + .../sql/issue1465_distributed_snapshot.sql | 49 ++++++++ 6 files changed, 241 insertions(+), 4 deletions(-) create mode 100644 src/test/isolation2/expected/issue1465_distributed_snapshot.out create mode 100644 src/test/isolation2/sql/issue1465_distributed_snapshot.sql 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..41494f38d07 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,45 @@ 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]; + + if (tmGxact == MyTmGxact || + tmGxact->gxid == InvalidDistributedTransactionId || + !tmGxact->commitInProgress) + continue; + + waitGxid = tmGxact->gxid; + break; + } + + if (waitGxid == InvalidDistributedTransactionId) + break; + + LWLockRelease(ProcArrayLock); + GxactLockTableWait(waitGxid); + } +} + /* * create distributed snapshot based on current visible distributed transaction */ @@ -3020,9 +3061,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: 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; From 4dc2724157746993281e31f992cb9b85f6e3fc52 Mon Sep 17 00:00:00 2001 From: zyf-gpu1 <205618093@qq.com> Date: Thu, 6 Aug 2026 06:53:03 +0000 Subject: [PATCH 2/2] Fix atomic GXID reads in snapshot waits Use pg_atomic_read_u64() when WaitForDtxCommit() inspects a DTX. The GXID is published through atomic_gxid, and a plain 64-bit read can tear on platforms where 64-bit accesses are not naturally atomic. Keep the snapshot wait aligned with the existing atomic GXID protocol. See: Issue#1465 --- src/backend/storage/ipc/procarray.c | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index 41494f38d07..9f634383192 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/src/backend/storage/ipc/procarray.c @@ -2649,12 +2649,15 @@ WaitForDtxCommit(void) int pgprocno = procArray->pgprocnos[i]; volatile TMGXACT *tmGxact = &allTmGxact[pgprocno]; + DistributedTransactionId gxid = + pg_atomic_read_u64(&tmGxact->atomic_gxid); + if (tmGxact == MyTmGxact || - tmGxact->gxid == InvalidDistributedTransactionId || + gxid == InvalidDistributedTransactionId || !tmGxact->commitInProgress) continue; - waitGxid = tmGxact->gxid; + waitGxid = gxid; break; } @@ -6719,4 +6722,4 @@ LoopBackendProc(BackendProcCallbackFunction func, void *args) (*func)(proc, args); } LWLockRelease(ProcArrayLock); -} \ No newline at end of file +}