handle prepare/commit_prepared in logical recvr

kelvich · kelvich · commit 04e3530b3631 · 2016-03-01T17:01:07.000+03:00
diff --git a/contrib/pglogical/pglogical_apply.c b/contrib/pglogical/pglogical_apply.c
@@ -216,6 +216,24 @@ handle_commit(StringInfo s)
 	pgstat_report_activity(STATE_IDLE, NULL);
 }
 
+static void
+handle_prepare(StringInfo s)
+{
+	BeginTransactionBlock();
+	CommitTransactionCommand();
+	StartTransactionCommand();
+	PrepareTransactionBlock("dumb_gid");
+	CommitTransactionCommand();
+}
+
+static void
+handle_commit_prepared(StringInfo s)
+{
+	StartTransactionCommand();
+	FinishPreparedTransaction("dumb_gid", true);
+	CommitTransactionCommand();
+}
+
 /*
  * Handle ORIGIN message.
  */
@@ -1026,10 +1044,18 @@ replication_handler(StringInfo s)
 		case 'C':
 			handle_commit(s);
 			break;
+		/* COMMIT PREPARED (FINISH) */
+		case 'F':
+			handle_commit_prepared(s);
+			break;
 		/* ORIGIN */
 		case 'O':
 			handle_origin(s);
 			break;
+		/* PREPARE */
+		case 'P':
+			handle_prepare(s);
+			break;
 		/* RELATION */
 		case 'R':
 			handle_relation(s);
diff --git a/contrib/pglogical_output/pglogical_proto_json.c b/contrib/pglogical_output/pglogical_proto_json.c
@@ -93,7 +93,7 @@ pglogical_json_write_commit(StringInfo out, PGLogicalOutputData *data, ReorderBu
 	if (txn->xact_action == XLOG_XACT_PREPARE)
 	{
 		appendStringInfoString(out, "\"action\":\"P\"");
-		appendStringInfo(out, ", \"gid\":\"%s\"", txn->gid);
+		appendStringInfo(out, ", \"gid\":\"%s\"", txn->gid); /* NB: Add \0 at the end, if we are using all 200 bytes in GID */
 	}
 	else if (txn->xact_action == XLOG_XACT_COMMIT_PREPARED)
 	{
diff --git a/contrib/pglogical_output/pglogical_proto_native.c b/contrib/pglogical_output/pglogical_proto_native.c
@@ -21,6 +21,7 @@
 #include "access/sysattr.h"
 #include "access/tuptoaster.h"
 #include "access/xact.h"
+#include "access/twophase.h"
 
 #include "catalog/catversion.h"
 #include "catalog/index.h"
@@ -196,7 +197,12 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 {
 	uint8 flags = 0;
 
-	pq_sendbyte(out, 'C');		/* sending COMMIT */
+	if (txn->xact_action == XLOG_XACT_PREPARE)
+		pq_sendbyte(out, 'P'); /* sending PREPARE */
+	else if (txn->xact_action == XLOG_XACT_COMMIT_PREPARED)
+		pq_sendbyte(out, 'F'); /* sending COMMIT_PREPARED (Finish 2PC) */
+	else
+		pq_sendbyte(out, 'C'); /* sending COMMIT */
 
 	/* send the flags field */
 	pq_sendbyte(out, flags);
@@ -205,6 +211,9 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 	pq_sendint64(out, commit_lsn);
 	pq_sendint64(out, txn->end_lsn);
 	pq_sendint64(out, txn->commit_time);
+
+	if (txn->xact_action == XLOG_XACT_PREPARE)
+		pq_sendbytes(out, txn->gid, GIDSIZE);
 }
 
 /*

Original file line number	Diff line number	Diff line change
`@@ -93,7 +93,7 @@ pglogical_json_write_commit(StringInfo out, PGLogicalOutputData *data, ReorderBu`
`93`	`93`	`if (txn->xact_action == XLOG_XACT_PREPARE)`
`94`	`94`	`{`
`95`	`95`	`appendStringInfoString(out, "\"action\":\"P\"");`
`96`		`- appendStringInfo(out, ", \"gid\":\"%s\"", txn->gid);`
	`96`	`+ appendStringInfo(out, ", \"gid\":\"%s\"", txn->gid); /* NB: Add \0 at the end, if we are using all 200 bytes in GID */`
`97`	`97`	`}`
`98`	`98`	`else if (txn->xact_action == XLOG_XACT_COMMIT_PREPARED)`
`99`	`99`	`{`