Fix bugs in 3PC implementation

Author: knizhnik

multimaster.c

@@ -1311,6 +1311,7 @@ void MtmSend2PCMessage(MtmTransState* ts, MtmMessageCode cmd)
 
 	if (MtmIsCoordinator(ts)) {
 		int i;
+		Assert(false); // All broadcasts are now done through logical decoding
 		for (i = 0; i < Mtm->nAllNodes; i++)
 		{
 			if (BIT_CHECK(ts->participantsMask & ~Mtm->disabledNodeMask & ~ts->votedMask, i))
@@ -3228,9 +3229,10 @@ bool MtmFilterTransaction(char* record, int size)
 		   origin_node != 0 &&
 		   (Mtm->status == MTM_RECOVERY || origin_node == replication_node));	
 
-    switch(PGLOGICAL_XACT_EVENT(flags))
+    switch (PGLOGICAL_XACT_EVENT(flags))
     {
 	  case PGLOGICAL_PREPARE:
+	  case PGLOGICAL_PRECOMMIT_PREPARED:
 	  case PGLOGICAL_ABORT_PREPARED:
 		gid = pq_getmsgstring(&s);
 		break;

multimaster.h

@@ -39,7 +39,7 @@
 #define MTM_TXTRACE(tx, event)
 #else
 #define MTM_TXTRACE(tx, event) \
-		fprintf(stderr, "[MTM_TXTRACE], %s, %lld, %s, %d\n", tx->gid, (long long)MtmGetSystemTime(), event, getpid())
+		fprintf(stderr, "[MTM_TXTRACE], %s, %lld, %s, %d\n", tx->gid, (long long)MtmGetSystemTime(), event, MyProcPid)
 #endif
 
 #define MULTIMASTER_NAME                "multimaster"

pglogical_apply.c

@@ -625,7 +625,9 @@ process_remote_commit(StringInfo in)
 	{
 	    case PGLOGICAL_PRECOMMIT_PREPARED:
 		{
+			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
+			MTM_LOG2("%d: PGLOGICAL_PRECOMMIT_PREPARED %s", MyProcPid, gid);
 			MtmPrecommitTransaction(gid);
 			return;
 		}

pglogical_proto.c

@@ -192,7 +192,8 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 	else
     	Assert(false);
 
-	if (flags == PGLOGICAL_COMMIT || flags == PGLOGICAL_PREPARE || flags == PGLOGICAL_PRECOMMIT_PREPARED) { 
+	if (flags == PGLOGICAL_COMMIT || flags == PGLOGICAL_PREPARE) {
+		/* COMMIT and PREPARE are preceded by BEGIN, which set MtmIsFilteredTxn flag */
 		if (MtmIsFilteredTxn) { 
 			Assert(MtmTransactionRecords == 0);
 			return;