Refactor setting XLP_FIRST_IS_OVERWRITE_CONTRECORD.

Set it directly in CreateOverwriteContrecordRecord(). That way,
AdvanceXLInsertBuffer() doesn't need the missingContrecPtr global
variable. This is in preparation for splitting xlog.c into multiple
files.

Reviewed-by: Robert Haas
Discussion: https://www.postgresql.org/message-id/a462d79c-cb5a-47cc-e9ac-616b5003965f%40iki.fi

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 8a02c56beb16f34ef15e7276d86840363dbe723c..57e6828feb294a93b6b5636f12853c3d74811beb 100644 (file)
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -913,7 +913,9 @@ static void VerifyOverwriteContrecord(xl_overwrite_contrecord *xlrec,
                                      XLogReaderState *state);
 static int LocalSetXLogInsertAllowed(void);
 static void CreateEndOfRecoveryRecord(void);
-static XLogRecPtr CreateOverwriteContrecordRecord(XLogRecPtr aborted_lsn);
+static XLogRecPtr CreateOverwriteContrecordRecord(XLogRecPtr aborted_lsn,
+                                                 XLogRecPtr missingContrecPtr,
+                                                 TimeLineID newTLI);
 static void CheckPointGuts(XLogRecPtr checkPointRedo, int flags);
 static void KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo);
 static XLogRecPtr XLogGetReplicationSlotMinimumLSN(void);
@@ -2295,18 +2297,6 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, TimeLineID tli, bool opportunistic)
        if (!Insert->forcePageWrites)
            NewPage->xlp_info |= XLP_BKP_REMOVABLE;
 
-       /*
-        * If a record was found to be broken at the end of recovery, and
-        * we're going to write on the page where its first contrecord was
-        * lost, set the XLP_FIRST_IS_OVERWRITE_CONTRECORD flag on the page
-        * header.  See CreateOverwriteContrecordRecord().
-        */
-       if (missingContrecPtr == NewPageBeginPtr)
-       {
-           NewPage->xlp_info |= XLP_FIRST_IS_OVERWRITE_CONTRECORD;
-           missingContrecPtr = InvalidXLogRecPtr;
-       }
-
        /*
         * If first page of an XLOG segment file, make it a long header.
         */
@@ -8149,7 +8139,7 @@ StartupXLOG(void)
    if (!XLogRecPtrIsInvalid(abortedRecPtr))
    {
        Assert(!XLogRecPtrIsInvalid(missingContrecPtr));
-       CreateOverwriteContrecordRecord(abortedRecPtr);
+       CreateOverwriteContrecordRecord(abortedRecPtr, missingContrecPtr, newTLI);
        abortedRecPtr = InvalidXLogRecPtr;
        missingContrecPtr = InvalidXLogRecPtr;
    }
@@ -9530,27 +9520,70 @@ CreateEndOfRecoveryRecord(void)
  * skip the record it was reading, and pass back the LSN of the skipped
  * record, so that its caller can verify (on "replay" of that record) that the
  * XLOG_OVERWRITE_CONTRECORD matches what was effectively overwritten.
+ *
+ * 'aborted_lsn' is the beginning position of the record that was incomplete.
+ * It is included in the WAL record.  'pagePtr' and 'newTLI' point to the
+ * beginning of the XLOG page where the record is to be inserted.  They must
+ * match the current WAL insert position, they're passed here just so that we
+ * can verify that.
  */
 static XLogRecPtr
-CreateOverwriteContrecordRecord(XLogRecPtr aborted_lsn)
+CreateOverwriteContrecordRecord(XLogRecPtr aborted_lsn, XLogRecPtr pagePtr,
+                               TimeLineID newTLI)
 {
    xl_overwrite_contrecord xlrec;
    XLogRecPtr  recptr;
+   XLogPageHeader pagehdr;
+   XLogRecPtr  startPos;
 
-   /* sanity check */
+   /* sanity checks */
    if (!RecoveryInProgress())
        elog(ERROR, "can only be used at end of recovery");
-
-   xlrec.overwritten_lsn = aborted_lsn;
-   xlrec.overwrite_time = GetCurrentTimestamp();
+   if (pagePtr % XLOG_BLCKSZ != 0)
+       elog(ERROR, "invalid position for missing continuation record %X/%X",
+            LSN_FORMAT_ARGS(pagePtr));
+
+   /* The current WAL insert position should be right after the page header */
+   startPos = pagePtr;
+   if (XLogSegmentOffset(startPos, wal_segment_size) == 0)
+       startPos += SizeOfXLogLongPHD;
+   else
+       startPos += SizeOfXLogShortPHD;
+   recptr = GetXLogInsertRecPtr();
+   if (recptr != startPos)
+       elog(ERROR, "invalid WAL insert position %X/%X for OVERWRITE_CONTRECORD",
+            LSN_FORMAT_ARGS(recptr));
 
    START_CRIT_SECTION();
 
+   /*
+    * Initialize the XLOG page header (by GetXLogBuffer), and set the
+    * XLP_FIRST_IS_OVERWRITE_CONTRECORD flag.
+    *
+    * No other backend is allowed to write WAL yet, so acquiring the WAL
+    * insertion lock is just pro forma.
+    */
+   WALInsertLockAcquire();
+   pagehdr = (XLogPageHeader) GetXLogBuffer(pagePtr, newTLI);
+   pagehdr->xlp_info |= XLP_FIRST_IS_OVERWRITE_CONTRECORD;
+   WALInsertLockRelease();
+
+   /*
+    * Insert the XLOG_OVERWRITE_CONTRECORD record as the first record on the
+    * page.  We know it becomes the first record, because no other backend is
+    * allowed to write WAL yet.
+    */
    XLogBeginInsert();
+   xlrec.overwritten_lsn = aborted_lsn;
+   xlrec.overwrite_time = GetCurrentTimestamp();
    XLogRegisterData((char *) &xlrec, sizeof(xl_overwrite_contrecord));
-
    recptr = XLogInsert(RM_XLOG_ID, XLOG_OVERWRITE_CONTRECORD);
 
+   /* check that the record was inserted to the right place */
+   if (ProcLastRecPtr != startPos)
+       elog(ERROR, "OVERWRITE_CONTRECORD was inserted to unexpected position %X/%X",
+            LSN_FORMAT_ARGS(ProcLastRecPtr));
+
    XLogFlush(recptr);
 
    END_CRIT_SECTION();