Use origin LSN for recovery

Author: knizhnik

contrib/mmts/pglogical_apply.c

@@ -511,6 +511,7 @@ process_remote_commit(StringInfo in)
 	csn_t       csn;
 	const char *gid = NULL;	
 	XLogRecPtr  end_lsn;
+	XLogRecPtr  origin_lsn;
 	int         n_records;
 	/* read flags */
 	flags = pq_getmsgbyte(in);
@@ -525,6 +526,8 @@ process_remote_commit(StringInfo in)
 	replorigin_session_origin_lsn = pq_getmsgint64(in); /* commit_lsn */
 	end_lsn = pq_getmsgint64(in); /* end_lsn */
 	replorigin_session_origin_timestamp = pq_getmsgint64(in); /* commit_time */
+	origin_lsn = pq_getmsgint64(in); 
+	Mtm->nodes[MtmReplicationNodeId-1].restartLsn = origin_lsn;
 
 	Assert(replorigin_session_origin == InvalidRepOriginId);
 

contrib/mmts/pglogical_proto.c

@@ -107,7 +107,7 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 {
 	bool isRecovery = MtmIsRecoveredNode(MtmReplicationNodeId);
 	csn_t csn = MtmTransactionSnapshot(txn->xid);
-	MTM_LOG1("%d: pglogical_write_begin XID=%d node=%d CSN=%ld recovery=%d restart_decoding_lsn=%lx first_lsn=%lx end_lsn=%lx confirmed_flush=%lx", 
+	MTM_LOG3("%d: pglogical_write_begin XID=%d node=%d CSN=%ld recovery=%d restart_decoding_lsn=%lx first_lsn=%lx end_lsn=%lx confirmed_flush=%lx", 
 			 MyProcPid, txn->xid, MtmReplicationNodeId, csn, isRecovery, txn->restart_decoding_lsn, txn->first_lsn, txn->end_lsn, MyReplicationSlot->data.confirmed_flush);
 
 	if (csn == INVALID_CSN && !isRecovery) { 
@@ -131,7 +131,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 {
     uint8 flags = 0;
 
-	MTM_LOG1("%d: pglogical_write_commit XID=%d node=%d restart_decoding_lsn=%lx first_lsn=%lx end_lsn=%lx confirmed_flush=%lx", 
+	MTM_LOG3("%d: pglogical_write_commit XID=%d node=%d restart_decoding_lsn=%lx first_lsn=%lx end_lsn=%lx confirmed_flush=%lx", 
 			 MyProcPid, txn->xid, MtmReplicationNodeId, txn->restart_decoding_lsn, txn->first_lsn, txn->end_lsn, MyReplicationSlot->data.confirmed_flush);
 
 
@@ -178,11 +178,12 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 
 	Assert(txn->xact_action != XLOG_XACT_PREPARE || txn->xid < 1000 || MtmTransactionRecords >= 2);
 	pq_sendint(out, MtmTransactionRecords, 4);
-
+	
     /* send fixed fields */
     pq_sendint64(out, commit_lsn);
     pq_sendint64(out, txn->end_lsn);
     pq_sendint64(out, txn->commit_time);
+	pq_sendint64(out, txn->origin_lsn);
 
 	if (txn->xact_action == XLOG_XACT_COMMIT_PREPARED) { 
 		Assert(MtmTransactionRecords == 0);
@@ -220,7 +221,7 @@ pglogical_write_update(StringInfo out, PGLogicalOutputData *data,
     if (!MtmIsFilteredTxn) { 
 		MtmTransactionRecords += 1;
 
-		MTM_LOG1("%d: pglogical_write_update confirmed_flush=%lx", MyProcPid, MyReplicationSlot->data.confirmed_flush);
+		MTM_LOG3("%d: pglogical_write_update confirmed_flush=%lx", MyProcPid, MyReplicationSlot->data.confirmed_flush);
 
 
 		pq_sendbyte(out, 'U');		/* action UPDATE */

contrib/mmts/pglogical_receiver.c

@@ -210,7 +210,6 @@ pglogical_receiver_main(Datum main_arg)
 	MtmReplicationMode mode;
 
     ByteBuffer buf;
-	XLogRecPtr originStartPos = 0;
 	RepOriginId originId;
 	char* originName;
 	/* Buffer for COPY data */
@@ -248,6 +247,7 @@ pglogical_receiver_main(Datum main_arg)
 	{ 
 		int  count;
 		ConnStatusType status;
+		XLogRecPtr originStartPos = 0;
 
 		/* 
 		 * Determine when and how we should open replication slot.
@@ -272,7 +272,7 @@ pglogical_receiver_main(Datum main_arg)
 		}
 
 		query = createPQExpBuffer();
-#if 1 /* Do we need to recretate slot ? */		
+#if 1 /* Do we need to recreate slot ? */		
 		if (mode == REPLMODE_RECOVERED) { /* recreate slot */
 			appendPQExpBuffer(query, "DROP_REPLICATION_SLOT \"%s\"", slotName);
 			res = PQexec(conn, query->data);
@@ -305,8 +305,9 @@ pglogical_receiver_main(Datum main_arg)
 
 		/* Start logical replication at specified position */
 		if (mode == REPLMODE_RECOVERED) {
-			originStartPos = 0;
-		} else { 
+			originStartPos = Mtm->nodes[nodeId].restartLsn;
+		} 
+		if (originStartPos == 0) { 
 			StartTransactionCommand();
 			originName = psprintf(MULTIMASTER_SLOT_PATTERN, nodeId);
 			originId = replorigin_by_name(originName, true);
@@ -485,7 +486,14 @@ pglogical_receiver_main(Datum main_arg)
 				if (rc > hdr_len)
 				{
 					stmt = copybuf + hdr_len;
-				
+					if (mode == REPLMODE_RECOVERED) {
+						if (stmt[0] != 'B') {
+							output_written_lsn = Max(walEnd, output_written_lsn);
+							continue;
+						}
+						mode = REPLMODE_NORMAL;
+					}
+
 					if (buf.used >= MtmTransSpillThreshold*MB) { 
 						if (spill_file < 0) {
 							int file_id;

src/backend/replication/logical/decode.c

@@ -28,7 +28,6 @@
 #include <time.h>
 
 #include "postgres.h"
-#include "miscadmin.h"
 
 #include "access/heapam.h"
 #include "access/heapam_xlog.h"
@@ -424,8 +423,6 @@ DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
 	ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
 
-	elog(LOG, "%d: DecodeHeapOp XID=%d, info=%d", MyProcPid, xid, info);
-
 	/* no point in doing anything yet */
 	if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
 		return;
@@ -816,8 +813,6 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	char	   *data;
 	RelFileNode target_node;
 
-	elog(LOG, "%d: DecodeUpdate XID=%d", MyProcPid);
-
 	xlrec = (xl_heap_update *) XLogRecGetData(r);
 
 	/* only interested in our database */
@@ -827,7 +822,6 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 
 	/* output plugin doesn't look for this origin, no need to queue */
 	if (FilterByOrigin(ctx, XLogRecGetOrigin(r))) {
-		elog(LOG, "%d: DecodeUpdate XID=%d filtered by origin %lx", MyProcPid, XLogRecGetOrigin(r));
 		return;
 	}