Add GetReplayXlogPtrHook allowing extensions to adjust replay LSN

Useful when extension performs some async operations in the WAL record
redo function.

Author: akorotkov

src/backend/access/transam/xlogfuncs.c

@@ -363,7 +363,7 @@ pg_last_wal_replay_lsn(PG_FUNCTION_ARGS)
 {
 	XLogRecPtr	recptr;
 
-	recptr = GetXLogReplayRecPtr(NULL);
+	recptr = GetEffectiveXlogReplayRecPtr();
 
 	if (recptr == 0)
 		PG_RETURN_NULL();

src/backend/access/transam/xlogrecovery.c

@@ -4504,6 +4504,28 @@ GetXLogReplayRecPtr(TimeLineID *replayTLI)
 	return recptr;
 }
 
+GetReplayXlogPtrHookType GetReplayXlogPtrHook = NULL;
+
+/*
+ * Get effective latest redo apply position.
+ *
+ * Can be tuned by extensions processing WAL records asyncronously.
+ */
+XLogRecPtr
+GetEffectiveXlogReplayRecPtr(void)
+{
+	XLogRecPtr	recptr = InvalidXLogRecPtr;
+
+	SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+	if (GetReplayXlogPtrHook)
+		recptr = GetReplayXlogPtrHook();
+	if (recptr == InvalidXLogRecPtr)
+		recptr = XLogRecoveryCtl->lastReplayedEndRecPtr;
+	SpinLockRelease(&XLogRecoveryCtl->info_lck);
+
+	return recptr;
+}
+
 
 /*
  * Get position of last applied, or the record being applied.

src/backend/replication/walreceiver.c

@@ -1130,7 +1130,7 @@ XLogWalRcvSendReply(bool force, bool requestReply)
 	/* Construct a new message */
 	writePtr = LogstreamResult.Write;
 	flushPtr = LogstreamResult.Flush;
-	applyPtr = GetXLogReplayRecPtr(NULL);
+	applyPtr = GetEffectiveXlogReplayRecPtr();
 
 	resetStringInfo(&reply_message);
 	pq_sendbyte(&reply_message, 'r');

src/backend/replication/walreceiverfuncs.c

@@ -373,7 +373,7 @@ GetReplicationApplyDelay(void)
 	receivePtr = walrcv->flushedUpto;
 	SpinLockRelease(&walrcv->mutex);
 
-	replayPtr = GetXLogReplayRecPtr(NULL);
+	replayPtr = GetEffectiveXlogReplayRecPtr();
 
 	if (receivePtr == replayPtr)
 		return 0;

src/include/access/xlogrecovery.h

@@ -48,6 +48,8 @@ typedef enum RecoveryPauseState
 	RECOVERY_PAUSED				/* recovery is paused */
 } RecoveryPauseState;
 
+typedef XLogRecPtr (*GetReplayXlogPtrHookType) (void);
+
 /* User-settable GUC parameters */
 extern PGDLLIMPORT bool recoveryTargetInclusive;
 extern PGDLLIMPORT int recoveryTargetAction;
@@ -76,6 +78,9 @@ extern PGDLLIMPORT bool reachedConsistency;
 /* Are we currently in standby mode? */
 extern PGDLLIMPORT bool StandbyMode;
 
+/* Hook for extensions to tune replay xlog pointer */
+extern PGDLLIMPORT GetReplayXlogPtrHookType GetReplayXlogPtrHook;
+
 extern Size XLogRecoveryShmemSize(void);
 extern void XLogRecoveryShmemInit(void);
 
@@ -137,6 +142,7 @@ extern void RemovePromoteSignalFiles(void);
 
 extern bool HotStandbyActive(void);
 extern XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI);
+extern XLogRecPtr GetEffectiveXlogReplayRecPtr(void);
 extern RecoveryPauseState GetRecoveryPauseState(void);
 extern void SetRecoveryPause(bool recoveryPause);
 extern void GetXLogReceiptTime(TimestampTz *rtime, bool *fromStream);