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

