Fix record length computation in pg_waldump/xlogdump.

The current method of computing the record length (excluding the
lenght of full-page images) has been wrong since the WAL format has
been revamped in 2c03216d831160bedd72d45f712601b6f7d03f1c.  Only the
main record's length was counted, but that can be significantly too
little if there's data associated with further blocks.

Fix by computing the record length as total_lenght - fpi_length.

Reported-By: Chen Huajun
Bug: #14687
Reviewed-By: Heikki Linnakangas
Discussion: https://postgr.es/m/20170603165939.1436.58887@wrigleys.postgresql.org
Backpatch: 9.5-

diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 6bc27f4be360c56c9e3e6f55ac29f3cf639169cb..2578d4b692081d9c9630d366302805a4c92059ee 100644 (file)
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -363,23 +363,13 @@ XLogDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
 }
 
 /*
- * Store per-rmgr and per-record statistics for a given record.
+ * Calculate the size of a record, split into !FPI and FPI parts.
  */
 static void
-XLogDumpCountRecord(XLogDumpConfig *config, XLogDumpStats *stats,
-                                       XLogReaderState *record)
+XLogDumpRecordLen(XLogReaderState *record, uint32 *rec_len, uint32 *fpi_len)
 {
-       RmgrId          rmid;
-       uint8           recid;
-       uint32          rec_len;
-       uint32          fpi_len;
        int                     block_id;
 
-       stats->count++;
-
-       rmid = XLogRecGetRmid(record);
-       rec_len = XLogRecGetDataLen(record) + SizeOfXLogRecord;
-
        /*
         * Calculate the amount of FPI data in the record.
         *
@@ -387,13 +377,38 @@ XLogDumpCountRecord(XLogDumpConfig *config, XLogDumpStats *stats,
         * bimg_len indicating the length of FPI data. It doesn't seem worth it to
         * add an accessor macro for this.
         */
-       fpi_len = 0;
+       *fpi_len = 0;
        for (block_id = 0; block_id <= record->max_block_id; block_id++)
        {
                if (XLogRecHasBlockImage(record, block_id))
-                       fpi_len += record->blocks[block_id].bimg_len;
+                       *fpi_len += record->blocks[block_id].bimg_len;
        }
 
+       /*
+        * Calculate the length of the record as the total length - the length of
+        * all the block images.
+        */
+       *rec_len = XLogRecGetTotalLen(record) - *fpi_len;
+}
+
+/*
+ * Store per-rmgr and per-record statistics for a given record.
+ */
+static void
+XLogDumpCountRecord(XLogDumpConfig *config, XLogDumpStats *stats,
+                                       XLogReaderState *record)
+{
+       RmgrId          rmid;
+       uint8           recid;
+       uint32          rec_len;
+       uint32          fpi_len;
+
+       stats->count++;
+
+       rmid = XLogRecGetRmid(record);
+
+       XLogDumpRecordLen(record, &rec_len, &fpi_len);
+
        /* Update per-rmgr statistics */
 
        stats->rmgr_stats[rmid].count++;
@@ -422,6 +437,8 @@ XLogDumpDisplayRecord(XLogDumpConfig *config, XLogReaderState *record)
 {
        const char *id;
        const RmgrDescData *desc = &RmgrDescTable[XLogRecGetRmid(record)];
+       uint32          rec_len;
+       uint32          fpi_len;
        RelFileNode rnode;
        ForkNumber      forknum;
        BlockNumber blk;
@@ -429,13 +446,15 @@ XLogDumpDisplayRecord(XLogDumpConfig *config, XLogReaderState *record)
        uint8           info = XLogRecGetInfo(record);
        XLogRecPtr      xl_prev = XLogRecGetPrev(record);
 
+       XLogDumpRecordLen(record, &rec_len, &fpi_len);
+
        id = desc->rm_identify(info);
        if (id == NULL)
                id = psprintf("UNKNOWN (%x)", info & ~XLR_INFO_MASK);
 
        printf("rmgr: %-11s len (rec/tot): %6u/%6u, tx: %10u, lsn: %X/%08X, prev %X/%08X, ",
                   desc->rm_name,
-                  XLogRecGetDataLen(record), XLogRecGetTotalLen(record),
+                  rec_len, XLogRecGetTotalLen(record),
                   XLogRecGetXid(record),
                   (uint32) (record->ReadRecPtr >> 32), (uint32) record->ReadRecPtr,
                   (uint32) (xl_prev >> 32), (uint32) xl_prev);