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.

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++)