xlog.c: Adjust some more functions to pass the TLI around.

Adjust ReadRecord() to take the replayTLI as an argument. Use
that value, rather than ThisTimeLineID, to set the minimum
recovery point TLI. Also pass the value through to ReadRecord()
via XLogPageReadPrivate, and from there pass it down to
WaitForWALToBecomeAvailable() and rescanLatestTimeLine(),
so that the latter doesn't need the global variable
ThisTimeLineID either.

To make all this work, a few other changes are needed. First,
pass a TLI down to ReadCheckpointRecord so that it can pass it
on to ReadRecord. Second, have read_backup_label() return the
timeline it obtains from the backup_label file and install
that temporarily as ThisTimeLineID. The reason for the latter
change is that, previously, ThisTimeLineID could be
uninitialized when ReadRecord() was called, as when we were
reading the checkpoint record. Though that seems to have no
consequences about which we need to be concerned, it's better
to have a legal value all the time.

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index b6f30382ebafc474f56ea81d01e932b703cdb85c..8a650b875ea7d8505dc11ccb615fcb15e8546f45 100644 (file)
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -845,6 +845,7 @@ typedef struct XLogPageReadPrivate
        int                     emode;
        bool            fetching_ckpt;  /* are we fetching a checkpoint record? */
        bool            randAccess;
+       TimeLineID      replayTLI;
 } XLogPageReadPrivate;
 
 /*
@@ -931,7 +932,8 @@ static int  XLogFileReadAnyTLI(XLogSegNo segno, int emode, XLogSource source);
 static int     XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
                                                 int reqLen, XLogRecPtr targetRecPtr, char *readBuf);
 static bool WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
-                                                                               bool fetching_ckpt, XLogRecPtr tliRecPtr);
+                                                                               bool fetching_ckpt, XLogRecPtr tliRecPtr,
+                                                                               TimeLineID replayTLI);
 static void XLogShutdownWalRcv(void);
 static int     emode_for_corrupt_record(int emode, XLogRecPtr RecPtr);
 static void XLogFileClose(void);
@@ -946,12 +948,14 @@ static void ValidateXLOGDirectoryStructure(void);
 static void CleanupBackupHistory(void);
 static void UpdateMinRecoveryPoint(XLogRecPtr lsn, bool force);
 static XLogRecord *ReadRecord(XLogReaderState *xlogreader,
-                                                         int emode, bool fetching_ckpt);
+                                                         int emode, bool fetching_ckpt,
+                                                         TimeLineID replayTLI);
 static void CheckRecoveryConsistency(void);
 static bool PerformRecoveryXLogAction(void);
 static XLogRecord *ReadCheckpointRecord(XLogReaderState *xlogreader,
-                                                                               XLogRecPtr RecPtr, int whichChkpt, bool report);
-static bool rescanLatestTimeLine(void);
+                                                                               XLogRecPtr RecPtr, int whichChkpt, bool report,
+                                                                               TimeLineID replayTLI);
+static bool rescanLatestTimeLine(TimeLineID replayTLI);
 static void InitControlFile(uint64 sysidentifier);
 static void WriteControlFile(void);
 static void ReadControlFile(void);
@@ -967,6 +971,7 @@ static void xlog_outdesc(StringInfo buf, XLogReaderState *record);
 static void pg_start_backup_callback(int code, Datum arg);
 static void pg_stop_backup_callback(int code, Datum arg);
 static bool read_backup_label(XLogRecPtr *checkPointLoc,
+                                                         TimeLineID *backupLabelTLI,
                                                          bool *backupEndRequired, bool *backupFromStandby);
 static bool read_tablespace_map(List **tablespaces);
 
@@ -4447,7 +4452,7 @@ CleanupBackupHistory(void)
  */
 static XLogRecord *
 ReadRecord(XLogReaderState *xlogreader, int emode,
-                  bool fetching_ckpt)
+                  bool fetching_ckpt, TimeLineID replayTLI)
 {
        XLogRecord *record;
        XLogPageReadPrivate *private = (XLogPageReadPrivate *) xlogreader->private_data;
@@ -4456,6 +4461,7 @@ ReadRecord(XLogReaderState *xlogreader, int emode,
        private->fetching_ckpt = fetching_ckpt;
        private->emode = emode;
        private->randAccess = (xlogreader->ReadRecPtr == InvalidXLogRecPtr);
+       private->replayTLI = replayTLI;
 
        /* This is the first attempt to read this page. */
        lastSourceFailed = false;
@@ -4558,7 +4564,7 @@ ReadRecord(XLogReaderState *xlogreader, int emode,
                                if (ControlFile->minRecoveryPoint < EndRecPtr)
                                {
                                        ControlFile->minRecoveryPoint = EndRecPtr;
-                                       ControlFile->minRecoveryPointTLI = ThisTimeLineID;
+                                       ControlFile->minRecoveryPointTLI = replayTLI;
                                }
                                /* update local copy */
                                minRecoveryPoint = ControlFile->minRecoveryPoint;
@@ -4612,7 +4618,7 @@ ReadRecord(XLogReaderState *xlogreader, int emode,
  * one and returns 'true'.
  */
 static bool
-rescanLatestTimeLine(void)
+rescanLatestTimeLine(TimeLineID replayTLI)
 {
        List       *newExpectedTLEs;
        bool            found;
@@ -4654,7 +4660,7 @@ rescanLatestTimeLine(void)
                ereport(LOG,
                                (errmsg("new timeline %u is not a child of database system timeline %u",
                                                newtarget,
-                                               ThisTimeLineID)));
+                                               replayTLI)));
                return false;
        }
 
@@ -4668,7 +4674,7 @@ rescanLatestTimeLine(void)
                ereport(LOG,
                                (errmsg("new timeline %u forked off current database system timeline %u before current recovery point %X/%X",
                                                newtarget,
-                                               ThisTimeLineID,
+                                               replayTLI,
                                                LSN_FORMAT_ARGS(EndRecPtr))));
                return false;
        }
@@ -6870,7 +6876,7 @@ StartupXLOG(void)
        replay_image_masked = (char *) palloc(BLCKSZ);
        primary_image_masked = (char *) palloc(BLCKSZ);
 
-       if (read_backup_label(&checkPointLoc, &backupEndRequired,
+       if (read_backup_label(&checkPointLoc, &ThisTimeLineID, &backupEndRequired,
                                                  &backupFromStandby))
        {
                List       *tablespaces = NIL;
@@ -6888,7 +6894,8 @@ StartupXLOG(void)
                 * When a backup_label file is present, we want to roll forward from
                 * the checkpoint it identifies, rather than using pg_control.
                 */
-               record = ReadCheckpointRecord(xlogreader, checkPointLoc, 0, true);
+               record = ReadCheckpointRecord(xlogreader, checkPointLoc, 0, true,
+                                                                         ThisTimeLineID);
                if (record != NULL)
                {
                        memcpy(&checkPoint, XLogRecGetData(xlogreader), sizeof(CheckPoint));
@@ -6907,7 +6914,8 @@ StartupXLOG(void)
                        if (checkPoint.redo < checkPointLoc)
                        {
                                XLogBeginRead(xlogreader, checkPoint.redo);
-                               if (!ReadRecord(xlogreader, LOG, false))
+                               if (!ReadRecord(xlogreader, LOG, false,
+                                                               checkPoint.ThisTimeLineID))
                                        ereport(FATAL,
                                                        (errmsg("could not find redo location referenced by checkpoint record"),
                                                         errhint("If you are restoring from a backup, touch \"%s/recovery.signal\" and add required recovery options.\n"
@@ -7023,7 +7031,9 @@ StartupXLOG(void)
                /* Get the last valid checkpoint record. */
                checkPointLoc = ControlFile->checkPoint;
                RedoStartLSN = ControlFile->checkPointCopy.redo;
-               record = ReadCheckpointRecord(xlogreader, checkPointLoc, 1, true);
+               ThisTimeLineID = ControlFile->checkPointCopy.ThisTimeLineID;
+               record = ReadCheckpointRecord(xlogreader, checkPointLoc, 1, true,
+                                                                         ThisTimeLineID);
                if (record != NULL)
                {
                        ereport(DEBUG1,
@@ -7523,12 +7533,12 @@ StartupXLOG(void)
                {
                        /* back up to find the record */
                        XLogBeginRead(xlogreader, checkPoint.redo);
-                       record = ReadRecord(xlogreader, PANIC, false);
+                       record = ReadRecord(xlogreader, PANIC, false, ThisTimeLineID);
                }
                else
                {
                        /* just have to read next record after CheckPoint */
-                       record = ReadRecord(xlogreader, LOG, false);
+                       record = ReadRecord(xlogreader, LOG, false, ThisTimeLineID);
                }
 
                if (record != NULL)
@@ -7765,7 +7775,7 @@ StartupXLOG(void)
                                }
 
                                /* Else, try to fetch the next WAL record */
-                               record = ReadRecord(xlogreader, LOG, false);
+                               record = ReadRecord(xlogreader, LOG, false, ThisTimeLineID);
                        } while (record != NULL);
 
                        /*
@@ -7889,7 +7899,7 @@ StartupXLOG(void)
         * what we consider the valid portion of WAL.
         */
        XLogBeginRead(xlogreader, LastRec);
-       record = ReadRecord(xlogreader, PANIC, false);
+       record = ReadRecord(xlogreader, PANIC, false, ThisTimeLineID);
        EndOfLog = EndRecPtr;
 
        /*
@@ -8553,7 +8563,7 @@ LocalSetXLogInsertAllowed(void)
  */
 static XLogRecord *
 ReadCheckpointRecord(XLogReaderState *xlogreader, XLogRecPtr RecPtr,
-                                        int whichChkpt, bool report)
+                                        int whichChkpt, bool report, TimeLineID replayTLI)
 {
        XLogRecord *record;
        uint8           info;
@@ -8578,7 +8588,7 @@ ReadCheckpointRecord(XLogReaderState *xlogreader, XLogRecPtr RecPtr,
        }
 
        XLogBeginRead(xlogreader, RecPtr);
-       record = ReadRecord(xlogreader, LOG, true);
+       record = ReadRecord(xlogreader, LOG, true, replayTLI);
 
        if (record == NULL)
        {
@@ -12077,14 +12087,17 @@ GetOldestRestartPoint(XLogRecPtr *oldrecptr, TimeLineID *oldtli)
  * point, we will fail to restore a consistent database state.
  *
  * Returns true if a backup_label was found (and fills the checkpoint
- * location and its REDO location into *checkPointLoc and RedoStartLSN,
- * respectively); returns false if not. If this backup_label came from a
- * streamed backup, *backupEndRequired is set to true. If this backup_label
- * was created during recovery, *backupFromStandby is set to true.
+ * location and TLI into *checkPointLoc and *backupLabelTLI, respectively);
+ * returns false if not. If this backup_label came from a streamed backup,
+ * *backupEndRequired is set to true. If this backup_label was created during
+ * recovery, *backupFromStandby is set to true.
+ *
+ * Also sets the global variable RedoStartLSN with the LSN read from the
+ * backup file.
  */
 static bool
-read_backup_label(XLogRecPtr *checkPointLoc, bool *backupEndRequired,
-                                 bool *backupFromStandby)
+read_backup_label(XLogRecPtr *checkPointLoc, TimeLineID *backupLabelTLI,
+                                 bool *backupEndRequired, bool *backupFromStandby)
 {
        char            startxlogfilename[MAXFNAMELEN];
        TimeLineID      tli_from_walseg,
@@ -12193,6 +12206,8 @@ read_backup_label(XLogRecPtr *checkPointLoc, bool *backupEndRequired,
                                 errmsg("could not read file \"%s\": %m",
                                                BACKUP_LABEL_FILE)));
 
+       *backupLabelTLI = tli_from_walseg;
+
        return true;
 }
 
@@ -12469,7 +12484,8 @@ retry:
                if (!WaitForWALToBecomeAvailable(targetPagePtr + reqLen,
                                                                                 private->randAccess,
                                                                                 private->fetching_ckpt,
-                                                                                targetRecPtr))
+                                                                                targetRecPtr,
+                                                                                private->replayTLI))
                {
                        if (readFile >= 0)
                                close(readFile);
@@ -12633,7 +12649,8 @@ next_record_is_invalid:
  */
 static bool
 WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
-                                                       bool fetching_ckpt, XLogRecPtr tliRecPtr)
+                                                       bool fetching_ckpt, XLogRecPtr tliRecPtr,
+                                                       TimeLineID replayTLI)
 {
        static TimestampTz last_fail_time = 0;
        TimestampTz now;
@@ -12756,7 +12773,7 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
                                         */
                                        if (recoveryTargetTimeLineGoal == RECOVERY_TARGET_TIMELINE_LATEST)
                                        {
-                                               if (rescanLatestTimeLine())
+                                               if (rescanLatestTimeLine(replayTLI))
                                                {
                                                        currentSource = XLOG_FROM_ARCHIVE;
                                                        break;
@@ -12883,7 +12900,7 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
                                                 */
                                                if (recoveryTargetTimeLineGoal ==
                                                        RECOVERY_TARGET_TIMELINE_LATEST)
-                                                       rescanLatestTimeLine();
+                                                       rescanLatestTimeLine(replayTLI);
 
                                                startWalReceiver = true;
                                        }