Add asserts for origin

Author: knizhnik

contrib/mmts/pglogical_proto.c

@@ -199,7 +199,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 			elog(WARNING, "Failed to map origin %d", txn->origin_id);
 			i = MtmNodeId-1;
 		} else { 
-			//Assert(i == MtmNodeId-1 || txn->origin_lsn != InvalidXLogRecPtr);
+			Assert(i == MtmNodeId-1 || txn->origin_lsn != InvalidXLogRecPtr);
 		}
 		pq_sendbyte(out, i+1);
 	} else { 

src/backend/access/transam/twophase.c

@@ -1066,7 +1066,7 @@ EndPrepare(GlobalTransaction gxact)
 	if (replorigin_session_origin != InvalidRepOriginId)
 	{
 		xl_xinfo.xinfo |= XACT_XINFO_HAS_ORIGIN;
-
+		Assert(replorigin_session_origin_lsn != 0);
 		xl_origin.origin_lsn = replorigin_session_origin_lsn;
 		xl_origin.origin_timestamp = replorigin_session_origin_timestamp;
 	}
@@ -1316,6 +1316,7 @@ ParsePrepareRecord(uint8 info, char *xlrec, xl_xact_parsed_prepare *parsed)
 		/* we're only guaranteed 4 byte alignment, so copy onto stack */
 		memcpy(&xl_origin, bufptr, sizeof(xl_origin));
 		parsed->origin_lsn = xl_origin.origin_lsn;
+		Assert(parsed->origin_lsn != 0);
 		parsed->origin_timestamp = xl_origin.origin_timestamp;
 		bufptr += sizeof(xl_xact_origin);
 	}

src/backend/access/transam/xact.c

@@ -5196,7 +5196,6 @@ XactLogCommitRecord(TimestampTz commit_time,
 	if (replorigin_session_origin != InvalidRepOriginId)
 	{
 		xl_xinfo.xinfo |= XACT_XINFO_HAS_ORIGIN;
-
 		xl_origin.origin_lsn = replorigin_session_origin_lsn;
 		xl_origin.origin_timestamp = replorigin_session_origin_timestamp;
 	}

src/backend/replication/logical/decode.c

@@ -558,6 +558,7 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
 	{
 		origin_lsn = parsed->origin_lsn;
+		elog(LOG, "DecodeCommit: xid=%d, origin_lsn=%ld", xid, origin_lsn);
 		commit_time = parsed->origin_timestamp;
 	}
 
@@ -658,6 +659,7 @@ DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
 	{
 		origin_lsn = parsed->origin_lsn;
+		elog(LOG, "DecodePrepate: xid=%d (%s), origin_lsn=%ld", xid, parsed->twophase_gid, origin_lsn);
 		commit_time = parsed->origin_timestamp;
 	}