[Andrey V. Lepikhov <a.lepikhov@postgrespro.ru>]

Current state of the patch set rebased on master, 5aed6a1fc2.

It is development version. Here some problems with visibility still 
detected in two tests:
1. CSN Snapshot module - TAP test on time skew.
2. Clock SI implementation - TAP test on emulation of bank transaction.

-- 
regards,
Andrey Lepikhov
Postgres Professional

Attachments:

  [text/x-patch] 0001-CSN-Log.patch (44.1K, ../0e65f96f-bfd6-fb09-89b7-e07854c215e6@postgrespro.ru/2-0001-CSN-Log.patch)
  download | inline diff:
From 875606471aaae9ee422593da08be40f5db4305dc Mon Sep 17 00:00:00 2001
From: Andrey Lepikhov <a.lepikhov@postgrespro.ru>
Date: Mon, 22 Mar 2021 15:56:06 +0500
Subject: [PATCH 1/4] CSN Log

---
 src/backend/access/rmgrdesc/Makefile     |   1 +
 src/backend/access/rmgrdesc/csnlogdesc.c |  95 ++++
 src/backend/access/rmgrdesc/xlogdesc.c   |   6 +-
 src/backend/access/transam/Makefile      |   1 +
 src/backend/access/transam/csn_log.c     | 685 +++++++++++++++++++++++
 src/backend/access/transam/rmgr.c        |   1 +
 src/backend/access/transam/varsup.c      |   2 +
 src/backend/access/transam/xlog.c        |  21 +-
 src/backend/commands/vacuum.c            |   3 +-
 src/backend/storage/ipc/ipci.c           |   3 +
 src/backend/storage/ipc/procarray.c      |   3 +
 src/backend/storage/lmgr/lwlock.c        |   2 +
 src/backend/storage/lmgr/lwlocknames.txt |   1 +
 src/backend/storage/sync/sync.c          |   5 +
 src/backend/utils/misc/guc.c             |  18 +
 src/backend/utils/probes.d               |   2 +
 src/bin/initdb/initdb.c                  |   3 +-
 src/bin/pg_controldata/pg_controldata.c  |   2 +
 src/bin/pg_upgrade/pg_upgrade.c          |   5 +
 src/bin/pg_upgrade/pg_upgrade.h          |   2 +
 src/bin/pg_waldump/rmgrdesc.c            |   1 +
 src/include/access/csn_log.h             |  83 +++
 src/include/access/rmgrlist.h            |   1 +
 src/include/access/xlog_internal.h       |   1 +
 src/include/catalog/pg_control.h         |   3 +
 src/include/storage/lwlock.h             |   1 +
 src/include/storage/proc.h               |   1 +
 src/include/storage/sync.h               |   1 +
 src/include/utils/snapshot.h             |  11 +
 src/test/regress/expected/sysviews.out   |   4 +-
 30 files changed, 962 insertions(+), 6 deletions(-)
 create mode 100644 src/backend/access/rmgrdesc/csnlogdesc.c
 create mode 100644 src/backend/access/transam/csn_log.c
 create mode 100644 src/include/access/csn_log.h

diff --git a/src/backend/access/rmgrdesc/Makefile b/src/backend/access/rmgrdesc/Makefile
index f88d72fd86..15fc36f7b4 100644
--- a/src/backend/access/rmgrdesc/Makefile
+++ b/src/backend/access/rmgrdesc/Makefile
@@ -11,6 +11,7 @@ include $(top_builddir)/src/Makefile.global
 OBJS = \
 	brindesc.o \
 	clogdesc.o \
+	csnlogdesc.o \
 	committsdesc.o \
 	dbasedesc.o \
 	genericdesc.o \
diff --git a/src/backend/access/rmgrdesc/csnlogdesc.c b/src/backend/access/rmgrdesc/csnlogdesc.c
new file mode 100644
index 0000000000..0f52b2986d
--- /dev/null
+++ b/src/backend/access/rmgrdesc/csnlogdesc.c
@@ -0,0 +1,95 @@
+/*-------------------------------------------------------------------------
+ *
+ * clogdesc.c
+ *	  rmgr descriptor routines for access/transam/csn_log.c
+ *
+ * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/access/rmgrdesc/csnlogdesc.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/csn_log.h"
+
+
+void
+csnlog_desc(StringInfo buf, XLogReaderState *record)
+{
+	char	   *rec = XLogRecGetData(record);
+	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
+
+	if (info == XLOG_CSN_ZEROPAGE)
+	{
+		int pageno;
+
+		memcpy(&pageno, XLogRecGetData(record), sizeof(int));
+		appendStringInfo(buf, "pageno %d", pageno);
+	}
+	else if (info == XLOG_CSN_TRUNCATE)
+	{
+		int pageno;
+
+		memcpy(&pageno, XLogRecGetData(record), sizeof(int));
+		appendStringInfo(buf, "pageno %d", pageno);
+	}
+	else if (info == XLOG_CSN_ASSIGNMENT)
+	{
+		CSN csn;
+
+		memcpy(&csn, XLogRecGetData(record), sizeof(CSN));
+		appendStringInfo(buf, "assign "INT64_FORMAT"", csn);
+	}
+	else if (info == XLOG_CSN_SETCSN)
+	{
+		xl_csn_set *xlrec = (xl_csn_set *) rec;
+		int			  nsubxids;
+
+		appendStringInfo(buf, "set "INT64_FORMAT" for: %u",
+						 xlrec->csn,
+						 xlrec->xtop);
+		nsubxids = ((XLogRecGetDataLen(record) - MinSizeOfCSNSet) /
+					sizeof(TransactionId));
+		if (nsubxids > 0)
+		{
+			int			i;
+			TransactionId *subxids;
+
+			subxids = palloc(sizeof(TransactionId) * nsubxids);
+			memcpy(subxids,
+				   XLogRecGetData(record) + MinSizeOfCSNSet,
+				   sizeof(TransactionId) * nsubxids);
+			for (i = 0; i < nsubxids; i++)
+				appendStringInfo(buf, ", %u", subxids[i]);
+			pfree(subxids);
+		}
+	}
+}
+
+const char *
+csnlog_identify(uint8 info)
+{
+	const char *id = NULL;
+
+	switch (info & ~XLR_INFO_MASK)
+	{
+		case XLOG_CSN_ASSIGNMENT:
+			id = "ASSIGNMENT";
+			break;
+		case XLOG_CSN_SETCSN:
+			id = "SETCSN";
+			break;
+		case XLOG_CSN_ZEROPAGE:
+			id = "ZEROPAGE";
+			break;
+		case XLOG_CSN_TRUNCATE:
+			id = "TRUNCATE";
+			break;
+	}
+
+	return id;
+}
diff --git a/src/backend/access/rmgrdesc/xlogdesc.c b/src/backend/access/rmgrdesc/xlogdesc.c
index e6090a9dad..aeba6db37d 100644
--- a/src/backend/access/rmgrdesc/xlogdesc.c
+++ b/src/backend/access/rmgrdesc/xlogdesc.c
@@ -113,7 +113,8 @@ xlog_desc(StringInfo buf, XLogReaderState *record)
 		appendStringInfo(buf, "max_connections=%d max_worker_processes=%d "
 						 "max_wal_senders=%d max_prepared_xacts=%d "
 						 "max_locks_per_xact=%d wal_level=%s "
-						 "wal_log_hints=%s track_commit_timestamp=%s",
+						 "wal_log_hints=%s track_commit_timestamp=%s "
+						 "enable_csn_snapshot=%s",
 						 xlrec.MaxConnections,
 						 xlrec.max_worker_processes,
 						 xlrec.max_wal_senders,
@@ -121,7 +122,8 @@ xlog_desc(StringInfo buf, XLogReaderState *record)
 						 xlrec.max_locks_per_xact,
 						 wal_level_str,
 						 xlrec.wal_log_hints ? "on" : "off",
-						 xlrec.track_commit_timestamp ? "on" : "off");
+						 xlrec.track_commit_timestamp ? "on" : "off",
+						 xlrec.enable_csn_snapshot ? "on" : "off");
 	}
 	else if (info == XLOG_FPW_CHANGE)
 	{
diff --git a/src/backend/access/transam/Makefile b/src/backend/access/transam/Makefile
index 595e02de72..d1196cba87 100644
--- a/src/backend/access/transam/Makefile
+++ b/src/backend/access/transam/Makefile
@@ -15,6 +15,7 @@ include $(top_builddir)/src/Makefile.global
 OBJS = \
 	clog.o \
 	commit_ts.o \
+	csn_log.o \
 	generic_xlog.o \
 	multixact.o \
 	parallel.o \
diff --git a/src/backend/access/transam/csn_log.c b/src/backend/access/transam/csn_log.c
new file mode 100644
index 0000000000..7a6d63e953
--- /dev/null
+++ b/src/backend/access/transam/csn_log.c
@@ -0,0 +1,685 @@
+/*-----------------------------------------------------------------------------
+ *
+ * csn_log.c
+ *		Track commit sequence numbers of finished transactions
+ *
+ * This module provides SLRU to store CSN for each transaction.  This
+ * mapping need to be kept only for xid's greater then oldestXid, but
+ * that can require arbitrary large amounts of memory in case of long-lived
+ * transactions.  Because of same lifetime and persistancy requirements
+ * this module is quite similar to subtrans.c
+ *
+ * If we switch database from CSN-base snapshot to xid-base snapshot then,
+ * nothing wrong. But if we switch xid-base snapshot to CSN-base snapshot
+ * it should decide a new xid whwich begin csn-base check. It can not be
+ * oldestActiveXID because of prepared transaction.
+ *
+ * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/backend/access/transam/csn_log.c
+ *
+ *-----------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/csn_log.h"
+#include "access/slru.h"
+#include "access/subtrans.h"
+#include "access/transam.h"
+#include "access/xlog_internal.h"
+#include "miscadmin.h"
+#include "pg_trace.h"
+#include "storage/spin.h"
+#include "storage/shmem.h"
+#include "utils/snapmgr.h"
+
+
+/*
+ * We use csnActive to judge if csn snapshot enabled instead of by
+ * enable_csn_snapshot, this design is similar to 'track_commit_timestamp'.
+ *
+ * Because in process of replication if master change 'enable_csn_snapshot'
+ * in a database restart, standby should apply wal record for GUC changed,
+ * then it's difficult to notice all backends about that. So they can get
+ * the message by 'csnActive' which in share buffer. It will not
+ * acquire a lock, so without performance issue.
+ *
+ */
+typedef struct CSNShared
+{
+	bool				csnActive;
+	CSN					last_csn_log_wal; /* for interval we log the assign csn to wal */
+	CSN					last_max_csn; /* Record the max csn till now */
+	TransactionId		xmin_for_csn; /*'xmin_for_csn' for when turn xid-snapshot to csn-snapshot*/
+	volatile slock_t	lock;
+} CSNShared;
+
+
+static CSNShared *csnShared;
+
+bool enable_csn_snapshot;
+bool enable_csn_wal;
+
+/*
+ * Defines for CSNLog page sizes.  A page is the same BLCKSZ as is used
+ * everywhere else in Postgres.
+ *
+ * Note: because TransactionIds are 32 bits and wrap around at 0xFFFFFFFF,
+ * CSNLog page numbering also wraps around at
+ * 0xFFFFFFFF/CSN_LOG_XACTS_PER_PAGE, and CSNLog segment numbering at
+ * 0xFFFFFFFF/CLOG_XACTS_PER_PAGE/SLRU_PAGES_PER_SEGMENT.  We need take no
+ * explicit notice of that fact in this module, except when comparing segment
+ * and page numbers in TruncateCSNLog (see CSNLogPagePrecedes).
+ */
+
+/* We store the commit CSN for each xid */
+#define CSN_LOG_XACTS_PER_PAGE (BLCKSZ / sizeof(CSN))
+
+#define TransactionIdToPage(xid)	((xid) / (TransactionId) CSN_LOG_XACTS_PER_PAGE)
+#define TransactionIdToPgIndex(xid) ((xid) % (TransactionId) CSN_LOG_XACTS_PER_PAGE)
+
+/* Link to shared-memory data structures for CLOG control */
+static SlruCtlData CSNLogCtlData;
+#define CsnlogCtl (&CSNLogCtlData)
+
+
+static int	ZeroCSNLogPage(int pageno, bool write_xlog);
+static void ZeroTruncateCSNLogPage(int pageno, bool write_xlog);
+static bool CSNLogPagePrecedes(int page1, int page2);
+static void CSNLogSetPageStatus(TransactionId xid, int nsubxids,
+									  TransactionId *subxids,
+									  CSN csn, int pageno);
+static void CSNLogSetCSNInSlot(TransactionId xid, CSN csn, int slotno);
+
+static void WriteCSNXlogRec(TransactionId xid, int nsubxids,
+							TransactionId *subxids, CSN csn);
+static void WriteZeroCSNPageXlogRec(int pageno);
+static void WriteTruncateCSNXlogRec(int pageno);
+static void set_last_log_wal_csn(CSN csn);
+static CSN get_last_log_wal_csn(void);
+
+
+/*
+ * CSNLogSetCSN
+ *
+ * Record CSN of transaction and its subtransaction tree.
+ *
+ * xid is a single xid to set status for. This will typically be the top level
+ * transactionid for a top level commit or abort. It can also be a
+ * subtransaction when we record transaction aborts.
+ *
+ * subxids is an array of xids of length nsubxids, representing subtransactions
+ * in the tree of xid. In various cases nsubxids may be zero.
+ *
+ * csn is the commit sequence number of the transaction. It should be
+ * AbortedCSN for abort cases.
+ */
+void
+CSNLogSetCSN(TransactionId xid, int nsubxids, TransactionId *subxids, CSN csn,
+			 bool write_xlog)
+{
+	int pageno;
+	int i = 0;
+	int offset = 0;
+
+	Assert(TransactionIdIsValid(xid));
+
+	pageno = TransactionIdToPage(xid);		/* get page of parent */
+
+	if(write_xlog)
+		WriteCSNXlogRec(xid, nsubxids, subxids, csn);
+
+	for (;;)
+	{
+		int num_on_page = 0;
+
+		while (i < nsubxids && TransactionIdToPage(subxids[i]) == pageno)
+		{
+			num_on_page++;
+			i++;
+		}
+
+		CSNLogSetPageStatus(xid,
+							num_on_page, subxids + offset,
+							csn, pageno);
+		if (i >= nsubxids)
+			break;
+
+		offset = i;
+		pageno = TransactionIdToPage(subxids[offset]);
+		xid = InvalidTransactionId;
+	}
+}
+
+/*
+ * Record the final state of transaction entries in the csn log for
+ * all entries on a single page. Atomic only on this page.
+ *
+ * Otherwise API is same as TransactionIdSetTreeStatus()
+ */
+static void
+CSNLogSetPageStatus(TransactionId xid, int nsubxids,
+						   TransactionId *subxids,
+						   CSN csn, int pageno)
+{
+	int slotno;
+	int i;
+
+	LWLockAcquire(CSNLogControlLock, LW_EXCLUSIVE);
+
+	slotno = SimpleLruReadPage(CsnlogCtl, pageno, true, xid);
+
+	/* Subtransactions first, if needed ... */
+	for (i = 0; i < nsubxids; i++)
+	{
+		Assert(CsnlogCtl->shared->page_number[slotno] ==
+											TransactionIdToPage(subxids[i]));
+		CSNLogSetCSNInSlot(subxids[i], csn, slotno);
+	}
+
+	/* ... then the main transaction */
+	if (TransactionIdIsValid(xid))
+		CSNLogSetCSNInSlot(xid, csn, slotno);
+
+	CsnlogCtl->shared->page_dirty[slotno] = true;
+
+	LWLockRelease(CSNLogControlLock);
+}
+
+/*
+ * Sets the commit status of a single transaction.
+ */
+static void
+CSNLogSetCSNInSlot(TransactionId xid, CSN csn, int slotno)
+{
+	int entryno = TransactionIdToPgIndex(xid);
+	CSN *ptr;
+
+	Assert(LWLockHeldByMe(CSNLogControlLock));
+
+	ptr = (CSN *) (CsnlogCtl->shared->page_buffer[slotno] +
+														entryno * sizeof(CSN));
+
+	*ptr = csn;
+}
+
+/*
+ * Interrogate the state of a transaction in the log.
+ *
+ * NB: this is a low-level routine and is NOT the preferred entry point
+ * for most uses; TransactionIdGetCSN() in csn_snapshot.c is the
+ * intended caller.
+ */
+CSN
+CSNLogGetCSNByXid(TransactionId xid)
+{
+	int pageno = TransactionIdToPage(xid);
+	int entryno = TransactionIdToPgIndex(xid);
+	int slotno;
+	CSN csn;
+
+	/* lock is acquired by SimpleLruReadPage_ReadOnly */
+	slotno = SimpleLruReadPage_ReadOnly(CsnlogCtl, pageno, xid);
+	csn = *(CSN *) (CsnlogCtl->shared->page_buffer[slotno] +
+														entryno * sizeof(CSN));
+
+	LWLockRelease(CSNLogControlLock);
+
+	return csn;
+}
+
+/*
+ * Number of shared CSNLog buffers.
+ */
+static Size
+CSNLogShmemBuffers(void)
+{
+	return Min(32, Max(4, NBuffers / 512));
+}
+
+/*
+ * Reserve shared memory for CsnlogCtl.
+ */
+Size
+CSNLogShmemSize(void)
+{
+	return SimpleLruShmemSize(CSNLogShmemBuffers(), 0);
+}
+
+/*
+ * Initialization of shared memory for CSNLog.
+ */
+void
+CSNLogShmemInit(void)
+{
+	bool found;
+
+	CsnlogCtl->PagePrecedes = CSNLogPagePrecedes;
+	SimpleLruInit(CsnlogCtl, "CSNLog Ctl", CSNLogShmemBuffers(), 0,
+				  CSNLogControlLock, "pg_csn", LWTRANCHE_CSN_LOG_BUFFERS,
+				  SYNC_HANDLER_CSN_LOG);
+
+	csnShared = ShmemInitStruct("CSNlog shared",
+								sizeof(CSNShared),
+								&found);
+	if (!found)
+	{
+		set_last_max_csn(InvalidCSN, true);
+		csnShared->last_csn_log_wal = InvalidCSN;
+		csnShared->xmin_for_csn = InvalidTransactionId;
+		csnShared->csnActive = false;
+		SpinLockInit(&csnShared->lock);
+	}
+}
+
+/*
+ * Initialize (or reinitialize) a page of CSNLog to zeroes.
+ *
+ * The page is not actually written, just set up in shared memory.
+ * The slot number of the new page is returned.
+ *
+ * Control lock must be held at entry, and will be held at exit.
+ */
+static int
+ZeroCSNLogPage(int pageno, bool write_xlog)
+{
+	int slotno;
+
+	Assert(LWLockHeldByMe(CSNLogControlLock));
+
+	slotno = SimpleLruZeroPage(CsnlogCtl, pageno);
+
+	if(write_xlog)
+		WriteZeroCSNPageXlogRec(pageno);
+
+	return slotno;
+}
+
+static void
+ZeroTruncateCSNLogPage(int pageno, bool write_xlog)
+{
+	if(write_xlog)
+		WriteTruncateCSNXlogRec(pageno);
+
+	SimpleLruTruncate(CsnlogCtl, pageno);
+}
+
+void
+ActivateCSNlog(void)
+{
+	int				startPage;
+	TransactionId	nextXid = InvalidTransactionId;
+
+	if (csnShared->csnActive)
+		return;
+
+	nextXid = XidFromFullTransactionId(ShmemVariableCache->nextXid);
+	startPage = TransactionIdToPage(nextXid);
+
+	/* Create the current segment file, if necessary */
+	if (!SimpleLruDoesPhysicalPageExist(CsnlogCtl, startPage))
+	{
+		int			slotno;
+		LWLockAcquire(CSNLogControlLock, LW_EXCLUSIVE);
+		slotno = ZeroCSNLogPage(startPage, false);
+		SimpleLruWritePage(CsnlogCtl, slotno);
+		LWLockRelease(CSNLogControlLock);
+	}
+	csnShared->csnActive = true;
+}
+
+bool
+get_csnlog_status(void)
+{
+	if(!csnShared)
+	{
+		/* Should not arrived */
+		elog(ERROR, "We do not have csnShared point");
+	}
+	return csnShared->csnActive;
+}
+
+void
+DeactivateCSNlog(void)
+{
+	csnShared->csnActive = false;
+	LWLockAcquire(CSNLogControlLock, LW_EXCLUSIVE);
+	(void) SlruScanDirectory(CsnlogCtl, SlruScanDirCbDeleteAll, NULL);
+	LWLockRelease(CSNLogControlLock);
+}
+
+void
+StartupCSN(void)
+{
+	ActivateCSNlog();
+}
+
+void
+CompleteCSNInitialization(void)
+{
+	/*
+	 * If the feature is not enabled, turn it off for good.  This also removes
+	 * any leftover data.
+	 *
+	 * Conversely, we activate the module if the feature is enabled.  This is
+	 * necessary for primary and standby as the activation depends on the
+	 * control file contents at the beginning of recovery or when a
+	 * XLOG_PARAMETER_CHANGE is replayed.
+	 */
+	if (!get_csnlog_status())
+		DeactivateCSNlog();
+	else
+		ActivateCSNlog();
+}
+
+void
+CSNlogParameterChange(bool newvalue, bool oldvalue)
+{
+	if (newvalue)
+	{
+		if (!csnShared->csnActive)
+			ActivateCSNlog();
+	}
+	else if (csnShared->csnActive)
+		DeactivateCSNlog();
+}
+
+/*
+ * Perform a checkpoint --- either during shutdown, or on-the-fly
+ */
+void
+CheckPointCSNLog(void)
+{
+	if (!get_csnlog_status())
+		return;
+
+	/*
+	 * Flush dirty CSNLog pages to disk.
+	 *
+	 * This is not actually necessary from a correctness point of view. We do
+	 * it merely to improve the odds that writing of dirty pages is done by
+	 * the checkpoint process and not by backends.
+	 */
+	TRACE_POSTGRESQL_CSNLOG_CHECKPOINT_START(true);
+	SimpleLruWriteAll(CsnlogCtl, true);
+	TRACE_POSTGRESQL_CSNLOG_CHECKPOINT_DONE(true);
+}
+
+/*
+ * Make sure that CSNLog has room for a newly-allocated XID.
+ *
+ * NB: this is called while holding XidGenLock.  We want it to be very fast
+ * most of the time; even when it's not so fast, no actual I/O need happen
+ * unless we're forced to write out a dirty clog or xlog page to make room
+ * in shared memory.
+ */
+void
+ExtendCSNLog(TransactionId newestXact)
+{
+	int			pageno;
+
+	if (!get_csnlog_status())
+		return;
+
+	/*
+	 * No work except at first XID of a page.  But beware: just after
+	 * wraparound, the first XID of page zero is FirstNormalTransactionId.
+	 */
+	if (TransactionIdToPgIndex(newestXact) != 0 &&
+		!TransactionIdEquals(newestXact, FirstNormalTransactionId))
+		return;
+
+	pageno = TransactionIdToPage(newestXact);
+
+	LWLockAcquire(CSNLogControlLock, LW_EXCLUSIVE);
+
+	/* Zero the page and make an XLOG entry about it */
+	ZeroCSNLogPage(pageno, !InRecovery);
+
+	LWLockRelease(CSNLogControlLock);
+}
+
+/*
+ * Remove all CSNLog segments before the one holding the passed
+ * transaction ID.
+ *
+ * This is normally called during checkpoint, with oldestXact being the
+ * oldest TransactionXmin of any running transaction.
+ */
+void
+TruncateCSNLog(TransactionId oldestXact)
+{
+	int			cutoffPage;
+
+	if (!get_csnlog_status())
+		return;
+
+	/*
+	 * The cutoff point is the start of the segment containing oldestXact. We
+	 * pass the *page* containing oldestXact to SimpleLruTruncate. We step
+	 * back one transaction to avoid passing a cutoff page that hasn't been
+	 * created yet in the rare case that oldestXact would be the first item on
+	 * a page and oldestXact == next XID.  In that case, if we didn't subtract
+	 * one, we'd trigger SimpleLruTruncate's wraparound detection.
+	 */
+	TransactionIdRetreat(oldestXact);
+	cutoffPage = TransactionIdToPage(oldestXact);
+	ZeroTruncateCSNLogPage(cutoffPage, true);
+}
+
+/*
+ * Decide which of two CSNLog page numbers is "older" for truncation
+ * purposes.
+ *
+ * We need to use comparison of TransactionIds here in order to do the right
+ * thing with wraparound XID arithmetic.  However, if we are asked about
+ * page number zero, we don't want to hand InvalidTransactionId to
+ * TransactionIdPrecedes: it'll get weird about permanent xact IDs.  So,
+ * offset both xids by FirstNormalTransactionId to avoid that.
+ */
+static bool
+CSNLogPagePrecedes(int page1, int page2)
+{
+	TransactionId xid1;
+	TransactionId xid2;
+
+	xid1 = ((TransactionId) page1) * CSN_LOG_XACTS_PER_PAGE;
+	xid1 += FirstNormalTransactionId;
+	xid2 = ((TransactionId) page2) * CSN_LOG_XACTS_PER_PAGE;
+	xid2 += FirstNormalTransactionId;
+
+	return TransactionIdPrecedes(xid1, xid2);
+}
+
+void
+WriteAssignCSNXlogRec(CSN csn)
+{
+	CSN log_csn = 0;
+
+	if (!enable_csn_wal || csn <= get_last_log_wal_csn())
+		return;
+
+	/*
+	 * We log the CSN 5s greater than generated, you can see comments on
+	 * CSN_ASSIGN_TIME_INTERVAL define.
+	 */
+	log_csn = CSNAddByNanosec(csn, CSN_ASSIGN_TIME_INTERVAL);
+	set_last_log_wal_csn(log_csn);
+
+	XLogBeginInsert();
+	XLogRegisterData((char *) (&log_csn), sizeof(CSN));
+	XLogInsert(RM_CSNLOG_ID, XLOG_CSN_ASSIGNMENT);
+}
+
+static void
+WriteCSNXlogRec(TransactionId xid, int nsubxids,
+				TransactionId *subxids, CSN csn)
+{
+	xl_csn_set xlrec;
+
+	if (!enable_csn_wal)
+		return;
+
+	xlrec.xtop = xid;
+	xlrec.nsubxacts = nsubxids;
+	xlrec.csn = csn;
+
+	XLogBeginInsert();
+	XLogRegisterData((char *) &xlrec, MinSizeOfCSNSet);
+	XLogRegisterData((char *) subxids, nsubxids * sizeof(TransactionId));
+	(void) XLogInsert(RM_CSNLOG_ID, XLOG_CSN_SETCSN);
+}
+
+/*
+ * Write a ZEROPAGE xlog record
+ */
+static void
+WriteZeroCSNPageXlogRec(int pageno)
+{
+	if (!enable_csn_wal)
+		return;
+
+	XLogBeginInsert();
+	XLogRegisterData((char *) (&pageno), sizeof(int));
+	(void) XLogInsert(RM_CSNLOG_ID, XLOG_CSN_ZEROPAGE);
+}
+
+/*
+ * Write a TRUNCATE xlog record
+ */
+static void
+WriteTruncateCSNXlogRec(int pageno)
+{
+	if (!enable_csn_wal)
+		return;
+
+	XLogBeginInsert();
+	XLogRegisterData((char *) (&pageno), sizeof(int));
+	(void) XLogInsert(RM_CSNLOG_ID, XLOG_CSN_TRUNCATE);
+}
+
+void
+csnlog_redo(XLogReaderState *record)
+{
+	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
+
+	/* Backup blocks are not used in csnlog records */
+	Assert(!XLogRecHasAnyBlockRefs(record));
+
+	if (info == XLOG_CSN_ASSIGNMENT)
+	{
+		CSN csn;
+
+		memcpy(&csn, XLogRecGetData(record), sizeof(CSN));
+		set_last_max_csn(csn, true);
+	}
+	else if (info == XLOG_CSN_SETCSN)
+	{
+		xl_csn_set *xlrec = (xl_csn_set *) XLogRecGetData(record);
+		CSNLogSetCSN(xlrec->xtop, xlrec->nsubxacts, xlrec->xsub, xlrec->csn, false);
+	}
+	else if (info == XLOG_CSN_ZEROPAGE)
+	{
+		int			pageno;
+		int			slotno;
+
+		memcpy(&pageno, XLogRecGetData(record), sizeof(int));
+		LWLockAcquire(CSNLogControlLock, LW_EXCLUSIVE);
+		slotno = ZeroCSNLogPage(pageno, false);
+		SimpleLruWritePage(CsnlogCtl, slotno);
+		LWLockRelease(CSNLogControlLock);
+		Assert(!CsnlogCtl->shared->page_dirty[slotno]);
+
+	}
+	else if (info == XLOG_CSN_TRUNCATE)
+	{
+		int			pageno;
+
+		memcpy(&pageno, XLogRecGetData(record), sizeof(int));
+		CsnlogCtl->shared->latest_page_number = pageno;
+		ZeroTruncateCSNLogPage(pageno, false);
+	}
+	else
+		elog(PANIC, "csnlog_redo: unknown op code %u", info);
+}
+
+static void
+set_last_log_wal_csn(CSN csn)
+{
+	SpinLockAcquire(&csnShared->lock);
+	csnShared->last_csn_log_wal = csn;
+	SpinLockRelease(&csnShared->lock);
+}
+
+static CSN
+get_last_log_wal_csn(void)
+{
+	CSN csn;
+
+	SpinLockAcquire(&csnShared->lock);
+	csn = csnShared->last_csn_log_wal;
+	SpinLockRelease(&csnShared->lock);
+	return csn;
+}
+
+CSN
+set_last_max_csn(CSN csn, bool force)
+{
+	SpinLockAcquire(&csnShared->lock);
+	if (csn <= csnShared->last_max_csn && !force)
+		csn = csnShared->last_max_csn + 1;
+
+	csnShared->last_max_csn = csn;
+
+	SpinLockRelease(&csnShared->lock);
+	return csn;
+}
+
+CSN
+get_last_max_csn(void)
+{
+	CSN csn;
+
+	SpinLockAcquire(&csnShared->lock);
+	csn = csnShared->last_max_csn;
+	SpinLockRelease(&csnShared->lock);
+	return csn;
+}
+
+CSN
+get_xmin_for_csn(void)
+{
+	CSN csn;
+
+	SpinLockAcquire(&csnShared->lock);
+	csn = csnShared->xmin_for_csn;
+	SpinLockRelease(&csnShared->lock);
+	return csn;
+}
+
+void
+prepare_csn_env(bool enable)
+{
+	if (enable)
+	{
+		TransactionId nextxid =
+						XidFromFullTransactionId(ShmemVariableCache->nextXid);
+		/* 'xmin_for_csn' for when turn xid-snapshot to csn-snapshot */
+		csnShared->xmin_for_csn = nextxid;
+		/* produce the csnlog segment we want now and seek to current page */
+		ActivateCSNlog();
+	}
+	else
+		/* Try to drop all csnlog seg */
+		DeactivateCSNlog();
+}
+
+/*
+ * Entrypoint for sync.c to sync csnlog files.
+ */
+int
+csnlogsyncfiletag(const FileTag *ftag, char *path)
+{
+	return SlruSyncFileTag(CsnlogCtl, ftag, path);
+}
diff --git a/src/backend/access/transam/rmgr.c b/src/backend/access/transam/rmgr.c
index 58091f6b52..b1e5ec350e 100644
--- a/src/backend/access/transam/rmgr.c
+++ b/src/backend/access/transam/rmgr.c
@@ -28,6 +28,7 @@
 #include "replication/origin.h"
 #include "storage/standby.h"
 #include "utils/relmapper.h"
+#include "access/csn_log.h"
 
 /* must be kept in sync with RmgrData definition in xlog_internal.h */
 #define PG_RMGR(symname,name,redo,desc,identify,startup,cleanup,mask) \
diff --git a/src/backend/access/transam/varsup.c b/src/backend/access/transam/varsup.c
index 142da4aaff..5791a2e66e 100644
--- a/src/backend/access/transam/varsup.c
+++ b/src/backend/access/transam/varsup.c
@@ -15,6 +15,7 @@
 
 #include "access/clog.h"
 #include "access/commit_ts.h"
+#include "access/csn_log.h"
 #include "access/subtrans.h"
 #include "access/transam.h"
 #include "access/xact.h"
@@ -175,6 +176,7 @@ GetNewTransactionId(bool isSubXact)
 	 * Extend pg_subtrans and pg_commit_ts too.
 	 */
 	ExtendCLOG(xid);
+	ExtendCSNLog(xid);
 	ExtendCommitTs(xid);
 	ExtendSUBTRANS(xid);
 
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 6f8810e149..46f016dce7 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -24,6 +24,7 @@
 
 #include "access/clog.h"
 #include "access/commit_ts.h"
+#include "access/csn_log.h"
 #include "access/heaptoast.h"
 #include "access/multixact.h"
 #include "access/rewriteheap.h"
@@ -4649,6 +4650,7 @@ InitControlFile(uint64 sysidentifier)
 	ControlFile->wal_level = wal_level;
 	ControlFile->wal_log_hints = wal_log_hints;
 	ControlFile->track_commit_timestamp = track_commit_timestamp;
+	ControlFile->enable_csn_snapshot = enable_csn_snapshot;
 	ControlFile->data_checksum_version = bootstrap_data_checksum_version;
 }
 
@@ -6910,6 +6912,7 @@ StartupXLOG(void)
 	SetCommitTsLimit(checkPoint.oldestCommitTsXid,
 					 checkPoint.newestCommitTsXid);
 	XLogCtl->ckptFullXid = checkPoint.nextXid;
+	set_last_max_csn(checkPoint.oldestUsedCSN, true);
 
 	/*
 	 * Initialize replication slots, before there's a chance to remove
@@ -6945,6 +6948,9 @@ StartupXLOG(void)
 	if (ControlFile->track_commit_timestamp)
 		StartupCommitTs();
 
+	if(ControlFile->enable_csn_snapshot)
+		StartupCSN();
+
 	/*
 	 * Recover knowledge about replay progress of known replication partners.
 	 */
@@ -8062,6 +8068,7 @@ StartupXLOG(void)
 	 * commit timestamp.
 	 */
 	CompleteCommitTsInitialization();
+	CompleteCSNInitialization();
 
 	/*
 	 * All done with end-of-recovery actions.
@@ -8981,6 +8988,8 @@ CreateCheckPoint(int flags)
 	 */
 	last_important_lsn = GetLastImportantRecPtr();
 
+	checkPoint.oldestUsedCSN = get_last_max_csn();
+
 	/*
 	 * We must block concurrent insertions while examining insert state to
 	 * determine the checkpoint REDO pointer.
@@ -9370,6 +9379,7 @@ CheckPointGuts(XLogRecPtr checkPointRedo, int flags)
 	TRACE_POSTGRESQL_BUFFER_CHECKPOINT_START(flags);
 	CheckpointStats.ckpt_write_t = GetCurrentTimestamp();
 	CheckPointCLOG();
+	CheckPointCSNLog();
 	CheckPointCommitTs();
 	CheckPointSUBTRANS();
 	CheckPointMultiXact();
@@ -9922,7 +9932,8 @@ XLogReportParameters(void)
 		max_wal_senders != ControlFile->max_wal_senders ||
 		max_prepared_xacts != ControlFile->max_prepared_xacts ||
 		max_locks_per_xact != ControlFile->max_locks_per_xact ||
-		track_commit_timestamp != ControlFile->track_commit_timestamp)
+		track_commit_timestamp != ControlFile->track_commit_timestamp ||
+		enable_csn_snapshot != ControlFile->enable_csn_snapshot)
 	{
 		/*
 		 * The change in number of backend slots doesn't need to be WAL-logged
@@ -9944,6 +9955,7 @@ XLogReportParameters(void)
 			xlrec.wal_level = wal_level;
 			xlrec.wal_log_hints = wal_log_hints;
 			xlrec.track_commit_timestamp = track_commit_timestamp;
+			xlrec.enable_csn_snapshot = enable_csn_snapshot;
 
 			XLogBeginInsert();
 			XLogRegisterData((char *) &xlrec, sizeof(xlrec));
@@ -9954,6 +9966,9 @@ XLogReportParameters(void)
 
 		LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
 
+		if (enable_csn_snapshot != ControlFile->enable_csn_snapshot)
+			prepare_csn_env(enable_csn_snapshot);
+
 		ControlFile->MaxConnections = MaxConnections;
 		ControlFile->max_worker_processes = max_worker_processes;
 		ControlFile->max_wal_senders = max_wal_senders;
@@ -9962,6 +9977,7 @@ XLogReportParameters(void)
 		ControlFile->wal_level = wal_level;
 		ControlFile->wal_log_hints = wal_log_hints;
 		ControlFile->track_commit_timestamp = track_commit_timestamp;
+		ControlFile->enable_csn_snapshot = enable_csn_snapshot;
 		UpdateControlFile();
 
 		LWLockRelease(ControlFileLock);
@@ -10393,6 +10409,9 @@ xlog_redo(XLogReaderState *record)
 		CommitTsParameterChange(xlrec.track_commit_timestamp,
 								ControlFile->track_commit_timestamp);
 		ControlFile->track_commit_timestamp = xlrec.track_commit_timestamp;
+		CSNlogParameterChange(xlrec.enable_csn_snapshot,
+							  ControlFile->enable_csn_snapshot);
+		ControlFile->enable_csn_snapshot = xlrec.enable_csn_snapshot;
 
 		UpdateControlFile();
 		LWLockRelease(ControlFileLock);
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index c064352e23..515f2b3eb9 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -53,7 +53,7 @@
 #include "utils/memutils.h"
 #include "utils/snapmgr.h"
 #include "utils/syscache.h"
-
+#include "access/csn_log.h"
 
 /*
  * GUC parameters
@@ -1688,6 +1688,7 @@ vac_truncate_clog(TransactionId frozenXID,
 	 */
 	TruncateCLOG(frozenXID, oldestxid_datoid);
 	TruncateCommitTs(frozenXID);
+	TruncateCSNLog(frozenXID);
 	TruncateMultiXact(minMulti, minmulti_datoid);
 
 	/*
diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c
index 3e4ec53a97..e06bc13841 100644
--- a/src/backend/storage/ipc/ipci.c
+++ b/src/backend/storage/ipc/ipci.c
@@ -16,6 +16,7 @@
 
 #include "access/clog.h"
 #include "access/commit_ts.h"
+#include "access/csn_log.h"
 #include "access/heapam.h"
 #include "access/multixact.h"
 #include "access/nbtree.h"
@@ -127,6 +128,7 @@ CreateSharedMemoryAndSemaphores(void)
 		size = add_size(size, ProcGlobalShmemSize());
 		size = add_size(size, XLOGShmemSize());
 		size = add_size(size, CLOGShmemSize());
+		size = add_size(size, CSNLogShmemSize());
 		size = add_size(size, CommitTsShmemSize());
 		size = add_size(size, SUBTRANSShmemSize());
 		size = add_size(size, TwoPhaseShmemSize());
@@ -218,6 +220,7 @@ CreateSharedMemoryAndSemaphores(void)
 	 */
 	XLOGShmemInit();
 	CLOGShmemInit();
+	CSNLogShmemInit();
 	CommitTsShmemInit();
 	SUBTRANSShmemInit();
 	MultiXactShmemInit();
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index 4fc6ffb917..4a200b0815 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -48,6 +48,7 @@
 #include <signal.h>
 
 #include "access/clog.h"
+#include "access/csn_log.h"
 #include "access/subtrans.h"
 #include "access/transam.h"
 #include "access/twophase.h"
@@ -1170,6 +1171,7 @@ ProcArrayApplyRecoveryInfo(RunningTransactions running)
 	while (TransactionIdPrecedes(latestObservedXid, running->nextXid))
 	{
 		ExtendSUBTRANS(latestObservedXid);
+		ExtendCSNLog(latestObservedXid);
 		TransactionIdAdvance(latestObservedXid);
 	}
 	TransactionIdRetreat(latestObservedXid);	/* = running->nextXid - 1 */
@@ -4320,6 +4322,7 @@ RecordKnownAssignedTransactionIds(TransactionId xid)
 		while (TransactionIdPrecedes(next_expected_xid, xid))
 		{
 			TransactionIdAdvance(next_expected_xid);
+			ExtendCSNLog(next_expected_xid);
 			ExtendSUBTRANS(next_expected_xid);
 		}
 		Assert(next_expected_xid == xid);
diff --git a/src/backend/storage/lmgr/lwlock.c b/src/backend/storage/lmgr/lwlock.c
index 975d547f34..3489e51ef3 100644
--- a/src/backend/storage/lmgr/lwlock.c
+++ b/src/backend/storage/lmgr/lwlock.c
@@ -134,6 +134,8 @@ static const char *const BuiltinTrancheNames[] = {
 	"CommitTSBuffer",
 	/* LWTRANCHE_SUBTRANS_BUFFER: */
 	"SubtransBuffer",
+	/* LWTRANCHE_CSN_LOG_BUFFERS */
+	"CSNLogBuffer",
 	/* LWTRANCHE_MULTIXACTOFFSET_BUFFER: */
 	"MultiXactOffsetBuffer",
 	/* LWTRANCHE_MULTIXACTMEMBER_BUFFER: */
diff --git a/src/backend/storage/lmgr/lwlocknames.txt b/src/backend/storage/lmgr/lwlocknames.txt
index 6c7cf6c295..56196b2d0d 100644
--- a/src/backend/storage/lmgr/lwlocknames.txt
+++ b/src/backend/storage/lmgr/lwlocknames.txt
@@ -53,3 +53,4 @@ XactTruncationLock					44
 # 45 was XactTruncationLock until removal of BackendRandomLock
 WrapLimitsVacuumLock				46
 NotifyQueueTailLock					47
+CSNLogControlLock				    48
diff --git a/src/backend/storage/sync/sync.c b/src/backend/storage/sync/sync.c
index 708215614d..cbb2fd510b 100644
--- a/src/backend/storage/sync/sync.c
+++ b/src/backend/storage/sync/sync.c
@@ -20,6 +20,7 @@
 
 #include "access/commit_ts.h"
 #include "access/clog.h"
+#include "access/csn_log.h"
 #include "access/multixact.h"
 #include "access/xlog.h"
 #include "access/xlogutils.h"
@@ -118,6 +119,10 @@ static const SyncOps syncsw[] = {
 	/* pg_multixact/members */
 	[SYNC_HANDLER_MULTIXACT_MEMBER] = {
 		.sync_syncfiletag = multixactmemberssyncfiletag
+	},
+	/* CSN_Log */
+	[SYNC_HANDLER_CSN_LOG] = {
+		.sync_syncfiletag = csnlogsyncfiletag
 	}
 };
 
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 3b36a31a47..14063bcf9e 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -1170,6 +1170,24 @@ static struct config_bool ConfigureNamesBool[] =
 		false,
 		NULL, NULL, NULL
 	},
+	{
+		{"enable_csn_snapshot", PGC_POSTMASTER, RESOURCES_MEM,
+			gettext_noop("Enable csn-base snapshot."),
+			gettext_noop("Used to achieve REPEATEBLE READ isolation level for postgres_fdw transactions.")
+		},
+		&enable_csn_snapshot,
+		false,
+		NULL, NULL, NULL
+	},
+	{
+		{"enable_csn_wal", PGC_POSTMASTER, RESOURCES_MEM,
+			gettext_noop("Enable csn-wal record."),
+			gettext_noop("Used to enable csn-wal record")
+		},
+		&enable_csn_wal,
+		true,
+		NULL, NULL, NULL
+	},
 	{
 		{"ssl", PGC_SIGHUP, CONN_AUTH_SSL,
 			gettext_noop("Enables SSL connections."),
diff --git a/src/backend/utils/probes.d b/src/backend/utils/probes.d
index b0c50a3c7f..3fcd0f4ccf 100644
--- a/src/backend/utils/probes.d
+++ b/src/backend/utils/probes.d
@@ -77,6 +77,8 @@ provider postgresql {
 	probe clog__checkpoint__done(bool);
 	probe subtrans__checkpoint__start(bool);
 	probe subtrans__checkpoint__done(bool);
+	probe csnlog__checkpoint__start(bool);
+	probe csnlog__checkpoint__done(bool);
 	probe multixact__checkpoint__start(bool);
 	probe multixact__checkpoint__done(bool);
 	probe twophase__checkpoint__start();
diff --git a/src/bin/initdb/initdb.c b/src/bin/initdb/initdb.c
index 3c1cf78b4f..6b3b9ff504 100644
--- a/src/bin/initdb/initdb.c
+++ b/src/bin/initdb/initdb.c
@@ -223,7 +223,8 @@ static const char *const subdirs[] = {
 	"pg_xact",
 	"pg_logical",
 	"pg_logical/snapshots",
-	"pg_logical/mappings"
+	"pg_logical/mappings",
+	"pg_csn"
 };
 
 
diff --git a/src/bin/pg_controldata/pg_controldata.c b/src/bin/pg_controldata/pg_controldata.c
index f911f98d94..325e6a0e2b 100644
--- a/src/bin/pg_controldata/pg_controldata.c
+++ b/src/bin/pg_controldata/pg_controldata.c
@@ -300,6 +300,8 @@ main(int argc, char *argv[])
 		   ControlFile->max_locks_per_xact);
 	printf(_("track_commit_timestamp setting:       %s\n"),
 		   ControlFile->track_commit_timestamp ? _("on") : _("off"));
+	printf(_("enable_csn_snapshot setting:    	    %s\n"),
+		   ControlFile->enable_csn_snapshot ? 	 _("on") : _("off"));
 	printf(_("Maximum data alignment:               %u\n"),
 		   ControlFile->maxAlign);
 	/* we don't print floatFormat since can't say much useful about it */
diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.c
index e23b8ca88d..ed9112ed40 100644
--- a/src/bin/pg_upgrade/pg_upgrade.c
+++ b/src/bin/pg_upgrade/pg_upgrade.c
@@ -541,6 +541,11 @@ copy_xact_xlog_xid(void)
 		check_ok();
 	}
 
+	if(old_cluster.controldata.cat_ver > CSN_BASE_SNAPSHOT_ADD_VER)
+	{
+		copy_subdir_files("pg_csn", "pg_csn");
+	}
+
 	/* now reset the wal archives in the new cluster */
 	prep_status("Resetting WAL archives");
 	exec_prog(UTILITY_LOG_FILE, NULL, true, true,
diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.h
index 919a7849fd..77f5a030ae 100644
--- a/src/bin/pg_upgrade/pg_upgrade.h
+++ b/src/bin/pg_upgrade/pg_upgrade.h
@@ -123,6 +123,8 @@ extern char *output_files[];
  */
 #define JSONB_FORMAT_CHANGE_CAT_VER 201409291
 
+#define	CSN_BASE_SNAPSHOT_ADD_VER	202002010
+
 
 /*
  * Each relation is represented by a relinfo structure.
diff --git a/src/bin/pg_waldump/rmgrdesc.c b/src/bin/pg_waldump/rmgrdesc.c
index 852d8ca4b1..2d280ce940 100644
--- a/src/bin/pg_waldump/rmgrdesc.c
+++ b/src/bin/pg_waldump/rmgrdesc.c
@@ -11,6 +11,7 @@
 #include "access/brin_xlog.h"
 #include "access/clog.h"
 #include "access/commit_ts.h"
+#include "access/csn_log.h"
 #include "access/generic_xlog.h"
 #include "access/ginxlog.h"
 #include "access/gistxlog.h"
diff --git a/src/include/access/csn_log.h b/src/include/access/csn_log.h
new file mode 100644
index 0000000000..01fb0daad4
--- /dev/null
+++ b/src/include/access/csn_log.h
@@ -0,0 +1,83 @@
+/*
+ * csn_log.h
+ *
+ * Commit-Sequence-Number log.
+ *
+ * Portions Copyright (c) 1996-2014, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/access/csn_log.h
+ */
+#ifndef CSNLOG_H
+#define CSNLOG_H
+
+#include "access/xlog.h"
+#include "storage/sync.h"
+#include "utils/snapshot.h"
+
+
+typedef struct xl_csn_set
+{
+	CSN				csn;
+	TransactionId	xtop;			/* XID's top-level XID */
+	int				nsubxacts;		/* number of subtransaction XIDs */
+	TransactionId	xsub[FLEXIBLE_ARRAY_MEMBER];	/* assigned subxids */
+} xl_csn_set;
+
+
+/* XLOG stuff */
+#define XLOG_CSN_ASSIGNMENT	0x00
+#define XLOG_CSN_SETCSN		0x10
+#define XLOG_CSN_ZEROPAGE	0x20
+#define XLOG_CSN_TRUNCATE	0x30
+
+/*
+ * We should log MAX generated CSN to wal, so that database will not generate
+ * a historical CSN after database restart. This may appear when system time
+ * turned back.
+ *
+ * However we can not log the MAX CSN every time it generated, if so it will
+ * cause too many wal expend, so we log it 5s more in the future.
+ *
+ * As a trade off, when this database restart, there will be 5s bad performance
+ * for time synchronization among postgres instances.
+ *
+ * It looks like we can redefine this as a configure parameter, and the user
+ * can decide which way they prefer.
+ *
+ */
+#define	CSN_ASSIGN_TIME_INTERVAL	5
+
+#define MinSizeOfCSNSet offsetof(xl_csn_set, xsub)
+#define	CSNAddByNanosec(csn,second) (csn + second * 1000000000L)
+
+
+extern void CSNLogSetCSN(TransactionId xid, int nsubxids,
+						 TransactionId *subxids, CSN csn, bool write_xlog);
+extern CSN CSNLogGetCSNByXid(TransactionId xid);
+
+extern Size CSNLogShmemSize(void);
+extern void CSNLogShmemInit(void);
+extern void ShutdownCSNLog(void);
+extern void CheckPointCSNLog(void);
+extern void ExtendCSNLog(TransactionId newestXact);
+extern void TruncateCSNLog(TransactionId oldestXact);
+extern int csnlogsyncfiletag(const FileTag *ftag, char *path);
+
+extern void csnlog_redo(XLogReaderState *record);
+extern void csnlog_desc(StringInfo buf, XLogReaderState *record);
+extern const char *csnlog_identify(uint8 info);
+extern void WriteAssignCSNXlogRec(CSN csn);
+extern CSN set_last_max_csn(CSN csn, bool force);
+extern CSN get_last_max_csn(void);
+extern CSN get_xmin_for_csn(void);
+extern void prepare_csn_env(bool enable_csn_snapshot);
+extern void CatchCSNLog(void);
+extern void ActivateCSNlog(void);
+extern void DeactivateCSNlog(void);
+extern void StartupCSN(void);
+extern void CompleteCSNInitialization(void);
+extern void CSNlogParameterChange(bool newvalue, bool oldvalue);
+extern bool get_csnlog_status(void);
+
+#endif   /* CSNLOG_H */
diff --git a/src/include/access/rmgrlist.h b/src/include/access/rmgrlist.h
index f582cf535f..3cf0775176 100644
--- a/src/include/access/rmgrlist.h
+++ b/src/include/access/rmgrlist.h
@@ -47,3 +47,4 @@ PG_RMGR(RM_COMMIT_TS_ID, "CommitTs", commit_ts_redo, commit_ts_desc, commit_ts_i
 PG_RMGR(RM_REPLORIGIN_ID, "ReplicationOrigin", replorigin_redo, replorigin_desc, replorigin_identify, NULL, NULL, NULL)
 PG_RMGR(RM_GENERIC_ID, "Generic", generic_redo, generic_desc, generic_identify, NULL, NULL, generic_mask)
 PG_RMGR(RM_LOGICALMSG_ID, "LogicalMessage", logicalmsg_redo, logicalmsg_desc, logicalmsg_identify, NULL, NULL, NULL)
+PG_RMGR(RM_CSNLOG_ID, "CSN", csnlog_redo, csnlog_desc, csnlog_identify, NULL, NULL, NULL)
diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.h
index b23e286406..d9690a23bb 100644
--- a/src/include/access/xlog_internal.h
+++ b/src/include/access/xlog_internal.h
@@ -240,6 +240,7 @@ typedef struct xl_parameter_change
 	int			wal_level;
 	bool		wal_log_hints;
 	bool		track_commit_timestamp;
+	bool		enable_csn_snapshot;
 } xl_parameter_change;
 
 /* logs restore point */
diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.h
index e3f48158ce..194d100ed4 100644
--- a/src/include/catalog/pg_control.h
+++ b/src/include/catalog/pg_control.h
@@ -61,6 +61,8 @@ typedef struct CheckPoint
 	 * set to InvalidTransactionId.
 	 */
 	TransactionId oldestActiveXid;
+
+	uint64 oldestUsedCSN;
 } CheckPoint;
 
 /* XLOG info values for XLOG rmgr */
@@ -181,6 +183,7 @@ typedef struct ControlFileData
 	int			max_prepared_xacts;
 	int			max_locks_per_xact;
 	bool		track_commit_timestamp;
+	bool		enable_csn_snapshot;
 
 	/*
 	 * This data is used to check for hardware-architecture compatibility of
diff --git a/src/include/storage/lwlock.h b/src/include/storage/lwlock.h
index a8f052e484..65d1e49fb2 100644
--- a/src/include/storage/lwlock.h
+++ b/src/include/storage/lwlock.h
@@ -168,6 +168,7 @@ typedef enum BuiltinTrancheIds
 	LWTRANCHE_XACT_BUFFER = NUM_INDIVIDUAL_LWLOCKS,
 	LWTRANCHE_COMMITTS_BUFFER,
 	LWTRANCHE_SUBTRANS_BUFFER,
+	LWTRANCHE_CSN_LOG_BUFFERS,
 	LWTRANCHE_MULTIXACTOFFSET_BUFFER,
 	LWTRANCHE_MULTIXACTMEMBER_BUFFER,
 	LWTRANCHE_NOTIFY_BUFFER,
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index 2fd1ff09a7..f0e920f462 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -17,6 +17,7 @@
 #include "access/clog.h"
 #include "access/xlogdefs.h"
 #include "lib/ilist.h"
+#include "utils/snapshot.h"
 #include "storage/latch.h"
 #include "storage/lock.h"
 #include "storage/pg_sema.h"
diff --git a/src/include/storage/sync.h b/src/include/storage/sync.h
index fbdf34f762..e49dd48fe8 100644
--- a/src/include/storage/sync.h
+++ b/src/include/storage/sync.h
@@ -39,6 +39,7 @@ typedef enum SyncRequestHandler
 	SYNC_HANDLER_COMMIT_TS,
 	SYNC_HANDLER_MULTIXACT_OFFSET,
 	SYNC_HANDLER_MULTIXACT_MEMBER,
+	SYNC_HANDLER_CSN_LOG,
 	SYNC_HANDLER_NONE
 } SyncRequestHandler;
 
diff --git a/src/include/utils/snapshot.h b/src/include/utils/snapshot.h
index 6b60755c53..92575a207f 100644
--- a/src/include/utils/snapshot.h
+++ b/src/include/utils/snapshot.h
@@ -121,6 +121,11 @@ typedef enum SnapshotType
 typedef struct SnapshotData *Snapshot;
 
 #define InvalidSnapshot		((Snapshot) NULL)
+#define InvalidCSN			((CSN) 0)
+typedef uint64 CSN;
+
+extern bool enable_csn_snapshot;
+extern bool enable_csn_wal;
 
 /*
  * Struct representing all kind of possible snapshots.
@@ -208,6 +213,12 @@ typedef struct SnapshotData
 	TimestampTz whenTaken;		/* timestamp when snapshot was taken */
 	XLogRecPtr	lsn;			/* position in the WAL stream when taken */
 
+	/*
+	 * CSN for snapshot isolation support.
+	 * Will be used only if enable_csn_snapshot is enabled.
+	 */
+	CSN	csn;
+
 	/*
 	 * The transaction completion count at the time GetSnapshotData() built
 	 * this snapshot. Allows to avoid re-computing static snapshots when no
diff --git a/src/test/regress/expected/sysviews.out b/src/test/regress/expected/sysviews.out
index a62bf5dc92..e21403028d 100644
--- a/src/test/regress/expected/sysviews.out
+++ b/src/test/regress/expected/sysviews.out
@@ -96,6 +96,8 @@ select name, setting from pg_settings where name like 'enable%';
               name              | setting 
 --------------------------------+---------
  enable_bitmapscan              | on
+ enable_csn_snapshot            | off
+ enable_csn_wal                 | on
  enable_gathermerge             | on
  enable_hashagg                 | on
  enable_hashjoin                | on
@@ -114,7 +116,7 @@ select name, setting from pg_settings where name like 'enable%';
  enable_seqscan                 | on
  enable_sort                    | on
  enable_tidscan                 | on
-(19 rows)
+(21 rows)
 
 -- Test that the pg_timezone_names and pg_timezone_abbrevs views are
 -- more-or-less working.  We can't test their contents in any great detail
-- 
2.25.1

  [text/x-patch] 0002-CSN-Snapshot.patch (63.2K, ../0e65f96f-bfd6-fb09-89b7-e07854c215e6@postgrespro.ru/3-0002-CSN-Snapshot.patch)
  download | inline diff:
From 22d80716f06e1b6000f788b5b96da70c0d7aa581 Mon Sep 17 00:00:00 2001
From: Andrey Lepikhov <a.lepikhov@postgrespro.ru>
Date: Mon, 22 Mar 2021 16:36:38 +0500
Subject: [PATCH 2/4] CSN Snapshot

---
 doc/src/sgml/config.sgml                  |  50 +-
 src/backend/access/transam/Makefile       |   1 +
 src/backend/access/transam/csn_log.c      |  30 +-
 src/backend/access/transam/csn_snapshot.c | 724 ++++++++++++++++++++++
 src/backend/access/transam/twophase.c     | 158 +++++
 src/backend/access/transam/xact.c         |  29 +
 src/backend/access/transam/xlog.c         |  26 +-
 src/backend/storage/ipc/ipci.c            |   3 +
 src/backend/storage/ipc/procarray.c       |  75 +++
 src/backend/storage/lmgr/lwlocknames.txt  |   1 +
 src/backend/storage/lmgr/proc.c           |   5 +
 src/backend/utils/misc/guc.c              |  21 +
 src/backend/utils/time/snapmgr.c          | 146 ++++-
 src/bin/pg_waldump/csnlogdesc.c           |   1 +
 src/include/access/csn_log.h              |   3 +-
 src/include/access/csn_snapshot.h         |  65 ++
 src/include/catalog/pg_control.h          |   6 +
 src/include/catalog/pg_proc.dat           |  14 +
 src/include/datatype/timestamp.h          |   2 +
 src/include/fmgr.h                        |   1 +
 src/include/storage/proc.h                |  13 +
 src/include/storage/procarray.h           |   3 +
 src/include/utils/snapmgr.h               |   2 +
 src/include/utils/snapshot.h              |   3 +-
 24 files changed, 1367 insertions(+), 15 deletions(-)
 create mode 100644 src/backend/access/transam/csn_snapshot.c
 create mode 120000 src/bin/pg_waldump/csnlogdesc.c
 create mode 100644 src/include/access/csn_snapshot.h

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 5679b40dd5..fd421ea84d 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -9409,8 +9409,56 @@ dynamic_library_path = 'C:\tools\postgresql;H:\my_project\lib;$libdir'
      </varlistentry>
 
      </variablelist>
-   </sect1>
 
+    <sect2 id="runtime-config-CSN-base-snapshot">
+     <title>CSN Based Snapshot</title>
+
+      <para>
+       By default, The snapshots in <productname>PostgreSQL</productname> uses the
+       XID (TransactionID) to identify the status of the transaction, the in-progress
+       transactions, and the future transactions for all its visibility calculations.
+      </para>
+
+      <para>
+       <productname>PostgreSQL</productname> also provides the CSN (commit-sequence-number)
+        based mechanism to identify the past-transactions and the ones that are yet to
+        be started/committed.
+      </para>
+
+     <variablelist>
+      <varlistentry id="guc-enable-csn-snapshot" xreflabel="enable_csn_snapshot">
+       <term><varname>enable_csn_snapshot</varname> (<type>boolean</type>)
+        <indexterm>
+         <primary><varname>enable_csn_snapshot</varname> configuration parameter</primary>
+        </indexterm>
+       </term>
+       <listitem>
+
+        <para>
+        Enable/disable the CSN based transaction visibility tracking for the snapshot.
+        </para>
+
+        <para>
+        <productname>PostgreSQL</productname> uses the clock timestamp as a CSN,
+        so enabling the CSN based snapshots can be useful for implementing the global
+        snapshots and global transaction visibility.
+        </para>
+
+        <para>
+         when enabled <productname>PostgreSQL</productname> creates
+         <filename>pg_csn</filename> directory under <envar>PGDATA</envar> to keep
+         the track of CSN and XID mappings.
+        </para>
+
+        <para>
+         The default value is off.
+        </para>
+       </listitem>
+      </varlistentry>
+
+     </variablelist>
+    </sect2>
+   </sect1>
    <sect1 id="runtime-config-compatible">
     <title>Version and Platform Compatibility</title>
 
diff --git a/src/backend/access/transam/Makefile b/src/backend/access/transam/Makefile
index d1196cba87..fc0321ee6b 100644
--- a/src/backend/access/transam/Makefile
+++ b/src/backend/access/transam/Makefile
@@ -16,6 +16,7 @@ OBJS = \
 	clog.o \
 	commit_ts.o \
 	csn_log.o \
+	csn_snapshot.o \
 	generic_xlog.o \
 	multixact.o \
 	parallel.o \
diff --git a/src/backend/access/transam/csn_log.c b/src/backend/access/transam/csn_log.c
index 7a6d63e953..9511ecf48a 100644
--- a/src/backend/access/transam/csn_log.c
+++ b/src/backend/access/transam/csn_log.c
@@ -659,20 +659,38 @@ get_xmin_for_csn(void)
 }
 
 void
-prepare_csn_env(bool enable)
+prepare_csn_env(bool enable, bool same, TransactionId *xmin_for_csn_in_control)
 {
 	if (enable)
 	{
-		TransactionId nextxid =
+		if (same)
+		{
+			/*
+			 * Database startup with no enable_csn_snapshot change and value is true,
+			 * it can just transmit xmin_for_csn from pg_control to csnState->xmin_for_csn.
+			 */
+			csnShared->xmin_for_csn = *xmin_for_csn_in_control;
+		}
+		else
+		{
+			TransactionId nextxid =
 						XidFromFullTransactionId(ShmemVariableCache->nextXid);
-		/* 'xmin_for_csn' for when turn xid-snapshot to csn-snapshot */
-		csnShared->xmin_for_csn = nextxid;
-		/* produce the csnlog segment we want now and seek to current page */
-		ActivateCSNlog();
+
+			/* 'xmin_for_csn' for when turn xid-snapshot to csn-snapshot */
+			csnShared->xmin_for_csn = nextxid;
+			*xmin_for_csn_in_control = nextxid;
+
+			/* produce the csnlog segment we want now and seek to current page */
+			ActivateCSNlog();
+		}
 	}
 	else
+	{
 		/* Try to drop all csnlog seg */
 		DeactivateCSNlog();
+		/* Clear xmin_for_csn in pg_control because we are xid-base snaposhot now. */
+		*xmin_for_csn_in_control = InvalidTransactionId;
+	}
 }
 
 /*
diff --git a/src/backend/access/transam/csn_snapshot.c b/src/backend/access/transam/csn_snapshot.c
new file mode 100644
index 0000000000..33d769857f
--- /dev/null
+++ b/src/backend/access/transam/csn_snapshot.c
@@ -0,0 +1,724 @@
+/*-------------------------------------------------------------------------
+ *
+ * csn_snapshot.c
+ *		Support for cross-node snapshot isolation.
+ *
+ * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/backend/access/transam/csn_snapshot.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include <sys/time.h>
+
+#include "access/csn_log.h"
+#include "access/csn_snapshot.h"
+#include "access/transam.h"
+#include "access/twophase.h"
+#include "access/xact.h"
+#include "storage/lmgr.h"
+#include "storage/proc.h"
+#include "storage/procarray.h"
+#include "storage/shmem.h"
+#include "utils/builtins.h"
+#include "utils/guc.h"
+#include "utils/snapmgr.h"
+#include "miscadmin.h"
+
+/* Raise a warning if imported global_csn exceeds ours by this value. */
+#define SNAP_DESYNC_COMPLAIN (1*NSECS_PER_SEC) /* 1 second */
+
+TransactionId 	 xmin_for_csn = InvalidTransactionId;
+
+/*
+ * GUC to delay advance of oldestXid for this amount of time. Also determines
+ * the size CSNSnapshotXidMap circular buffer.
+ */
+int csn_snapshot_defer_time;
+
+/* DEBUG variable to simplify time skew modelling in tests. */
+int csn_time_shift;
+
+/*
+ * CSNSnapshotXidMap
+ *
+ * To be able to install csn snapshot that points to past we need to keep
+ * old versions of tuples and therefore delay advance of oldestXid.  Here we
+ * keep track of correspondence between snapshot's snapshot_csn and oldestXid
+ * that was set at the time when the snapshot was taken.  Much like the
+ * snapshot too old's OldSnapshotControlData does, but with finer granularity
+ * to seconds.
+ *
+ * Different strategies can be employed to hold oldestXid (e.g. we can track
+ * oldest csn-based snapshot among cluster nodes and map it oldestXid
+ * on each node).
+ *
+ * On each snapshot acquisition CSNSnapshotMapXmin() is called and stores
+ * correspondence between current snapshot_csn and oldestXmin in a sparse way:
+ * snapshot_csn is rounded to seconds (and here we use the fact that snapshot_csn
+ * is just a timestamp) and oldestXmin is stored in the circular buffer where
+ * rounded snapshot_csn acts as an offset from current circular buffer head.
+ * Size of the circular buffer is controlled by csn_snapshot_defer_time GUC.
+ *
+ * When csn snapshot arrives we check that its
+ * snapshot_csn is still in our map, otherwise we'll error out with "snapshot too
+ * old" message.  If snapshot_csn is successfully mapped to oldestXid we move
+ * backend's pgxact->xmin to proc->originalXmin and fill pgxact->xmin to
+ * mapped oldestXid.  That way GetOldestXmin() can take into account backends
+ * with imported csn snapshot and old tuple versions will be preserved.
+ *
+ * Also while calculating oldestXmin for our map in presence of imported
+ * csn snapshots we should use proc->originalXmin instead of pgxact->xmin
+ * that was set during import.  Otherwise, we can create a feedback loop:
+ * xmin's of imported csn snapshots were calculated using our map and new
+ * entries in map going to be calculated based on that xmin's, and there is
+ * a risk to stuck forever with one non-increasing oldestXmin.  All other
+ * callers of GetOldestXmin() are using pgxact->xmin so the old tuple versions
+ * are preserved.
+ */
+typedef struct CSNSnapshotXidMap
+{
+	int				 head;				/* offset of current freshest value */
+	int				 size;				/* total size of circular buffer */
+	CSN_atomic		 last_csn_seconds;	/* last rounded csn that changed
+										 * xmin_by_second[] */
+	TransactionId   *xmin_by_second;	/* circular buffer of oldestXmin's */
+}
+CSNSnapshotXidMap;
+
+static CSNSnapshotXidMap *csnXidMap;
+
+/* Estimate shared memory space needed */
+Size
+CSNSnapshotShmemSize(void)
+{
+	Size	size = 0;
+
+	if (csn_snapshot_defer_time > 0)
+	{
+		size += sizeof(CSNSnapshotXidMap);
+		size += csn_snapshot_defer_time * sizeof(TransactionId);
+		size = MAXALIGN(size);
+	}
+
+	return size;
+}
+
+/* Init shared memory structures */
+void
+CSNSnapshotShmemInit(void)
+{
+	bool found;
+
+	if (csn_snapshot_defer_time > 0)
+	{
+		csnXidMap = ShmemInitStruct("gsXidMap",
+								   sizeof(CSNSnapshotXidMap),
+								   &found);
+		if (!found)
+		{
+			int i;
+
+			pg_atomic_init_u64(&csnXidMap->last_csn_seconds,
+							   (uint64) get_last_max_csn());
+			csnXidMap->head = 0;
+			csnXidMap->size = csn_snapshot_defer_time;
+			csnXidMap->xmin_by_second =
+							ShmemAlloc(sizeof(TransactionId) * csnXidMap->size);
+
+			for (i = 0; i < csnXidMap->size; i++)
+				csnXidMap->xmin_by_second[i] = InvalidTransactionId;
+		}
+	}
+}
+
+/*
+ * GenerateCSN
+ *
+ * Generate CSN which is actually a local time. Also we are forcing
+ * this time to be always increasing. For portability reasons gettimeofday()
+ * and milliseconds resolution are used, but always increasing time helps to
+ * handle multiple transactions in one ms.
+ */
+CSN
+GenerateCSN(CSN assign)
+{
+	struct timeval  current_time;
+	CSN	csn;
+
+	Assert(get_csnlog_status() || csn_snapshot_defer_time > 0);
+
+	if(assign != InvalidCSN && assign < get_last_max_csn())
+	{
+		/*
+		 * While assign not 0, we just want to make sure we log the max csn
+		 * in last_max_csn. So we not care about the return value.
+		 *
+		 * We can return an Invalid value.
+		 */
+		return InvalidCSN;
+	}
+
+	/*
+	 * TODO: create some macro that add small random shift to current time.
+	 */
+	gettimeofday(&current_time, NULL);
+	csn = (CSN) (((uint64) (current_time.tv_sec + csn_time_shift) *
+		(uint64) 1000000000) + (uint64) (current_time).tv_usec * (uint64) 1000);
+
+	if (assign != InvalidCSN && csn < assign)
+		csn = assign;
+
+	csn = set_last_max_csn(csn, false);
+	WriteAssignCSNXlogRec(csn);
+	return csn;
+}
+
+/*
+ * CSNSnapshotPrepareCurrent
+ *
+ * Set InDoubt state for currently active transaction and return commit's
+ * global snapshot.
+ */
+CSN
+CSNSnapshotPrepareCurrent(void)
+{
+	TransactionId xid = GetCurrentTransactionIdIfAny();
+
+	if (!enable_csn_snapshot)
+		ereport(ERROR,
+			(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				errmsg("could not prepare transaction for global commit"),
+				errhint("Make sure the configuration parameter \"%s\" is enabled.",
+						"enable_csn_snapshot")));
+
+	if (TransactionIdIsValid(xid))
+	{
+		TransactionId *subxids;
+		int nsubxids = xactGetCommittedChildren(&subxids);
+		CSNLogSetCSN(xid, nsubxids, subxids, InDoubtCSN, true);
+	}
+
+	/* Nothing to write if we don't have xid */
+
+	return GenerateCSN(InvalidCSN);
+}
+
+
+/*
+ * CSNSnapshotAssignCurrent
+ *
+ * Assign CSN to the currently active transaction. CSN is supposedly
+ * maximal among of values returned by CSNSnapshotPrepareCurrent and
+ * pg_csn_snapshot_prepare.
+ */
+void
+CSNSnapshotAssignCurrent(CSN csn)
+{
+	if (!enable_csn_snapshot)
+		ereport(ERROR,
+			(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				errmsg("could not prepare transaction for global commit"),
+				errhint("Make sure the configuration parameter \"%s\" is enabled.",
+						"enable_csn_snapshot")));
+
+	if (!CSNIsNormal(csn))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("pg_csn_snapshot_assign expects normal csn")));
+
+	Assert(csn != InvalidCSN);
+	/* We do not care the Generate result, we just want to make sure max
+	 * csnState->last_max_csn value.
+	 */
+	GenerateCSN(csn);
+
+	/* Set csn and defuse ProcArrayEndTransaction from assigning one */
+	pg_atomic_write_u64(&MyProc->assignedCSN, csn);
+}
+
+/*
+ * CSNSnapshotSync
+ *
+ * Due to time desynchronization on different nodes we can receive snapshot_csn
+ * which is greater than snapshot_csn on this node. To preserve proper isolation
+ * this node needs to wait when such snapshot_csn comes on local clock.
+ *
+ * This should happend relatively rare if nodes have running NTP/PTP/etc.
+ * Complain if wait time is more than SNAP_SYNC_COMPLAIN.
+ */
+void
+CSNSnapshotSync(CSN remote_csn)
+{
+	CSN	local_csn;
+	CSN	delta;
+
+	Assert(enable_csn_snapshot);
+
+	for(;;)
+	{
+		if (get_last_max_csn() > remote_csn)
+			/* Everything is fine */
+			return;
+		else if ((local_csn = GenerateCSN(InvalidCSN)) >= remote_csn)
+			/*
+			 * Everything is fine too, but last_max_csn wasn't updated for
+			 * some time.
+			 */
+			return;
+
+		/* Okay we need to sleep now */
+		delta = remote_csn - local_csn;
+		if (delta > SNAP_DESYNC_COMPLAIN)
+			ereport(WARNING,
+				(errmsg("remote global snapshot exceeds ours by more than a second"),
+				 errhint("Consider running NTPd on servers participating in global transaction")));
+
+		/* TODO: report this sleeptime somewhere? */
+		pg_usleep((long) (delta/NSECS_PER_USEC));
+
+		/*
+		 * Loop that checks to ensure that we actually slept for specified
+		 * amount of time.
+		 */
+	}
+
+	Assert(false); /* Should not happend */
+	return;
+}
+
+/*
+ * TransactionIdGetCSN
+ *
+ * Get CSN for specified TransactionId taking care about special xids,
+ * xids beyond TransactionXmin and InDoubt states.
+ */
+CSN
+TransactionIdGetCSN(TransactionId xid)
+{
+	CSN csn;
+
+	Assert(get_csnlog_status());
+
+	/* Handle permanent TransactionId's for which we don't have mapping */
+	if (!TransactionIdIsNormal(xid))
+	{
+		if (xid == InvalidTransactionId)
+			return AbortedCSN;
+		if (xid == FrozenTransactionId || xid == BootstrapTransactionId)
+			return FrozenCSN;
+		Assert(false); /* Should not happend */
+	}
+
+	/*
+	 * If we just switch a xid-snapsot to a csn_snapshot, we should handle a start
+	 * xid for csn base check. Just in case we have prepared transaction which
+	 * hold the TransactionXmin but without CSN.
+	 */
+	if (xmin_for_csn == InvalidTransactionId)
+	{
+		csn = get_xmin_for_csn();
+
+		/*
+		 * If the instance started with enabled csn snapshots, use csn of its
+		 * start point. If csn snapshots were enabled in this backend only, use
+		 * xmin of this transaction.
+		 */
+		xmin_for_csn = (csn != InvalidTransactionId) ? csn : TransactionXmin;
+	}
+
+	/*
+	 * For the xid with 'xid >= TransactionXmin and xid < xmin_for_csn',
+	 * it defined as unclear csn which follow xid-snapshot result.
+	 */
+	if(!TransactionIdPrecedes(xid, TransactionXmin) &&
+							TransactionIdPrecedes(xid, xmin_for_csn))
+		return UnclearCSN;
+
+	/*
+	 * For xids which less then TransactionXmin CSNLog can be already
+	 * trimmed but we know that such transaction is definetly not concurrently
+	 * running according to any snapshot including timetravel ones. Callers
+	 * should check TransactionDidCommit after.
+	 */
+	if (TransactionIdPrecedes(xid, xmin_for_csn))
+		return FrozenCSN;
+
+	/* Read CSN from SLRU */
+	csn = CSNLogGetCSNByXid(xid);
+
+	/*
+	 * If we faced InDoubt state then transaction is being committed and we
+	 * should wait until CSN will be assigned so that visibility check
+	 * could decide whether tuple is in snapshot. See also comments in
+	 * CSNSnapshotPrecommit().
+	 */
+	if (CSNIsInDoubt(csn))
+	{
+		XactLockTableWait(xid, NULL, NULL, XLTW_None);
+		csn = CSNLogGetCSNByXid(xid);
+		Assert(CSNIsNormal(csn) || CSNIsAborted(csn));
+	}
+
+	Assert(CSNIsNormal(csn) || CSNIsInProgress(csn) || CSNIsAborted(csn));
+
+	return csn;
+}
+
+/*
+ * CSNSnapshotStartup
+ *
+ * Set csnXidMap entries to oldestActiveXID during startup.
+ */
+void
+CSNSnapshotStartup(TransactionId oldestActiveXID)
+{
+	/*
+	 * Run only if we have initialized shared memory and csnXidMap
+	 * is enabled.
+	 */
+	if (IsNormalProcessingMode() &&
+		enable_csn_snapshot && csn_snapshot_defer_time > 0)
+	{
+		int i;
+
+		Assert(TransactionIdIsValid(oldestActiveXID));
+		for (i = 0; i < csnXidMap->size; i++)
+			csnXidMap->xmin_by_second[i] = oldestActiveXID;
+
+		ProcArraySetCSNSnapshotXmin(oldestActiveXID);
+	}
+}
+
+/*
+ * CSNSnapshotMapXmin
+ *
+ * Maintain circular buffer of oldestXmins for several seconds in past. This
+ * buffer allows to shift oldestXmin in the past when backend is importing
+ * CSN snapshot. Otherwise old versions of tuples that were needed for
+ * this transaction can be recycled by other processes (vacuum, HOT, etc).
+ *
+ * Locking here is not trivial. Called upon each snapshot creation after
+ * ProcArrayLock is released. Such usage creates several race conditions. It
+ * is possible that backend who got csn called CSNSnapshotMapXmin()
+ * only after other backends managed to get snapshot and complete
+ * CSNSnapshotMapXmin() call, or even committed. This is safe because
+ *
+ *		* We already hold our xmin in MyProc, so our snapshot will not be
+ *		  harmed even though ProcArrayLock is released.
+ *
+ *		* snapshot_csn is always pessimistically rounded up to the next
+ *		  second.
+ *
+ *		* For performance reasons, xmin value for particular second is filled
+ *		  only once. Because of that instead of writing to buffer just our
+ *		  xmin (which is enough for our snapshot), we bump oldestXmin there --
+ *		  it mitigates the possibility of damaging someone else's snapshot by
+ *		  writing to the buffer too advanced value in case of slowness of
+ *		  another backend who generated csn earlier, but didn't manage to
+ *		  insert it before us.
+ *
+ *		* if CSNSnapshotMapXmin() founds a gap in several seconds between
+ *		  current call and latest completed call then it should fill that gap
+ *		  with latest known values instead of new one. Otherwise it is
+ *		  possible (however highly unlikely) that this gap also happend
+ *		  between taking snapshot and call to CSNSnapshotMapXmin() for some
+ *		  backend. And we are at risk to fill circullar buffer with
+ *		  oldestXmin's that are bigger then they actually were.
+ */
+void
+CSNSnapshotMapXmin(CSN snapshot_csn)
+{
+	int offset, gap, i;
+	CSN csn_seconds;
+	CSN last_csn_seconds;
+	volatile TransactionId oldest_deferred_xmin;
+	TransactionId current_oldest_xmin;
+	TransactionId previous_oldest_xmin;
+
+	/* Callers should check config values */
+	Assert(csn_snapshot_defer_time > 0);
+	Assert(csnXidMap != NULL);
+	/*
+	 * Round up snapshot_csn to the next second -- pessimistically and safely.
+	 */
+	csn_seconds = (snapshot_csn / NSECS_PER_SEC + 1);
+
+	/*
+	 * Fast-path check. Avoid taking exclusive CSNSnapshotXidMapLock lock
+	 * if oldestXid was already written to xmin_by_second[] for this rounded
+	 * snapshot_csn.
+	 */
+	if (pg_atomic_read_u64(&csnXidMap->last_csn_seconds) >= csn_seconds)
+		return;
+
+	/* Ok, we have new entry (or entries) */
+	LWLockAcquire(CSNSnapshotXidMapLock, LW_EXCLUSIVE);
+
+	/* Re-check last_csn_seconds under lock */
+	last_csn_seconds = pg_atomic_read_u64(&csnXidMap->last_csn_seconds);
+	if (last_csn_seconds >= csn_seconds)
+	{
+		LWLockRelease(CSNSnapshotXidMapLock);
+		return;
+	}
+	pg_atomic_write_u64(&csnXidMap->last_csn_seconds, csn_seconds);
+
+	/*
+	 * Count oldest_xmin.
+	 *
+	 * It was possible to calculate oldest_xmin during corresponding snapshot
+	 * creation, but GetSnapshotData() intentionally reads only PgXact, but not
+	 * PgProc. And we need info about originalXmin (see comment to csnXidMap)
+	 * which is stored in PgProc because of threats in comments around PgXact
+	 * about extending it with new fields. So just calculate oldest_xmin again,
+	 * that anyway happens quite rarely.
+	 */
+	current_oldest_xmin = GetOldestNonRemovableTransactionId(NULL);
+	Assert(TransactionIdIsNormal(current_oldest_xmin));
+
+	previous_oldest_xmin = csnXidMap->xmin_by_second[csnXidMap->head];
+	Assert(TransactionIdIsNormal(previous_oldest_xmin) || !enable_csn_snapshot);
+
+	gap = csn_seconds - last_csn_seconds;
+	offset = csn_seconds % csnXidMap->size;
+
+	/* Sanity check before we update head and gap */
+	Assert( gap >= 1 );
+	Assert( (csnXidMap->head + gap) % csnXidMap->size == offset );
+
+	gap = gap > csnXidMap->size ? csnXidMap->size : gap;
+	csnXidMap->head = offset;
+
+	/* Fill new entry with current_oldest_xmin */
+	csnXidMap->xmin_by_second[offset] = current_oldest_xmin;
+
+	/*
+	 * If we have gap then fill it with previous_oldest_xmin for reasons
+	 * outlined in comment above this function.
+	 */
+	for (i = 1; i < gap; i++)
+	{
+		offset = (offset + csnXidMap->size - 1) % csnXidMap->size;
+		csnXidMap->xmin_by_second[offset] = previous_oldest_xmin;
+	}
+
+	oldest_deferred_xmin =
+		csnXidMap->xmin_by_second[ (csnXidMap->head + 1) % csnXidMap->size ];
+
+	LWLockRelease(CSNSnapshotXidMapLock);
+
+	/*
+	 * Advance procArray->csn_snapshot_xmin after we released
+	 * CSNSnapshotXidMapLock. Since we gather not xmin but oldestXmin, it
+	 * never goes backwards regardless of how slow we can do that.
+	 */
+	Assert(TransactionIdFollowsOrEquals(oldest_deferred_xmin,
+										ProcArrayGetCSNSnapshotXmin()));
+	ProcArraySetCSNSnapshotXmin(oldest_deferred_xmin);
+}
+
+
+/*
+ * CSNSnapshotToXmin
+ *
+ * Get oldestXmin that took place when snapshot_csn was taken.
+ */
+TransactionId
+CSNSnapshotToXmin(CSN snapshot_csn)
+{
+	TransactionId xmin;
+	CSN csn_seconds;
+	volatile CSN last_csn_seconds;
+
+	/* Callers should check config values */
+	Assert(csn_snapshot_defer_time > 0);
+	Assert(csnXidMap != NULL);
+
+	/* Round down to get conservative estimates */
+	csn_seconds = (snapshot_csn / NSECS_PER_SEC);
+
+	LWLockAcquire(CSNSnapshotXidMapLock, LW_SHARED);
+	last_csn_seconds = pg_atomic_read_u64(&csnXidMap->last_csn_seconds);
+	if (csn_seconds > last_csn_seconds)
+	{
+		/* we don't have entry for this snapshot_csn yet, return latest known */
+		xmin = csnXidMap->xmin_by_second[csnXidMap->head];
+	}
+	else if (last_csn_seconds - csn_seconds < csnXidMap->size)
+	{
+		/* we are good, retrieve value from our map */
+		Assert(last_csn_seconds % csnXidMap->size == csnXidMap->head);
+		xmin = csnXidMap->xmin_by_second[csn_seconds % csnXidMap->size];
+	}
+	else
+	{
+		/* requested snapshot_csn is too old, let caller know */
+		xmin = InvalidTransactionId;
+	}
+	LWLockRelease(CSNSnapshotXidMapLock);
+
+	return xmin;
+}
+
+/*
+ * XidInvisibleInCSNSnapshot
+ *
+ * Version of XidInMVCCSnapshot for transactions. For non-imported
+ * csn snapshots this should give same results as XidInLocalMVCCSnapshot
+ * (except that aborts will be shown as invisible without going to clog) and to
+ * ensure such behaviour XidInMVCCSnapshot is coated with asserts that checks
+ * identicalness of XidInvisibleInCSNSnapshot/XidInLocalMVCCSnapshot in
+ * case of ordinary snapshot.
+ */
+bool
+XidInvisibleInCSNSnapshot(TransactionId xid, Snapshot snapshot)
+{
+	CSN csn;
+
+	Assert(get_csnlog_status());
+
+	csn = TransactionIdGetCSN(xid);
+
+	if (CSNIsNormal(csn))
+		return (csn < snapshot->csn) ? false : true;
+	else if (CSNIsFrozen(csn))
+		/* It is bootstrap or frozen transaction */
+		return false;
+	else if (CSNIsUnclear(csn))
+		/*
+		 * Some xid can not figure out csn because of snapshot switch,
+		 * and we can follow xid-base result.
+		 */
+		return true;
+	else
+	{
+		/* It is aborted or in-progress */
+		Assert(CSNIsAborted(csn) || CSNIsInProgress(csn));
+		if (CSNIsAborted(csn))
+			Assert(TransactionIdDidAbort(xid));
+		return true;
+	}
+}
+
+
+/*****************************************************************************
+ * Functions to handle transactions commit.
+ *
+ * For local transactions CSNSnapshotPrecommit sets InDoubt state before
+ * ProcArrayEndTransaction is called and transaction data potetntially becomes
+ * visible to other backends. ProcArrayEndTransaction (or ProcArrayRemove in
+ * twophase case) then acquires csn under ProcArray lock and stores it
+ * in proc->assignedCSN. It's important that csn for commit is
+ * generated under ProcArray lock, otherwise snapshots won't
+ * be equivalent. Consequent call to CSNSnapshotCommit will write
+ * proc->assignedCSN to CSNLog.
+ *
+ *
+ * CSNSnapshotAbort is slightly different comparing to commit because abort
+ * can skip InDoubt phase and can be called for transaction subtree.
+ *****************************************************************************/
+
+
+/*
+ * CSNSnapshotAbort
+ *
+ * Abort transaction in CsnLog. We can skip InDoubt state for aborts
+ * since no concurrent transactions allowed to see aborted data anyway.
+ */
+void
+CSNSnapshotAbort(PGPROC *proc, TransactionId xid,
+					int nsubxids, TransactionId *subxids)
+{
+	if (!get_csnlog_status())
+		return;
+
+	CSNLogSetCSN(xid, nsubxids, subxids, AbortedCSN, true);
+
+	/*
+	 * Clean assignedCSN anyway, as it was possibly set in
+	 * XidSnapshotAssignCsnCurrent.
+	 */
+	pg_atomic_write_u64(&proc->assignedCSN, InProgressCSN);
+}
+
+/*
+ * CSNSnapshotPrecommit
+ *
+ * Set InDoubt status for local transaction that we are going to commit.
+ * This step is needed to achieve consistency between local snapshots and
+ * csn-based snapshots. We don't hold ProcArray lock while writing
+ * csn for transaction in SLRU but instead we set InDoubt status before
+ * transaction is deleted from ProcArray so the readers who will read csn
+ * in the gap between ProcArray removal and CSN assignment can wait
+ * until CSN is finally assigned. See also TransactionIdGetCSN().
+ *
+ * This should be called only from parallel group leader before backend is
+ * deleted from ProcArray.
+ */
+void
+CSNSnapshotPrecommit(PGPROC *proc, TransactionId xid,
+					int nsubxids, TransactionId *subxids)
+{
+	CSN oldassignedCSN = InProgressCSN;
+	bool in_progress;
+
+	if (!get_csnlog_status())
+		return;
+
+	/* Set InDoubt status if it is local transaction */
+	in_progress = pg_atomic_compare_exchange_u64(&proc->assignedCSN,
+												 &oldassignedCSN,
+												 InDoubtCSN);
+	if (in_progress)
+	{
+		Assert(CSNIsInProgress(oldassignedCSN));
+		CSNLogSetCSN(xid, nsubxids,
+						   subxids, InDoubtCSN, true);
+	}
+	else
+	{
+		/* Otherwise we should have valid CSN by this time */
+		Assert(CSNIsNormal(oldassignedCSN));
+		Assert(CSNIsInDoubt(CSNLogGetCSNByXid(xid)));
+	}
+}
+
+/*
+ * CSNSnapshotCommit
+ *
+ * Write CSN that were acquired earlier to CsnLog. Should be
+ * preceded by CSNSnapshotPrecommit() so readers can wait until we finally
+ * finished writing to SLRU.
+ *
+ * Should be called after ProcArrayEndTransaction, but before releasing
+ * transaction locks, so that TransactionIdGetCSN can wait on this
+ * lock for CSN.
+ */
+void
+CSNSnapshotCommit(PGPROC *proc, TransactionId xid,
+				  int nsubxids, TransactionId *subxids)
+{
+	volatile CSN assignedCSN;
+
+	if (!get_csnlog_status())
+		return;
+
+	if (!TransactionIdIsValid(xid))
+	{
+		assignedCSN = pg_atomic_read_u64(&proc->assignedCSN);
+		Assert(CSNIsInProgress(assignedCSN));
+		return;
+	}
+
+	/* Finally write resulting CSN in SLRU */
+	assignedCSN = pg_atomic_read_u64(&proc->assignedCSN);
+	Assert(CSNIsNormal(assignedCSN));
+	CSNLogSetCSN(xid, nsubxids, subxids, assignedCSN, true);
+
+	/* Reset for next transaction */
+	pg_atomic_write_u64(&proc->assignedCSN, InProgressCSN);
+}
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 89335b64a2..6b91953193 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -77,6 +77,8 @@
 #include <unistd.h>
 
 #include "access/commit_ts.h"
+#include "access/csn_log.h"
+#include "access/csn_snapshot.h"
 #include "access/htup_details.h"
 #include "access/subtrans.h"
 #include "access/transam.h"
@@ -1483,8 +1485,34 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
 									   hdr->nabortrels, abortrels,
 									   gid);
 
+	/*
+	 * CSNSnapshot callbacks that should be called right before we are
+	 * going to become visible. Details in comments to this functions.
+	 */
+	if (isCommit)
+		CSNSnapshotPrecommit(proc, xid, hdr->nsubxacts, children);
+	else
+		CSNSnapshotAbort(proc, xid, hdr->nsubxacts, children);
+
+
 	ProcArrayRemove(proc, latestXid);
 
+	/*
+	 * Stamp our transaction with CSN in CSNLog.
+	 * Should be called after ProcArrayEndTransaction, but before releasing
+	 * transaction locks, since TransactionIdGetCSN relies on
+	 * XactLockTableWait to await csn.
+	 */
+	if (isCommit)
+	{
+		CSNSnapshotCommit(proc, xid, hdr->nsubxacts, children);
+	}
+	else
+	{
+		Assert(CSNIsInProgress(
+				   pg_atomic_read_u64(&proc->assignedCSN)));
+	}
+
 	/*
 	 * In case we fail while running the callbacks, mark the gxact invalid so
 	 * no one else will try to commit/rollback, and so it will be recycled if
@@ -2458,3 +2486,133 @@ PrepareRedoRemove(TransactionId xid, bool giveWarning)
 		RemoveTwoPhaseFile(xid, giveWarning);
 	RemoveGXact(gxact);
 }
+
+/*
+ * CSNSnapshotPrepareTwophase
+ *
+ * Set InDoubt state for currently active transaction and return commit's
+ * global snapshot.
+ */
+static CSN
+CSNSnapshotPrepareTwophase(const char *gid)
+{
+	GlobalTransaction gxact;
+	PGPROC	   *proc;
+	char	   *buf;
+	TransactionId xid;
+	xl_xact_parsed_prepare parsed;
+
+	if (!enable_csn_snapshot)
+		ereport(ERROR,
+			(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				errmsg("could not prepare transaction for global commit"),
+				errhint("Make sure the configuration parameter \"%s\" is enabled.",
+						"enable_csn_snapshot")));
+
+	/*
+	 * Validate the GID, and lock the GXACT to ensure that two backends do not
+	 * try to access the same GID at once.
+	 */
+	gxact = LockGXact(gid, GetUserId());
+	proc = &ProcGlobal->allProcs[gxact->pgprocno];
+	xid = proc->xid;
+
+	if (gxact->ondisk)
+		buf = ReadTwoPhaseFile(xid, true);
+	else
+		XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL);
+
+	ParsePrepareRecord(0, (xl_xact_prepare *)buf, &parsed);
+
+	CSNLogSetCSN(xid, parsed.nsubxacts,
+					parsed.subxacts, InDoubtCSN, true);
+
+	/* Unlock our GXACT */
+	LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
+	gxact->locking_backend = InvalidBackendId;
+	LWLockRelease(TwoPhaseStateLock);
+
+	pfree(buf);
+
+	return GenerateCSN(InvalidCSN);
+}
+
+/*
+ * CSNSnapshotAssignTwoPhase
+ *
+ * Asign CSN for currently active transaction. CSN is supposedly
+ * maximal among of values returned by CSNSnapshotPrepareCurrent and
+ * pg_csn_snapshot_prepare.
+ *
+ * This function is a counterpart of CSNSnapshotAssignCurrent() for
+ * twophase transactions.
+ */
+static void
+CSNSnapshotAssignTwoPhase(const char *gid, CSN csn)
+{
+	GlobalTransaction	gxact;
+	PGPROC				*proc;
+
+	if (!enable_csn_snapshot)
+		ereport(ERROR,
+			(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				errmsg("could not prepare transaction for global commit"),
+				errhint("Make sure the configuration parameter \"%s\" is enabled.",
+						"enable_csn_snapshot")));
+
+	if (!CSNIsNormal(csn))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("pg_csn_snapshot_assign expects normal csn")));
+
+	/*
+	 * Validate the GID, and lock the GXACT to ensure that two backends do not
+	 * try to access the same GID at once.
+	 */
+	gxact = LockGXact(gid, GetUserId());
+	proc = &ProcGlobal->allProcs[gxact->pgprocno];
+
+	Assert(csn != InvalidCSN);
+	/* We do not care the Generate result, we just want to make sure max
+	 * csnState->last_max_csn value.
+	 */
+	GenerateCSN(csn);
+	/* Set csn and defuse ProcArrayRemove from assigning one. */
+	pg_atomic_write_u64(&proc->assignedCSN, csn);
+
+	/* Unlock our GXACT */
+	LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
+	gxact->locking_backend = InvalidBackendId;
+	LWLockRelease(TwoPhaseStateLock);
+}
+
+/*
+ * SQL interface to CSNSnapshotPrepareTwophase()
+ *
+ * TODO: Rewrite this as PREPARE TRANSACTION 'gid' RETURNING SNAPSHOT
+ */
+Datum
+pg_csn_snapshot_prepare(PG_FUNCTION_ARGS)
+{
+	const char 	*gid = text_to_cstring(PG_GETARG_TEXT_PP(0));
+	CSN	csn;
+
+	csn = CSNSnapshotPrepareTwophase(gid);
+
+	PG_RETURN_INT64(csn);
+}
+
+/*
+ * SQL interface to CSNSnapshotAssignTwoPhase()
+ *
+ * TODO: Rewrite this as COMMIT PREPARED 'gid' SNAPSHOT 'csn'
+ */
+Datum
+pg_csn_snapshot_assign(PG_FUNCTION_ARGS)
+{
+	const char *gid = text_to_cstring(PG_GETARG_TEXT_PP(0));
+	CSN	csn = PG_GETARG_INT64(1);
+
+	CSNSnapshotAssignTwoPhase(gid, csn);
+	PG_RETURN_VOID();
+}
\ No newline at end of file
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 6395a9b240..deabedbe37 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -21,6 +21,7 @@
 #include <unistd.h>
 
 #include "access/commit_ts.h"
+#include "access/csn_snapshot.h"
 #include "access/multixact.h"
 #include "access/parallel.h"
 #include "access/subtrans.h"
@@ -1486,6 +1487,14 @@ RecordTransactionCommit(void)
 
 	/* Reset XactLastRecEnd until the next transaction writes something */
 	XactLastRecEnd = 0;
+
+	/*
+	 * Mark our transaction as InDoubt in CsnLog and get ready for
+	 * commit.
+	 */
+	if (markXidCommitted)
+		CSNSnapshotPrecommit(MyProc, xid, nchildren, children);
+
 cleanup:
 	/* Clean up local data */
 	if (rels)
@@ -1747,6 +1756,11 @@ RecordTransactionAbort(bool isSubXact)
 	 */
 	TransactionIdAbortTree(xid, nchildren, children);
 
+	/*
+	 * Mark our transaction as Aborted in CsnLog.
+	 */
+	CSNSnapshotAbort(MyProc, xid, nchildren, children);
+
 	END_CRIT_SECTION();
 
 	/* Compute latestXid while we have the child XIDs handy */
@@ -2237,6 +2251,21 @@ CommitTransaction(void)
 	 */
 	ProcArrayEndTransaction(MyProc, latestXid);
 
+	/*
+	 * Stamp our transaction with CSN in CsnLog.
+	 * Should be called after ProcArrayEndTransaction, but before releasing
+	 * transaction locks.
+	 */
+	if (!is_parallel_worker)
+	{
+		TransactionId xid = GetTopTransactionIdIfAny();
+		TransactionId *subxids;
+		int nsubxids;
+
+		nsubxids = xactGetCommittedChildren(&subxids);
+		CSNSnapshotCommit(MyProc, xid, nsubxids, subxids);
+	}
+
 	/*
 	 * This is all post-commit cleanup.  Note that if an error is raised here,
 	 * it's too late to abort the transaction.  This should be just
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 46f016dce7..200c387084 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -4652,6 +4652,7 @@ InitControlFile(uint64 sysidentifier)
 	ControlFile->track_commit_timestamp = track_commit_timestamp;
 	ControlFile->enable_csn_snapshot = enable_csn_snapshot;
 	ControlFile->data_checksum_version = bootstrap_data_checksum_version;
+	ControlFile->xmin_for_csn = InvalidTransactionId;
 }
 
 static void
@@ -7211,6 +7212,7 @@ StartupXLOG(void)
 			 * maintained during recovery and need not be started yet.
 			 */
 			StartupSUBTRANS(oldestActiveXID);
+			CSNSnapshotStartup(oldestActiveXID);
 
 			/*
 			 * If we're beginning at a shutdown checkpoint, we know that
@@ -8030,7 +8032,10 @@ StartupXLOG(void)
 	 * timestamps are started below, if necessary.)
 	 */
 	if (standbyState == STANDBY_DISABLED)
+	{
 		StartupSUBTRANS(oldestActiveXID);
+		CSNSnapshotStartup(oldestActiveXID);
+	}
 
 	/*
 	 * Perform end of recovery actions for any SLRUs that need it.
@@ -9925,6 +9930,8 @@ XLogRestorePoint(const char *rpName)
 static void
 XLogReportParameters(void)
 {
+	TransactionId xmin_for_csn = ControlFile->xmin_for_csn;
+
 	if (wal_level != ControlFile->wal_level ||
 		wal_log_hints != ControlFile->wal_log_hints ||
 		MaxConnections != ControlFile->MaxConnections ||
@@ -9964,11 +9971,12 @@ XLogReportParameters(void)
 			XLogFlush(recptr);
 		}
 
-		LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
-
-		if (enable_csn_snapshot != ControlFile->enable_csn_snapshot)
-			prepare_csn_env(enable_csn_snapshot);
+		prepare_csn_env(enable_csn_snapshot,
+						enable_csn_snapshot == ControlFile->enable_csn_snapshot,
+						&xmin_for_csn);
 
+		LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
+		ControlFile->xmin_for_csn = xmin_for_csn;
 		ControlFile->MaxConnections = MaxConnections;
 		ControlFile->max_worker_processes = max_worker_processes;
 		ControlFile->max_wal_senders = max_wal_senders;
@@ -9982,6 +9990,16 @@ XLogReportParameters(void)
 
 		LWLockRelease(ControlFileLock);
 	}
+	else
+	{
+		/*
+		 * When no GUC change, but for xmin_for_csn it should transmit the xmin_for_csn
+		 * from pg_control to csnState->xmin_for_csn. Or it will cause issue when prepare
+		 * transaction exists and with 'xid-snapshot start  ->  csn-snapshot start  ->
+		 * csn-snapshot start' sequence.
+		 */
+		prepare_csn_env(enable_csn_snapshot, true, &xmin_for_csn);
+	}
 }
 
 /*
diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c
index e06bc13841..bbc5f1adf7 100644
--- a/src/backend/storage/ipc/ipci.c
+++ b/src/backend/storage/ipc/ipci.c
@@ -17,6 +17,7 @@
 #include "access/clog.h"
 #include "access/commit_ts.h"
 #include "access/csn_log.h"
+#include "access/csn_snapshot.h"
 #include "access/heapam.h"
 #include "access/multixact.h"
 #include "access/nbtree.h"
@@ -152,6 +153,7 @@ CreateSharedMemoryAndSemaphores(void)
 		size = add_size(size, BTreeShmemSize());
 		size = add_size(size, SyncScanShmemSize());
 		size = add_size(size, AsyncShmemSize());
+		size = add_size(size, CSNSnapshotShmemSize());
 #ifdef EXEC_BACKEND
 		size = add_size(size, ShmemBackendArraySize());
 #endif
@@ -272,6 +274,7 @@ CreateSharedMemoryAndSemaphores(void)
 	BTreeShmemInit();
 	SyncScanShmemInit();
 	AsyncShmemInit();
+	CSNSnapshotShmemInit();
 
 #ifdef EXEC_BACKEND
 
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index 4a200b0815..62138fddcc 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -49,6 +49,7 @@
 
 #include "access/clog.h"
 #include "access/csn_log.h"
+#include "access/csn_snapshot.h"
 #include "access/subtrans.h"
 #include "access/transam.h"
 #include "access/twophase.h"
@@ -98,6 +99,9 @@ typedef struct ProcArrayStruct
 	/* oldest catalog xmin of any replication slot */
 	TransactionId replication_slot_catalog_xmin;
 
+	/* xmin of oldest active csn snapshot */
+	TransactionId csn_snapshot_xmin;
+
 	/* indexes into allProcs[], has PROCARRAY_MAXPROCS entries */
 	int			pgprocnos[FLEXIBLE_ARRAY_MEMBER];
 } ProcArrayStruct;
@@ -419,6 +423,7 @@ CreateSharedProcArray(void)
 		procArray->lastOverflowedXid = InvalidTransactionId;
 		procArray->replication_slot_xmin = InvalidTransactionId;
 		procArray->replication_slot_catalog_xmin = InvalidTransactionId;
+		procArray->csn_snapshot_xmin = InvalidTransactionId;
 		ShmemVariableCache->xactCompletionCount = 1;
 	}
 
@@ -547,6 +552,14 @@ ProcArrayRemove(PGPROC *proc, TransactionId latestXid)
 		/* Advance global latestCompletedXid while holding the lock */
 		MaintainLatestCompletedXid(latestXid);
 
+		/*
+		 * Assign xid csn while holding ProcArrayLock for non-distributed
+		 * COMMIT PREPARED. After lock is released consequent
+		 * CSNSnapshotCommit() will write this value to CsnLog.
+		 */
+		if (CSNIsInDoubt(pg_atomic_read_u64(&proc->assignedCSN)))
+			pg_atomic_write_u64(&proc->assignedCSN, GenerateCSN(InvalidCSN));
+
 		/* Same with xactCompletionCount  */
 		ShmemVariableCache->xactCompletionCount++;
 
@@ -656,6 +669,7 @@ ProcArrayEndTransaction(PGPROC *proc, TransactionId latestXid)
 
 		proc->lxid = InvalidLocalTransactionId;
 		proc->xmin = InvalidTransactionId;
+		proc->originalXmin = InvalidTransactionId;
 		proc->delayChkpt = false;	/* be sure this is cleared in abort */
 		proc->recoveryConflictPending = false;
 
@@ -695,6 +709,7 @@ ProcArrayEndTransactionInternal(PGPROC *proc, TransactionId latestXid)
 	proc->xid = InvalidTransactionId;
 	proc->lxid = InvalidLocalTransactionId;
 	proc->xmin = InvalidTransactionId;
+	proc->originalXmin = InvalidTransactionId;
 	proc->delayChkpt = false;	/* be sure this is cleared in abort */
 	proc->recoveryConflictPending = false;
 
@@ -720,6 +735,16 @@ ProcArrayEndTransactionInternal(PGPROC *proc, TransactionId latestXid)
 	/* Also advance global latestCompletedXid while holding the lock */
 	MaintainLatestCompletedXid(latestXid);
 
+	/*
+	 * Assign xid csn while holding ProcArrayLock for
+	 * COMMIT.
+	 *
+	 * TODO: in case of group commit we can generate one CSNSnapshot for
+	 * whole group to save time on timestamp aquisition.
+	 */
+	if (CSNIsInDoubt(pg_atomic_read_u64(&proc->assignedCSN)))
+		pg_atomic_write_u64(&proc->assignedCSN, GenerateCSN(InvalidCSN));
+
 	/* Same with xactCompletionCount  */
 	ShmemVariableCache->xactCompletionCount++;
 }
@@ -878,6 +903,7 @@ ProcArrayClearTransaction(PGPROC *proc)
 
 	proc->lxid = InvalidLocalTransactionId;
 	proc->xmin = InvalidTransactionId;
+	proc->originalXmin = InvalidTransactionId;
 	proc->recoveryConflictPending = false;
 
 	Assert(!(proc->statusFlags & PROC_VACUUM_STATE_MASK));
@@ -1667,6 +1693,7 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h)
 	TransactionId kaxmin;
 	bool		in_recovery = RecoveryInProgress();
 	TransactionId *other_xids = ProcGlobal->xids;
+	TransactionId csn_snapshot_xmin = InvalidTransactionId;
 
 	LWLockAcquire(ProcArrayLock, LW_SHARED);
 
@@ -1715,6 +1742,7 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h)
 	 */
 	h->slot_xmin = procArray->replication_slot_xmin;
 	h->slot_catalog_xmin = procArray->replication_slot_catalog_xmin;
+	csn_snapshot_xmin = ProcArrayGetCSNSnapshotXmin();
 
 	for (int index = 0; index < arrayP->numProcs; index++)
 	{
@@ -1723,6 +1751,7 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h)
 		int8		statusFlags = ProcGlobal->statusFlags[index];
 		TransactionId xid;
 		TransactionId xmin;
+		TransactionId original_xmin = UINT32_ACCESS_ONCE(proc->originalXmin);
 
 		/* Fetch xid just once - see GetNewTransactionId */
 		xid = UINT32_ACCESS_ONCE(other_xids[index]);
@@ -1735,6 +1764,9 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h)
 		 * (yet) an Xid; conversely, if it has an Xid, that could determine
 		 * some not-yet-set Xmin.
 		 */
+		if (TransactionIdIsValid(original_xmin))
+			xmin = original_xmin;
+
 		xmin = TransactionIdOlder(xmin, xid);
 
 		/* if neither is set, this proc doesn't influence the horizon */
@@ -1854,6 +1886,10 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h)
 		/* defer doesn't apply to temp relations */
 	}
 
+	if (TransactionIdIsValid(csn_snapshot_xmin) &&
+		TransactionIdOlder(csn_snapshot_xmin, h->shared_oldest_nonremovable))
+		h->shared_oldest_nonremovable = csn_snapshot_xmin;
+
 	/*
 	 * Check whether there are replication slots requiring an older xmin.
 	 */
@@ -2064,6 +2100,8 @@ GetSnapshotDataReuse(Snapshot snapshot)
 	if (curXactCompletionCount != snapshot->snapXactCompletionCount)
 		return false;
 
+	if (get_csnlog_status())
+		return false;
 	/*
 	 * If the current xactCompletionCount is still the same as it was at the
 	 * time the snapshot was built, we can be sure that rebuilding the
@@ -2143,6 +2181,7 @@ GetSnapshotData(Snapshot snapshot)
 	size_t		count = 0;
 	int			subcount = 0;
 	bool		suboverflowed = false;
+	CSN	csn = FrozenCSN;
 	FullTransactionId latest_completed;
 	TransactionId oldestxid;
 	int			mypgxactoff;
@@ -2151,6 +2190,7 @@ GetSnapshotData(Snapshot snapshot)
 
 	TransactionId replication_slot_xmin = InvalidTransactionId;
 	TransactionId replication_slot_catalog_xmin = InvalidTransactionId;
+	TransactionId csn_snapshot_xmin = InvalidTransactionId;
 
 	Assert(snapshot != NULL);
 
@@ -2371,10 +2411,18 @@ GetSnapshotData(Snapshot snapshot)
 	 */
 	replication_slot_xmin = procArray->replication_slot_xmin;
 	replication_slot_catalog_xmin = procArray->replication_slot_catalog_xmin;
+	csn_snapshot_xmin = ProcArrayGetCSNSnapshotXmin();
 
 	if (!TransactionIdIsValid(MyProc->xmin))
 		MyProc->xmin = TransactionXmin = xmin;
 
+	/*
+	 * Take CSN under ProcArrayLock so the snapshot stays
+	 * synchronized.
+	 */
+	if (!snapshot->takenDuringRecovery && get_csnlog_status())
+		csn = GenerateCSN(InvalidCSN);
+
 	LWLockRelease(ProcArrayLock);
 
 	/* maintain state for GlobalVis* */
@@ -2396,6 +2444,9 @@ GetSnapshotData(Snapshot snapshot)
 		def_vis_xid_data =
 			TransactionIdRetreatedBy(xmin, vacuum_defer_cleanup_age);
 
+		if (TransactionIdIsValid(csn_snapshot_xmin))
+			def_vis_xid_data = TransactionIdOlder(csn_snapshot_xmin, def_vis_xid_data);
+
 		/* Check whether there's a replication slot requiring an older xmin. */
 		def_vis_xid_data =
 			TransactionIdOlder(def_vis_xid_data, replication_slot_xmin);
@@ -2480,6 +2531,11 @@ GetSnapshotData(Snapshot snapshot)
 	snapshot->active_count = 0;
 	snapshot->regd_count = 0;
 	snapshot->copied = false;
+	snapshot->imported_csn = false;
+	snapshot->csn = csn;
+
+	if (csn_snapshot_defer_time > 0 && IsUnderPostmaster)
+		CSNSnapshotMapXmin(snapshot->csn);
 
 	GetSnapshotDataInitOldSnapshot(snapshot);
 
@@ -3832,6 +3888,25 @@ ProcArrayGetReplicationSlotXmin(TransactionId *xmin,
 	LWLockRelease(ProcArrayLock);
 }
 
+/*
+ * ProcArraySetCSNSnapshotXmin
+ */
+void
+ProcArraySetCSNSnapshotXmin(TransactionId xmin)
+{
+	/* We rely on atomic fetch/store of xid */
+	procArray->csn_snapshot_xmin = xmin;
+}
+
+/*
+ * ProcArrayGetCSNSnapshotXmin
+ */
+TransactionId
+ProcArrayGetCSNSnapshotXmin(void)
+{
+	return procArray->csn_snapshot_xmin;
+}
+
 /*
  * XidCacheRemoveRunningXids
  *
diff --git a/src/backend/storage/lmgr/lwlocknames.txt b/src/backend/storage/lmgr/lwlocknames.txt
index 56196b2d0d..ac0e6a33fe 100644
--- a/src/backend/storage/lmgr/lwlocknames.txt
+++ b/src/backend/storage/lmgr/lwlocknames.txt
@@ -54,3 +54,4 @@ XactTruncationLock					44
 WrapLimitsVacuumLock				46
 NotifyQueueTailLock					47
 CSNLogControlLock				    48
+CSNSnapshotXidMapLock               49
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index 897045ee27..6e66801001 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -37,6 +37,7 @@
 
 #include "access/transam.h"
 #include "access/twophase.h"
+#include "access/csn_snapshot.h"
 #include "access/xact.h"
 #include "miscadmin.h"
 #include "pgstat.h"
@@ -440,6 +441,9 @@ InitProcess(void)
 	MyProc->clogGroupMemberLsn = InvalidXLogRecPtr;
 	Assert(pg_atomic_read_u32(&MyProc->clogGroupNext) == INVALID_PGPROCNO);
 
+	MyProc->originalXmin = InvalidTransactionId;
+	pg_atomic_init_u64(&MyProc->assignedCSN, InProgressCSN);
+
 	/*
 	 * Acquire ownership of the PGPROC's latch, so that we can use WaitLatch
 	 * on it.  That allows us to repoint the process latch, which so far
@@ -582,6 +586,7 @@ InitAuxiliaryProcess(void)
 	MyProc->lwWaitMode = 0;
 	MyProc->waitLock = NULL;
 	MyProc->waitProcLock = NULL;
+	MyProc->originalXmin = InvalidTransactionId;
 	pg_atomic_write_u64(&MyProc->waitStart, 0);
 #ifdef USE_ASSERT_CHECKING
 	{
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 14063bcf9e..cc01d72d64 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -30,6 +30,7 @@
 #include <unistd.h>
 
 #include "access/commit_ts.h"
+#include "access/csn_snapshot.h"
 #include "access/gin.h"
 #include "access/rmgr.h"
 #include "access/tableam.h"
@@ -3140,6 +3141,26 @@ static struct config_int ConfigureNamesInt[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"csn_snapshot_defer_time", PGC_POSTMASTER, PRESET_OPTIONS,
+			gettext_noop("Minimal age of records which allowed to be vacuumed, in seconds."),
+			NULL
+		},
+		&csn_snapshot_defer_time,
+		5, 0, INT_MAX,
+		NULL, NULL, NULL
+	},
+
+	{
+		{"csn_time_shift", PGC_USERSET, RESOURCES_MEM,
+			gettext_noop("Do the time shift in the CSN generator."),
+			gettext_noop("Used for debug purposes.")
+		},
+		&csn_time_shift,
+		0, INT_MIN, INT_MAX,
+		NULL, NULL, NULL
+	},
+
 	{
 		{"block_size", PGC_INTERNAL, PRESET_OPTIONS,
 			gettext_noop("Shows the size of a disk block."),
diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c
index 95704265b6..2d4c5f53af 100644
--- a/src/backend/utils/time/snapmgr.c
+++ b/src/backend/utils/time/snapmgr.c
@@ -48,6 +48,7 @@
 #include <sys/stat.h>
 #include <unistd.h>
 
+#include "access/csn_log.h"
 #include "access/subtrans.h"
 #include "access/transam.h"
 #include "access/xact.h"
@@ -173,6 +174,7 @@ static TimestampTz AlignTimestampToMinuteBoundary(TimestampTz ts);
 static Snapshot CopySnapshot(Snapshot snapshot);
 static void FreeSnapshot(Snapshot snapshot);
 static void SnapshotResetXmin(void);
+static bool XidInLocalMVCCSnapshot(TransactionId xid, Snapshot snapshot);
 
 /*
  * Snapshot fields to be serialized.
@@ -191,6 +193,8 @@ typedef struct SerializedSnapshotData
 	CommandId	curcid;
 	TimestampTz whenTaken;
 	XLogRecPtr	lsn;
+	CSN			csn;
+	bool		imported_csn;
 } SerializedSnapshotData;
 
 Size
@@ -291,6 +295,7 @@ GetTransactionSnapshot(void)
 				CurrentSnapshot = GetSerializableTransactionSnapshot(&CurrentSnapshotData);
 			else
 				CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData);
+
 			/* Make a saved copy */
 			CurrentSnapshot = CopySnapshot(CurrentSnapshot);
 			FirstXactSnapshot = CurrentSnapshot;
@@ -2115,6 +2120,8 @@ SerializeSnapshot(Snapshot snapshot, char *start_address)
 	serialized_snapshot.curcid = snapshot->curcid;
 	serialized_snapshot.whenTaken = snapshot->whenTaken;
 	serialized_snapshot.lsn = snapshot->lsn;
+	serialized_snapshot.csn = snapshot->csn;
+	serialized_snapshot.imported_csn = snapshot->imported_csn;
 
 	/*
 	 * Ignore the SubXID array if it has overflowed, unless the snapshot was
@@ -2190,6 +2197,8 @@ RestoreSnapshot(char *start_address)
 	snapshot->whenTaken = serialized_snapshot.whenTaken;
 	snapshot->lsn = serialized_snapshot.lsn;
 	snapshot->snapXactCompletionCount = 0;
+	snapshot->csn = serialized_snapshot.csn;
+	snapshot->imported_csn = serialized_snapshot.imported_csn;
 
 	/* Copy XIDs, if present. */
 	if (serialized_snapshot.xcnt > 0)
@@ -2230,6 +2239,44 @@ RestoreTransactionSnapshot(Snapshot snapshot, void *source_pgproc)
 
 /*
  * XidInMVCCSnapshot
+ *
+ * Check whether this xid is in snapshot. When enable_csn_snapshot is
+ * switched off just call XidInLocalMVCCSnapshot().
+ */
+bool
+XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
+{
+	bool in_snapshot;
+
+	if (snapshot->imported_csn)
+	{
+		Assert(enable_csn_snapshot);
+		/* No point to using snapshot info except CSN */
+		return XidInvisibleInCSNSnapshot(xid, snapshot);
+	}
+
+	in_snapshot = XidInLocalMVCCSnapshot(xid, snapshot);
+
+	if (!get_csnlog_status())
+	{
+		Assert(CSNIsFrozen(snapshot->csn));
+		return in_snapshot;
+	}
+
+	if (in_snapshot)
+	{
+		/*
+		 * This xid may be already in unknown state and in that case
+		 * we must wait and recheck.
+		 */
+		return XidInvisibleInCSNSnapshot(xid, snapshot);
+	}
+	else
+		return false;
+}
+
+/*
+ * XidInLocalMVCCSnapshot
  *		Is the given XID still-in-progress according to the snapshot?
  *
  * Note: GetSnapshotData never stores either top xid or subxids of our own
@@ -2238,8 +2285,8 @@ RestoreTransactionSnapshot(Snapshot snapshot, void *source_pgproc)
  * TransactionIdIsCurrentTransactionId first, except when it's known the
  * XID could not be ours anyway.
  */
-bool
-XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
+static bool
+XidInLocalMVCCSnapshot(TransactionId xid, Snapshot snapshot)
 {
 	uint32		i;
 
@@ -2349,3 +2396,98 @@ XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
 
 	return false;
 }
+
+/*
+ * ExportCSNSnapshot
+ *
+ * Export csn so that caller can expand this transaction to other
+ * nodes.
+ *
+ * TODO: it's better to do this through EXPORT/IMPORT SNAPSHOT syntax and
+ * add some additional checks that transaction did not yet acquired xid, but
+ * for current iteration of this patch I don't want to hack on parser.
+ */
+CSN
+ExportCSNSnapshot()
+{
+	if (!get_csnlog_status())
+		ereport(ERROR,
+			(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+			 errmsg("could not export csn snapshot"),
+			 errhint("Make sure the configuration parameter \"%s\" is enabled.",
+					 "enable_csn_snapshot")));
+
+	return CurrentSnapshot->csn;
+}
+
+/* SQL accessor to ExportCSNSnapshot() */
+Datum
+pg_csn_snapshot_export(PG_FUNCTION_ARGS)
+{
+	CSN	export_csn = ExportCSNSnapshot();
+	PG_RETURN_UINT64(export_csn);
+}
+
+/*
+ * ImportCSNSnapshot
+ *
+ * Import csn and retract this backends xmin to the value that was
+ * actual when we had such csn.
+ *
+ * TODO: it's better to do this through EXPORT/IMPORT SNAPSHOT syntax and
+ * add some additional checks that transaction did not yet acquired xid, but
+ * for current iteration of this patch I don't want to hack on parser.
+ */
+void
+ImportCSNSnapshot(CSN csn)
+{
+	volatile TransactionId xmin;
+
+	if (!get_csnlog_status())
+		ereport(ERROR,
+			(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+			 errmsg("could not import csn snapshot"),
+			 errhint("Make sure the configuration parameter \"%s\" is enabled.",
+					 "enable_csn_snapshot")));
+
+	if (csn_snapshot_defer_time <= 0)
+		ereport(ERROR,
+			(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+			 errmsg("could not import csn snapshot"),
+			 errhint("Make sure the configuration parameter \"%s\" is positive.",
+					 "csn_snapshot_defer_time")));
+
+	/*
+	 * Call CSNSnapshotToXmin under ProcArrayLock to avoid situation that
+	 * resulting xmin will be evicted from map before we will set it into our
+	 * backend's xmin.
+	 */
+	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	xmin = CSNSnapshotToXmin(csn);
+
+	if (!TransactionIdIsValid(xmin))
+	{
+		LWLockRelease(ProcArrayLock);
+		elog(ERROR, "CSNSnapshotToXmin: csn snapshot too old: %lu", csn);
+	}
+
+	MyProc->originalXmin = MyProc->xmin;
+	MyProc->xmin = TransactionXmin = xmin;
+	LWLockRelease(ProcArrayLock);
+
+	CurrentSnapshot->xmin = xmin; /* defuse SnapshotResetXmin() */
+	CurrentSnapshot->csn = csn;
+	CurrentSnapshot->imported_csn = true;
+
+	CSNSnapshotSync(csn);
+}
+
+/* SQL accessor to ImportCSNSnapshot() */
+Datum
+pg_csn_snapshot_import(PG_FUNCTION_ARGS)
+{
+	CSN	csn = PG_GETARG_UINT64(0);
+
+	ImportCSNSnapshot(csn);
+	PG_RETURN_VOID();
+}
diff --git a/src/bin/pg_waldump/csnlogdesc.c b/src/bin/pg_waldump/csnlogdesc.c
new file mode 120000
index 0000000000..dcde44b3ee
--- /dev/null
+++ b/src/bin/pg_waldump/csnlogdesc.c
@@ -0,0 +1 @@
+../../../src/backend/access/rmgrdesc/csnlogdesc.c
\ No newline at end of file
diff --git a/src/include/access/csn_log.h b/src/include/access/csn_log.h
index 01fb0daad4..9857724dca 100644
--- a/src/include/access/csn_log.h
+++ b/src/include/access/csn_log.h
@@ -71,7 +71,8 @@ extern void WriteAssignCSNXlogRec(CSN csn);
 extern CSN set_last_max_csn(CSN csn, bool force);
 extern CSN get_last_max_csn(void);
 extern CSN get_xmin_for_csn(void);
-extern void prepare_csn_env(bool enable_csn_snapshot);
+extern void prepare_csn_env(bool enable_csn_snapshot, bool same,
+							TransactionId *xmin_for_csn_in_control);
 extern void CatchCSNLog(void);
 extern void ActivateCSNlog(void);
 extern void DeactivateCSNlog(void);
diff --git a/src/include/access/csn_snapshot.h b/src/include/access/csn_snapshot.h
new file mode 100644
index 0000000000..3a739e6744
--- /dev/null
+++ b/src/include/access/csn_snapshot.h
@@ -0,0 +1,65 @@
+/*-------------------------------------------------------------------------
+ *
+ * csn_snapshot.h
+ *	  Support for cross-node snapshot isolation.
+ *
+ * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/access/csn_snapshot.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef CSN_SNAPSHOT_H
+#define CSN_SNAPSHOT_H
+
+#include "port/atomics.h"
+#include "storage/lock.h"
+#include "utils/snapshot.h"
+#include "utils/guc.h"
+
+/*
+ * snapshot.h is used in frontend code so atomic variant of CSN type
+ * is defined here.
+ */
+typedef pg_atomic_uint64 CSN_atomic;
+
+#define InProgressCSN	 	UINT64CONST(0x0)
+#define AbortedCSN	 		UINT64CONST(0x1)
+#define FrozenCSN		 	UINT64CONST(0x2)
+#define InDoubtCSN	 		UINT64CONST(0x3)
+#define UnclearCSN	 		UINT64CONST(0x4)
+#define FirstNormalCSN 		UINT64CONST(0x5)
+
+#define CSNIsInProgress(csn)	((csn) == InProgressCSN)
+#define CSNIsAborted(csn)		((csn) == AbortedCSN)
+#define CSNIsFrozen(csn)		((csn) == FrozenCSN)
+#define CSNIsInDoubt(csn)		((csn) == InDoubtCSN)
+#define CSNIsUnclear(csn)		((csn) == UnclearCSN)
+#define CSNIsNormal(csn)		((csn) >= FirstNormalCSN)
+
+
+extern int csn_snapshot_defer_time;
+extern int csn_time_shift;
+
+
+extern Size CSNSnapshotShmemSize(void);
+extern void CSNSnapshotShmemInit(void);
+
+extern void CSNSnapshotStartup(TransactionId oldestActiveXID);
+extern void CSNSnapshotMapXmin(CSN snapshot_csn);
+extern TransactionId CSNSnapshotToXmin(CSN snapshot_csn);
+extern CSN GenerateCSN(CSN assign);
+extern bool XidInvisibleInCSNSnapshot(TransactionId xid, Snapshot snapshot);
+extern CSN TransactionIdGetCSN(TransactionId xid);
+extern void CSNSnapshotAbort(PGPROC *proc, TransactionId xid, int nsubxids,
+								TransactionId *subxids);
+extern void CSNSnapshotPrecommit(PGPROC *proc, TransactionId xid, int nsubxids,
+									TransactionId *subxids);
+extern void CSNSnapshotCommit(PGPROC *proc, TransactionId xid, int nsubxids,
+									TransactionId *subxids);
+extern void CSNSnapshotAssignCurrent(CSN snapshot_csn);
+extern CSN CSNSnapshotPrepareCurrent(void);
+extern void CSNSnapshotSync(CSN remote_csn);
+
+#endif							/* CSN_SNAPSHOT_H */
diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.h
index 194d100ed4..fbd63f50d9 100644
--- a/src/include/catalog/pg_control.h
+++ b/src/include/catalog/pg_control.h
@@ -185,6 +185,12 @@ typedef struct ControlFileData
 	bool		track_commit_timestamp;
 	bool		enable_csn_snapshot;
 
+	/*
+	 * Used to record a xmin when database startup with a snapshot-switch to csn snapshot,
+	 * and will hold the value until it switch to xid-snapshot.
+	 */
+	TransactionId xmin_for_csn;
+
 	/*
 	 * This data is used to check for hardware-architecture compatibility of
 	 * the database and the backend executable.  We need not check endianness
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index b9f4afba05..8585464c5b 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11411,4 +11411,18 @@
   proname => 'is_normalized', prorettype => 'bool', proargtypes => 'text text',
   prosrc => 'unicode_is_normalized' },
 
+# csn shnapshot handling
+{ oid => '10001', descr => 'export csn snapshot',
+  proname => 'pg_csn_snapshot_export', provolatile => 'v', proparallel => 'u',
+  prorettype => 'int8', proargtypes => '', prosrc => 'pg_csn_snapshot_export' },
+{ oid => '10002', descr => 'import csn snapshot',
+  proname => 'pg_csn_snapshot_import', provolatile => 'v', proparallel => 'u',
+  prorettype => 'void', proargtypes => 'int8', prosrc => 'pg_csn_snapshot_import' },
+{ oid => '10003', descr => 'prepare distributed transaction for commit, get csn',
+  proname => 'pg_csn_snapshot_prepare', provolatile => 'v', proparallel => 'u',
+  prorettype => 'int8', proargtypes => 'text', prosrc => 'pg_csn_snapshot_prepare' },
+{ oid => '10004', descr => 'assign csn to distributed transaction',
+  proname => 'pg_csn_snapshot_assign', provolatile => 'v', proparallel => 'u',
+  prorettype => 'void', proargtypes => 'text int8', prosrc => 'pg_csn_snapshot_assign' },
+
 ]
diff --git a/src/include/datatype/timestamp.h b/src/include/datatype/timestamp.h
index 99873497a6..d8a681aaa4 100644
--- a/src/include/datatype/timestamp.h
+++ b/src/include/datatype/timestamp.h
@@ -92,6 +92,8 @@ typedef struct
 #define USECS_PER_HOUR	INT64CONST(3600000000)
 #define USECS_PER_MINUTE INT64CONST(60000000)
 #define USECS_PER_SEC	INT64CONST(1000000)
+#define NSECS_PER_SEC	INT64CONST(1000000000)
+#define NSECS_PER_USEC	INT64CONST(1000)
 
 /*
  * We allow numeric timezone offsets up to 15:59:59 either way from Greenwich.
diff --git a/src/include/fmgr.h b/src/include/fmgr.h
index ab7b85c86e..f08999740b 100644
--- a/src/include/fmgr.h
+++ b/src/include/fmgr.h
@@ -281,6 +281,7 @@ extern struct varlena *pg_detoast_datum_packed(struct varlena *datum);
 #define PG_GETARG_FLOAT4(n)  DatumGetFloat4(PG_GETARG_DATUM(n))
 #define PG_GETARG_FLOAT8(n)  DatumGetFloat8(PG_GETARG_DATUM(n))
 #define PG_GETARG_INT64(n)	 DatumGetInt64(PG_GETARG_DATUM(n))
+#define PG_GETARG_UINT64(n)	 DatumGetUInt64(PG_GETARG_DATUM(n))
 /* use this if you want the raw, possibly-toasted input datum: */
 #define PG_GETARG_RAW_VARLENA_P(n)	((struct varlena *) PG_GETARG_POINTER(n))
 /* use this if you want the input datum de-toasted: */
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index f0e920f462..6edb492638 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -15,6 +15,7 @@
 #define _PROC_H_
 
 #include "access/clog.h"
+#include "access/csn_snapshot.h"
 #include "access/xlogdefs.h"
 #include "lib/ilist.h"
 #include "utils/snapshot.h"
@@ -252,6 +253,18 @@ struct PGPROC
 	PGPROC	   *lockGroupLeader;	/* lock group leader, if I'm a member */
 	dlist_head	lockGroupMembers;	/* list of members, if I'm a leader */
 	dlist_node	lockGroupLink;	/* my member link, if I'm a member */
+
+	/*
+	 * assignedCSN holds CSN for this transaction.  It is generated
+	 * under a ProcArray lock and later is writter to a CSNLog.  This
+	 * variable defined as atomic only for case of group commit, in all other
+	 * scenarios only backend responsible for this proc entry is working with
+	 * this variable.
+	 */
+	CSN_atomic assignedCSN;
+
+	/* Original xmin of this backend before csn snapshot was imported */
+	TransactionId originalXmin;
 };
 
 /* NOTE: "typedef struct PGPROC PGPROC" appears in storage/lock.h. */
diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h
index b01fa52139..755a0743b6 100644
--- a/src/include/storage/procarray.h
+++ b/src/include/storage/procarray.h
@@ -94,4 +94,7 @@ extern void ProcArraySetReplicationSlotXmin(TransactionId xmin,
 extern void ProcArrayGetReplicationSlotXmin(TransactionId *xmin,
 											TransactionId *catalog_xmin);
 
+extern void ProcArraySetCSNSnapshotXmin(TransactionId xmin);
+extern TransactionId ProcArrayGetCSNSnapshotXmin(void);
+
 #endif							/* PROCARRAY_H */
diff --git a/src/include/utils/snapmgr.h b/src/include/utils/snapmgr.h
index f66ac58188..c339eb7384 100644
--- a/src/include/utils/snapmgr.h
+++ b/src/include/utils/snapmgr.h
@@ -129,6 +129,8 @@ extern void AtSubCommit_Snapshot(int level);
 extern void AtSubAbort_Snapshot(int level);
 extern void AtEOXact_Snapshot(bool isCommit, bool resetXmin);
 
+extern CSN ExportCSNSnapshot(void);
+extern void ImportCSNSnapshot(CSN csn);
 extern void ImportSnapshot(const char *idstr);
 extern bool XactHasExportedSnapshots(void);
 extern void DeleteAllExportedSnapshotFiles(void);
diff --git a/src/include/utils/snapshot.h b/src/include/utils/snapshot.h
index 92575a207f..98cff1d872 100644
--- a/src/include/utils/snapshot.h
+++ b/src/include/utils/snapshot.h
@@ -217,7 +217,8 @@ typedef struct SnapshotData
 	 * CSN for snapshot isolation support.
 	 * Will be used only if enable_csn_snapshot is enabled.
 	 */
-	CSN	csn;
+	CSN			csn;
+	bool		imported_csn;
 
 	/*
 	 * The transaction completion count at the time GetSnapshotData() built
-- 
2.25.1

  [text/x-patch] 0003-CSN-Snapshot-Tests.patch (21.9K, ../0e65f96f-bfd6-fb09-89b7-e07854c215e6@postgrespro.ru/4-0003-CSN-Snapshot-Tests.patch)
  download | inline diff:
From fa93fd78d70742d254d1972cc23d05006c19d4af Mon Sep 17 00:00:00 2001
From: Andrey Lepikhov <a.lepikhov@postgrespro.ru>
Date: Fri, 29 Jan 2021 10:37:43 +0500
Subject: [PATCH 3/4] CSN Snapshot Tests

---
 src/test/modules/Makefile                     |   1 +
 src/test/modules/csnsnapshot/Makefile         |  22 ++
 .../modules/csnsnapshot/csn_snapshot.conf     |   1 +
 src/test/modules/csnsnapshot/t/001_base.pl    | 119 ++++++++++
 src/test/modules/csnsnapshot/t/002_standby.pl |  66 ++++++
 .../modules/csnsnapshot/t/003_time_skew.pl    | 211 ++++++++++++++++++
 .../csnsnapshot/t/004_read_committed.pl       |  97 ++++++++
 7 files changed, 517 insertions(+)
 create mode 100644 src/test/modules/csnsnapshot/Makefile
 create mode 100644 src/test/modules/csnsnapshot/csn_snapshot.conf
 create mode 100644 src/test/modules/csnsnapshot/t/001_base.pl
 create mode 100644 src/test/modules/csnsnapshot/t/002_standby.pl
 create mode 100644 src/test/modules/csnsnapshot/t/003_time_skew.pl
 create mode 100644 src/test/modules/csnsnapshot/t/004_read_committed.pl

diff --git a/src/test/modules/Makefile b/src/test/modules/Makefile
index 93e7829c67..353c57eb51 100644
--- a/src/test/modules/Makefile
+++ b/src/test/modules/Makefile
@@ -7,6 +7,7 @@ include $(top_builddir)/src/Makefile.global
 SUBDIRS = \
 		  brin \
 		  commit_ts \
+		  csnsnapshot \
 		  delay_execution \
 		  dummy_index_am \
 		  dummy_seclabel \
diff --git a/src/test/modules/csnsnapshot/Makefile b/src/test/modules/csnsnapshot/Makefile
new file mode 100644
index 0000000000..e7c1195d0f
--- /dev/null
+++ b/src/test/modules/csnsnapshot/Makefile
@@ -0,0 +1,22 @@
+# src/test/modules/csnsnapshot/Makefile
+
+REGRESS_OPTS = --temp-config=$(top_srcdir)/src/test/modules/csnsnapshot/csn_snapshot.conf
+NO_INSTALLCHECK = 1
+
+TAP_TESTS = 1
+EXTRA_INSTALL=contrib/postgres_fdw
+
+ifndef CSN_ALL
+PROVE_TESTS=t/001_base.pl t/002_standby.pl t/003_time_skew.pl
+endif
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = src/test/modules/csnsnapshot
+top_builddir = ../../../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/src/test/modules/csnsnapshot/csn_snapshot.conf b/src/test/modules/csnsnapshot/csn_snapshot.conf
new file mode 100644
index 0000000000..e9d3c35756
--- /dev/null
+++ b/src/test/modules/csnsnapshot/csn_snapshot.conf
@@ -0,0 +1 @@
+track_commit_timestamp = on
diff --git a/src/test/modules/csnsnapshot/t/001_base.pl b/src/test/modules/csnsnapshot/t/001_base.pl
new file mode 100644
index 0000000000..a89b1beeea
--- /dev/null
+++ b/src/test/modules/csnsnapshot/t/001_base.pl
@@ -0,0 +1,119 @@
+# Single-node test: value can be set, and is still present after recovery
+
+use strict;
+use warnings;
+
+use TestLib;
+use Test::More tests => 6;
+use PostgresNode;
+
+my $node = get_new_node('csntest');
+$node->init;
+$node->append_conf('postgresql.conf', qq{
+					enable_csn_snapshot = on
+					csn_snapshot_defer_time = 10
+					max_prepared_transactions = 10
+					});
+
+my $result1;
+my $result2;
+$node->start;
+
+# Check CSN increased monotonically after restart
+$result1 = $node->safe_psql('postgres', "SHOW csn_time_shift");
+note("csn_time_shift: $result1");
+$result1 = $node->safe_psql('postgres', 'SELECT pg_csn_snapshot_export()');
+note("Snapshot CSN: $result1");
+$node->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = -100");
+$node->restart;
+$result2 = $node->safe_psql('postgres', 'SHOW csn_time_shift');
+note("csn_time_shift: $result2");
+$result2 = $node->safe_psql('postgres', 'SELECT pg_csn_snapshot_export()');
+note("Snapshot CSN after restart: $result2");
+is($result1 < $result2, 1, 'CSN monotonically increases');
+
+$node->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 0");
+$node->restart;
+
+# Create a table
+$node->safe_psql('postgres', 'create table t1(i int, j int)');
+
+# insert test record
+$node->safe_psql('postgres', 'insert into t1 values(1,1)');
+# export csn snapshot
+my $test_snapshot = $node->safe_psql('postgres', 'select pg_csn_snapshot_export()');
+# insert test record
+$node->safe_psql('postgres', 'insert into t1 values(2,1)');
+
+my $count1 = $node->safe_psql('postgres', "select count(*) from t1");
+is($count1, '2', 'Get right number in normal query');
+my $count2 = $node->safe_psql('postgres', "
+			begin transaction isolation level repeatable read;
+			select pg_csn_snapshot_import($test_snapshot);
+			select count(*) from t1;
+			commit;"
+			);
+
+is($count2, '
+1', 'Get right number in csn import query');
+
+#prepare transaction test
+$node->safe_psql('postgres', "
+						begin;
+						insert into t1 values(3,1);
+						insert into t1 values(3,2);
+						prepare	transaction 'pt3';
+						");
+$node->safe_psql('postgres', "
+						begin;
+						insert into t1 values(4,1);
+						insert into t1 values(4,2);
+						prepare	transaction 'pt4';
+						");
+$node->safe_psql('postgres', "
+						begin;
+						insert into t1 values(5,1);
+						insert into t1 values(5,2);
+						prepare	transaction 'pt5';
+						");
+$node->safe_psql('postgres', "
+						begin;
+						insert into t1 values(6,1);
+						insert into t1 values(6,2);
+						prepare	transaction 'pt6';
+						");
+$node->safe_psql('postgres', "commit prepared 'pt4';");
+
+# restart with enable_csn_snapshot off
+$node->append_conf('postgresql.conf', "enable_csn_snapshot = off");
+$node->restart;
+$node->safe_psql('postgres', "
+						insert into t1 values(7,1);
+						insert into t1 values(7,2);
+						");
+$node->safe_psql('postgres', "commit prepared 'pt3';");
+$count1 = $node->safe_psql('postgres', "select count(*) from t1");
+is($count1, '8', 'Get right number in normal query');
+
+
+# restart with enable_csn_snapshot on
+$node->append_conf('postgresql.conf', "enable_csn_snapshot = on");
+$node->restart;
+$node->safe_psql('postgres', "
+						insert into t1 values(8,1);
+						insert into t1 values(8,2);
+						");
+$node->safe_psql('postgres', "commit prepared 'pt5';");
+$count1 = $node->safe_psql('postgres', "select count(*) from t1");
+is($count1, '12', 'Get right number in normal query');
+
+# restart with enable_csn_snapshot off
+$node->append_conf('postgresql.conf', "enable_csn_snapshot = on");
+$node->restart;
+$node->safe_psql('postgres', "
+						insert into t1 values(9,1);
+						insert into t1 values(9,2);
+						");
+$node->safe_psql('postgres', "commit prepared 'pt6';");
+$count1 = $node->safe_psql('postgres', "select count(*) from t1");
+is($count1, '16', 'Get right number in normal query');
diff --git a/src/test/modules/csnsnapshot/t/002_standby.pl b/src/test/modules/csnsnapshot/t/002_standby.pl
new file mode 100644
index 0000000000..b7c4ea93b2
--- /dev/null
+++ b/src/test/modules/csnsnapshot/t/002_standby.pl
@@ -0,0 +1,66 @@
+# Test simple scenario involving a standby
+
+use strict;
+use warnings;
+
+use TestLib;
+use Test::More tests => 6;
+use PostgresNode;
+
+my $bkplabel = 'backup';
+my $master   = get_new_node('master');
+$master->init(allows_streaming => 1);
+
+$master->append_conf(
+	'postgresql.conf', qq{
+	enable_csn_snapshot = on
+	max_wal_senders = 5
+	});
+$master->start;
+$master->backup($bkplabel);
+
+my $standby = get_new_node('standby');
+$standby->init_from_backup($master, $bkplabel, has_streaming => 1);
+$standby->start;
+
+$master->safe_psql('postgres', "create table t1(i int, j int)");
+
+my $guc_on_master = $master->safe_psql('postgres', 'show enable_csn_snapshot');
+is($guc_on_master, 'on', "GUC on master");
+
+my $guc_on_standby = $standby->safe_psql('postgres', 'show enable_csn_snapshot');
+is($guc_on_standby, 'on', "GUC on standby");
+
+$master->append_conf('postgresql.conf', 'enable_csn_snapshot = off');
+$master->restart;
+
+$guc_on_master = $master->safe_psql('postgres', 'show enable_csn_snapshot');
+is($guc_on_master, 'off', "GUC off master");
+
+$guc_on_standby = $standby->safe_psql('postgres', 'show enable_csn_snapshot');
+is($guc_on_standby, 'on', "GUC on standby");
+
+# We consume a large number of transaction,for skip page
+for my $i (1 .. 4096) #4096
+{
+	$master->safe_psql('postgres', "insert into t1 values(1,$i)");
+}
+$master->safe_psql('postgres', "select pg_sleep(2)");
+$master->append_conf('postgresql.conf', 'enable_csn_snapshot = on');
+$master->restart;
+
+my $count_standby = $standby->safe_psql('postgres', 'select count(*) from t1');
+is($count_standby, '4096', "Ok for siwtch xid-base > csn-base"); #4096
+
+# We consume a large number of transaction,for skip page
+for my $i (1 .. 4096) #4096
+{
+	$master->safe_psql('postgres', "insert into t1 values(1,$i)");
+}
+$master->safe_psql('postgres', "select pg_sleep(2)");
+
+$master->append_conf('postgresql.conf', 'enable_csn_snapshot = off');
+$master->restart;
+
+$count_standby = $standby->safe_psql('postgres', 'select count(*) from t1');
+is($count_standby, '8192', "Ok for siwtch csn-base > xid-base"); #8192
\ No newline at end of file
diff --git a/src/test/modules/csnsnapshot/t/003_time_skew.pl b/src/test/modules/csnsnapshot/t/003_time_skew.pl
new file mode 100644
index 0000000000..b01ad1a7c0
--- /dev/null
+++ b/src/test/modules/csnsnapshot/t/003_time_skew.pl
@@ -0,0 +1,211 @@
+use strict;
+use warnings;
+
+use TestLib;
+use Test::More tests => 13;
+use PostgresNode;
+
+my $node1 = get_new_node('csn1');
+$node1->init;
+$node1->append_conf('postgresql.conf', qq{
+					enable_csn_snapshot = on
+					csn_snapshot_defer_time = 10
+					max_prepared_transactions = 10
+					csn_time_shift = 0
+					shared_preload_libraries = 'postgres_fdw'
+					postgres_fdw.use_csn_snapshots = true
+					});
+$node1->start;
+my $node2 = get_new_node('csn2');
+$node2->init;
+$node2->append_conf('postgresql.conf', qq{
+					enable_csn_snapshot = on
+					csn_snapshot_defer_time = 10
+					max_prepared_transactions = 10
+					csn_time_shift = 0
+					shared_preload_libraries = 'postgres_fdw'
+					postgres_fdw.use_csn_snapshots = true
+					});
+$node2->start;
+
+$node1->safe_psql('postgres', "
+	CREATE EXTENSION postgres_fdw;
+	CREATE SERVER remote FOREIGN DATA WRAPPER postgres_fdw	OPTIONS (port '".$node2->port."');
+	CREATE USER MAPPING FOR PUBLIC SERVER remote;
+	CREATE TABLE summary(value int, ntrans int);
+	INSERT INTO summary (value, ntrans) VALUES (0, 0);
+");
+$node2->safe_psql('postgres', "
+	CREATE EXTENSION postgres_fdw;
+	CREATE SERVER remote FOREIGN DATA WRAPPER postgres_fdw	OPTIONS (port '".$node1->port."');
+	CREATE USER MAPPING FOR PUBLIC SERVER remote;
+	CREATE FOREIGN TABLE summary(value int, ntrans int) SERVER remote;
+");
+
+$node1->safe_psql('postgres', "
+	CREATE TABLE t (id int, payload int) PARTITION BY HASH(id);
+	CREATE TABLE t_0 PARTITION OF t FOR VALUES WITH (modulus 2, remainder 0);
+	CREATE FOREIGN TABLE t_1 PARTITION OF t FOR VALUES WITH (modulus 2, remainder 1) SERVER remote;
+");
+$node2->safe_psql('postgres', "
+	CREATE TABLE t (id serial, payload int) PARTITION BY HASH(id);
+	CREATE TABLE t_1 PARTITION OF t FOR VALUES WITH (modulus 2, remainder 1);
+	CREATE FOREIGN TABLE t_0 PARTITION OF t FOR VALUES WITH (modulus 2, remainder 0) SERVER remote;
+");
+
+$node1->safe_psql('postgres', "INSERT INTO t(id, payload) (SELECT gs.*, 1 FROM generate_series(1,100) AS gs)");
+$node2->safe_psql('postgres', "INSERT INTO t(id, payload) (SELECT gs.*, 2 FROM generate_series(101,200) AS gs)");
+my $count1 = $node1->safe_psql('postgres', "SELECT SUM(payload) FROM t");
+my $count2 = $node2->safe_psql('postgres', "SELECT SUM(payload) FROM t");
+is( (($count1 == 300) and ($count1 == $count2)), 1, 'Correct insert');
+
+# ##############################################################################
+#
+# Basic test. Check REPEATABLE READ anomaly.
+# ntrans is needed to control that some transactions were committed.
+#
+# ##############################################################################
+
+my $q1 = File::Temp->new();
+append_to_file($q1, q{
+	START TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+	UPDATE summary SET value = value + (SELECT SUM(payload) FROM t);
+	UPDATE summary SET value = value - (SELECT SUM(payload) FROM t);
+	UPDATE summary SET ntrans = ntrans + 1;
+	COMMIT;
+});
+my $q2 = File::Temp->new();
+append_to_file($q2, q{
+	BEGIN;
+	\set pl random(-100, 100)
+	\set id random(1, 200)
+	UPDATE t SET payload = :pl WHERE id = :id;
+	COMMIT;
+});
+
+my $seconds = 5;
+my $pgb_handle1 = $node1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $q1, 'postgres' );
+my $pgb_handle2 = $node2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $q2, 'postgres' );
+$node1->pgbench_await($pgb_handle1);
+$node2->pgbench_await($pgb_handle2);
+
+$count1 = $node1->safe_psql('postgres', "SELECT SUM(value) FROM summary");
+$count2 = $node2->safe_psql('postgres', "SELECT SUM(value) FROM summary");
+my $ntrans = $node2->safe_psql('postgres', "SELECT SUM(ntrans) FROM summary");
+note("$count1, $count2, $ntrans");
+is( ( ($ntrans > 0) and ($count1 == 0) and ($count1 == $count2)), 1, 'Correct update');
+
+# ##############################################################################
+#
+# Test on 'snapshot too old'
+#
+# ##############################################################################
+$node1->safe_psql('postgres', "UPDATE summary SET ntrans = 0;");
+$node1->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 20");
+$node1->restart();
+
+# READ COMMITTED transactions ignores the time skew.
+$node2->psql('postgres', "UPDATE summary SET ntrans = 1");
+$ntrans = $node1->safe_psql('postgres', "SELECT ntrans FROM summary");
+note("$ntrans");
+is( $ntrans, 1, 'Read committed behavior if snapshot turn sour');
+
+# But REPEATABLE READ transactions isn't
+my $err = '';
+$node1->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 40");
+$node1->restart();
+$node2->psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; UPDATE summary SET ntrans = 2; COMMIT;", stderr => \$err);
+$ntrans = $node1->safe_psql('postgres', "SELECT ntrans FROM summary");
+note("$ntrans");
+is( (($ntrans == 1) and (index($err, 'csn snapshot too old') != -1)), 1, 'Read committed can\'t update if snapshot turn sour');
+
+# ##############################################################################
+#
+# Test on issue #1:
+# 'xact confirmed as committed, so any following xact must see its effects'.
+#
+# ##############################################################################
+$node1->safe_psql('postgres', "delete from t");
+$node1->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 45");
+$node2->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 40");
+$node1->restart();
+$node2->restart();
+
+my $st_sec; my $end_sec;
+my $time_diff;
+
+$node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; INSERT INTO t VALUES(1,1), (3,1); COMMIT;");
+$ntrans = $node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT count(*) FROM t; COMMIT;");
+is( $ntrans, 2, 'Slow node can see mix node data change');
+$ntrans = $node1->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT count(*) FROM t; COMMIT;");
+is( $ntrans, 2, 'Fast node can see mix node data change');
+
+$node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; INSERT INTO t VALUES(1,1); COMMIT;");
+$ntrans = $node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT count(*) FROM t; COMMIT;");
+is( $ntrans, 3, 'CURRENTLY FAILED:Data change to fast node on slow node, and slow node can see data change');
+
+# READ COMMITED mode ignores the time skew.
+$node1->safe_psql('postgres', "UPDATE summary SET ntrans = 1");
+$ntrans = $node2->safe_psql('postgres', "SELECT ntrans FROM summary");
+note("ntrans: $ntrans\n");
+is( $ntrans, 1, 'See committed values in the READ COMMITTED mode');
+
+# Access from the future
+$node1->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; UPDATE summary SET ntrans = ntrans + 1; COMMIT;");
+$ntrans = $node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT ntrans FROM summary; COMMIT;");
+note("ntrans: $ntrans\n");
+is( $ntrans, 1, 'Do not see values, committed in the future at the REPEATABLE READ mode');
+
+# But...
+$node1->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 50");
+$node2->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 55");
+$node1->restart();
+$node2->restart();
+
+# Check READ COMMITED mode
+$node2->safe_psql('postgres', "UPDATE summary SET ntrans = 2");
+$ntrans = $node1->safe_psql('postgres', "SELECT ntrans FROM summary");
+note("ntrans: $ntrans\n");
+is( $ntrans, 2, 'See committed values in the READ COMMITTED mode, step 2');
+
+# Node from the future will wait for a time before UPDATE table.
+($st_sec) = localtime();
+$node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; UPDATE summary SET ntrans = 3; COMMIT;");
+($end_sec) = localtime(); $time_diff = $end_sec - $st_sec;
+$ntrans = $node1->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT ntrans FROM summary; COMMIT;");
+note("ntrans: $ntrans, Test time: $time_diff seconds");
+is( ($ntrans == 3), 1, 'The test execution time correlates with the time offset.');
+
+# Node from the future will wait for a time before SELECT from a table.
+$node1->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; UPDATE summary SET ntrans = 4; COMMIT;");
+($st_sec) = localtime();
+$ntrans = $node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT ntrans FROM summary; COMMIT;");
+($end_sec) = localtime(); $time_diff = $end_sec - $st_sec;
+note("ntrans: $ntrans, Test time: $time_diff seconds ($end_sec, $st_sec)");
+is( ($ntrans == 4), 1, 'See values, committed in the past. The test execution time correlates with the time offset.');
+
+$node1->safe_psql('postgres', "UPDATE summary SET ntrans = 0, value = 0");
+$q1 = File::Temp->new();
+append_to_file($q1, q{
+	UPDATE summary SET value = value + 1, ntrans = ntrans + 1;
+});
+$q2 = File::Temp->new();
+append_to_file($q2, q{
+	START TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+	UPDATE summary SET value = value + (SELECT SUM(ntrans) FROM summary);
+	UPDATE summary SET value = value - (SELECT SUM(ntrans) FROM summary);
+	COMMIT;
+});
+$seconds = 3;
+$pgb_handle1 = $node1->pgbench_async(-n, -c => 1, -T => $seconds, -f => $q1, 'postgres' );
+$pgb_handle2 = $node2->pgbench_async(-n, -c => 1, -T => $seconds, -f => $q2, 'postgres' );
+$node1->pgbench_await($pgb_handle1);
+$node2->pgbench_await($pgb_handle2);
+
+$count1 = $node1->safe_psql('postgres', "SELECT SUM(value) FROM summary");
+$count2 = $node1->safe_psql('postgres', "SELECT SUM(ntrans) FROM summary");
+note("$count1, $count2");
+is( ( ($count1 > 0) and ($count1 == $count2)), 1, 'Skew test');
+
+$node1->stop();
+$node2->stop();
diff --git a/src/test/modules/csnsnapshot/t/004_read_committed.pl b/src/test/modules/csnsnapshot/t/004_read_committed.pl
new file mode 100644
index 0000000000..0918212bd8
--- /dev/null
+++ b/src/test/modules/csnsnapshot/t/004_read_committed.pl
@@ -0,0 +1,97 @@
+use strict;
+use warnings;
+
+use TestLib;
+use Test::More tests => 2;
+use PostgresNode;
+
+my $node1 = get_new_node('csn1');
+$node1->init;
+$node1->append_conf('postgresql.conf', qq{
+					max_prepared_transactions = 20
+					shared_preload_libraries = 'postgres_fdw'
+					enable_csn_snapshot = on
+					csn_snapshot_defer_time = 10
+					postgres_fdw.use_csn_snapshots = true
+					csn_time_shift = 0
+					});
+$node1->start;
+my $node2 = get_new_node('csn2');
+$node2->init;
+$node2->append_conf('postgresql.conf', qq{
+					max_prepared_transactions = 20
+					shared_preload_libraries = 'postgres_fdw'
+					enable_csn_snapshot = on
+					csn_snapshot_defer_time = 10
+					postgres_fdw.use_csn_snapshots = true
+					csn_time_shift = 0
+					});
+$node2->start;
+
+# Create foreign servers
+$node1->safe_psql('postgres', "
+	CREATE EXTENSION postgres_fdw;
+	CREATE SERVER remote FOREIGN DATA WRAPPER postgres_fdw	OPTIONS (port '".$node2->port."');
+	CREATE USER MAPPING FOR PUBLIC SERVER remote;
+");
+$node2->safe_psql('postgres', "
+	CREATE EXTENSION postgres_fdw;
+	CREATE SERVER remote FOREIGN DATA WRAPPER postgres_fdw	OPTIONS (port '".$node1->port."');
+	CREATE USER MAPPING FOR PUBLIC SERVER remote;
+");
+
+# Create sharded table
+$node1->safe_psql('postgres', "
+	CREATE TABLE dept1(name TEXT);
+	CREATE FOREIGN TABLE dept2 (name TEXT) SERVER remote;
+");
+$node2->safe_psql('postgres', "
+	CREATE TABLE dept2(name TEXT);
+	CREATE FOREIGN TABLE dept1 (name TEXT) SERVER remote;
+	CREATE TABLE results(success_tx int);
+	INSERT INTO results (success_tx) VALUES (0);
+");
+
+# Fill the table
+$node1->safe_psql('postgres', "INSERT INTO dept1 (name) VALUES ('Jonathan')");
+$node1->safe_psql('postgres', "INSERT INTO dept2 (name) VALUES ('Hoshi')");
+$node2->safe_psql('postgres', "INSERT INTO dept1 (name) VALUES ('Leonard')");
+my $count1 = $node1->safe_psql('postgres', "SELECT count(*) FROM ((SELECT * FROM dept1) UNION (SELECT * FROM dept2)) AS a");
+my $count2 = $node2->safe_psql('postgres', "SELECT count(*) FROM ((SELECT * FROM dept1) UNION (SELECT * FROM dept2)) AS a");
+note("$count1, $count2");
+is( (($count1 == 3) and ($count1 == $count2)), 1, 'Correct insert');
+
+# Queries
+my $q1 = File::Temp->new();
+append_to_file($q1, q{
+	BEGIN;
+	SELECT count(*) AS cnt FROM dept1; \gset
+	\if :cnt > 0
+		INSERT INTO dept2 (SELECT * FROM dept1);
+		DELETE FROM dept1;
+	\else
+		INSERT INTO dept1 (SELECT * FROM dept2);
+		DELETE FROM dept2;
+	\endif
+
+	COMMIT;
+});
+my $q2 = File::Temp->new();
+append_to_file($q2, q{
+	SELECT count(*) AS cnt FROM ((SELECT * FROM dept1) UNION (SELECT * FROM dept2)) AS a; \gset
+	\if :cnt = 3
+		UPDATE results SET success_tx = success_tx + 1;
+	\endif
+});
+my $transactions = 1000;
+my $pgb_handle1 = $node1->pgbench_async(-n, -c => 1, -t => $transactions, -f => $q1, 'postgres' );
+my $pgb_handle2 = $node2->pgbench_async(-n, -c => 20, -t => $transactions, -f => $q2, 'postgres' );
+$node1->pgbench_await($pgb_handle1);
+$node2->pgbench_await($pgb_handle2);
+
+$count2 = $node2->safe_psql('postgres', "SELECT success_tx FROM results");
+note("$count2");
+is( $count2, 20*$transactions, 'Correct READ COMMITTED updates');
+
+$node1->stop();
+$node2->stop();
-- 
2.25.1

  [text/x-patch] 0004-Clock-SI-implementation.patch (33.5K, ../0e65f96f-bfd6-fb09-89b7-e07854c215e6@postgrespro.ru/5-0004-Clock-SI-implementation.patch)
  download | inline diff:
From 52425146d9484b1b19197a5b362ce9dfd2d69b23 Mon Sep 17 00:00:00 2001
From: Andrey Lepikhov <a.lepikhov@postgrespro.ru>
Date: Mon, 22 Mar 2021 16:37:39 +0500
Subject: [PATCH 4/4] Clock SI implementation

---
 contrib/postgres_fdw/Makefile                 |   9 +
 contrib/postgres_fdw/connection.c             | 323 ++++++++++++++++--
 contrib/postgres_fdw/postgres_fdw.c           |  12 +
 contrib/postgres_fdw/postgres_fdw.h           |   2 +
 .../postgres_fdw/t/001_bank_coordinator.pl    | 264 ++++++++++++++
 .../postgres_fdw/t/002_bank_participant.pl    | 240 +++++++++++++
 src/backend/access/transam/xact.c             |   7 +
 src/include/catalog/pg_proc.dat               |   3 +
 src/test/perl/PostgresNode.pm                 |  35 ++
 9 files changed, 871 insertions(+), 24 deletions(-)
 create mode 100644 contrib/postgres_fdw/t/001_bank_coordinator.pl
 create mode 100644 contrib/postgres_fdw/t/002_bank_participant.pl

diff --git a/contrib/postgres_fdw/Makefile b/contrib/postgres_fdw/Makefile
index c1b0cad453..539b7c4c55 100644
--- a/contrib/postgres_fdw/Makefile
+++ b/contrib/postgres_fdw/Makefile
@@ -29,3 +29,12 @@ top_builddir = ../..
 include $(top_builddir)/src/Makefile.global
 include $(top_srcdir)/contrib/contrib-global.mk
 endif
+
+# Global makefile will do temp-install for 'check'. Since REGRESS is defined,
+# PGXS (included from contrib-global.mk or directly) will care to add
+# postgres_fdw to it as EXTRA_INSTALL and build pg_regress. It will also
+# actually run pg_regress, so the only thing left is tap tests.
+check: tapcheck
+
+tapcheck: temp-install
+	$(prove_check)
diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c
index ee0b4acf0b..0fadef60ac 100644
--- a/contrib/postgres_fdw/connection.c
+++ b/contrib/postgres_fdw/connection.c
@@ -12,8 +12,10 @@
  */
 #include "postgres.h"
 
+#include "access/csn_snapshot.h"
 #include "access/htup_details.h"
 #include "access/xact.h"
+#include "access/xlog.h" /* GetSystemIdentifier() */
 #include "catalog/pg_user_mapping.h"
 #include "commands/defrem.h"
 #include "funcapi.h"
@@ -28,6 +30,8 @@
 #include "utils/hsearch.h"
 #include "utils/inval.h"
 #include "utils/memutils.h"
+#include "utils/snapmgr.h"
+#include "utils/snapshot.h"
 #include "utils/syscache.h"
 
 /*
@@ -69,6 +73,21 @@ typedef struct ConnCacheEntry
  */
 static HTAB *ConnectionHash = NULL;
 
+/*
+ * FdwTransactionState
+ *
+ * Holds number of open remote transactions and shared state
+ * needed for all connection entries.
+ */
+typedef struct FdwTransactionState
+{
+	char		*gid;
+	int			nparticipants;
+	CSN			csn;
+	bool		two_phase_commit;
+} FdwTransactionState;
+static FdwTransactionState *fdwTransState;
+
 /* for assigning cursor numbers and prepared statement numbers */
 static unsigned int cursor_number = 0;
 static unsigned int prep_stmt_number = 0;
@@ -76,6 +95,9 @@ static unsigned int prep_stmt_number = 0;
 /* tracks whether any work is needed in callback functions */
 static bool xact_got_connection = false;
 
+/* counter of prepared tx made by this backend */
+static int two_phase_xact_count = 0;
+
 /*
  * SQL functions
  */
@@ -92,6 +114,7 @@ static void configure_remote_session(PGconn *conn);
 static void do_sql_command(PGconn *conn, const char *sql);
 static void begin_remote_xact(ConnCacheEntry *entry);
 static void pgfdw_xact_callback(XactEvent event, void *arg);
+static void deallocate_prepared_stmts(ConnCacheEntry *entry);
 static void pgfdw_subxact_callback(SubXactEvent event,
 								   SubTransactionId mySubid,
 								   SubTransactionId parentSubid,
@@ -148,6 +171,15 @@ GetConnection(UserMapping *user, bool will_prep_stmt)
 									  pgfdw_inval_callback, (Datum) 0);
 	}
 
+	/* allocate FdwTransactionState */
+	if (fdwTransState == NULL)
+	{
+		MemoryContext oldcxt;
+		oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
+		fdwTransState = palloc0(sizeof(FdwTransactionState));
+		MemoryContextSwitchTo(oldcxt);
+	}
+
 	/* Set flag that we did GetConnection during the current transaction */
 	xact_got_connection = true;
 
@@ -531,7 +563,8 @@ configure_remote_session(PGconn *conn)
 }
 
 /*
- * Convenience subroutine to issue a non-data-returning SQL command to remote
+ * Convenience subroutine to issue a non-data-returning SQL command or
+ * statement to remote node.
  */
 static void
 do_sql_command(PGconn *conn, const char *sql)
@@ -541,7 +574,8 @@ do_sql_command(PGconn *conn, const char *sql)
 	if (!PQsendQuery(conn, sql))
 		pgfdw_report_error(ERROR, NULL, conn, false, sql);
 	res = pgfdw_get_result(conn, sql);
-	if (PQresultStatus(res) != PGRES_COMMAND_OK)
+	if (PQresultStatus(res) != PGRES_COMMAND_OK &&
+			PQresultStatus(res) != PGRES_TUPLES_OK)
 		pgfdw_report_error(ERROR, res, conn, true, sql);
 	PQclear(res);
 }
@@ -569,6 +603,10 @@ begin_remote_xact(ConnCacheEntry *entry)
 		elog(DEBUG3, "starting remote transaction on connection %p",
 			 entry->conn);
 
+		if (UseCSNSnapshots && (/*!IsolationUsesXactSnapshot() ||*/
+								   IsolationIsSerializable()))
+			elog(ERROR, "Global snapshots support only REPEATABLE READ");
+
 		if (IsolationIsSerializable())
 			sql = "START TRANSACTION ISOLATION LEVEL SERIALIZABLE";
 		else
@@ -577,6 +615,23 @@ begin_remote_xact(ConnCacheEntry *entry)
 		do_sql_command(entry->conn, sql);
 		entry->xact_depth = 1;
 		entry->changing_xact_state = false;
+
+		if (UseCSNSnapshots && IsolationUsesXactSnapshot())
+		{
+			char import_sql[128];
+
+			/* Export our snapshot */
+			if (fdwTransState->csn == 0)
+				fdwTransState->csn = ExportCSNSnapshot();
+
+			snprintf(import_sql, sizeof(import_sql),
+				"SELECT pg_csn_snapshot_import("UINT64_FORMAT")",
+				fdwTransState->csn);
+
+			do_sql_command(entry->conn, import_sql);
+		}
+
+		fdwTransState->nparticipants += 1;
 	}
 
 	/*
@@ -784,6 +839,98 @@ pgfdw_report_error(int elevel, PGresult *res, PGconn *conn,
 	PG_END_TRY();
 }
 
+/* Callback typedef for BroadcastStmt */
+typedef bool (*BroadcastCmdResHandler) (PGresult *result, void *arg);
+
+/*
+ * Broadcast sql in parallel to all ConnectionHash entries
+ */
+static bool
+BroadcastStmt(char const * sql, unsigned expectedStatus,
+				BroadcastCmdResHandler handler, void *arg)
+{
+	HASH_SEQ_STATUS scan;
+	ConnCacheEntry *entry;
+	bool		allOk = true;
+
+	/* Broadcast sql */
+	hash_seq_init(&scan, ConnectionHash);
+	while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
+	{
+		pgfdw_reject_incomplete_xact_state_change(entry);
+
+		if (entry->xact_depth > 0 && entry->conn != NULL)
+		{
+			if (!PQsendQuery(entry->conn, sql))
+			{
+				PGresult   *res = PQgetResult(entry->conn);
+
+				elog(WARNING, "Failed to send command %s", sql);
+				pgfdw_report_error(WARNING, res, entry->conn, true, sql);
+				PQclear(res);
+			}
+		}
+	}
+
+	/* Collect responses */
+	hash_seq_init(&scan, ConnectionHash);
+	while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
+	{
+		if (entry->xact_depth > 0 && entry->conn != NULL)
+		{
+			PGresult *result = PQgetResult(entry->conn);
+
+			if (PQresultStatus(result) != expectedStatus ||
+				(handler && !handler(result, arg)))
+			{
+				elog(WARNING,
+					 "Failed command %s: status=%d, expected status=%d",
+					 sql, PQresultStatus(result), expectedStatus);
+				pgfdw_report_error(ERROR, result, entry->conn, true, sql);
+				allOk = false;
+			}
+			PQclear(result);
+			PQgetResult(entry->conn);	/* consume NULL result */
+		}
+	}
+
+	return allOk;
+}
+
+/* Wrapper for broadcasting commands */
+static bool
+BroadcastCmd(char const *sql)
+{
+	return BroadcastStmt(sql, PGRES_COMMAND_OK, NULL, NULL);
+}
+
+/* Wrapper for broadcasting statements */
+static bool
+BroadcastFunc(char const *sql)
+{
+	return BroadcastStmt(sql, PGRES_TUPLES_OK, NULL, NULL);
+}
+
+/* Callback for selecting maximal csn */
+static bool
+MaxCsnCB(PGresult *result, void *arg)
+{
+	char		   *resp;
+	CSN	   *max_csn = (CSN *) arg;
+	CSN		csn = 0;
+
+	resp = PQgetvalue(result, 0, 0);
+
+	if (resp == NULL || (*resp) == '\0' ||
+			sscanf(resp, UINT64_FORMAT, &csn) != 1)
+		return false;
+
+	if (*max_csn < csn)
+		*max_csn = csn;
+
+	return true;
+}
+
 /*
  * pgfdw_xact_callback --- cleanup at main-transaction end.
  *
@@ -801,6 +948,104 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 	if (!xact_got_connection)
 		return;
 
+	/* Handle possible two-phase commit */
+	if (event == XACT_EVENT_PARALLEL_PRE_COMMIT || event == XACT_EVENT_PRE_COMMIT)
+	{
+		bool include_local_tx = false;
+
+		/* Should we take into account this node? */
+		if (TransactionIdIsValid(GetCurrentTransactionIdIfAny()))
+		{
+			include_local_tx = true;
+			fdwTransState->nparticipants += 1;
+		}
+
+		/* Switch to 2PC mode there were more than one participant */
+		if (UseCSNSnapshots && fdwTransState->nparticipants > 1)
+			fdwTransState->two_phase_commit = true;
+
+		/*
+		 * Data change to fast node on slow node, and slow node can see data change.
+		 * In order to implement above we should change code here.
+		 */
+		if (fdwTransState->two_phase_commit)
+		{
+			CSN max_csn = InProgressCSN;
+			CSN my_csn = InProgressCSN;
+			bool res;
+			char *sql;
+
+			fdwTransState->gid = psprintf("pgfdw:%lld:%llu:%d:%u:%d:%d",
+										  (long long) GetCurrentTimestamp(),
+										  (long long) GetSystemIdentifier(),
+										  MyProcPid,
+										  GetCurrentTransactionIdIfAny(),
+										  ++two_phase_xact_count,
+										  fdwTransState->nparticipants);
+
+			/* Broadcast PREPARE */
+			sql = psprintf("PREPARE TRANSACTION '%s'", fdwTransState->gid);
+			res = BroadcastCmd(sql);
+
+			if (IsolationUsesXactSnapshot())
+			{
+				if (!res)
+					goto error;
+
+				/* Broadcast pg_csn_snapshot_prepare() */
+				if (include_local_tx)
+					my_csn = CSNSnapshotPrepareCurrent();
+
+				sql = psprintf("SELECT pg_csn_snapshot_prepare('%s')",
+															fdwTransState->gid);
+				res = BroadcastStmt(sql, PGRES_TUPLES_OK, MaxCsnCB, &max_csn);
+				if (!res)
+					goto error;
+
+				/* select maximal global csn */
+				if (include_local_tx && my_csn > max_csn)
+					max_csn = my_csn;
+
+				/*
+				 * We should always notice local node to update the csn, so local can
+				 * see the change next transaction.
+				 */
+				if (include_local_tx)
+					CSNSnapshotAssignCurrent(max_csn);
+				else
+					/*
+					 * Read-only transactions haven't assigned xid csn. We only
+					 * increase the last csn value.
+					 */
+					GenerateCSN(max_csn);
+
+				sql = psprintf("SELECT pg_csn_snapshot_assign('%s',"UINT64_FORMAT")",
+								fdwTransState->gid, max_csn);
+				res = BroadcastFunc(sql);
+			}
+
+error:
+			if (!res)
+			{
+				sql = psprintf("ABORT PREPARED '%s'", fdwTransState->gid);
+				BroadcastCmd(sql);
+				elog(ERROR, "Failed to PREPARE transaction on remote node");
+			}
+
+			/*
+			 * Do not fall down. Consequent COMMIT event will clean thing up.
+			 */
+			return;
+		}
+	}
+
+	/* COMMIT open transaction of we were doing 2PC */
+	if (fdwTransState->two_phase_commit &&
+		(event == XACT_EVENT_PARALLEL_COMMIT || event == XACT_EVENT_COMMIT))
+	{
+		BroadcastCmd(psprintf("COMMIT PREPARED '%s'", fdwTransState->gid));
+	}
+
 	/*
 	 * Scan all connection cache entries to find open remote transactions, and
 	 * close them.
@@ -808,8 +1053,6 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 	hash_seq_init(&scan, ConnectionHash);
 	while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
 	{
-		PGresult   *res;
-
 		/* Ignore cache entry if no open connection right now */
 		if (entry->conn == NULL)
 			continue;
@@ -826,6 +1069,7 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 			{
 				case XACT_EVENT_PARALLEL_PRE_COMMIT:
 				case XACT_EVENT_PRE_COMMIT:
+					Assert(!fdwTransState->two_phase_commit);
 
 					/*
 					 * If abort cleanup previously failed for this connection,
@@ -838,28 +1082,22 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 					do_sql_command(entry->conn, "COMMIT TRANSACTION");
 					entry->changing_xact_state = false;
 
-					/*
-					 * If there were any errors in subtransactions, and we
-					 * made prepared statements, do a DEALLOCATE ALL to make
-					 * sure we get rid of all prepared statements. This is
-					 * annoying and not terribly bulletproof, but it's
-					 * probably not worth trying harder.
-					 *
-					 * DEALLOCATE ALL only exists in 8.3 and later, so this
-					 * constrains how old a server postgres_fdw can
-					 * communicate with.  We intentionally ignore errors in
-					 * the DEALLOCATE, so that we can hobble along to some
-					 * extent with older servers (leaking prepared statements
-					 * as we go; but we don't really support update operations
-					 * pre-8.3 anyway).
-					 */
-					if (entry->have_prep_stmt && entry->have_error)
+					if (UseCSNSnapshots)
 					{
-						res = PQexec(entry->conn, "DEALLOCATE ALL");
-						PQclear(res);
+						CSN			csn = InvalidCSN;
+						PGresult	*res;
+
+						res = pgfdw_exec_query(entry->conn, "SELECT pg_current_csn()");
+						if (PQresultStatus(res) == PGRES_TUPLES_OK)
+						{
+							sscanf(PQgetvalue(res, 0, 0), "%lu", &csn);
+
+							if (csn != InvalidCSN)
+								GenerateCSN(csn);
+						}
 					}
-					entry->have_prep_stmt = false;
-					entry->have_error = false;
+
+					deallocate_prepared_stmts(entry);
 					break;
 				case XACT_EVENT_PRE_PREPARE:
 
@@ -878,6 +1116,11 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 					break;
 				case XACT_EVENT_PARALLEL_COMMIT:
 				case XACT_EVENT_COMMIT:
+					if (fdwTransState->two_phase_commit)
+						deallocate_prepared_stmts(entry);
+					else /* Pre-commit should have closed the open transaction */
+						elog(ERROR, "missed cleaning up connection during pre-commit");
+					break;
 				case XACT_EVENT_PREPARE:
 					/* Pre-commit should have closed the open transaction */
 					elog(ERROR, "missed cleaning up connection during pre-commit");
@@ -975,6 +1218,38 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 
 	/* Also reset cursor numbering for next transaction */
 	cursor_number = 0;
+
+	/* Reset fdwTransState */
+	memset(fdwTransState, '\0', sizeof(FdwTransactionState));
+}
+
+/*
+ * If there were any errors in subtransactions, and we
+ * made prepared statements, do a DEALLOCATE ALL to make
+ * sure we get rid of all prepared statements. This is
+ * annoying and not terribly bulletproof, but it's
+ * probably not worth trying harder.
+ *
+ * DEALLOCATE ALL only exists in 8.3 and later, so this
+ * constrains how old a server postgres_fdw can
+ * communicate with.  We intentionally ignore errors in
+ * the DEALLOCATE, so that we can hobble along to some
+ * extent with older servers (leaking prepared statements
+ * as we go; but we don't really support update operations
+ * pre-8.3 anyway).
+ */
+static void
+deallocate_prepared_stmts(ConnCacheEntry *entry)
+{
+	PGresult   *res;
+
+	if (entry->have_prep_stmt && entry->have_error)
+	{
+		res = PQexec(entry->conn, "DEALLOCATE ALL");
+		PQclear(res);
+	}
+	entry->have_prep_stmt = false;
+	entry->have_error = false;
 }
 
 /*
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index 35b48575c5..f59f78a770 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -312,6 +312,9 @@ typedef struct
 	List	   *already_used;	/* expressions already dealt with */
 } ec_member_foreign_arg;
 
+bool		UseCSNSnapshots;
+void		_PG_init(void);
+
 /*
  * SQL functions
  */
@@ -6813,3 +6816,12 @@ get_batch_size_option(Relation rel)
 
 	return batch_size;
 }
+
+void
+_PG_init(void)
+{
+	DefineCustomBoolVariable("postgres_fdw.use_csn_snapshots",
+							 "Use global snapshots for FDW transactions", NULL,
+							 &UseCSNSnapshots, false, PGC_USERSET, 0, NULL,
+							 NULL, NULL);
+}
\ No newline at end of file
diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h
index 1f67b4d9fd..87fb9bf8c3 100644
--- a/contrib/postgres_fdw/postgres_fdw.h
+++ b/contrib/postgres_fdw/postgres_fdw.h
@@ -211,4 +211,6 @@ extern const char *get_jointype_name(JoinType jointype);
 extern bool is_builtin(Oid objectId);
 extern bool is_shippable(Oid objectId, Oid classId, PgFdwRelationInfo *fpinfo);
 
+extern bool UseCSNSnapshots;
+
 #endif							/* POSTGRES_FDW_H */
diff --git a/contrib/postgres_fdw/t/001_bank_coordinator.pl b/contrib/postgres_fdw/t/001_bank_coordinator.pl
new file mode 100644
index 0000000000..a32637f581
--- /dev/null
+++ b/contrib/postgres_fdw/t/001_bank_coordinator.pl
@@ -0,0 +1,264 @@
+use strict;
+use warnings;
+
+use PostgresNode;
+use TestLib;
+use Test::More tests => 3;
+
+my $master = get_new_node("master");
+$master->init;
+$master->append_conf('postgresql.conf', qq(
+	max_prepared_transactions = 30
+	log_checkpoints = true
+	postgres_fdw.use_csn_snapshots = on
+	enable_csn_snapshot = on
+	default_transaction_isolation = 'REPEATABLE READ'
+));
+$master->start;
+
+my $shard1 = get_new_node("shard1");
+$shard1->init;
+$shard1->append_conf('postgresql.conf', qq(
+	max_prepared_transactions = 30
+	csn_snapshot_defer_time = 15
+	enable_csn_snapshot = on
+));
+$shard1->start;
+
+my $shard2 = get_new_node("shard2");
+$shard2->init;
+$shard2->append_conf('postgresql.conf', qq(
+	max_prepared_transactions = 30
+	csn_snapshot_defer_time = 15
+	enable_csn_snapshot = on
+));
+$shard2->start;
+
+###############################################################################
+# Prepare nodes
+###############################################################################
+
+$master->safe_psql('postgres', qq[
+	CREATE EXTENSION postgres_fdw;
+	CREATE TABLE accounts(id integer primary key, amount integer);
+	CREATE TABLE global_transactions(tx_time timestamp);
+]);
+
+foreach my $node ($shard1, $shard2)
+{
+	my $port = $node->port;
+	my $host = $node->host;
+
+	$node->safe_psql('postgres',
+			"CREATE TABLE accounts(id integer primary key, amount integer)");
+
+	$master->safe_psql('postgres', qq[
+		CREATE SERVER shard_$port FOREIGN DATA WRAPPER postgres_fdw options(dbname 'postgres', host '$host', port '$port');
+		CREATE FOREIGN TABLE accounts_fdw_$port() inherits (accounts) server shard_$port options(table_name 'accounts');
+		CREATE USER MAPPING for CURRENT_USER SERVER shard_$port;
+	])
+}
+
+$shard1->safe_psql('postgres', qq[
+	insert into accounts select 2*id-1, 0 from generate_series(1, 10010) as id;
+	CREATE TABLE local_transactions(tx_time timestamp);
+]);
+
+$shard2->safe_psql('postgres', qq[
+	insert into accounts select 2*id, 0 from generate_series(1, 10010) as id;
+	CREATE TABLE local_transactions(tx_time timestamp);
+]);
+
+diag("master: @{[$master->connstr('postgres')]}");
+diag("shard1: @{[$shard1->connstr('postgres')]}");
+diag("shard2: @{[$shard2->connstr('postgres')]}");
+
+###############################################################################
+# pgbench scripts
+###############################################################################
+
+my $bank = File::Temp->new();
+append_to_file($bank, q{
+	\set id random(1, 20000)
+	BEGIN;
+	WITH upd AS (UPDATE accounts SET amount = amount - 1 WHERE id = :id RETURNING *)
+		INSERT into global_transactions SELECT now() FROM upd;
+	UPDATE accounts SET amount = amount + 1 WHERE id = (:id + 1);
+	COMMIT;
+});
+
+my $bank1 = File::Temp->new();
+append_to_file($bank1, q{
+	\set id random(1, 10000)
+	BEGIN;
+	WITH upd AS (UPDATE accounts SET amount = amount - 1 WHERE id = (2*:id + 1) RETURNING *)
+		INSERT into local_transactions SELECT now() FROM upd;
+	UPDATE accounts SET amount = amount + 1 WHERE id = (2*:id + 3);
+	COMMIT;
+});
+
+my $bank2 = File::Temp->new();
+append_to_file($bank2, q{
+	\set id random(1, 10000)
+
+	BEGIN;
+	WITH upd AS (UPDATE accounts SET amount = amount - 1 WHERE id = 2*:id RETURNING *)
+		INSERT into local_transactions SELECT now() FROM upd;
+	UPDATE accounts SET amount = amount + 1 WHERE id = (2*:id + 2);
+	COMMIT;
+});
+
+###############################################################################
+# Helpers
+###############################################################################
+
+sub count_and_delete_rows
+{
+	my ($node, $table) = @_;
+	my $count;
+
+	$count = $node->safe_psql('postgres',"select count(*) from $table");
+	$node->safe_psql('postgres',"delete from $table");
+	diag($node->name, ": completed $count transactions");
+	return $count;
+}
+
+###############################################################################
+# Concurrent global transactions
+###############################################################################
+
+my ($err, $rc);
+my $started;
+my $seconds = 30;
+my $selects;
+my $total = '0';
+my $oldtotal = '0';
+my $isolation_errors = 0;
+
+
+my $pgb_handle;
+
+$pgb_handle = $master->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+
+$started = time();
+$selects = 0;
+while (time() - $started < $seconds)
+{
+	$total = $master->safe_psql('postgres', "select sum(amount) from accounts");
+	if ( ($total ne $oldtotal) and ($total ne '') )
+	{
+		$isolation_errors++;
+		$oldtotal = $total;
+		diag("Isolation error. Total = $total");
+	}
+	if ($total ne '') { $selects++; }
+}
+
+$master->pgbench_await($pgb_handle);
+
+# sanity check
+diag("completed $selects selects");
+die "no actual transactions happend" unless ( $selects > 0 &&
+	count_and_delete_rows($master, 'global_transactions') > 0);
+
+is($isolation_errors, 0, 'isolation between concurrent global transaction');
+
+###############################################################################
+# Concurrent global and local transactions
+###############################################################################
+
+my ($pgb_handle1, $pgb_handle2, $pgb_handle3);
+
+# global txses
+$pgb_handle1 = $master->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+
+# concurrent local
+$pgb_handle2 = $shard1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank1, 'postgres' );
+$pgb_handle3 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank2, 'postgres' );
+
+$started = time();
+$selects = 0;
+$oldtotal = 0;
+while (time() - $started < $seconds)
+{
+	$total = $master->safe_psql('postgres', "select sum(amount) from accounts");
+	if ( ($total ne $oldtotal) and ($total ne '') )
+	{
+		$isolation_errors++;
+		$oldtotal = $total;
+		diag("Isolation error. Total = $total");
+	}
+	if ($total ne '') { $selects++; }
+}
+
+diag("selects = $selects");
+$master->pgbench_await($pgb_handle1);
+$shard1->pgbench_await($pgb_handle2);
+$shard2->pgbench_await($pgb_handle3);
+
+diag("completed $selects selects");
+die "" unless ( $selects > 0 &&
+	count_and_delete_rows($master, 'global_transactions') > 0 &&
+	count_and_delete_rows($shard1, 'local_transactions') > 0 &&
+	count_and_delete_rows($shard2, 'local_transactions') > 0);
+
+is($isolation_errors, 0, 'isolation between concurrent global and local transactions');
+
+
+###############################################################################
+# Snapshot stability
+###############################################################################
+
+my ($hashes, $hash1, $hash2);
+my $stability_errors = 0;
+
+# global txses
+$pgb_handle1 = $master->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+# concurrent local
+$pgb_handle2 = $shard1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank1, 'postgres' );
+$pgb_handle3 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank2, 'postgres' );
+
+$selects = 0;
+$started = time();
+while (time() - $started < $seconds)
+{
+	foreach my $node ($master, $shard1, $shard2)
+	{
+		($hash1, $_, $hash2) = split "\n", $node->safe_psql('postgres', qq[
+			begin isolation level repeatable read;
+			select md5(array_agg((t.*)::text)::text) from (select * from accounts order by id) as t;
+			select pg_sleep(3);
+			select md5(array_agg((t.*)::text)::text) from (select * from accounts order by id) as t;
+			commit;
+		]);
+
+		if ($hash1 ne $hash2)
+		{
+			diag("oops");
+			$stability_errors++;
+		}
+		elsif ($hash1 eq '' or $hash2 eq '')
+		{
+			die;
+		}
+		else
+		{
+			$selects++;
+		}
+	}
+}
+
+$master->pgbench_await($pgb_handle1);
+$shard1->pgbench_await($pgb_handle2);
+$shard2->pgbench_await($pgb_handle3);
+
+die "" unless ( $selects > 0 &&
+	count_and_delete_rows($master, 'global_transactions') > 0 &&
+	count_and_delete_rows($shard1, 'local_transactions') > 0 &&
+	count_and_delete_rows($shard2, 'local_transactions') > 0);
+
+is($stability_errors, 0, 'snapshot is stable during concurrent global and local transactions');
+
+$master->stop;
+$shard1->stop;
+$shard2->stop;
diff --git a/contrib/postgres_fdw/t/002_bank_participant.pl b/contrib/postgres_fdw/t/002_bank_participant.pl
new file mode 100644
index 0000000000..e5a5a721c6
--- /dev/null
+++ b/contrib/postgres_fdw/t/002_bank_participant.pl
@@ -0,0 +1,240 @@
+use strict;
+use warnings;
+
+use PostgresNode;
+use TestLib;
+use Test::More tests => 3;
+
+my $shard1 = get_new_node("shard1");
+$shard1->init;
+$shard1->append_conf('postgresql.conf', qq(
+	max_prepared_transactions = 30
+	postgres_fdw.use_csn_snapshots = on
+	csn_snapshot_defer_time = 15
+	enable_csn_snapshot = on
+	default_transaction_isolation = 'REPEATABLE READ'
+));
+$shard1->start;
+
+my $shard2 = get_new_node("shard2");
+$shard2->init;
+$shard2->append_conf('postgresql.conf', qq(
+	max_prepared_transactions = 30
+	postgres_fdw.use_csn_snapshots = on
+	csn_snapshot_defer_time = 15
+	enable_csn_snapshot = on
+	default_transaction_isolation = 'REPEATABLE READ'
+));
+$shard2->start;
+
+###############################################################################
+# Prepare nodes
+###############################################################################
+
+my @shards = ($shard1, $shard2);
+
+foreach my $node (@shards)
+{
+	$node->safe_psql('postgres', qq[
+		CREATE EXTENSION postgres_fdw;
+		CREATE TABLE accounts(id integer primary key, amount integer);
+		CREATE TABLE accounts_local() inherits(accounts);
+		CREATE TABLE global_transactions(tx_time timestamp);
+		CREATE TABLE local_transactions(tx_time timestamp);
+	]);
+
+	foreach my $neighbor (@shards)
+	{
+		next if ($neighbor eq $node);
+
+		my $port = $neighbor->port;
+		my $host = $neighbor->host;
+
+		$node->safe_psql('postgres', qq[
+			CREATE SERVER shard_$port FOREIGN DATA WRAPPER postgres_fdw
+					options(dbname 'postgres', host '$host', port '$port');
+			CREATE FOREIGN TABLE accounts_fdw_$port() inherits (accounts)
+					server shard_$port options(table_name 'accounts_local');
+			CREATE USER MAPPING for CURRENT_USER SERVER shard_$port;
+		]);
+	}
+}
+
+$shard1->psql('postgres', "insert into accounts_local select 2*id-1, 0 from generate_series(1, 10010) as id;");
+$shard2->psql('postgres', "insert into accounts_local select 2*id,   0 from generate_series(1, 10010) as id;");
+
+###############################################################################
+# pgbench scripts
+###############################################################################
+
+my $bank = File::Temp->new();
+append_to_file($bank, q{
+	\set id random(1, 20000)
+	BEGIN;
+	WITH upd AS (UPDATE accounts SET amount = amount - 1 WHERE id = :id RETURNING *)
+		INSERT into global_transactions SELECT now() FROM upd;
+	UPDATE accounts SET amount = amount + 1 WHERE id = (:id + 1);
+	COMMIT;
+});
+
+###############################################################################
+# Helpers
+###############################################################################
+
+sub count_and_delete_rows
+{
+	my ($node, $table) = @_;
+	my $count;
+
+	$count = $node->safe_psql('postgres',"select count(*) from $table");
+	$node->safe_psql('postgres',"delete from $table");
+	diag($node->name, ": completed $count transactions");
+	return $count;
+}
+
+###############################################################################
+# Concurrent global transactions
+###############################################################################
+
+my ($err, $rc);
+my $started;
+my $seconds = 30;
+my $selects;
+my $total = '0';
+my $oldtotal = '0';
+my $isolation_errors = 0;
+my $i;
+
+
+my ($pgb_handle1, $pgb_handle2);
+
+$pgb_handle1 = $shard1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+$pgb_handle2 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+
+$started = time();
+$selects = 0;
+$i = 0;
+while (time() - $started < $seconds)
+{
+	my $shard = $shard1;
+	foreach my $shard (@shards)
+	{
+		$total = $shard->safe_psql('postgres', "select sum(amount) from accounts");
+		if ( ($total ne $oldtotal) and ($total ne '') )
+		{
+			$isolation_errors++;
+			$oldtotal = $total;
+			diag("$i: Isolation error. Total = $total");
+		}
+		if ($total ne '') { $selects++; }
+	}
+	$i++;
+}
+
+$shard1->pgbench_await($pgb_handle1);
+$shard2->pgbench_await($pgb_handle2);
+
+# sanity check
+diag("completed $selects selects");
+die "no actual transactions happend" unless ( $selects > 0 &&
+	count_and_delete_rows($shard1, 'global_transactions') > 0 &&
+	count_and_delete_rows($shard2, 'global_transactions') > 0);
+
+is($isolation_errors, 0, 'isolation between concurrent global transaction');
+
+###############################################################################
+# And do the same after soft restart
+###############################################################################
+
+$shard1->restart;
+$shard2->restart;
+$shard1->poll_query_until('postgres', "select 't'")
+	or die "Timed out waiting for shard1 to became online";
+$shard2->poll_query_until('postgres', "select 't'")
+	or die "Timed out waiting for shard2 to became online";
+
+$seconds = 15;
+$pgb_handle1 = $shard1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+$pgb_handle2 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+
+$started = time();
+$selects = 0;
+$i = 0;
+
+while (time() - $started < $seconds)
+{
+	my $shard = $shard1;
+	foreach my $shard (@shards)
+	{
+		$total = $shard->safe_psql('postgres', "select sum(amount) from accounts");
+		if ( ($total ne $oldtotal) and ($total ne '') )
+		{
+			$isolation_errors++;
+			$oldtotal = $total;
+			diag("$i: Isolation error. Total = $total");
+		}
+		if ($total ne '') { $selects++; }
+	}
+	$i++;
+}
+
+$shard1->pgbench_await($pgb_handle1);
+$shard2->pgbench_await($pgb_handle2);
+
+# sanity check
+diag("completed $selects selects");
+die "no actual transactions happend" unless ( $selects > 0 &&
+	count_and_delete_rows($shard1, 'global_transactions') > 0 &&
+	count_and_delete_rows($shard2, 'global_transactions') > 0);
+
+is($isolation_errors, 0, 'isolation between concurrent global transaction after restart');
+
+###############################################################################
+# And do the same after hard restart
+###############################################################################
+
+$shard1->teardown_node;
+$shard2->teardown_node;
+$shard1->start;
+$shard2->start;
+$shard1->poll_query_until('postgres', "select 't'")
+	or die "Timed out waiting for shard1 to became online";
+$shard2->poll_query_until('postgres', "select 't'")
+	or die "Timed out waiting for shard2 to became online";
+
+
+$seconds = 15;
+$pgb_handle1 = $shard1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+$pgb_handle2 = $shard2->pgbench_async(-n, -c => 5, -T => $seconds, -f => $bank, 'postgres' );
+
+$started = time();
+$selects = 0;
+$i = 0;
+
+while (time() - $started < $seconds)
+{
+	my $shard = $shard1;
+	foreach my $shard (@shards)
+	{
+		$total = $shard->safe_psql('postgres', "select sum(amount) from accounts");
+		if ( ($total ne $oldtotal) and ($total ne '') )
+		{
+			$isolation_errors++;
+			$oldtotal = $total;
+			diag("$i: Isolation error. Total = $total");
+		}
+		if ($total ne '') { $selects++; }
+	}
+	$i++;
+}
+
+$shard1->pgbench_await($pgb_handle1);
+$shard2->pgbench_await($pgb_handle2);
+
+# sanity check
+diag("completed $selects selects");
+die "no actual transactions happend" unless ( $selects > 0 &&
+	count_and_delete_rows($shard1, 'global_transactions') > 0 &&
+	count_and_delete_rows($shard2, 'global_transactions') > 0);
+
+is($isolation_errors, 0, 'isolation between concurrent global transaction after hard restart');
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index deabedbe37..d375eacc8f 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -2103,6 +2103,13 @@ StartTransaction(void)
 	ShowTransactionState("StartTransaction");
 }
 
+Datum
+pg_current_csn(PG_FUNCTION_ARGS)
+{
+	CSN	csn = GenerateCSN(InvalidCSN);
+
+	PG_RETURN_INT64(csn);
+}
 
 /*
  *	CommitTransaction
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 8585464c5b..ed776e5166 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11424,5 +11424,8 @@
 { oid => '10004', descr => 'assign csn to distributed transaction',
   proname => 'pg_csn_snapshot_assign', provolatile => 'v', proparallel => 'u',
   prorettype => 'void', proargtypes => 'text int8', prosrc => 'pg_csn_snapshot_assign' },
+{ oid => '10005', descr => 'get current CSN',
+  proname => 'pg_current_csn', provolatile => 'v', proparallel => 'u',
+  prorettype => 'int8', proargtypes => '', prosrc => 'pg_current_csn' },
 
 ]
diff --git a/src/test/perl/PostgresNode.pm b/src/test/perl/PostgresNode.pm
index 97e05993be..ea251be0b7 100644
--- a/src/test/perl/PostgresNode.pm
+++ b/src/test/perl/PostgresNode.pm
@@ -2243,6 +2243,41 @@ sub pg_recvlogical_upto
 	}
 }
 
+sub pgbench()
+{
+	my ($self, $node, @args) = @_;
+	my $pgbench_handle = $self->pgbench_async($node, @args);
+	$self->pgbench_await($pgbench_handle);
+}
+
+sub pgbench_async()
+{
+	my ($self, @args) = @_;
+
+	my ($in, $out, $err, $rc);
+	$in = '';
+	$out = '';
+
+	my @pgbench_command = (
+		'pgbench',
+		-h => $self->host,
+		-p => $self->port,
+		@args
+	);
+	my $handle = IPC::Run::start(\@pgbench_command, $in, $out);
+	return $handle;
+}
+
+sub pgbench_await()
+{
+	my ($self, $pgbench_handle) = @_;
+
+	# During run some pgbench threads can exit (for example due to
+	# serialization error). That will set non-zero returning code.
+	# So don't check return code here and leave it to a caller.
+	my $rc = IPC::Run::finish($pgbench_handle);
+}
+
 =pod
 
 =back
-- 
2.25.1