Fix pg_xlogdump so that it handles cross-page XLP_FIRST_IS_CONTRECORD record.

Previously pg_xlogdump failed to dump the contents of the WAL file
if the file starts with the continuation WAL record which spans
more than one pages. Since pg_xlogdump assumed that the continuation
record always fits on a page, it could not find the valid WAL record to
start reading from in that case.

This patch changes pg_xlogdump so that it can handle a continuation
WAL record which crosses a page boundary and find the valid record
to start reading from.

Back-patch to 9.3 where pg_xlogdump was introduced.

Author: Pavan Deolasee
Reviewed-By: Michael Paquier and Craig Ringer
Discussion: CABOikdPsPByMiG6J01DKq6om2+BNkxHTPkOyqHM2a4oYwGKsqQ@mail.gmail.com

diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index dcf747c63342b44b42bda5a502469ac5da0d645e..f2da505892408bcf31d9ab1bb4bb4f63e5a4804c 100644 (file)
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -866,46 +866,83 @@ XLogRecPtr
 XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
 {
    XLogReaderState saved_state = *state;
-   XLogRecPtr  targetPagePtr;
    XLogRecPtr  tmpRecPtr;
-   int         targetRecOff;
    XLogRecPtr  found = InvalidXLogRecPtr;
-   uint32      pageHeaderSize;
    XLogPageHeader header;
-   int         readLen;
    char       *errormsg;
 
    Assert(!XLogRecPtrIsInvalid(RecPtr));
 
-   targetRecOff = RecPtr % XLOG_BLCKSZ;
+   /*
+    * skip over potential continuation data, keeping in mind that it may span
+    * multiple pages
+    */
+   tmpRecPtr = RecPtr;
+   while (true)
+   {
+       XLogRecPtr  targetPagePtr;
+       int         targetRecOff;
+       uint32      pageHeaderSize;
+       int         readLen;
 
-   /* scroll back to page boundary */
-   targetPagePtr = RecPtr - targetRecOff;
+       /*
+        * Compute targetRecOff. It should typically be equal or greater than
+        * short page-header since a valid record can't start anywhere before
+        * that, except when caller has explicitly specified the offset that
+        * falls somewhere there or when we are skipping multi-page
+        * continuation record. It doesn't matter though because
+        * ReadPageInternal() is prepared to handle that and will read at least
+        * short page-header worth of data
+        */
+       targetRecOff = tmpRecPtr % XLOG_BLCKSZ;
 
-   /* Read the page containing the record */
-   readLen = ReadPageInternal(state, targetPagePtr, targetRecOff);
-   if (readLen < 0)
-       goto err;
+       /* scroll back to page boundary */
+       targetPagePtr = tmpRecPtr - targetRecOff;
 
-   header = (XLogPageHeader) state->readBuf;
+       /* Read the page containing the record */
+       readLen = ReadPageInternal(state, targetPagePtr, targetRecOff);
+       if (readLen < 0)
+           goto err;
 
-   pageHeaderSize = XLogPageHeaderSize(header);
+       header = (XLogPageHeader) state->readBuf;
 
-   /* make sure we have enough data for the page header */
-   readLen = ReadPageInternal(state, targetPagePtr, pageHeaderSize);
-   if (readLen < 0)
-       goto err;
+       pageHeaderSize = XLogPageHeaderSize(header);
 
-   /* skip over potential continuation data */
-   if (header->xlp_info & XLP_FIRST_IS_CONTRECORD)
-   {
-       /* record headers are MAXALIGN'ed */
-       tmpRecPtr = targetPagePtr + pageHeaderSize
-           + MAXALIGN(header->xlp_rem_len);
-   }
-   else
-   {
-       tmpRecPtr = targetPagePtr + pageHeaderSize;
+       /* make sure we have enough data for the page header */
+       readLen = ReadPageInternal(state, targetPagePtr, pageHeaderSize);
+       if (readLen < 0)
+           goto err;
+
+       /* skip over potential continuation data */
+       if (header->xlp_info & XLP_FIRST_IS_CONTRECORD)
+       {
+           /*
+            * If the length of the remaining continuation data is more than
+            * what can fit in this page, the continuation record crosses over
+            * this page. Read the next page and try again. xlp_rem_len in the
+            * next page header will contain the remaining length of the
+            * continuation data
+            *
+            * Note that record headers are MAXALIGN'ed
+            */
+           if (MAXALIGN(header->xlp_rem_len) > (XLOG_BLCKSZ - pageHeaderSize))
+               tmpRecPtr = targetPagePtr + XLOG_BLCKSZ;
+           else
+           {
+               /*
+                * The previous continuation record ends in this page. Set
+                * tmpRecPtr to point to the first valid record
+                */
+               tmpRecPtr = targetPagePtr + pageHeaderSize
+                   + MAXALIGN(header->xlp_rem_len);
+               break;
+           }
+       }
+       else
+       {
+           tmpRecPtr = targetPagePtr + pageHeaderSize;
+           break;
+       }
    }
 
    /*