125 "array length mismatch");
128#define ReplicationSlotOnDiskConstantSize \
129 offsetof(ReplicationSlotOnDisk, slotdata)
131#define ReplicationSlotOnDiskNotChecksummedSize \
132 offsetof(ReplicationSlotOnDisk, version)
134#define ReplicationSlotOnDiskChecksummedSize \
135 sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskNotChecksummedSize
137#define ReplicationSlotOnDiskV2Size \
138 sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize
140#define SLOT_MAGIC 0x1051CA1
141#define SLOT_VERSION 5
271 if (strlen(
name) == 0)
274 (
errcode(ERRCODE_INVALID_NAME),
275 errmsg(
"replication slot name \"%s\" is too short",
283 (
errcode(ERRCODE_NAME_TOO_LONG),
284 errmsg(
"replication slot name \"%s\" is too long",
289 for (cp =
name; *cp; cp++)
291 if (!((*cp >=
'a' && *cp <=
'z')
292 || (*cp >=
'0' && *cp <=
'9')
296 (
errcode(ERRCODE_INVALID_NAME),
297 errmsg(
"replication slot name \"%s\" contains invalid character",
299 errhint(
"Replication slot names may only contain lower case letters, numbers, and the underscore character.")));
326 bool two_phase,
bool failover,
bool synced)
347 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
348 errmsg(
"cannot enable failover for a replication slot created on the standby"));
359 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
360 errmsg(
"cannot enable failover for a temporary replication slot"));
385 errmsg(
"replication slot \"%s\" already exists",
name)));
386 if (!s->
in_use && slot == NULL)
394 (
errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
395 errmsg(
"all replication slots are in use"),
396 errhint(
"Free one or increase \"max_replication_slots\".")));
573 if (s == NULL || !s->
in_use)
578 (
errcode(ERRCODE_UNDEFINED_OBJECT),
579 errmsg(
"replication slot \"%s\" does not exist",
627 WAIT_EVENT_REPLICATION_SLOT_DROP);
633 (
errcode(ERRCODE_OBJECT_IN_USE),
634 errmsg(
"replication slot \"%s\" is active for PID %d",
650 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
651 errmsg(
"can no longer access replication slot \"%s\"",
653 errdetail(
"This replication slot has been invalidated due to \"%s\".",
672 ?
errmsg(
"acquired logical replication slot \"%s\"",
674 :
errmsg(
"acquired physical replication slot \"%s\"",
689 char *slotname = NULL;
690 bool is_logical =
false;
759 ?
errmsg(
"released logical replication slot \"%s\"",
761 :
errmsg(
"released physical replication slot \"%s\"",
826 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
827 errmsg(
"cannot drop replication slot \"%s\"",
name),
828 errdetail(
"This replication slot is being synchronized from the primary server."));
840 bool update_slot =
false;
849 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
850 errmsg(
"cannot use %s with a physical replication slot",
851 "ALTER_REPLICATION_SLOT"));
861 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
862 errmsg(
"cannot alter replication slot \"%s\"",
name),
863 errdetail(
"This replication slot is being synchronized from the primary server."));
869 if (failover && *failover)
871 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
872 errmsg(
"cannot enable failover for a replication slot"
884 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
885 errmsg(
"cannot enable failover for a temporary replication slot"));
959 if (rename(path, tmppath) == 0)
987 errmsg(
"could not rename file \"%s\" to \"%s\": %m",
1018 if (!
rmtree(tmppath,
true))
1020 (
errmsg(
"could not remove directory \"%s\"", tmppath)));
1134 agg_xmin = effective_xmin;
1140 agg_catalog_xmin = effective_catalog_xmin;
1184 restart_lsn < min_required))
1185 min_required = restart_lsn;
1245 restart_lsn < result)
1246 result = restart_lsn;
1267 *nslots = *nactive = 0;
1357 if (active_pid == 0)
1387 (
errcode(ERRCODE_OBJECT_IN_USE),
1388 errmsg(
"replication slot \"%s\" is active for PID %d",
1389 slotname, active_pid)));
1422 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1423 errmsg(
"replication slots can only be used if \"max_replication_slots\" > 0")));
1427 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1428 errmsg(
"replication slots can only be used if \"wal_level\" >= \"replica\"")));
1439 (
errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1440 errmsg(
"permission denied to use replication slots"),
1441 errdetail(
"Only roles with the %s attribute may use replication slots.",
1535 long slot_idle_seconds)
1547 unsigned long long ex = oldestLSN - restart_lsn;
1550 ngettext(
"The slot's restart_lsn %X/%X exceeds the limit by %llu byte.",
1551 "The slot's restart_lsn %X/%X exceeds the limit by %llu bytes.",
1557 "max_slot_wal_keep_size");
1562 snapshotConflictHorizon);
1566 appendStringInfoString(&err_detail,
_(
"Logical decoding on standby requires \"wal_level\" >= \"logical\" on the primary server."));
1575 appendStringInfo(&err_detail,
_(
"The slot's idle time of %dmin %02ds exceeds the configured \"%s\" duration of %dmin."),
1576 minutes, secs,
"idle_replication_slot_timeout",
1580 "idle_replication_slot_timeout");
1589 errmsg(
"terminating process %d to release replication slot \"%s\"",
1591 errmsg(
"invalidating obsolete replication slot \"%s\"",
1642 initial_restart_lsn < oldestLSN)
1654 snapshotConflictHorizon))
1658 snapshotConflictHorizon))
1680#ifdef USE_INJECTION_POINTS
1683 *inactive_since = 0;
1724 int last_signaled_pid = 0;
1725 bool released_lock =
false;
1726 bool terminated =
false;
1740 long slot_idle_secs = 0;
1795 snapshotConflictHorizon,
1796 initial_effective_xmin,
1797 initial_catalog_effective_xmin,
1798 initial_restart_lsn,
1808 invalidation_cause_prev != invalidation_cause));
1827 if (active_pid == 0)
1841 *invalidated =
true;
1861 int slot_idle_usecs;
1867 if (active_pid != 0)
1877 released_lock =
true;
1891 if (last_signaled_pid != active_pid)
1894 slotname, restart_lsn,
1895 oldestLSN, snapshotConflictHorizon,
1903 (
void)
kill(active_pid, SIGTERM);
1905 last_signaled_pid = active_pid;
1907 invalidation_cause_prev = invalidation_cause;
1912 WAIT_EVENT_REPLICATION_SLOT_DROP);
1933 released_lock =
true;
1941 slotname, restart_lsn,
1942 oldestLSN, snapshotConflictHorizon,
1952 return released_lock;
1981 bool invalidated =
false;
2002 snapshotConflictHorizon,
2036 elog(
DEBUG1,
"performing replication slot checkpoint");
2091 DIR *replication_dir;
2092 struct dirent *replication_de;
2094 elog(
DEBUG1,
"starting up replication slots");
2103 if (strcmp(replication_de->
d_name,
".") == 0 ||
2104 strcmp(replication_de->
d_name,
"..") == 0)
2120 (
errmsg(
"could not remove directory \"%s\"",
2179 errmsg(
"could not create directory \"%s\": %m",
2188 if (rename(tmppath, path) != 0)
2191 errmsg(
"could not rename file \"%s\" to \"%s\": %m",
2221 was_dirty = slot->
dirty;
2234 sprintf(tmppath,
"%s/state.tmp", dir);
2235 sprintf(path,
"%s/state", dir);
2246 int save_errno = errno;
2252 errmsg(
"could not create file \"%s\": %m",
2275 if ((
write(
fd, &cp,
sizeof(cp))) !=
sizeof(cp))
2277 int save_errno = errno;
2284 errno = save_errno ? save_errno : ENOSPC;
2287 errmsg(
"could not write to file \"%s\": %m",
2297 int save_errno = errno;
2305 errmsg(
"could not fsync file \"%s\": %m",
2313 int save_errno = errno;
2319 errmsg(
"could not close file \"%s\": %m",
2325 if (rename(tmppath, path) != 0)
2327 int save_errno = errno;
2333 errmsg(
"could not rename file \"%s\" to \"%s\": %m",
2355 slot->
dirty =
false;
2373 bool restored =
false;
2382 sprintf(path,
"%s/state.tmp", slotdir);
2383 if (unlink(path) < 0 && errno != ENOENT)
2386 errmsg(
"could not remove file \"%s\": %m", path)));
2388 sprintf(path,
"%s/state", slotdir);
2390 elog(
DEBUG1,
"restoring replication slot from \"%s\"", path);
2402 errmsg(
"could not open file \"%s\": %m", path)));
2412 errmsg(
"could not fsync file \"%s\": %m",
2430 errmsg(
"could not read file \"%s\": %m", path)));
2434 errmsg(
"could not read file \"%s\": read %d of %zu",
2443 errmsg(
"replication slot file \"%s\" has wrong magic number: %u instead of %u",
2450 errmsg(
"replication slot file \"%s\" has unsupported version %u",
2457 errmsg(
"replication slot file \"%s\" has corrupted length %u",
2466 if (readBytes != cp.
length)
2471 errmsg(
"could not read file \"%s\": %m", path)));
2475 errmsg(
"could not read file \"%s\": read %d of %zu",
2482 errmsg(
"could not close file \"%s\": %m", path)));
2493 (
errmsg(
"checksum mismatch for replication slot file \"%s\": is %u, should be %u",
2502 if (!
rmtree(slotdir,
true))
2505 (
errmsg(
"could not remove directory \"%s\"",
2528 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2529 errmsg(
"logical replication slot \"%s\" exists, but \"wal_level\" < \"logical\"",
2531 errhint(
"Change \"wal_level\" to be \"logical\" or higher.")));
2542 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2543 errmsg(
"logical replication slot \"%s\" exists on the standby, but \"hot_standby\" = \"off\"",
2545 errhint(
"Change \"hot_standby\" to be \"on\".")));
2549 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2550 errmsg(
"physical replication slot \"%s\" exists, but \"wal_level\" < \"replica\"",
2552 errhint(
"Change \"wal_level\" to be \"replica\" or higher.")));
2598 (
errmsg(
"too many replication slots active before shutdown"),
2599 errhint(
"Increase \"max_replication_slots\" and try again.")));
2710 if ((*
newval)[0] ==
'\0')
2719 if (!ok || elemlist ==
NIL)
2729 size += strlen(slot_name) + 1;
2740 strcpy(ptr, slot_name);
2741 ptr += strlen(slot_name) + 1;
2772 const char *standby_slot_name;
2786 if (strcmp(standby_slot_name, slot_name) == 0)
2789 standby_slot_name += strlen(standby_slot_name) + 1;
2806 int caught_up_slot_num = 0;
2860 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2861 errmsg(
"replication slot \"%s\" specified in parameter \"%s\" does not exist",
2862 name,
"synchronized_standby_slots"),
2863 errdetail(
"Logical replication is waiting on the standby associated with replication slot \"%s\".",
2865 errhint(
"Create the replication slot \"%s\" or amend parameter \"%s\".",
2866 name,
"synchronized_standby_slots"));
2874 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2875 errmsg(
"cannot specify logical replication slot \"%s\" in parameter \"%s\"",
2876 name,
"synchronized_standby_slots"),
2877 errdetail(
"Logical replication is waiting for correction on replication slot \"%s\".",
2879 errhint(
"Remove the logical replication slot \"%s\" from parameter \"%s\".",
2880 name,
"synchronized_standby_slots"));
2894 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2895 errmsg(
"physical replication slot \"%s\" specified in parameter \"%s\" has been invalidated",
2896 name,
"synchronized_standby_slots"),
2897 errdetail(
"Logical replication is waiting on the standby associated with replication slot \"%s\".",
2899 errhint(
"Drop and recreate the replication slot \"%s\", or amend parameter \"%s\".",
2900 name,
"synchronized_standby_slots"));
2909 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2910 errmsg(
"replication slot \"%s\" specified in parameter \"%s\" does not have active_pid",
2911 name,
"synchronized_standby_slots"),
2912 errdetail(
"Logical replication is waiting on the standby associated with replication slot \"%s\".",
2914 errhint(
"Start the standby associated with the replication slot \"%s\", or amend parameter \"%s\".",
2915 name,
"synchronized_standby_slots"));
2921 Assert(restart_lsn >= wait_for_lsn);
2924 min_restart_lsn > restart_lsn)
2925 min_restart_lsn = restart_lsn;
2927 caught_up_slot_num++;
2989 WAIT_EVENT_WAIT_FOR_STANDBY_CONFIRMATION);
3007 "idle_replication_slot_timeout");
void TimestampDifference(TimestampTz start_time, TimestampTz stop_time, long *secs, int *microsecs)
bool TimestampDifferenceExceedsSeconds(TimestampTz start_time, TimestampTz stop_time, int threshold_sec)
TimestampTz GetCurrentTimestamp(void)
Datum now(PG_FUNCTION_ARGS)
#define ngettext(s, p, n)
#define PG_USED_FOR_ASSERTS_ONLY
#define FLEXIBLE_ARRAY_MEMBER
#define MemSet(start, val, len)
bool ConditionVariableCancelSleep(void)
bool ConditionVariableTimedSleep(ConditionVariable *cv, long timeout, uint32 wait_event_info)
void ConditionVariableBroadcast(ConditionVariable *cv)
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
void ConditionVariableInit(ConditionVariable *cv)
void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)
int errdetail_internal(const char *fmt,...)
int errcode_for_file_access(void)
int errdetail(const char *fmt,...)
int errhint(const char *fmt,...)
int errcode(int sqlerrcode)
int errmsg(const char *fmt,...)
#define ereport(elevel,...)
int MakePGDirectory(const char *directoryName)
int CloseTransientFile(int fd)
void fsync_fname(const char *fname, bool isdir)
DIR * AllocateDir(const char *dirname)
struct dirent * ReadDir(DIR *dir, const char *dirname)
int OpenTransientFile(const char *fileName, int fileFlags)
PGFileType get_dirent_type(const char *path, const struct dirent *de, bool look_through_symlinks, int elevel)
void ProcessConfigFile(GucContext context)
void * guc_malloc(int elevel, size_t size)
#define GUC_check_errdetail
Assert(PointerIsAligned(start, uint64))
#define IS_INJECTION_POINT_ATTACHED(name)
volatile sig_atomic_t ConfigReloadPending
void before_shmem_exit(pg_on_exit_callback function, Datum arg)
void list_free(List *list)
bool LWLockHeldByMe(LWLock *lock)
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
bool LWLockHeldByMeInMode(LWLock *lock, LWLockMode mode)
void LWLockRelease(LWLock *lock)
void LWLockInitialize(LWLock *lock, int tranche_id)
@ LWTRANCHE_REPLICATION_SLOT_IO
char * pstrdup(const char *in)
void pfree(void *pointer)
#define START_CRIT_SECTION()
#define CHECK_FOR_INTERRUPTS()
#define END_CRIT_SECTION()
BackendType MyBackendType
bool has_rolreplication(Oid roleid)
void namestrcpy(Name name, const char *str)
#define ERRCODE_DATA_CORRUPTED
#define COMP_CRC32C(crc, data, len)
#define EQ_CRC32C(c1, c2)
static int list_length(const List *l)
#define foreach_ptr(type, var, lst)
static rewind_source * source
void pgstat_create_replslot(ReplicationSlot *slot)
void pgstat_acquire_replslot(ReplicationSlot *slot)
void pgstat_drop_replslot(ReplicationSlot *slot)
static int fd(const char *x, int i)
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
#define INVALID_PROC_NUMBER
int SendProcSignal(pid_t pid, ProcSignalReason reason, ProcNumber procNumber)
@ PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT
bool rmtree(const char *path, bool rmtopdir)
Size add_size(Size s1, Size s2)
Size mul_size(Size s1, Size s2)
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
int ReplicationSlotIndex(ReplicationSlot *slot)
void ReplicationSlotAcquire(const char *name, bool nowait, bool error_if_invalid)
static const SlotInvalidationCauseMap SlotInvalidationCauses[]
char * synchronized_standby_slots
void assign_synchronized_standby_slots(const char *newval, void *extra)
#define ReplicationSlotOnDiskChecksummedSize
void CheckPointReplicationSlots(bool is_shutdown)
void ReplicationSlotCreate(const char *name, bool db_specific, ReplicationSlotPersistency persistency, bool two_phase, bool failover, bool synced)
void ReplicationSlotDropAcquired(void)
void ReplicationSlotMarkDirty(void)
void ReplicationSlotReserveWal(void)
bool ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive)
struct SlotInvalidationCauseMap SlotInvalidationCauseMap
static XLogRecPtr ss_oldest_flush_lsn
static ReplicationSlotInvalidationCause DetermineSlotInvalidationCause(uint32 possible_causes, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, TransactionId initial_effective_xmin, TransactionId initial_catalog_effective_xmin, XLogRecPtr initial_restart_lsn, TimestampTz *inactive_since, TimestampTz now)
void ReplicationSlotsDropDBSlots(Oid dboid)
#define ReplicationSlotOnDiskNotChecksummedSize
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void)
ReplicationSlotInvalidationCause GetSlotInvalidationCause(const char *cause_name)
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
static void RestoreSlotFromDisk(const char *name)
int idle_replication_slot_timeout_mins
void ReplicationSlotPersist(void)
static void ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, bool terminating, int pid, NameData slotname, XLogRecPtr restart_lsn, XLogRecPtr oldestLSN, TransactionId snapshotConflictHorizon, long slot_idle_seconds)
ReplicationSlot * MyReplicationSlot
static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
void ReplicationSlotDrop(const char *name, bool nowait)
bool SlotExistsInSyncStandbySlots(const char *slot_name)
static bool validate_sync_standby_slots(char *rawname, List **elemlist)
void ReplicationSlotSave(void)
ReplicationSlot * SearchNamedReplicationSlot(const char *name, bool need_lock)
static void CreateSlotOnDisk(ReplicationSlot *slot)
#define ReplicationSlotOnDiskV2Size
void CheckSlotPermissions(void)
bool ReplicationSlotName(int index, Name name)
bool check_synchronized_standby_slots(char **newval, void **extra, GucSource source)
void ReplicationSlotsShmemInit(void)
void ReplicationSlotAlter(const char *name, const bool *failover, const bool *two_phase)
void ReplicationSlotRelease(void)
int max_replication_slots
StaticAssertDecl(lengthof(SlotInvalidationCauses)==(RS_INVAL_MAX_CAUSES+1), "array length mismatch")
ReplicationSlotCtlData * ReplicationSlotCtl
struct ReplicationSlotOnDisk ReplicationSlotOnDisk
void WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
bool StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
void ReplicationSlotsComputeRequiredLSN(void)
void ReplicationSlotCleanup(bool synced_only)
void ReplicationSlotInitialize(void)
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
void StartupReplicationSlots(void)
static bool InvalidatePossiblyObsoleteSlot(uint32 possible_causes, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, bool *invalidated)
static bool CanInvalidateIdleSlot(ReplicationSlot *s)
void CheckSlotRequirements(void)
bool InvalidateObsoleteReplicationSlots(uint32 possible_causes, XLogSegNo oldestSegno, Oid dboid, TransactionId snapshotConflictHorizon)
static SyncStandbySlotsConfigData * synchronized_standby_slots_config
#define ReplicationSlotOnDiskConstantSize
Size ReplicationSlotsShmemSize(void)
const char * GetSlotInvalidationCauseName(ReplicationSlotInvalidationCause cause)
bool ReplicationSlotValidateName(const char *name, int elevel)
static void ReplicationSlotShmemExit(int code, Datum arg)
bool check_idle_replication_slot_timeout(int *newval, void **extra, GucSource source)
#define RS_INVAL_MAX_CAUSES
ReplicationSlotPersistency
#define SlotIsPhysical(slot)
ReplicationSlotInvalidationCause
#define SlotIsLogical(slot)
static void ReplicationSlotSetInactiveSince(ReplicationSlot *s, TimestampTz ts, bool acquire_lock)
bool IsSyncingReplicationSlots(void)
#define SpinLockInit(lock)
#define SpinLockRelease(lock)
#define SpinLockAcquire(lock)
XLogRecPtr LogStandbySnapshot(void)
#define ERRCODE_DUPLICATE_OBJECT
bool pg_str_endswith(const char *str, const char *end)
void appendStringInfo(StringInfo str, const char *fmt,...)
void appendStringInfoString(StringInfo str, const char *s)
void initStringInfo(StringInfo str)
ReplicationSlot replication_slots[1]
ReplicationSlotPersistentData slotdata
TransactionId catalog_xmin
XLogRecPtr confirmed_flush
ReplicationSlotPersistency persistency
ReplicationSlotInvalidationCause invalidated
XLogRecPtr candidate_xmin_lsn
TransactionId effective_catalog_xmin
XLogRecPtr candidate_restart_valid
XLogRecPtr last_saved_confirmed_flush
TransactionId effective_xmin
XLogRecPtr candidate_restart_lsn
LWLock io_in_progress_lock
ConditionVariable active_cv
TransactionId candidate_catalog_xmin
ReplicationSlotPersistentData data
TimestampTz inactive_since
ReplicationSlotInvalidationCause cause
char slot_names[FLEXIBLE_ARRAY_MEMBER]
ConditionVariable wal_confirm_rcv_cv
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
#define InvalidTransactionId
#define TransactionIdIsValid(xid)
bool SplitIdentifierString(char *rawstring, char separator, List **namelist)
static void pgstat_report_wait_start(uint32 wait_event_info)
static void pgstat_report_wait_end(void)
bool log_replication_commands
WalSndCtlData * WalSndCtl
bool RecoveryInProgress(void)
XLogSegNo XLogGetLastRemovedSegno(void)
XLogRecPtr GetRedoRecPtr(void)
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
XLogRecPtr GetXLogInsertRecPtr(void)
void XLogFlush(XLogRecPtr record)
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)
#define LSN_FORMAT_ARGS(lsn)
#define XLogRecPtrIsInvalid(r)
#define InvalidXLogRecPtr
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)