Fix a recursive AbortTransaction issue

When the error happens after ProcArrayEndTransaction, it will recurse back to
AbortTransaction, we need to make sure it will not generate extra WAL record
and not fail the assertions.

gpMgmt/bin/gppylib/programs/clsInjectFault.py

@@ -438,6 +438,7 @@ class GpInjectFaultProgram:
                   "finish_prepared_after_record_commit_prepared (inject fault in FinishPreparedTransaction() after recording the commit prepared record), " \
                   "resgroup_assigned_on_master (inject fault in AssignResGroupOnMaster() after slot is assigned), " \
                   "copy_from_high_processed (inject fault to pretend copying from very high number of processed rows), " \
+                  "abort_after_procarray_end (inject fault in AbortTransaction after ProcArrayEndTransaction), " \
 			      "all (affects all faults injected, used for 'status' and 'reset'), ") 
         addTo.add_option("-c", "--ddl_statement", dest="ddlStatement", type="string",
                          metavar="ddlStatement",

src/backend/access/transam/xact.c

@@ -1762,7 +1762,9 @@ RecordTransactionAbort(bool isSubXact)
 	 * wrote a commit record for it, there's no turning back. The Distributed
 	 * Transaction Manager will take care of completing the transaction for us.
 	 */
-	if (isQEReader || getCurrentDtxState() == DTX_STATE_NOTIFYING_COMMIT_PREPARED)
+	if (isQEReader ||
+		getCurrentDtxState() == DTX_STATE_NOTIFYING_COMMIT_PREPARED ||
+		MyProc->localDistribXactData.state == LOCALDISTRIBXACT_STATE_ABORTED)
 		xid = InvalidTransactionId;
 	else
 		xid = GetCurrentTransactionIdIfAny();
@@ -2315,6 +2317,12 @@ StartTransaction(void)
 		case DTX_CONTEXT_QD_RETRY_PHASE_2:
 		case DTX_CONTEXT_QE_FINISH_PREPARED:
 		{
+			if (DistributedTransactionContext == DTX_CONTEXT_LOCAL_ONLY &&
+				Gp_role == GP_ROLE_UTILITY)
+			{
+				LocalDistribXactData *ele = &MyProc->localDistribXactData;
+				ele->state = LOCALDISTRIBXACT_STATE_ACTIVE;
+			}
 			/*
 			 * MPP: we're in utility-mode or a QE starting a pure-local
 			 * transaction without any synchronization to segmates!
@@ -2345,6 +2353,8 @@ StartTransaction(void)
 						(errmsg("setting SharedLocalSnapshotSlot->startTimestamp = " INT64_FORMAT "[old=" INT64_FORMAT "])",
 								stmtStartTimestamp, oldStartTimestamp)));
 			}
+			LocalDistribXactData *ele = &MyProc->localDistribXactData;
+			ele->state = LOCALDISTRIBXACT_STATE_ACTIVE;
 		}
 		break;
 
@@ -2804,6 +2814,8 @@ CommitTransaction(void)
 	MyProc->inCommit = false;
 	MyProc->lxid = InvalidLocalTransactionId;
 
+	MyProc->localDistribXactData.state = LOCALDISTRIBXACT_STATE_NONE;
+
 	AtEOXact_MultiXact();
 
 	ResourceOwnerRelease(TopTransactionResourceOwner,
@@ -3286,6 +3298,7 @@ AbortTransaction(void)
 	 */
 	ProcArrayEndTransaction(MyProc, latestXid, false);
 
+	SIMPLE_FAULT_INJECTOR(AbortAfterProcarrayEnd);
 	/*
 	 * Post-abort cleanup.	See notes in CommitTransaction() concerning
 	 * ordering.  We can skip all of it if the transaction failed before

src/backend/cdb/cdblocaldistribxact.c

@@ -85,7 +85,8 @@ LocalDistribXact_ChangeState(PGPROC *proc,
 		break;
 
 	case LOCALDISTRIBXACT_STATE_COMMITTED:
-		if (oldState != LOCALDISTRIBXACT_STATE_ACTIVE)
+		if (oldState != LOCALDISTRIBXACT_STATE_ACTIVE &&
+			oldState != LOCALDISTRIBXACT_STATE_PREPARED)
 			elog(PANIC,
 			     "Expected distributed transaction xid = %u to local element to be in state \"Active\" or \"Commit Delivery\" and "
 			     "found state \"%s\"",
@@ -94,7 +95,8 @@ LocalDistribXact_ChangeState(PGPROC *proc,
 		break;
 
 	case LOCALDISTRIBXACT_STATE_ABORTED:
-		if (oldState != LOCALDISTRIBXACT_STATE_ACTIVE)
+		if (oldState != LOCALDISTRIBXACT_STATE_ACTIVE &&
+			oldState != LOCALDISTRIBXACT_STATE_ABORTED)
 			elog(PANIC,
 			     "Expected distributed transaction xid = %u to local element to be in state \"Active\" or \"Abort Delivery\" and "
 			     "found state \"%s\"",

src/backend/storage/ipc/procarray.c

@@ -248,43 +248,6 @@ ProcArrayEndTransaction(PGPROC *proc, TransactionId latestXid, bool isCommit)
 {
 	bool needNotifyCommittedDtxTransaction;
 
-	/*
-	 * MyProc->localDistribXactData is only used for debugging purpose by
-	 * backend itself on segments only hence okay to modify without holding
-	 * the lock.
-	 */
-	if (MyProc->localDistribXactData.state != LOCALDISTRIBXACT_STATE_NONE)
-	{
-		switch (DistributedTransactionContext)
-		{
-			case DTX_CONTEXT_QE_TWO_PHASE_EXPLICIT_WRITER:
-			case DTX_CONTEXT_QE_TWO_PHASE_IMPLICIT_WRITER:
-			case DTX_CONTEXT_QE_AUTO_COMMIT_IMPLICIT:
-				LocalDistribXact_ChangeState(MyProc,
-											 isCommit ?
-											 LOCALDISTRIBXACT_STATE_COMMITTED :
-											 LOCALDISTRIBXACT_STATE_ABORTED);
-				break;
-
-			case DTX_CONTEXT_QE_READER:
-			case DTX_CONTEXT_QE_ENTRY_DB_SINGLETON:
-				// QD or QE Writer will handle it.
-				break;
-
-			case DTX_CONTEXT_QD_DISTRIBUTED_CAPABLE:
-			case DTX_CONTEXT_QD_RETRY_PHASE_2:
-			case DTX_CONTEXT_QE_PREPARED:
-			case DTX_CONTEXT_QE_FINISH_PREPARED:
-				elog(PANIC, "Unexpected distribute transaction context: '%s'",
-					 DtxContextToString(DistributedTransactionContext));
-				break;
-
-			default:
-				elog(PANIC, "Unrecognized DTX transaction context: %d",
-					 (int) DistributedTransactionContext);
-		}
-	}
-
 	if (isCommit && notifyCommittedDtxTransactionIsNeeded())
 		needNotifyCommittedDtxTransaction = true;
 	else
@@ -351,6 +314,43 @@ ProcArrayEndTransaction(PGPROC *proc, TransactionId latestXid, bool isCommit)
 		Assert(proc->subxids.overflowed == false);
 	}
 
+	/*
+	 * MyProc->localDistribXactData is used by backend itself hence okay to
+	 * modify without holding the lock.
+	 */
+	if (MyProc->localDistribXactData.state != LOCALDISTRIBXACT_STATE_NONE)
+	{
+		switch (DistributedTransactionContext)
+		{
+			case DTX_CONTEXT_QE_TWO_PHASE_EXPLICIT_WRITER:
+			case DTX_CONTEXT_QE_TWO_PHASE_IMPLICIT_WRITER:
+			case DTX_CONTEXT_QE_AUTO_COMMIT_IMPLICIT:
+			case DTX_CONTEXT_QD_DISTRIBUTED_CAPABLE:
+			case DTX_CONTEXT_QD_RETRY_PHASE_2:
+			case DTX_CONTEXT_LOCAL_ONLY:
+			case DTX_CONTEXT_QE_FINISH_PREPARED:
+				LocalDistribXact_ChangeState(MyProc,
+											 isCommit ?
+											 LOCALDISTRIBXACT_STATE_COMMITTED :
+											 LOCALDISTRIBXACT_STATE_ABORTED);
+				break;
+
+			case DTX_CONTEXT_QE_READER:
+			case DTX_CONTEXT_QE_ENTRY_DB_SINGLETON:
+				// QD or QE Writer will handle it.
+				break;
+
+			case DTX_CONTEXT_QE_PREPARED:
+				elog(PANIC, "Unexpected distribute transaction context: '%s'",
+					 DtxContextToString(DistributedTransactionContext));
+				break;
+
+			default:
+				elog(PANIC, "Unrecognized DTX transaction context: %d",
+					 (int) DistributedTransactionContext);
+		}
+	}
+
 	return needNotifyCommittedDtxTransaction;
 }
 
@@ -376,8 +376,6 @@ ProcArrayClearTransaction(PGPROC *proc, bool commit)
 	proc->xid = InvalidTransactionId;
 	proc->xmin = InvalidTransactionId;
 
-	proc->localDistribXactData.state = LOCALDISTRIBXACT_STATE_NONE;
-
 	/* redundant, but just in case */
 	proc->vacuumFlags &= ~PROC_VACUUM_STATE_MASK;
 	proc->serializableIsoLevel = false;

src/backend/utils/misc/faultinjector.c

@@ -371,6 +371,8 @@ FaultInjectorIdentifierEnumToString[] = {
 		/* inject fault inside dynamic index scan after context reset */
 	_("dynamic_index_scan_context_reset"),
 		/* inject fault when creating new TOAST tables, to modify the chunk size */
+	_("abort_after_procarray_end"),
+		/* inject fault in AbortTransaction after ProcArrayEndTransaction */
 	_("not recognized"),
 };
 
@@ -1082,6 +1084,7 @@ FaultInjector_NewHashEntry(
 			case DecreaseToastMaxChunkSize:
 			case ProcessStartupPacketFault:
 			case DynamicIndexScanContextReset:
+			case AbortAfterProcarrayEnd:
 
 				break;
 			default:

src/include/utils/faultinjector.h

@@ -252,6 +252,8 @@ typedef enum FaultInjectorIdentifier_e {
 
 	DynamicIndexScanContextReset,
 
+	AbortAfterProcarrayEnd,
+
 	/* INSERT has to be done before that line */
 	FaultInjectorIdMax,
 	

src/test/isolation2/expected/distributed_transactions.out

+-- Test error after ProcArrayEndTransaction
+
+CREATE EXTENSION IF NOT EXISTS gp_inject_fault;
+CREATE
+
+-- abort fail on QD
+SELECT gp_inject_fault_new( 'abort_after_procarray_end', 'error', 1);
+gp_inject_fault_new
+-------------------
+t                  
+(1 row)
+BEGIN;
+BEGIN
+CREATE TABLE test_xact_abort_failure(a int);
+CREATE
+ABORT;
+ERROR:  fault triggered, fault name:'abort_after_procarray_end' fault type:'error'
+SELECT gp_inject_fault_new( 'abort_after_procarray_end', 'reset', 1);
+gp_inject_fault_new
+-------------------
+t                  
+(1 row)
+
+-- abort fail on QE
+SELECT gp_inject_fault_new( 'abort_after_procarray_end', 'error', dbid) from gp_segment_configuration where role = 'p' and content = 0;
+gp_inject_fault_new
+-------------------
+t                  
+(1 row)
+BEGIN;
+BEGIN
+CREATE TABLE test_xact_abort_failure(a int);
+CREATE
+ABORT;
+ABORT
+SELECT gp_inject_fault_new( 'abort_after_procarray_end', 'reset', dbid) from gp_segment_configuration where role = 'p' and content = 0;
+gp_inject_fault_new
+-------------------
+t                  
+(1 row)
+
+-- abort fail in local transaction
+SELECT gp_inject_fault_new( 'abort_after_procarray_end', 'error', dbid) from gp_segment_configuration where role = 'p' and content = 0;
+gp_inject_fault_new
+-------------------
+t                  
+(1 row)
+2U: BEGIN;
+BEGIN
+2U: CREATE TABLE test_xact_abort_failure(a int);
+CREATE
+2U: ABORT;
+ERROR:  fault triggered, fault name:'abort_after_procarray_end' fault type:'error'
+SELECT gp_inject_fault_new( 'abort_after_procarray_end', 'reset', dbid) from gp_segment_configuration where role = 'p' and content = 0;
+gp_inject_fault_new
+-------------------
+t                  
+(1 row)

src/test/isolation2/isolation2_schedule

@@ -130,3 +130,5 @@ test: packcore
 
 # Cancel test
 test: cancel_plpython
+
+test: distributed_transactions

src/test/isolation2/sql/distributed_transactions.sql

+-- Test error after ProcArrayEndTransaction
+
+CREATE EXTENSION IF NOT EXISTS gp_inject_fault;
+
+-- abort fail on QD
+SELECT gp_inject_fault_new( 'abort_after_procarray_end', 'error', 1);
+BEGIN;
+CREATE TABLE test_xact_abort_failure(a int);
+ABORT;
+SELECT gp_inject_fault_new( 'abort_after_procarray_end', 'reset', 1);
+
+-- abort fail on QE
+SELECT gp_inject_fault_new( 'abort_after_procarray_end', 'error', dbid) from gp_segment_configuration where role = 'p' and content = 0;
+BEGIN;
+CREATE TABLE test_xact_abort_failure(a int);
+ABORT;
+SELECT gp_inject_fault_new( 'abort_after_procarray_end', 'reset', dbid) from gp_segment_configuration where role = 'p' and content = 0;
+
+-- abort fail in local transaction
+SELECT gp_inject_fault_new( 'abort_after_procarray_end', 'error', dbid) from gp_segment_configuration where role = 'p' and content = 0;
+2U: BEGIN;
+2U: CREATE TABLE test_xact_abort_failure(a int);
+2U: ABORT;
+SELECT gp_inject_fault_new( 'abort_after_procarray_end', 'reset', dbid) from gp_segment_configuration where role = 'p' and content = 0;