Use origin LSN for recovery

Author: knizhnik

pglogical_apply.c

@@ -543,6 +543,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);
@@ -557,6 +558,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);
 

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) { 
@@ -140,7 +140,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);
 
 
@@ -187,11 +187,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);
@@ -229,7 +230,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 */

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,25 +486,22 @@ pglogical_receiver_main(Datum main_arg)
 				if (rc > hdr_len)
 				{
 					stmt = copybuf + hdr_len;
-				
-				if (buf.used >= MtmTransSpillThreshold*MB) { 
-					if (spill_file < 0) {
-						int file_id;
-						spill_file = MtmCreateSpillFile(nodeId, &file_id);
-						pq_sendbyte(&spill_info, 'F');
-						pq_sendint(&spill_info, nodeId, 4);
-						pq_sendint(&spill_info, file_id, 4);
+					if (mode == REPLMODE_RECOVERED) {
+						if (stmt[0] != 'B') {
+							output_written_lsn = Max(walEnd, output_written_lsn);
+							continue;
+						}
+						mode = REPLMODE_NORMAL;
 					}
-					ByteBufferAppend(&buf, ")", 1);
-					pq_sendbyte(&spill_info, '(');
-					pq_sendint(&spill_info, buf.used, 4);
-					MtmSpillToFile(spill_file, buf.data, buf.used);
-					ByteBufferReset(&buf);
-				}
-                ByteBufferAppend(&buf, stmt, rc - hdr_len);
-				if (stmt[0] == 'C') /* commit|prepare */
-                {
-					if (spill_file >= 0) { 
+
+					if (buf.used >= MtmTransSpillThreshold*MB) { 
+						if (spill_file < 0) {
+							int file_id;
+							spill_file = MtmCreateSpillFile(nodeId, &file_id);
+							pq_sendbyte(&spill_info, 'F');
+							pq_sendint(&spill_info, nodeId, 4);
+							pq_sendint(&spill_info, file_id, 4);
+						}
 						ByteBufferAppend(&buf, ")", 1);
 						pq_sendbyte(&spill_info, '(');
 						pq_sendint(&spill_info, buf.used, 4);