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;
+               }
        }
 
        /*