Merge branch 'twophase_decoding'

Author: kelvich

contrib/mmts/arbiter.c

@@ -555,7 +555,7 @@ static bool MtmRecovery()
             }
         }
     }
-	return recorvered;
+	return recovered;
 }
 #endif
 
@@ -602,8 +602,8 @@ static void MtmTransReceiver(Datum arg)
 		do { 
 			struct timeval tv;
 			events = inset;
-			tv.tv_sec = MtmKeepAliveTimeout/USEC;
-			tv.tv_usec = MtmKeepAliveTimeout%USEC;
+			tv.tv_sec = MtmKeepaliveTimeout/USEC;
+			tv.tv_usec = MtmKeepaliveTimeout%USEC;
 			n = select(max_fd+1, &events, NULL, NULL, &tv);
 		} while (n < 0 && MtmRecovery());
 

contrib/mmts/multimaster.h

@@ -29,6 +29,13 @@
 typedef uint64 csn_t; /* commit serial number */
 #define INVALID_CSN  ((csn_t)-1)
 
+#define PGLOGICAL_COMMIT			0x00
+#define PGLOGICAL_PREPARE			0x01
+#define PGLOGICAL_COMMIT_PREPARED	0x02
+#define PGLOGICAL_ABORT_PREPARED	0x03
+
+#define PGLOGICAL_XACT_EVENT(flags)	(flags & 0x03)
+
 typedef uint64 timestamp_t;
 
 /* Identifier of global transaction */

contrib/mmts/pglogical_apply.c

@@ -464,9 +464,61 @@ read_rel(StringInfo s, LOCKMODE mode)
 }
 
 static void
-process_remote_commit(StringInfo s)
+process_remote_commit(StringInfo in)
 {
-	CommitTransactionCommand();
+	XLogRecPtr		commit_lsn;
+	XLogRecPtr		end_lsn;
+	TimestampTz		commit_time;
+	uint8 			flags;
+	const char	   *gid;
+
+	/* read flags */
+	flags = pq_getmsgbyte(in);
+
+	/* read fields */
+	commit_lsn = pq_getmsgint64(in);
+	end_lsn = pq_getmsgint64(in);
+	commit_time = pq_getmsgint64(in);
+
+	if (PGLOGICAL_XACT_EVENT(flags) != PGLOGICAL_COMMIT)
+		gid = pq_getmsgstring(in);
+
+	switch(PGLOGICAL_XACT_EVENT(flags))
+	{
+		case PGLOGICAL_COMMIT:
+		{
+			if (IsTransactionState())
+				CommitTransactionCommand();
+			break;
+		}
+		case PGLOGICAL_PREPARE:
+		{
+			/* prepare TBLOCK_INPROGRESS state for PrepareTransactionBlock() */
+			BeginTransactionBlock();
+			CommitTransactionCommand();
+			StartTransactionCommand();
+			/* PREPARE itself */
+			PrepareTransactionBlock(gid);
+			CommitTransactionCommand();
+			break;
+		}
+		case PGLOGICAL_COMMIT_PREPARED:
+		{
+			StartTransactionCommand();
+			FinishPreparedTransaction(gid, true);
+			CommitTransactionCommand();
+			break;
+		}
+		case PGLOGICAL_ABORT_PREPARED:
+		{
+			StartTransactionCommand();
+			FinishPreparedTransaction(gid, false);
+			CommitTransactionCommand();
+			break;
+		}
+		default:
+			Assert(false);
+	}
 }
 
 static void

contrib/mmts/pglogical_proto.c

@@ -127,9 +127,37 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 					   ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
 {
     PGLogicalProtoMM* mm = (PGLogicalProtoMM*)data->api;
-    if (!mm->isLocal) { 
-        pq_sendbyte(out, 'C');		/* sending COMMIT */
-    }
+    uint8 flags = 0;
+
+    if (mm->isLocal)
+    	return;
+
+    pq_sendbyte(out, 'C');		/* sending COMMIT */
+
+    if (txn->xact_action == XLOG_XACT_COMMIT)
+    	flags = PGLOGICAL_COMMIT;
+    else if (txn->xact_action == XLOG_XACT_PREPARE)
+    	flags = PGLOGICAL_PREPARE;
+    else if (txn->xact_action == XLOG_XACT_COMMIT_PREPARED)
+    	flags = PGLOGICAL_COMMIT_PREPARED;
+    else if (txn->xact_action == XLOG_XACT_ABORT_PREPARED)
+    	flags = PGLOGICAL_ABORT_PREPARED;
+    else
+    	Assert(false);
+
+    /* send the flags field */
+    pq_sendbyte(out, flags);
+
+    /* send fixed fields */
+    pq_sendint64(out, commit_lsn);
+    pq_sendint64(out, txn->end_lsn);
+    pq_sendint64(out, txn->commit_time);
+
+    if (txn->xact_action == XLOG_XACT_PREPARE ||
+    		txn->xact_action == XLOG_XACT_COMMIT_PREPARED ||
+    		txn->xact_action == XLOG_XACT_ABORT_PREPARED)
+    	pq_sendstring(out, txn->gid);
+
 }
 
 /*

contrib/mmts/tests/reinit-mm.sh

@@ -1,5 +1,5 @@
 n_nodes=3
-export PATH=~/postgres_cluster/dist/bin/:$PATH
+export PATH=~/code/postgres_cluster/install/bin/:$PATH
 ulimit -c unlimited
 pkill -9 postgres
 pkill -9 arbiter