Rework handling of running transactions by extensions.

 * RunningTransactionsExtension structure as a part of xl_running_xacts and
   RunningTransactionsData to be filled by extensions.
 * getRunningTransactionsExtension hook to fill RunningTransactionsExtension.
 * Add nextXid field to CSNSnapshotData.
 * SnapBuildGetCSNSnaphot() function to modify CSNSnapshotData in the
   SnapBuild.
 * waitSnapshotHook hook to wait for transactions inside the snapshot
   builder.

Author: akorotkov

src/backend/replication/logical/snapbuild.c

@@ -1280,8 +1280,9 @@ SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xact
 	ReorderBufferTXN *txn;
 	TransactionId xmin;
 
-	builder->csnSnapshotData.snapshotcsn = running->csn;
-	builder->csnSnapshotData.xmin = 0;
+	builder->csnSnapshotData.snapshotcsn = running->extension.csn;
+	builder->csnSnapshotData.xmin = running->extension.runXmin;
+	builder->csnSnapshotData.nextXid = running->extension.nextXid;
 	builder->csnSnapshotData.xlogptr = lsn;
 
 	/*
@@ -1311,6 +1312,10 @@ SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xact
 	 * we hit fast paths in heapam_visibility.c.
 	 */
 	builder->xmin = running->oldestRunningXid;
+	builder->csnSnapshotData.snapshotcsn = running->extension.csn;
+	builder->csnSnapshotData.xmin = running->extension.runXmin;
+	builder->csnSnapshotData.nextXid = running->extension.nextXid;
+	builder->csnSnapshotData.xlogptr = lsn;
 
 	/* Remove transactions we don't need to keep track off anymore */
 	SnapBuildPurgeOlderTxn(builder);
@@ -1439,7 +1444,8 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
 	 * NB: We might have already started to incrementally assemble a snapshot,
 	 * so we need to be careful to deal with that.
 	 */
-	if (running->oldestRunningXid == running->nextXid)
+	if (running->oldestRunningXid == running->nextXid &&
+		running->extension.runXmin == running->extension.nextXid)
 	{
 		if (builder->start_decoding_at == InvalidXLogRecPtr ||
 			builder->start_decoding_at <= lsn)
@@ -1568,6 +1574,11 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
 	return true;
 }
 
+/*
+ * Hook for custom waits in SnapBuildWaitSnapshot() provided by extensions.
+ */
+WaitSnapshotHookType waitSnapshotHook = NULL;
+
 /* ---
  * Iterate through xids in record, wait for all older than the cutoff to
  * finish.  Then, if possible, log a new xl_running_xacts record.
@@ -1602,6 +1613,12 @@ SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff)
 		XactLockTableWait(xid, NULL, NULL, XLTW_None);
 	}
 
+	/*
+	 * Give extensions chance for their custom waits.
+	 */
+	if (waitSnapshotHook)
+		waitSnapshotHook(&running->extension);
+
 	/*
 	 * All transactions we needed to finish finished - try to ensure there is
 	 * another xl_running_xacts record in a timely manner, without having to
@@ -2209,9 +2226,8 @@ CheckPointSnapBuild(void)
 	FreeDir(snap_dir);
 }
 
-void
-SnapBuildUpdateCSNSnaphot(SnapBuild *builder,
-						  CSNSnapshotData *csnSnapshotData)
+CSNSnapshotData *
+SnapBuildGetCSNSnaphot(SnapBuild *builder)
 {
-	builder->csnSnapshotData = *csnSnapshotData;
+	return &builder->csnSnapshotData;
 }

src/backend/storage/ipc/procarray.c

@@ -2699,6 +2699,11 @@ ProcArrayInstallRestoredXmin(TransactionId xmin, PGPROC *proc)
 	return result;
 }
 
+/*
+ * A hook for filling RunningTransactionsExtension structure by extensions.
+ */
+GetRunningTransactionsExtensionHookType getRunningTransactionsExtension = NULL;
+
 /*
  * GetRunningTransactionData -- returns information about running transactions.
  *
@@ -2880,7 +2885,20 @@ GetRunningTransactionData(void)
 	CurrentRunningXacts->nextXid = XidFromFullTransactionId(ShmemVariableCache->nextXid);
 	CurrentRunningXacts->oldestRunningXid = oldestRunningXid;
 	CurrentRunningXacts->latestCompletedXid = latestCompletedXid;
-	CurrentRunningXacts->csn = pg_atomic_read_u64(&ShmemVariableCache->nextCommitSeqNo);
+
+	/*
+	 * Give extensions chance to fill their structs.
+	 */
+	if (getRunningTransactionsExtension)
+	{
+		getRunningTransactionsExtension(&CurrentRunningXacts->extension);
+	}
+	else
+	{
+		CurrentRunningXacts->extension.csn = 0;
+		CurrentRunningXacts->extension.nextXid = 0;
+		CurrentRunningXacts->extension.runXmin = 0;
+	}
 
 	Assert(TransactionIdIsValid(CurrentRunningXacts->nextXid));
 	Assert(TransactionIdIsValid(CurrentRunningXacts->oldestRunningXid));

src/backend/storage/ipc/standby.c

@@ -1355,7 +1355,7 @@ LogCurrentRunningXacts(RunningTransactions CurrRunningXacts)
 	xlrec.nextXid = CurrRunningXacts->nextXid;
 	xlrec.oldestRunningXid = CurrRunningXacts->oldestRunningXid;
 	xlrec.latestCompletedXid = CurrRunningXacts->latestCompletedXid;
-	xlrec.csn = CurrRunningXacts->csn;
+	xlrec.extension = CurrRunningXacts->extension;
 
 	/* Header */
 	XLogBeginInsert();

src/backend/utils/time/snapmgr.c

@@ -2194,9 +2194,7 @@ SerializeSnapshot(Snapshot snapshot, char *start_address)
 	serialized_snapshot.curcid = snapshot->curcid;
 	serialized_snapshot.whenTaken = snapshot->whenTaken;
 	serialized_snapshot.lsn = snapshot->lsn;
-	serialized_snapshot.csnSnapshotData.xmin = snapshot->csnSnapshotData.xmin;
-	serialized_snapshot.csnSnapshotData.snapshotcsn = snapshot->csnSnapshotData.snapshotcsn;
-	serialized_snapshot.csnSnapshotData.xlogptr = snapshot->csnSnapshotData.xlogptr;
+	serialized_snapshot.csnSnapshotData = snapshot->csnSnapshotData;
 	serialized_snapshot.undoRegularRowLocation = snapshot->undoRegularRowLocationPhNode.undoLocation;
 	serialized_snapshot.undoRegularPageLocation = snapshot->undoRegularPageLocationPhNode.undoLocation;
 	serialized_snapshot.undoSystemLocation = snapshot->undoSystemLocationPhNode.undoLocation;
@@ -2275,9 +2273,7 @@ RestoreSnapshot(char *start_address)
 	snapshot->whenTaken = serialized_snapshot.whenTaken;
 	snapshot->lsn = serialized_snapshot.lsn;
 	snapshot->snapXactCompletionCount = 0;
-	snapshot->csnSnapshotData.xmin = serialized_snapshot.csnSnapshotData.xmin;
-	snapshot->csnSnapshotData.snapshotcsn = serialized_snapshot.csnSnapshotData.snapshotcsn;
-	snapshot->csnSnapshotData.xlogptr = serialized_snapshot.csnSnapshotData.xlogptr;
+	snapshot->csnSnapshotData = serialized_snapshot.csnSnapshotData;
 	snapshot->undoRegularRowLocationPhNode.undoLocation = serialized_snapshot.undoRegularRowLocation;
 	snapshot->undoRegularPageLocationPhNode.undoLocation = serialized_snapshot.undoRegularPageLocation;
 	snapshot->undoSystemLocationPhNode.undoLocation = serialized_snapshot.undoSystemLocation;

src/include/replication/snapbuild.h

@@ -13,6 +13,7 @@
 #define SNAPBUILD_H
 
 #include "access/xlogdefs.h"
+#include "storage/standbydefs.h"
 #include "utils/snapmgr.h"
 
 typedef enum
@@ -57,6 +58,10 @@ struct ReorderBuffer;
 struct xl_heap_new_cid;
 struct xl_running_xacts;
 
+typedef void (*WaitSnapshotHookType) (RunningTransactionsExtension *extension);
+
+extern PGDLLIMPORT WaitSnapshotHookType waitSnapshotHook;
+
 extern void CheckPointSnapBuild(void);
 
 extern SnapBuild *AllocateSnapshotBuilder(struct ReorderBuffer *reorder,
@@ -91,7 +96,6 @@ extern void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid,
 extern void SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn,
 										 struct xl_running_xacts *running);
 extern void SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn);
-extern void SnapBuildUpdateCSNSnaphot(SnapBuild *builder,
-									  CSNSnapshotData *csnSnapshotData);
+extern CSNSnapshotData *SnapBuildGetCSNSnaphot(SnapBuild *builder);
 
 #endif							/* SNAPBUILD_H */

src/include/storage/standby.h

@@ -91,13 +91,17 @@ typedef struct RunningTransactionsData
 	TransactionId nextXid;		/* xid from ShmemVariableCache->nextXid */
 	TransactionId oldestRunningXid; /* *not* oldestXmin */
 	TransactionId latestCompletedXid;	/* so we can set xmax */
-	CommitSeqNo csn;	/* current csn */
+	RunningTransactionsExtension extension;
 
 	TransactionId *xids;		/* array of (sub)xids still running */
 } RunningTransactionsData;
 
 typedef RunningTransactionsData *RunningTransactions;
 
+typedef void (*GetRunningTransactionsExtensionHookType) (RunningTransactionsExtension *extension);
+
+extern PGDLLIMPORT GetRunningTransactionsExtensionHookType getRunningTransactionsExtension;
+
 extern void LogAccessExclusiveLock(Oid dbOid, Oid relOid);
 extern void LogAccessExclusiveLockPrepare(void);
 

src/include/storage/standbydefs.h

@@ -41,6 +41,17 @@ typedef struct xl_standby_locks
 	xl_standby_lock locks[FLEXIBLE_ARRAY_MEMBER];
 } xl_standby_locks;
 
+/*
+ * A part of xl_running_xacts and RunningTransactionsData to be filled by
+ * extensions.
+ */
+typedef struct
+{
+	uint64		nextXid;
+	uint64		runXmin;
+	CommitSeqNo csn;	/* current csn */
+} RunningTransactionsExtension;
+
 /*
  * When we write running xact data to WAL, we use this structure.
  */
@@ -52,7 +63,7 @@ typedef struct xl_running_xacts
 	TransactionId nextXid;		/* xid from ShmemVariableCache->nextXid */
 	TransactionId oldestRunningXid; /* *not* oldestXmin */
 	TransactionId latestCompletedXid;	/* so we can set xmax */
-	CommitSeqNo csn;	/* current csn */
+	RunningTransactionsExtension extension;
 
 	TransactionId xids[FLEXIBLE_ARRAY_MEMBER];
 } xl_running_xacts;

src/include/utils/snapshot.h

@@ -131,6 +131,7 @@ typedef struct
 typedef struct CSNSnapshotData
 {
 	uint64			xmin;
+	uint64			nextXid;
 	CommitSeqNo		snapshotcsn;
 	XLogRecPtr		xlogptr;
 } CSNSnapshotData;