origin info in commit messages

Author: kelvich

contrib/test_decoding/sql/prepared.sql

@@ -48,6 +48,6 @@ DROP TABLE test_prepared2;
 SELECT data FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
 
 -- same but with twophase decoding
-SELECT data FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'twophase-decoding', '1') ;
+SELECT data FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'twophase-decoding', '1');
 
 SELECT pg_drop_replication_slot('regression_slot');

src/backend/replication/logical/decode.c

@@ -651,6 +651,12 @@ DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	int			i;
 	TransactionId xid = parsed->twophase_xid;
 
+	if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
+	{
+		origin_lsn = parsed->origin_lsn;
+		commit_time = parsed->origin_timestamp;
+	}
+
 	/*
 	 * Process invalidation messages, even if we're not interested in the
 	 * transaction's contents, since the various caches need to always be
@@ -703,9 +709,15 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 {
 	int			i;
 	XLogRecPtr	origin_lsn = InvalidXLogRecPtr;
-	XLogRecPtr	commit_time = InvalidXLogRecPtr;
+	TimestampTz	commit_time = 0;
 	XLogRecPtr	origin_id = XLogRecGetOrigin(buf->record);
 
+	if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
+	{
+		origin_lsn = parsed->origin_lsn;
+		commit_time = parsed->origin_timestamp;
+	}
+
 	/*
 	 * If that is ROLLBACK PREPARED than send that to callbacks.
 	 */