From ad9197b6bd4d60c227b3ca818177a1b2dc721969 Mon Sep 17 00:00:00 2001
From: Pavan Deolasee
Date: Mon, 15 Feb 2016 18:26:13 +0530
Subject: Handle errors while PREPARing a transaction gracefully.

If an error occurs after PREPARE TRANSACTION command is sent, we don't know if
the command is successful or not. So the coordinator will go ahead and abort
the transaction. But the node which failed to run PREPARE TRANSACTION may not
be even reachable on the coordinator. So we don't try to rollback transaction
on such nodes.
---
 src/backend/access/transam/xact.c   | 26 ++++++++++++++++++++++----
 src/backend/pgxc/pool/execRemote.c  | 26 ++++++++++++++++++++++++--
 src/backend/storage/ipc/procarray.c |  6 +++++-
 3 files changed, 51 insertions(+), 7 deletions(-)

diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 9b9a4c8409..0b0662bf7d 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -2463,7 +2463,10 @@ CommitTransaction(void)
 		 * We need to mark our XIDs as committed in pg_clog.  This is where we
 		 * durably commit.
 		 */
-		latestXid = RecordTransactionCommit();
+#ifdef XCP
+		if (!IsConnFromDatanode())
+#endif
+			latestXid = RecordTransactionCommit();
 	}
 	else
 	{
@@ -3236,7 +3239,12 @@ AbortTransaction(void)
 	 * record.
 	 */
 	if (!is_parallel_worker)
-		latestXid = RecordTransactionAbort(false);
+	{
+#ifdef XCP
+		if (!IsConnFromDatanode())
+#endif
+			latestXid = RecordTransactionAbort(false);
+	}
 	else
 	{
 		latestXid = InvalidTransactionId;
@@ -6766,8 +6774,18 @@ SetTopTransactionId(GlobalTransactionId xid)
 	TransactionState s = CurrentTransactionState;
 	Assert(!GlobalTransactionIdIsValid(s->transactionId) ||
 			GlobalTransactionIdEquals(s->transactionId, xid));
-	XactTopTransactionId = s->transactionId = xid;
-	elog(DEBUG2, "Assigning XID received from the remote node - %d", xid);
+
+	if (!IsConnFromDatanode())
+	{
+		XactTopTransactionId = s->transactionId = xid;
+		elog(DEBUG2, "Assigning XID received from the remote node - %d", xid);
+	}
+	else if (!TransactionIdIsValid(GetNextTransactionId()))
+	{
+		SetNextTransactionId(xid);
+		if (whereToSendOutput == DestRemote)
+			pq_putmessage('x', (const char *) &xid, sizeof (GlobalTransactionId));
+	}
 }
 #endif
 #endif
diff --git a/src/backend/pgxc/pool/execRemote.c b/src/backend/pgxc/pool/execRemote.c
index 913563b5e1..9657ab0157 100644
--- a/src/backend/pgxc/pool/execRemote.c
+++ b/src/backend/pgxc/pool/execRemote.c
@@ -2319,9 +2319,20 @@ prepare_err:
 		if (conn->ck_resp_rollback)
 		{
 			conn->ck_resp_rollback = false;
+
+			if (conn->state != DN_CONNECTION_STATE_IDLE)
+			{
+				ereport(WARNING,
+						(errcode(ERRCODE_INTERNAL_ERROR),
+						 errmsg("Error while PREPARING transaction %s on "
+							 "node %s. Administrative action may be required "
+							 "to abort this transaction on the node",
+							 prepareGID, conn->nodename)));
+				continue;
+			}
+
 			/* sanity checks */
 			Assert(conn->sock != NO_SOCKET);
-			Assert(conn->state == DN_CONNECTION_STATE_IDLE);
 			/* Send down abort prepared command */
 			if (pgxc_node_send_gxid(conn, auxXid))
 			{
@@ -2359,9 +2370,20 @@ prepare_err:
 		if (conn->ck_resp_rollback)
 		{
 			conn->ck_resp_rollback = false;
+
+			if (conn->state != DN_CONNECTION_STATE_IDLE)
+			{
+				ereport(WARNING,
+						(errcode(ERRCODE_INTERNAL_ERROR),
+						 errmsg("Error while PREPARING transaction %s on "
+							 "node %s. Administrative action may be required "
+							 "to abort this transaction on the node",
+							 prepareGID, conn->nodename)));
+				continue;
+			}
+
 			/* sanity checks */
 			Assert(conn->sock != NO_SOCKET);
-			Assert(conn->state == DN_CONNECTION_STATE_IDLE);
 			/* Send down abort prepared command */
 			if (pgxc_node_send_gxid(conn, auxXid))
 			{
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index 84b6a87d0d..b7fc75d52d 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -496,6 +496,10 @@ ProcArrayEndTransaction(PGPROC *proc, TransactionId latestXid)
 		 * anyone else's calculation of a snapshot.  We might change their
 		 * estimate of global xmin, but that's OK.
 		 */
+#ifdef XCP
+		if (IsConnFromDatanode())
+			allPgXact[proc->pgprocno].xid = InvalidTransactionId;
+#endif
 		Assert(!TransactionIdIsValid(allPgXact[proc->pgprocno].xid));
 
 		proc->lxid = InvalidLocalTransactionId;
@@ -4371,7 +4375,7 @@ KnownAssignedXidsReset(void)
 void
 ProcArrayCheckXminConsistency(TransactionId global_xmin)
 {
-	ProcArrayStruct *arrayP = procArray;
+	volatile ProcArrayStruct *arrayP = procArray;
 	int			index;
 
 	for (index = 0; index < arrayP->numProcs; index++)
-- 
cgit v1.2.3