Add guc gp_hot_standby_snapshot_restore_point_name

For hot standby cluster, only at the restore_point
it can reach a consistent state.  So the snapshot
on hot standby should be token at thre restore_point.

Add the guc gp_hot_standby_snapshot_restore_point_name,
it sets the restore point name that the snapshot should
be token at.

Add a new type of xlog: XLOG_RESTORE_POINT_RUNNING_XACTS,
it records the running xids and the restore point name.
On hot standby, when doing recovery, it the xlogRecord's
restore point name equals the
gp_hot_standby_snapshot_restore_point_name,
then we will update the hot standby snapshot.

Beside the XLOG_RESTORE_POINT_RUNNING_XACTS, I think the
for checkpoint shutdown xlogrecord, the hot standby
cluster is also consistent, we can update the snapshot
too.

TODO:
I have added the test hot_standby/snapshot. But the test
failed. It is expected to fail when query table hs_sh
on hot standby before the restore point "r1" created.
It means this commit cannot work correctly, I must miss
something.

Author: fanfuxiaoran

src/backend/access/rmgrdesc/standbydesc.c

@@ -36,6 +36,13 @@ standby_desc_running_xacts(StringInfo buf, xl_running_xacts *xlrec)
 		appendStringInfoString(buf, "; subxid ovf");
 }
 
+static void
+standby_desc_rp_running_xacts(StringInfo buf, xl_running_xacts *xlrec, const char *rpName)
+{
+	standby_desc_running_xacts(buf, xlrec);
+	appendStringInfo(buf, "restore_point %s", rpName);
+}
+
 void
 standby_desc(StringInfo buf, XLogReaderState *record)
 {
@@ -73,6 +80,24 @@ standby_desc(StringInfo buf, XLogReaderState *record)
 		gxid = *((DistributedTransactionId *) rec);
 		appendStringInfo(buf, UINT64_FORMAT, gxid);
 	}
+	else if (info == XLOG_RESTORE_POINT_RUNNING_XACTS)
+	{
+		xl_restore_point_running_xacts *xlrec
+			= (xl_restore_point_running_xacts *) rec;
+		xl_running_xacts *xlrecRunning= (xl_running_xacts*) palloc(offsetof(xl_running_xacts, xids) +
+				(xlrec->xcnt + xlrec->subxcnt) * sizeof(TransactionId));
+		xlrecRunning->xcnt = xlrec->xcnt;
+		xlrecRunning->subxcnt = xlrec->subxcnt;
+		xlrecRunning->subxid_overflow = xlrec->subxid_overflow;
+		xlrecRunning->nextXid = xlrec->nextXid;
+		xlrecRunning->oldestRunningXid = xlrec->oldestRunningXid;
+		xlrecRunning->latestCompletedXid = xlrec->latestCompletedXid;
+		memcpy(xlrecRunning->xids, xlrec->xids,
+				(xlrec->xcnt + xlrec->subxcnt) * sizeof(TransactionId));
+
+		standby_desc_rp_running_xacts(buf, xlrecRunning, xlrec->rpName);
+		pfree(xlrecRunning);
+	}
 
 }
 
@@ -95,6 +120,9 @@ standby_identify(uint8 info)
 		case XLOG_LATESTCOMPLETED_GXID:
 			id = "XLOG_LATESTCOMPLETED_GXID";
 			break;
+		case XLOG_RESTORE_POINT_RUNNING_XACTS:
+			id = "XLOG_RESTORE_POINT_RUNNING_XACTS";
+			break;
 	}
 
 	return id;

src/backend/access/transam/xlog.c

@@ -10757,6 +10757,9 @@ XLogRestorePoint(const char *rpName)
 	xlrec.rp_time = GetCurrentTimestamp();
 	strlcpy(xlrec.rp_name, rpName, MAXFNAMELEN);
 
+	/* LogHotStandby for the restore here */
+	LogRestorePointStandbySnapshot(rpName);
+
 	XLogBeginInsert();
 	XLogRegisterData((char *) &xlrec, sizeof(xl_restore_point));
 

src/backend/replication/logical/decode.c

@@ -374,6 +374,9 @@ standby_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 		case XLOG_LATESTCOMPLETED_GXID:
 			/* FIXME: need to decode this part? */
 			break;
+		case XLOG_RESTORE_POINT_RUNNING_XACTS:
+			/* FIXME: same as XLOG_LATESTCOMPLETED_GXID */
+			break;
 		default:
 			elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
 	}

src/backend/storage/ipc/standby.c

@@ -42,6 +42,7 @@ int			vacuum_defer_cleanup_age;
 int			max_standby_archive_delay = 30 * 1000;
 int			max_standby_streaming_delay = 30 * 1000;
 bool		log_recovery_conflict_waits = false;
+char		*gp_hot_standby_snapshot_restore_point_name = NULL;
 
 static HTAB *RecoveryLockLists;
 
@@ -55,7 +56,9 @@ static void ResolveRecoveryConflictWithVirtualXIDs(VirtualTransactionId *waitlis
 												   uint32 wait_event_info,
 												   bool report_waiting);
 static void SendRecoveryConflictWithBufferPin(ProcSignalReason reason);
+static XLogRecPtr LogStandbySnapshotImpl(const char *rpName);
 static XLogRecPtr LogCurrentRunningXacts(RunningTransactions CurrRunningXacts);
+static XLogRecPtr CbdbLogRestorePointRunningXacts(RunningTransactions CurrRunningXacts, const char *rpName);
 static void LogAccessExclusiveLocks(int nlocks, xl_standby_lock *locks);
 static const char *get_recovery_conflict_desc(ProcSignalReason reason);
 
@@ -1129,6 +1132,12 @@ standby_redo(XLogReaderState *record)
 	}
 	else if (info == XLOG_RUNNING_XACTS)
 	{
+		if (EnableHotStandby && gp_hot_standby_snapshot_restore_point_name)
+		{
+			elog(LOG, "not calling redo");
+			return;
+		}
+
 		xl_running_xacts *xlrec = (xl_running_xacts *) XLogRecGetData(record);
 		RunningTransactionsData running;
 
@@ -1169,6 +1178,31 @@ standby_redo(XLogReaderState *record)
 			LWLockRelease(ProcArrayLock);
 		}
 	}
+	else if (info == XLOG_RESTORE_POINT_RUNNING_XACTS)
+	{
+		if (!gp_hot_standby_snapshot_restore_point_name)
+			return;
+		xl_restore_point_running_xacts *xlrec =
+			(xl_restore_point_running_xacts *) XLogRecGetData(record);
+		char *rpName = gp_hot_standby_snapshot_restore_point_name;
+
+		if (EnableHotStandby && rpName &&
+				strcmp(xlrec->rpName, rpName) == 0)
+		{
+			RunningTransactionsData running;
+
+			running.xcnt = xlrec->xcnt;
+			running.subxcnt = xlrec->subxcnt;
+			running.subxid_overflow = xlrec->subxid_overflow;
+			running.nextXid = xlrec->nextXid;
+			running.latestCompletedXid = xlrec->latestCompletedXid;
+			running.oldestRunningXid = xlrec->oldestRunningXid;
+			running.xids = xlrec->xids;
+
+			elog(LOG, "update running xact %s", rpName);
+			ProcArrayApplyRecoveryInfo(&running);
+		}
+	}
 	else
 		elog(PANIC, "standby_redo: unknown op code %u", info);
 }
@@ -1241,6 +1275,12 @@ standby_redo(XLogReaderState *record)
  */
 XLogRecPtr
 LogStandbySnapshot(void)
+{
+	return LogStandbySnapshotImpl(NULL);
+}
+
+static XLogRecPtr
+LogStandbySnapshotImpl(const char *rpName)
 {
 	XLogRecPtr	recptr;
 	RunningTransactions running;
@@ -1278,7 +1318,15 @@ LogStandbySnapshot(void)
 	if (wal_level < WAL_LEVEL_LOGICAL)
 		LWLockRelease(ProcArrayLock);
 
-	recptr = LogCurrentRunningXacts(running);
+	if (rpName == NULL)
+	{
+		recptr = LogCurrentRunningXacts(running);
+	}
+	else
+	{
+		recptr = CbdbLogRestorePointRunningXacts(running, rpName);
+	}
+
 
 	/* Release lock if we kept it longer ... */
 	if (wal_level >= WAL_LEVEL_LOGICAL)
@@ -1289,9 +1337,9 @@ LogStandbySnapshot(void)
 	if (IS_QUERY_DISPATCHER())
 	{
 		/*
-		 * GPDB: write latestCompletedGxid too, because the standby needs this 
+		 * GPDB: write latestCompletedGxid too, because the standby needs this
 		 * value for creating distributed snapshot. The standby cannot rely on
-		 * the nextGxid value to set latestCompletedGxid during restart (which 
+		 * the nextGxid value to set latestCompletedGxid during restart (which
 		 * the primary does) because nextGxid was bumped in the checkpoint.
 		 */
 		LWLockAcquire(ProcArrayLock, LW_SHARED);
@@ -1370,6 +1418,69 @@ LogCurrentRunningXacts(RunningTransactions CurrRunningXacts)
 	return recptr;
 }
 
+XLogRecPtr
+LogRestorePointStandbySnapshot(const char *rpName)
+{
+	return LogStandbySnapshotImpl(rpName);
+}
+
+static XLogRecPtr
+CbdbLogRestorePointRunningXacts(RunningTransactions CurrRunningXacts, const char *rpName)
+{
+	xl_restore_point_running_xacts	xlrec;
+	XLogRecPtr	recptr;
+
+	xlrec.xcnt = CurrRunningXacts->xcnt;
+	xlrec.subxcnt = CurrRunningXacts->subxcnt;
+	xlrec.subxid_overflow = CurrRunningXacts->subxid_overflow;
+	xlrec.nextXid = CurrRunningXacts->nextXid;
+	xlrec.oldestRunningXid = CurrRunningXacts->oldestRunningXid;
+	xlrec.latestCompletedXid = CurrRunningXacts->latestCompletedXid;
+	strlcpy(xlrec.rpName, rpName, MAXFNAMELEN);
+
+	/* Header */
+	XLogBeginInsert();
+	XLogSetRecordFlags(XLOG_MARK_UNIMPORTANT);
+	XLogRegisterData((char *) (&xlrec), MinSizeOfXactRpRunningXacts);
+
+	/* array of TransactionIds */
+	if (xlrec.xcnt > 0)
+		XLogRegisterData((char *) CurrRunningXacts->xids,
+						 (xlrec.xcnt + xlrec.subxcnt) * sizeof(TransactionId));
+
+	recptr = XLogInsert(RM_STANDBY_ID, XLOG_RESTORE_POINT_RUNNING_XACTS);
+
+	/* report error if subxid_overflow */
+	if (CurrRunningXacts->subxid_overflow)
+		elog(ERROR,
+			 "snapshot of %u running transactions overflowed (lsn %X/%X oldest xid %u latest complete %u next xid %u)",
+			 CurrRunningXacts->xcnt,
+			 LSN_FORMAT_ARGS(recptr),
+			 CurrRunningXacts->oldestRunningXid,
+			 CurrRunningXacts->latestCompletedXid,
+			 CurrRunningXacts->nextXid);
+	else
+		elog(trace_recovery(DEBUG2),
+			 "snapshot of %u+%u running transaction ids (lsn %X/%X oldest xid %u latest complete %u next xid %u)",
+			 CurrRunningXacts->xcnt, CurrRunningXacts->subxcnt,
+			 LSN_FORMAT_ARGS(recptr),
+			 CurrRunningXacts->oldestRunningXid,
+			 CurrRunningXacts->latestCompletedXid,
+			 CurrRunningXacts->nextXid);
+
+	/*
+	 * Ensure running_xacts information is synced to disk not too far in the
+	 * future. We don't want to stall anything though (i.e. use XLogFlush()),
+	 * so we let the wal writer do it during normal operation.
+	 * XLogSetAsyncXactLSN() conveniently will mark the LSN as to-be-synced
+	 * and nudge the WALWriter into action if sleeping. Check
+	 * XLogBackgroundFlush() for details why a record might not be flushed
+	 * without it.
+	 */
+	XLogSetAsyncXactLSN(recptr);
+
+	return recptr;
+}
 /*
  * Wholesale logging of AccessExclusiveLocks. Other lock types need not be
  * logged, as described in backend/storage/lmgr/README.

src/backend/utils/misc/guc_gp.c

@@ -47,6 +47,7 @@
 #include "postmaster/postmaster.h"
 #include "replication/walsender.h"
 #include "storage/proc.h"
+#include "storage/standby.h"
 #include "task/pg_cron.h"
 #include "tcop/idle_resource_cleaner.h"
 #include "utils/builtins.h"
@@ -5006,6 +5007,16 @@ struct config_string ConfigureNamesString_gp[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"gp_hot_standby_snapshot_restore_point_name", PGC_SUSET, DEVELOPER_OPTIONS,
+			gettext_noop("Specifies the hot standby snapshot's restore point name."),
+			gettext_noop(""),
+			GUC_NOT_IN_SAMPLE
+		},
+		&gp_hot_standby_snapshot_restore_point_name,
+		"",
+		NULL, NULL, NULL
+	},
 	/* End-of-list marker */
 	{
 		{NULL, 0, 0, NULL, NULL}, NULL, NULL, NULL, NULL

src/include/storage/standby.h

@@ -25,6 +25,7 @@ extern int	vacuum_defer_cleanup_age;
 extern int	max_standby_archive_delay;
 extern int	max_standby_streaming_delay;
 extern bool log_recovery_conflict_waits;
+extern char *gp_hot_standby_snapshot_restore_point_name;
 
 extern void InitRecoveryTransactionEnvironment(void);
 extern void ShutdownRecoveryTransactionEnvironment(void);
@@ -60,6 +61,7 @@ extern void StandbyReleaseAllLocks(void);
 extern void StandbyReleaseOldLocks(TransactionId oldxid);
 
 #define MinSizeOfXactRunningXacts offsetof(xl_running_xacts, xids)
+#define MinSizeOfXactRpRunningXacts offsetof(xl_restore_point_running_xacts, xids)
 
 
 /*
@@ -84,6 +86,7 @@ typedef struct RunningTransactionsData
 	TransactionId latestCompletedXid;	/* so we can set xmax */
 
 	TransactionId *xids;		/* array of (sub)xids still running */
+	char		*restore_point_name;
 } RunningTransactionsData;
 
 typedef RunningTransactionsData *RunningTransactions;
@@ -95,4 +98,6 @@ extern XLogRecPtr LogStandbySnapshot(void);
 extern void LogStandbyInvalidations(int nmsgs, SharedInvalidationMessage *msgs,
 									bool relcacheInitFileInval);
 
+extern XLogRecPtr LogRestorePointStandbySnapshot(const char *rpName);
+
 #endif							/* STANDBY_H */

src/include/storage/standbydefs.h

@@ -31,10 +31,11 @@ extern void standby_desc_invalidations(StringInfo buf,
 /*
  * XLOG message types
  */
-#define XLOG_STANDBY_LOCK			0x00
-#define XLOG_RUNNING_XACTS			0x10
-#define XLOG_INVALIDATIONS			0x20
-#define XLOG_LATESTCOMPLETED_GXID   0xF0
+#define XLOG_STANDBY_LOCK							0x00
+#define XLOG_RUNNING_XACTS							0x10
+#define XLOG_INVALIDATIONS							0x20
+#define XLOG_RESTORE_POINT_RUNNING_XACTS			0xE0
+#define XLOG_LATESTCOMPLETED_GXID					0xF0
 
 typedef struct xl_standby_locks
 {
@@ -57,6 +58,20 @@ typedef struct xl_running_xacts
 	TransactionId xids[FLEXIBLE_ARRAY_MEMBER];
 } xl_running_xacts;
 
+#define MAXFNAMELEN		64
+typedef struct xl_restore_point_running_xacts
+{
+	int			xcnt;			/* # of xact ids in xids[] */
+	int			subxcnt;		/* # of subxact ids in xids[] */
+	bool		subxid_overflow;	/* snapshot overflowed, subxids missing */
+	TransactionId nextXid;		/* xid from ShmemVariableCache->nextXid */
+	TransactionId oldestRunningXid; /* *not* oldestXmin */
+	TransactionId latestCompletedXid;	/* so we can set xmax */
+	char		rpName[MAXFNAMELEN]; /* cbdr: log the running xact with restore_point_name */
+
+	TransactionId xids[FLEXIBLE_ARRAY_MEMBER];
+} xl_restore_point_running_xacts;
+
 /*
  * Invalidations for standby, currently only when transactions without an
  * assigned xid commit.

src/include/utils/unsync_guc_name.h

@@ -239,6 +239,7 @@
 		"gp_max_slices",
 		"gp_motion_cost_per_row",
 		"gp_pause_on_restore_point_replay",
+		"gp_hot_standby_snapshot_restore_point_name",
 		"gp_predicate_pushdown_sample_rows",
 		"gp_print_create_gang_time",
 		"gp_qd_hostname",

src/test/isolation2/hot_standby_schedule

@@ -4,3 +4,4 @@ test: hot_standby/transaction_isolation
 test: hot_standby/query_conflict
 test: hot_standby/faults
 test: hot_standby/teardown
+test: hot_standby/snapshot

src/test/isolation2/sql/hot_standby/snapshot.sql

@@ -0,0 +1,20 @@
+----------------------------------------------------------------
+-- Test guc gp_hot_standby_snapshot_restore_point_name
+-- when the guc set, for example rp1, the hot_standby query only can see
+-- the data commited before rp1.
+----------------------------------------------------------------
+!\retcode gpconfig -c gp_hot_standby_snapshot_restore_point_name -v "rp1";
+!\retcode gpstop -ar;
+
+1: show gp_hot_standby_snapshot_restore_point_name;
+1: create table hs_sh(a int);
+1: insert into hs_sh select * from generate_series(1,10);
+
+-1S: select * from hs_sh;;
+
+1: select gp_create_restore_point('rp1');
+
+-1S: select * from hs_sh;
+
+
+