Advance replorigin during prepare

kelvich · kelvich · commit b4a5fe836d6e · 2016-10-25T17:07:38.000+03:00
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
@@ -1025,6 +1025,10 @@ EndPrepare(GlobalTransaction gxact)
 	xl_xact_origin xl_origin;
 	xl_xact_xinfo xl_xinfo;
 	uint8		info = XLOG_XACT_PREPARE;
+	bool		replorigin;
+
+	replorigin = (replorigin_session_origin != InvalidRepOriginId &&
+				  replorigin_session_origin != DoNotReplicateId);
 
 	/* Add the end sentinel to the list of 2PC records */
 	RegisterTwoPhaseRecord(TWOPHASE_RM_END_ID, 0,
@@ -1089,6 +1093,12 @@ EndPrepare(GlobalTransaction gxact)
 	XLogIncludeOrigin();
 
 	gxact->prepare_end_lsn = XLogInsert(RM_XACT_ID, info);
+
+	if (replorigin)
+		/* Move LSNs forward for this replication origin */
+		replorigin_session_advance(replorigin_session_origin_lsn,
+								   XactLastRecEnd);
+
 	XLogFlush(gxact->prepare_end_lsn);
 
 	/* If we crash now, we have prepared: WAL replay will fix things */
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
@@ -5657,9 +5657,20 @@ xact_redo(XLogReaderState *record)
 	}
 	else if (info == XLOG_XACT_PREPARE)
 	{
+		xl_xact_parsed_prepare parsed;
+
+		ParsePrepareRecord(XLogRecGetXid(record), XLogRecGetData(record), &parsed);
+
 		/* the record contents are exactly the 2PC file */
 		RecreateTwoPhaseFile(XLogRecGetXid(record),
 						  XLogRecGetData(record), XLogRecGetDataLen(record));
+
+		if (parsed.xinfo & XACT_XINFO_HAS_ORIGIN)
+		{
+			/* recover apply progress */
+			replorigin_advance(XLogRecGetOrigin(record), parsed.origin_lsn,
+						record->EndRecPtr, false /* backward */ , false /* WAL */ );
+		}
 	}
 	else if (info == XLOG_XACT_ASSIGNMENT)
 	{

Original file line number	Diff line number	Diff line change
`@@ -5657,9 +5657,20 @@ xact_redo(XLogReaderState *record)`
`5657`	`5657`	`}`
`5658`	`5658`	`else if (info == XLOG_XACT_PREPARE)`
`5659`	`5659`	`{`
	`5660`	`+ xl_xact_parsed_prepare parsed;`
	`5661`	`+`
	`5662`	`+ ParsePrepareRecord(XLogRecGetXid(record), XLogRecGetData(record), &parsed);`
	`5663`	`+`
`5660`	`5664`	`/* the record contents are exactly the 2PC file */`
`5661`	`5665`	`RecreateTwoPhaseFile(XLogRecGetXid(record),`
`5662`	`5666`	`XLogRecGetData(record), XLogRecGetDataLen(record));`
	`5667`	`+`
	`5668`	`+ if (parsed.xinfo & XACT_XINFO_HAS_ORIGIN)`
	`5669`	`+ {`
	`5670`	`+ /* recover apply progress */`
	`5671`	`+ replorigin_advance(XLogRecGetOrigin(record), parsed.origin_lsn,`
	`5672`	`+ record->EndRecPtr, false /* backward / , false / WAL */ );`
	`5673`	`+ }`
`5663`	`5674`	`}`
`5664`	`5675`	`else if (info == XLOG_XACT_ASSIGNMENT)`
`5665`	`5676`	`{`