Update 2PC support

Author: knizhnik

contrib/mmts/multimaster.c

@@ -1116,6 +1116,11 @@ MtmGetState(void)
 	return dtm;
 }
 
+TransactionId MtmGetCurrentTransaction(void)
+{
+	return dtmTx.xid;
+}
+
 static void 
 MtmShmemStartup(void)
 {

contrib/mmts/multimaster.h

@@ -157,6 +157,7 @@ extern void  MtmSleep(timestamp_t interval);
 extern void  MtmSetCurrentTransactionGID(char const* gid);
 extern void  MtmSetCurrentTransactionCSN(csn_t csn);
 extern csn_t MtmGetTransactionCSN(TransactionId xid);
+extern TransactionId MtmGetCurrentTransactionId(void);
 extern bool  MtmIsRecoveredNode(int nodeId);
 extern void  MtmRefreshClusterStatus(bool nowait);
 extern void  MtmSwitchClusterMode(MtmNodeStatus mode);

contrib/mmts/pglogical_apply.c

@@ -484,12 +484,15 @@ process_remote_commit(StringInfo in)
 	{
 		case PGLOGICAL_COMMIT:
 		{
-			if (IsTransactionState())
+			if (IsTransactionState()) {
+				Assert(TransactionIdIsValid(MtmGetCurrentTransactionId()));
 				CommitTransactionCommand();
+			}
 			break;
 		}
 		case PGLOGICAL_PREPARE:
 		{
+			Assert(IsTransactionState() && TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
 			/* prepare TBLOCK_INPROGRESS state for PrepareTransactionBlock() */
 			BeginTransactionBlock();
@@ -503,6 +506,7 @@ process_remote_commit(StringInfo in)
 		}
 		case PGLOGICAL_COMMIT_PREPARED:
 		{
+			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			csn = pq_getmsgint64(in);
 			gid = pq_getmsgstring(in);
 			StartTransactionCommand();
@@ -514,6 +518,7 @@ process_remote_commit(StringInfo in)
 		}
 		case PGLOGICAL_ABORT_PREPARED:
 		{
+			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
 			StartTransactionCommand();
 			MtmSetCurrentTransactionGID(gid);
@@ -889,8 +894,9 @@ void MtmExecutor(int id, void* work, size_t size)
     {
 		EmitErrorReport();
         FlushErrorState();
-		MTM_TRACE("%d: REMOTE abort transaction %d\n", MyProcPid, GetCurrentTransactionId());
+		MTM_TRACE("%d: REMOTE begin abort transaction %d\n", MyProcPid, MtmGetCurrentTransactionId());
         AbortCurrentTransaction();
+		MTM_TRACE("%d: REMOTE end abort transaction %d\n", MyProcPid, MtmGetCurrentTransactionId());
     }
     PG_END_TRY();