pg_waldump.c

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * pg_waldump.c - decode and display WAL
 *
 * Copyright (c) 2013-2023, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *        src/bin/pg_waldump/pg_waldump.c
 *-------------------------------------------------------------------------
 */
 
#define FRONTEND 1
#include "postgres.h"
 
#include <dirent.h>
#include <limits.h>
#include <signal.h>
#include <sys/stat.h>
#include <unistd.h>
 
#include "access/transam.h"
#include "access/xlog_internal.h"
#include "access/xlogreader.h"
#include "access/xlogrecord.h"
#include "access/xlogstats.h"
#include "common/fe_memutils.h"
#include "common/file_perm.h"
#include "common/file_utils.h"
#include "common/logging.h"
#include "common/relpath.h"
#include "getopt_long.h"
#include "rmgrdesc.h"
#include "storage/bufpage.h"
 
/*
 * NOTE: For any code change or issue fix here, it is highly recommended to
 * give a thought about doing the same in pg_walinspect contrib module as well.
 */
 
static const char *progname;
 
static int  WalSegSz;
static volatile sig_atomic_t time_to_stop = false;
 
static const RelFileLocator emptyRelFileLocator = {0, 0, 0};
 
typedef struct XLogDumpPrivate
{
    TimeLineID  timeline;
    XLogRecPtr  startptr;
    XLogRecPtr  endptr;
    bool        endptr_reached;
} XLogDumpPrivate;
 
typedef struct XLogDumpConfig
{
    /* display options */
    bool        quiet;
    bool        bkp_details;
    int         stop_after_records;
    int         already_displayed_records;
    bool        follow;
    bool        stats;
    bool        stats_per_record;
 
    /* filter options */
    bool        filter_by_rmgr[RM_MAX_ID + 1];
    bool        filter_by_rmgr_enabled;
    TransactionId filter_by_xid;
    bool        filter_by_xid_enabled;
    RelFileLocator filter_by_relation;
    bool        filter_by_extended;
    bool        filter_by_relation_enabled;
    BlockNumber filter_by_relation_block;
    bool        filter_by_relation_block_enabled;
    ForkNumber  filter_by_relation_forknum;
    bool        filter_by_fpw;
 
    /* save options */
    char       *save_fullpage_path;
} XLogDumpConfig;
 
 
/*
 * When sigint is called, just tell the system to exit at the next possible
 * moment.
 */
#ifndef WIN32
 
static void
sigint_handler(SIGNAL_ARGS)
{
    time_to_stop = true;
}
#endif
 
static void
print_rmgr_list(void)
{
    int         i;
 
    for (i = 0; i <= RM_MAX_BUILTIN_ID; i++)
    {
        printf("%s\n", GetRmgrDesc(i)->rm_name);
    }
}
 
/*
 * Check whether directory exists and whether we can open it. Keep errno set so
 * that the caller can report errors somewhat more accurately.
 */
static bool
verify_directory(const char *directory)
{
    DIR        *dir = opendir(directory);
 
    if (dir == NULL)
        return false;
    closedir(dir);
    return true;
}
 
/*
 * Create if necessary the directory storing the full-page images extracted
 * from the WAL records read.
 */
static void
create_fullpage_directory(char *path)
{
    int         ret;
 
    switch ((ret = pg_check_dir(path)))
    {
        case 0:
            /* Does not exist, so create it */
            if (pg_mkdir_p(path, pg_dir_create_mode) < 0)
                pg_fatal("could not create directory \"%s\": %m", path);
            break;
        case 1:
            /* Present and empty, so do nothing */
            break;
        case 2:
        case 3:
        case 4:
            /* Exists and not empty */
            pg_fatal("directory \"%s\" exists but is not empty", path);
            break;
        default:
            /* Trouble accessing directory */
            pg_fatal("could not access directory \"%s\": %m", path);
    }
}
 
/*
 * Split a pathname as dirname(1) and basename(1) would.
 *
 * XXX this probably doesn't do very well on Windows.  We probably need to
 * apply canonicalize_path(), at the very least.
 */
static void
split_path(const char *path, char **dir, char **fname)
{
    char       *sep;
 
    /* split filepath into directory & filename */
    sep = strrchr(path, '/');
 
    /* directory path */
    if (sep != NULL)
    {
        *dir = pnstrdup(path, sep - path);
        *fname = pg_strdup(sep + 1);
    }
    /* local directory */
    else
    {
        *dir = NULL;
        *fname = pg_strdup(path);
    }
}
 
/*
 * Open the file in the valid target directory.
 *
 * return a read only fd
 */
static int
open_file_in_directory(const char *directory, const char *fname)
{
    int         fd = -1;
    char        fpath[MAXPGPATH];
 
    Assert(directory != NULL);
 
    snprintf(fpath, MAXPGPATH, "%s/%s", directory, fname);
    fd = open(fpath, O_RDONLY | PG_BINARY, 0);
 
    if (fd < 0 && errno != ENOENT)
        pg_fatal("could not open file \"%s\": %m", fname);
    return fd;
}
 
/*
 * Try to find fname in the given directory. Returns true if it is found,
 * false otherwise. If fname is NULL, search the complete directory for any
 * file with a valid WAL file name. If file is successfully opened, set the
 * wal segment size.
 */
static bool
search_directory(const char *directory, const char *fname)
{
    int         fd = -1;
    DIR        *xldir;
 
    /* open file if valid filename is provided */
    if (fname != NULL)
        fd = open_file_in_directory(directory, fname);
 
    /*
     * A valid file name is not passed, so search the complete directory.  If
     * we find any file whose name is a valid WAL file name then try to open
     * it.  If we cannot open it, bail out.
     */
    else if ((xldir = opendir(directory)) != NULL)
    {
        struct dirent *xlde;
 
        while ((xlde = readdir(xldir)) != NULL)
        {
            if (IsXLogFileName(xlde->d_name))
            {
                fd = open_file_in_directory(directory, xlde->d_name);
                fname = pg_strdup(xlde->d_name);
                break;
            }
        }
 
        closedir(xldir);
    }
 
    /* set WalSegSz if file is successfully opened */
    if (fd >= 0)
    {
        PGAlignedXLogBlock buf;
        int         r;
 
        r = read(fd, buf.data, XLOG_BLCKSZ);
        if (r == XLOG_BLCKSZ)
        {
            XLogLongPageHeader longhdr = (XLogLongPageHeader) buf.data;
 
            WalSegSz = longhdr->xlp_seg_size;
 
            if (!IsValidWalSegSize(WalSegSz))
            {
                pg_log_error(ngettext("invalid WAL segment size in WAL file \"%s\" (%d byte)",
                                      "invalid WAL segment size in WAL file \"%s\" (%d bytes)",
                                      WalSegSz),
                             fname, WalSegSz);
                pg_log_error_detail("The WAL segment size must be a power of two between 1 MB and 1 GB.");
                exit(1);
            }
        }
        else if (r < 0)
            pg_fatal("could not read file \"%s\": %m",
                     fname);
        else
            pg_fatal("could not read file \"%s\": read %d of %d",
                     fname, r, XLOG_BLCKSZ);
        close(fd);
        return true;
    }
 
    return false;
}
 
/*
 * Identify the target directory.
 *
 * Try to find the file in several places:
 * if directory != NULL:
 *   directory /
 *   directory / XLOGDIR /
 * else
 *   .
 *   XLOGDIR /
 *   $PGDATA / XLOGDIR /
 *
 * The valid target directory is returned.
 */
static char *
identify_target_directory(char *directory, char *fname)
{
    char        fpath[MAXPGPATH];
 
    if (directory != NULL)
    {
        if (search_directory(directory, fname))
            return pg_strdup(directory);
 
        /* directory / XLOGDIR */
        snprintf(fpath, MAXPGPATH, "%s/%s", directory, XLOGDIR);
        if (search_directory(fpath, fname))
            return pg_strdup(fpath);
    }
    else
    {
        const char *datadir;
 
        /* current directory */
        if (search_directory(".", fname))
            return pg_strdup(".");
        /* XLOGDIR */
        if (search_directory(XLOGDIR, fname))
            return pg_strdup(XLOGDIR);
 
        datadir = getenv("PGDATA");
        /* $PGDATA / XLOGDIR */
        if (datadir != NULL)
        {
            snprintf(fpath, MAXPGPATH, "%s/%s", datadir, XLOGDIR);
            if (search_directory(fpath, fname))
                return pg_strdup(fpath);
        }
    }
 
    /* could not locate WAL file */
    if (fname)
        pg_fatal("could not locate WAL file \"%s\"", fname);
    else
        pg_fatal("could not find any WAL file");
 
    return NULL;                /* not reached */
}
 
/* pg_waldump's XLogReaderRoutine->segment_open callback */
static void
WALDumpOpenSegment(XLogReaderState *state, XLogSegNo nextSegNo,
                   TimeLineID *tli_p)
{
    TimeLineID  tli = *tli_p;
    char        fname[MAXPGPATH];
    int         tries;
 
    XLogFileName(fname, tli, nextSegNo, state->segcxt.ws_segsize);
 
    /*
     * In follow mode there is a short period of time after the server has
     * written the end of the previous file before the new file is available.
     * So we loop for 5 seconds looking for the file to appear before giving
     * up.
     */
    for (tries = 0; tries < 10; tries++)
    {
        state->seg.ws_file = open_file_in_directory(state->segcxt.ws_dir, fname);
        if (state->seg.ws_file >= 0)
            return;
        if (errno == ENOENT)
        {
            int         save_errno = errno;
 
            /* File not there yet, try again */
            pg_usleep(500 * 1000);
 
            errno = save_errno;
            continue;
        }
        /* Any other error, fall through and fail */
        break;
    }
 
    pg_fatal("could not find file \"%s\": %m", fname);
}
 
/*
 * pg_waldump's XLogReaderRoutine->segment_close callback.  Same as
 * wal_segment_close
 */
static void
WALDumpCloseSegment(XLogReaderState *state)
{
    close(state->seg.ws_file);
    /* need to check errno? */
    state->seg.ws_file = -1;
}
 
/* pg_waldump's XLogReaderRoutine->page_read callback */
static int
WALDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
                XLogRecPtr targetPtr, char *readBuff)
{
    XLogDumpPrivate *private = state->private_data;
    int         count = XLOG_BLCKSZ;
    WALReadError errinfo;
 
    if (private->endptr != InvalidXLogRecPtr)
    {
        if (targetPagePtr + XLOG_BLCKSZ <= private->endptr)
            count = XLOG_BLCKSZ;
        else if (targetPagePtr + reqLen <= private->endptr)
            count = private->endptr - targetPagePtr;
        else
        {
            private->endptr_reached = true;
            return -1;
        }
    }
 
    if (!WALRead(state, readBuff, targetPagePtr, count, private->timeline,
                 &errinfo))
    {
        WALOpenSegment *seg = &errinfo.wre_seg;
        char        fname[MAXPGPATH];
 
        XLogFileName(fname, seg->ws_tli, seg->ws_segno,
                     state->segcxt.ws_segsize);
 
        if (errinfo.wre_errno != 0)
        {
            errno = errinfo.wre_errno;
            pg_fatal("could not read from file \"%s\", offset %d: %m",
                     fname, errinfo.wre_off);
        }
        else
            pg_fatal("could not read from file \"%s\", offset %d: read %d of %d",
                     fname, errinfo.wre_off, errinfo.wre_read,
                     errinfo.wre_req);
    }
 
    return count;
}
 
/*
 * Boolean to return whether the given WAL record matches a specific relation
 * and optionally block.
 */
static bool
XLogRecordMatchesRelationBlock(XLogReaderState *record,
                               RelFileLocator matchRlocator,
                               BlockNumber matchBlock,
                               ForkNumber matchFork)
{
    int         block_id;
 
    for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
    {
        RelFileLocator rlocator;
        ForkNumber  forknum;
        BlockNumber blk;
 
        if (!XLogRecGetBlockTagExtended(record, block_id,
                                        &rlocator, &forknum, &blk, NULL))
            continue;
 
        if ((matchFork == InvalidForkNumber || matchFork == forknum) &&
            (RelFileLocatorEquals(matchRlocator, emptyRelFileLocator) ||
             RelFileLocatorEquals(matchRlocator, rlocator)) &&
            (matchBlock == InvalidBlockNumber || matchBlock == blk))
            return true;
    }
 
    return false;
}
 
/*
 * Boolean to return whether the given WAL record contains a full page write.
 */
static bool
XLogRecordHasFPW(XLogReaderState *record)
{
    int         block_id;
 
    for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
    {
        if (!XLogRecHasBlockRef(record, block_id))
            continue;
 
        if (XLogRecHasBlockImage(record, block_id))
            return true;
    }
 
    return false;
}
 
/*
 * Function to externally save all FPWs stored in the given WAL record.
 * Decompression is applied to all the blocks saved, if necessary.
 */
static void
XLogRecordSaveFPWs(XLogReaderState *record, const char *savepath)
{
    int         block_id;
 
    for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
    {
        PGAlignedBlock buf;
        Page        page;
        char        filename[MAXPGPATH];
        char        forkname[FORKNAMECHARS + 2];    /* _ + terminating zero */
        FILE       *file;
        BlockNumber blk;
        RelFileLocator rnode;
        ForkNumber  fork;
 
        if (!XLogRecHasBlockRef(record, block_id))
            continue;
 
        if (!XLogRecHasBlockImage(record, block_id))
            continue;
 
        page = (Page) buf.data;
 
        /* Full page exists, so let's save it */
        if (!RestoreBlockImage(record, block_id, page))
            pg_fatal("%s", record->errormsg_buf);
 
        (void) XLogRecGetBlockTagExtended(record, block_id,
                                          &rnode, &fork, &blk, NULL);
 
        if (fork >= 0 && fork <= MAX_FORKNUM)
            sprintf(forkname, "_%s", forkNames[fork]);
        else
            pg_fatal("invalid fork number: %u", fork);
 
        snprintf(filename, MAXPGPATH, "%s/%08X-%08X-%08X.%u.%u.%u.%u%s", savepath,
                 record->seg.ws_tli,
                 LSN_FORMAT_ARGS(record->ReadRecPtr),
                 rnode.spcOid, rnode.dbOid, rnode.relNumber, blk, forkname);
 
        file = fopen(filename, PG_BINARY_W);
        if (!file)
            pg_fatal("could not open file \"%s\": %m", filename);
 
        if (fwrite(page, BLCKSZ, 1, file) != 1)
            pg_fatal("could not write file \"%s\": %m", filename);
 
        if (fclose(file) != 0)
            pg_fatal("could not close file \"%s\": %m", filename);
    }
}
 
/*
 * Print a record to stdout
 */
static void
XLogDumpDisplayRecord(XLogDumpConfig *config, XLogReaderState *record)
{
    const char *id;
    const RmgrDescData *desc = GetRmgrDesc(XLogRecGetRmid(record));
    uint32      rec_len;
    uint32      fpi_len;
    uint8       info = XLogRecGetInfo(record);
    XLogRecPtr  xl_prev = XLogRecGetPrev(record);
    StringInfoData s;
 
    XLogRecGetLen(record, &rec_len, &fpi_len);
 
    printf("rmgr: %-11s len (rec/tot): %6u/%6u, tx: %10u, lsn: %X/%08X, prev %X/%08X, ",
           desc->rm_name,
           rec_len, XLogRecGetTotalLen(record),
           XLogRecGetXid(record),
           LSN_FORMAT_ARGS(record->ReadRecPtr),
           LSN_FORMAT_ARGS(xl_prev));
 
    id = desc->rm_identify(info);
    if (id == NULL)
        printf("desc: UNKNOWN (%x) ", info & ~XLR_INFO_MASK);
    else
        printf("desc: %s ", id);
 
    initStringInfo(&s);
    desc->rm_desc(&s, record);
    printf("%s", s.data);
 
    resetStringInfo(&s);
    XLogRecGetBlockRefInfo(record, true, config->bkp_details, &s, NULL);
    printf("%s", s.data);
    pfree(s.data);
}
 
/*
 * Display a single row of record counts and sizes for an rmgr or record.
 */
static void
XLogDumpStatsRow(const char *name,
                 uint64 n, uint64 total_count,
                 uint64 rec_len, uint64 total_rec_len,
                 uint64 fpi_len, uint64 total_fpi_len,
                 uint64 tot_len, uint64 total_len)
{
    double      n_pct,
                rec_len_pct,
                fpi_len_pct,
                tot_len_pct;
 
    n_pct = 0;
    if (total_count != 0)
        n_pct = 100 * (double) n / total_count;
 
    rec_len_pct = 0;
    if (total_rec_len != 0)
        rec_len_pct = 100 * (double) rec_len / total_rec_len;
 
    fpi_len_pct = 0;
    if (total_fpi_len != 0)
        fpi_len_pct = 100 * (double) fpi_len / total_fpi_len;
 
    tot_len_pct = 0;
    if (total_len != 0)
        tot_len_pct = 100 * (double) tot_len / total_len;
 
    printf("%-27s "
           "%20" INT64_MODIFIER "u (%6.02f) "
           "%20" INT64_MODIFIER "u (%6.02f) "
           "%20" INT64_MODIFIER "u (%6.02f) "
           "%20" INT64_MODIFIER "u (%6.02f)\n",
           name, n, n_pct, rec_len, rec_len_pct, fpi_len, fpi_len_pct,
           tot_len, tot_len_pct);
}
 
 
/*
 * Display summary statistics about the records seen so far.
 */
static void
XLogDumpDisplayStats(XLogDumpConfig *config, XLogStats *stats)
{
    int         ri,
                rj;
    uint64      total_count = 0;
    uint64      total_rec_len = 0;
    uint64      total_fpi_len = 0;
    uint64      total_len = 0;
    double      rec_len_pct,
                fpi_len_pct;
 
    /*
     * Leave if no stats have been computed yet, as tracked by the end LSN.
     */
    if (XLogRecPtrIsInvalid(stats->endptr))
        return;
 
    /*
     * Each row shows its percentages of the total, so make a first pass to
     * calculate column totals.
     */
 
    for (ri = 0; ri <= RM_MAX_ID; ri++)
    {
        if (!RmgrIdIsValid(ri))
            continue;
 
        total_count += stats->rmgr_stats[ri].count;
        total_rec_len += stats->rmgr_stats[ri].rec_len;
        total_fpi_len += stats->rmgr_stats[ri].fpi_len;
    }
    total_len = total_rec_len + total_fpi_len;
 
    printf("WAL statistics between %X/%X and %X/%X:\n",
           LSN_FORMAT_ARGS(stats->startptr), LSN_FORMAT_ARGS(stats->endptr));
 
    /*
     * 27 is strlen("Transaction/COMMIT_PREPARED"), 20 is strlen(2^64), 8 is
     * strlen("(100.00%)")
     */
 
    printf("%-27s %20s %8s %20s %8s %20s %8s %20s %8s\n"
           "%-27s %20s %8s %20s %8s %20s %8s %20s %8s\n",
           "Type", "N", "(%)", "Record size", "(%)", "FPI size", "(%)", "Combined size", "(%)",
           "----", "-", "---", "-----------", "---", "--------", "---", "-------------", "---");
 
    for (ri = 0; ri <= RM_MAX_ID; ri++)
    {
        uint64      count,
                    rec_len,
                    fpi_len,
                    tot_len;
        const RmgrDescData *desc;
 
        if (!RmgrIdIsValid(ri))
            continue;
 
        desc = GetRmgrDesc(ri);
 
        if (!config->stats_per_record)
        {
            count = stats->rmgr_stats[ri].count;
            rec_len = stats->rmgr_stats[ri].rec_len;
            fpi_len = stats->rmgr_stats[ri].fpi_len;
            tot_len = rec_len + fpi_len;
 
            if (RmgrIdIsCustom(ri) && count == 0)
                continue;
 
            XLogDumpStatsRow(desc->rm_name,
                             count, total_count, rec_len, total_rec_len,
                             fpi_len, total_fpi_len, tot_len, total_len);
        }
        else
        {
            for (rj = 0; rj < MAX_XLINFO_TYPES; rj++)
            {
                const char *id;
 
                count = stats->record_stats[ri][rj].count;
                rec_len = stats->record_stats[ri][rj].rec_len;
                fpi_len = stats->record_stats[ri][rj].fpi_len;
                tot_len = rec_len + fpi_len;
 
                /* Skip undefined combinations and ones that didn't occur */
                if (count == 0)
                    continue;
 
                /* the upper four bits in xl_info are the rmgr's */
                id = desc->rm_identify(rj << 4);
                if (id == NULL)
                    id = psprintf("UNKNOWN (%x)", rj << 4);
 
                XLogDumpStatsRow(psprintf("%s/%s", desc->rm_name, id),
                                 count, total_count, rec_len, total_rec_len,
                                 fpi_len, total_fpi_len, tot_len, total_len);
            }
        }
    }
 
    printf("%-27s %20s %8s %20s %8s %20s %8s %20s\n",
           "", "--------", "", "--------", "", "--------", "", "--------");
 
    /*
     * The percentages in earlier rows were calculated against the column
     * total, but the ones that follow are against the row total. Note that
     * these are displayed with a % symbol to differentiate them from the
     * earlier ones, and are thus up to 9 characters long.
     */
 
    rec_len_pct = 0;
    if (total_len != 0)
        rec_len_pct = 100 * (double) total_rec_len / total_len;
 
    fpi_len_pct = 0;
    if (total_len != 0)
        fpi_len_pct = 100 * (double) total_fpi_len / total_len;
 
    printf("%-27s "
           "%20" INT64_MODIFIER "u %-9s"
           "%20" INT64_MODIFIER "u %-9s"
           "%20" INT64_MODIFIER "u %-9s"
           "%20" INT64_MODIFIER "u %-6s\n",
           "Total", stats->count, "",
           total_rec_len, psprintf("[%.02f%%]", rec_len_pct),
           total_fpi_len, psprintf("[%.02f%%]", fpi_len_pct),
           total_len, "[100%]");
}
 
static void
usage(void)
{
    printf(_("%s decodes and displays PostgreSQL write-ahead logs for debugging.\n\n"),
           progname);
    printf(_("Usage:\n"));
    printf(_("  %s [OPTION]... [STARTSEG [ENDSEG]]\n"), progname);
    printf(_("\nOptions:\n"));
    printf(_("  -b, --bkp-details      output detailed information about backup blocks\n"));
    printf(_("  -B, --block=N          with --relation, only show records that modify block N\n"));
    printf(_("  -e, --end=RECPTR       stop reading at WAL location RECPTR\n"));
    printf(_("  -f, --follow           keep retrying after reaching end of WAL\n"));
    printf(_("  -F, --fork=FORK        only show records that modify blocks in fork FORK;\n"
             "                         valid names are main, fsm, vm, init\n"));
    printf(_("  -n, --limit=N          number of records to display\n"));
    printf(_("  -p, --path=PATH        directory in which to find WAL segment files or a\n"
             "                         directory with a ./pg_wal that contains such files\n"
             "                         (default: current directory, ./pg_wal, $PGDATA/pg_wal)\n"));
    printf(_("  -q, --quiet            do not print any output, except for errors\n"));
    printf(_("  -r, --rmgr=RMGR        only show records generated by resource manager RMGR;\n"
             "                         use --rmgr=list to list valid resource manager names\n"));
    printf(_("  -R, --relation=T/D/R   only show records that modify blocks in relation T/D/R\n"));
    printf(_("  -s, --start=RECPTR     start reading at WAL location RECPTR\n"));
    printf(_("  -t, --timeline=TLI     timeline from which to read WAL records\n"
             "                         (default: 1 or the value used in STARTSEG)\n"));
    printf(_("  -V, --version          output version information, then exit\n"));
    printf(_("  -w, --fullpage         only show records with a full page write\n"));
    printf(_("  -x, --xid=XID          only show records with transaction ID XID\n"));
    printf(_("  -z, --stats[=record]   show statistics instead of records\n"
             "                         (optionally, show per-record statistics)\n"));
    printf(_("  --save-fullpage=DIR    save full page images to DIR\n"));
    printf(_("  -?, --help             show this help, then exit\n"));
    printf(_("\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT);
    printf(_("%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
}
 
int
main(int argc, char **argv)
{
    uint32      xlogid;
    uint32      xrecoff;
    XLogReaderState *xlogreader_state;
    XLogDumpPrivate private;
    XLogDumpConfig config;
    XLogStats   stats;
    XLogRecord *record;
    XLogRecPtr  first_record;
    char       *waldir = NULL;
    char       *errormsg;
 
    static struct option long_options[] = {
        {"bkp-details", no_argument, NULL, 'b'},
        {"block", required_argument, NULL, 'B'},
        {"end", required_argument, NULL, 'e'},
        {"follow", no_argument, NULL, 'f'},
        {"fork", required_argument, NULL, 'F'},
        {"fullpage", no_argument, NULL, 'w'},
        {"help", no_argument, NULL, '?'},
        {"limit", required_argument, NULL, 'n'},
        {"path", required_argument, NULL, 'p'},
        {"quiet", no_argument, NULL, 'q'},
        {"relation", required_argument, NULL, 'R'},
        {"rmgr", required_argument, NULL, 'r'},
        {"start", required_argument, NULL, 's'},
        {"timeline", required_argument, NULL, 't'},
        {"xid", required_argument, NULL, 'x'},
        {"version", no_argument, NULL, 'V'},
        {"stats", optional_argument, NULL, 'z'},
        {"save-fullpage", required_argument, NULL, 1},
        {NULL, 0, NULL, 0}
    };
 
    int         option;
    int         optindex = 0;
 
#ifndef WIN32
    pqsignal(SIGINT, sigint_handler);
#endif
 
    pg_logging_init(argv[0]);
    set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_waldump"));
    progname = get_progname(argv[0]);
 
    if (argc > 1)
    {
        if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
        {
            usage();
            exit(0);
        }
        if (strcmp(argv[1], "--version") == 0 || strcmp(argv[1], "-V") == 0)
        {
            puts("pg_waldump (PostgreSQL) " PG_VERSION);
            exit(0);
        }
    }
 
    memset(&private, 0, sizeof(XLogDumpPrivate));
    memset(&config, 0, sizeof(XLogDumpConfig));
    memset(&stats, 0, sizeof(XLogStats));
 
    private.timeline = 1;
    private.startptr = InvalidXLogRecPtr;
    private.endptr = InvalidXLogRecPtr;
    private.endptr_reached = false;
 
    config.quiet = false;
    config.bkp_details = false;
    config.stop_after_records = -1;
    config.already_displayed_records = 0;
    config.follow = false;
    /* filter_by_rmgr array was zeroed by memset above */
    config.filter_by_rmgr_enabled = false;
    config.filter_by_xid = InvalidTransactionId;
    config.filter_by_xid_enabled = false;
    config.filter_by_extended = false;
    config.filter_by_relation_enabled = false;
    config.filter_by_relation_block_enabled = false;
    config.filter_by_relation_forknum = InvalidForkNumber;
    config.filter_by_fpw = false;
    config.save_fullpage_path = NULL;
    config.stats = false;
    config.stats_per_record = false;
 
    stats.startptr = InvalidXLogRecPtr;
    stats.endptr = InvalidXLogRecPtr;
 
    if (argc <= 1)
    {
        pg_log_error("no arguments specified");
        goto bad_argument;
    }
 
    while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wx:z",
                                 long_options, &optindex)) != -1)
    {
        switch (option)
        {
            case 'b':
                config.bkp_details = true;
                break;
            case 'B':
                if (sscanf(optarg, "%u", &config.filter_by_relation_block) != 1 ||
                    !BlockNumberIsValid(config.filter_by_relation_block))
                {
                    pg_log_error("invalid block number: \"%s\"", optarg);
                    goto bad_argument;
                }
                config.filter_by_relation_block_enabled = true;
                config.filter_by_extended = true;
                break;
            case 'e':
                if (sscanf(optarg, "%X/%X", &xlogid, &xrecoff) != 2)
                {
                    pg_log_error("invalid WAL location: \"%s\"",
                                 optarg);
                    goto bad_argument;
                }
                private.endptr = (uint64) xlogid << 32 | xrecoff;
                break;
            case 'f':
                config.follow = true;
                break;
            case 'F':
                config.filter_by_relation_forknum = forkname_to_number(optarg);
                if (config.filter_by_relation_forknum == InvalidForkNumber)
                {
                    pg_log_error("invalid fork name: \"%s\"", optarg);
                    goto bad_argument;
                }
                config.filter_by_extended = true;
                break;
            case 'n':
                if (sscanf(optarg, "%d", &config.stop_after_records) != 1)
                {
                    pg_log_error("invalid value \"%s\" for option %s", optarg, "-n/--limit");
                    goto bad_argument;
                }
                break;
            case 'p':
                waldir = pg_strdup(optarg);
                break;
            case 'q':
                config.quiet = true;
                break;
            case 'r':
                {
                    int         rmid;
 
                    if (pg_strcasecmp(optarg, "list") == 0)
                    {
                        print_rmgr_list();
                        exit(EXIT_SUCCESS);
                    }
 
                    /*
                     * First look for the generated name of a custom rmgr, of
                     * the form "custom###". We accept this form, because the
                     * custom rmgr module is not loaded, so there's no way to
                     * know the real name. This convention should be
                     * consistent with that in rmgrdesc.c.
                     */
                    if (sscanf(optarg, "custom%03d", &rmid) == 1)
                    {
                        if (!RmgrIdIsCustom(rmid))
                        {
                            pg_log_error("custom resource manager \"%s\" does not exist",
                                         optarg);
                            goto bad_argument;
                        }
                        config.filter_by_rmgr[rmid] = true;
                        config.filter_by_rmgr_enabled = true;
                    }
                    else
                    {
                        /* then look for builtin rmgrs */
                        for (rmid = 0; rmid <= RM_MAX_BUILTIN_ID; rmid++)
                        {
                            if (pg_strcasecmp(optarg, GetRmgrDesc(rmid)->rm_name) == 0)
                            {
                                config.filter_by_rmgr[rmid] = true;
                                config.filter_by_rmgr_enabled = true;
                                break;
                            }
                        }
                        if (rmid > RM_MAX_BUILTIN_ID)
                        {
                            pg_log_error("resource manager \"%s\" does not exist",
                                         optarg);
                            goto bad_argument;
                        }
                    }
                }
                break;
            case 'R':
                if (sscanf(optarg, "%u/%u/%u",
                           &config.filter_by_relation.spcOid,
                           &config.filter_by_relation.dbOid,
                           &config.filter_by_relation.relNumber) != 3 ||
                    !OidIsValid(config.filter_by_relation.spcOid) ||
                    !RelFileNumberIsValid(config.filter_by_relation.relNumber))
                {
                    pg_log_error("invalid relation specification: \"%s\"", optarg);
                    pg_log_error_detail("Expecting \"tablespace OID/database OID/relation filenode\".");
                    goto bad_argument;
                }
                config.filter_by_relation_enabled = true;
                config.filter_by_extended = true;
                break;
            case 's':
                if (sscanf(optarg, "%X/%X", &xlogid, &xrecoff) != 2)
                {
                    pg_log_error("invalid WAL location: \"%s\"",
                                 optarg);
                    goto bad_argument;
                }
                else
                    private.startptr = (uint64) xlogid << 32 | xrecoff;
                break;
            case 't':
 
                /*
                 * This is like option_parse_int() but needs to handle
                 * unsigned 32-bit int.  Also, we accept both decimal and
                 * hexadecimal specifications here.
                 */
                {
                    char       *endptr;
                    unsigned long val;
 
                    errno = 0;
                    val = strtoul(optarg, &endptr, 0);
 
                    while (*endptr != '\0' && isspace((unsigned char) *endptr))
                        endptr++;
 
                    if (*endptr != '\0')
                    {
                        pg_log_error("invalid value \"%s\" for option %s",
                                     optarg, "-t/--timeline");
                        goto bad_argument;
                    }
 
                    if (errno == ERANGE || val < 1 || val > UINT_MAX)
                    {
                        pg_log_error("%s must be in range %u..%u",
                                     "-t/--timeline", 1, UINT_MAX);
                        goto bad_argument;
                    }
 
                    private.timeline = val;
 
                    break;
                }
            case 'w':
                config.filter_by_fpw = true;
                break;
            case 'x':
                if (sscanf(optarg, "%u", &config.filter_by_xid) != 1)
                {
                    pg_log_error("invalid transaction ID specification: \"%s\"",
                                 optarg);
                    goto bad_argument;
                }
                config.filter_by_xid_enabled = true;
                break;
            case 'z':
                config.stats = true;
                config.stats_per_record = false;
                if (optarg)
                {
                    if (strcmp(optarg, "record") == 0)
                        config.stats_per_record = true;
                    else if (strcmp(optarg, "rmgr") != 0)
                    {
                        pg_log_error("unrecognized value for option %s: %s",
                                     "--stats", optarg);
                        goto bad_argument;
                    }
                }
                break;
            case 1:
                config.save_fullpage_path = pg_strdup(optarg);
                break;
            default:
                goto bad_argument;
        }
    }
 
    if (config.filter_by_relation_block_enabled &&
        !config.filter_by_relation_enabled)
    {
        pg_log_error("option %s requires option %s to be specified",
                     "-B/--block", "-R/--relation");
        goto bad_argument;
    }
 
    if ((optind + 2) < argc)
    {
        pg_log_error("too many command-line arguments (first is \"%s\")",
                     argv[optind + 2]);
        goto bad_argument;
    }
 
    if (waldir != NULL)
    {
        /* validate path points to directory */
        if (!verify_directory(waldir))
        {
            pg_log_error("could not open directory \"%s\": %m", waldir);
            goto bad_argument;
        }
    }
 
    if (config.save_fullpage_path != NULL)
        create_fullpage_directory(config.save_fullpage_path);
 
    /* parse files as start/end boundaries, extract path if not specified */
    if (optind < argc)
    {
        char       *directory = NULL;
        char       *fname = NULL;
        int         fd;
        XLogSegNo   segno;
 
        split_path(argv[optind], &directory, &fname);
 
        if (waldir == NULL && directory != NULL)
        {
            waldir = directory;
 
            if (!verify_directory(waldir))
                pg_fatal("could not open directory \"%s\": %m", waldir);
        }
 
        waldir = identify_target_directory(waldir, fname);
        fd = open_file_in_directory(waldir, fname);
        if (fd < 0)
            pg_fatal("could not open file \"%s\"", fname);
        close(fd);
 
        /* parse position from file */
        XLogFromFileName(fname, &private.timeline, &segno, WalSegSz);
 
        if (XLogRecPtrIsInvalid(private.startptr))
            XLogSegNoOffsetToRecPtr(segno, 0, WalSegSz, private.startptr);
        else if (!XLByteInSeg(private.startptr, segno, WalSegSz))
        {
            pg_log_error("start WAL location %X/%X is not inside file \"%s\"",
                         LSN_FORMAT_ARGS(private.startptr),
                         fname);
            goto bad_argument;
        }
 
        /* no second file specified, set end position */
        if (!(optind + 1 < argc) && XLogRecPtrIsInvalid(private.endptr))
            XLogSegNoOffsetToRecPtr(segno + 1, 0, WalSegSz, private.endptr);
 
        /* parse ENDSEG if passed */
        if (optind + 1 < argc)
        {
            XLogSegNo   endsegno;
 
            /* ignore directory, already have that */
            split_path(argv[optind + 1], &directory, &fname);
 
            fd = open_file_in_directory(waldir, fname);
            if (fd < 0)
                pg_fatal("could not open file \"%s\"", fname);
            close(fd);
 
            /* parse position from file */
            XLogFromFileName(fname, &private.timeline, &endsegno, WalSegSz);
 
            if (endsegno < segno)
                pg_fatal("ENDSEG %s is before STARTSEG %s",
                         argv[optind + 1], argv[optind]);
 
            if (XLogRecPtrIsInvalid(private.endptr))
                XLogSegNoOffsetToRecPtr(endsegno + 1, 0, WalSegSz,
                                        private.endptr);
 
            /* set segno to endsegno for check of --end */
            segno = endsegno;
        }
 
 
        if (!XLByteInSeg(private.endptr, segno, WalSegSz) &&
            private.endptr != (segno + 1) * WalSegSz)
        {
            pg_log_error("end WAL location %X/%X is not inside file \"%s\"",
                         LSN_FORMAT_ARGS(private.endptr),
                         argv[argc - 1]);
            goto bad_argument;
        }
    }
    else
        waldir = identify_target_directory(waldir, NULL);
 
    /* we don't know what to print */
    if (XLogRecPtrIsInvalid(private.startptr))
    {
        pg_log_error("no start WAL location given");
        goto bad_argument;
    }
 
    /* done with argument parsing, do the actual work */
 
    /* we have everything we need, start reading */
    xlogreader_state =
        XLogReaderAllocate(WalSegSz, waldir,
                           XL_ROUTINE(.page_read = WALDumpReadPage,
                                      .segment_open = WALDumpOpenSegment,
                                      .segment_close = WALDumpCloseSegment),
                           &private);
    if (!xlogreader_state)
        pg_fatal("out of memory while allocating a WAL reading processor");
 
    /* first find a valid recptr to start from */
    first_record = XLogFindNextRecord(xlogreader_state, private.startptr);
 
    if (first_record == InvalidXLogRecPtr)
        pg_fatal("could not find a valid record after %X/%X",
                 LSN_FORMAT_ARGS(private.startptr));
 
    /*
     * Display a message that we're skipping data if `from` wasn't a pointer
     * to the start of a record and also wasn't a pointer to the beginning of
     * a segment (e.g. we were used in file mode).
     */
    if (first_record != private.startptr &&
        XLogSegmentOffset(private.startptr, WalSegSz) != 0)
        pg_log_info(ngettext("first record is after %X/%X, at %X/%X, skipping over %u byte",
                             "first record is after %X/%X, at %X/%X, skipping over %u bytes",
                             (first_record - private.startptr)),
                    LSN_FORMAT_ARGS(private.startptr),
                    LSN_FORMAT_ARGS(first_record),
                    (uint32) (first_record - private.startptr));
 
    if (config.stats == true && !config.quiet)
        stats.startptr = first_record;
 
    for (;;)
    {
        if (time_to_stop)
        {
            /* We've been Ctrl-C'ed, so leave */
            break;
        }
 
        /* try to read the next record */
        record = XLogReadRecord(xlogreader_state, &errormsg);
        if (!record)
        {
            if (!config.follow || private.endptr_reached)
                break;
            else
            {
                pg_usleep(1000000L);    /* 1 second */
                continue;
            }
        }
 
        /* apply all specified filters */
        if (config.filter_by_rmgr_enabled &&
            !config.filter_by_rmgr[record->xl_rmid])
            continue;
 
        if (config.filter_by_xid_enabled &&
            config.filter_by_xid != record->xl_xid)
            continue;
 
        /* check for extended filtering */
        if (config.filter_by_extended &&
            !XLogRecordMatchesRelationBlock(xlogreader_state,
                                            config.filter_by_relation_enabled ?
                                            config.filter_by_relation :
                                            emptyRelFileLocator,
                                            config.filter_by_relation_block_enabled ?
                                            config.filter_by_relation_block :
                                            InvalidBlockNumber,
                                            config.filter_by_relation_forknum))
            continue;
 
        if (config.filter_by_fpw && !XLogRecordHasFPW(xlogreader_state))
            continue;
 
        /* perform any per-record work */
        if (!config.quiet)
        {
            if (config.stats == true)
            {
                XLogRecStoreStats(&stats, xlogreader_state);
                stats.endptr = xlogreader_state->EndRecPtr;
            }
            else
                XLogDumpDisplayRecord(&config, xlogreader_state);
        }
 
        /* save full pages if requested */
        if (config.save_fullpage_path != NULL)
            XLogRecordSaveFPWs(xlogreader_state, config.save_fullpage_path);
 
        /* check whether we printed enough */
        config.already_displayed_records++;
        if (config.stop_after_records > 0 &&
            config.already_displayed_records >= config.stop_after_records)
            break;
    }
 
    if (config.stats == true && !config.quiet)
        XLogDumpDisplayStats(&config, &stats);
 
    if (time_to_stop)
        exit(0);
 
    if (errormsg)
        pg_fatal("error in WAL record at %X/%X: %s",
                 LSN_FORMAT_ARGS(xlogreader_state->ReadRecPtr),
                 errormsg);
 
    XLogReaderFree(xlogreader_state);
 
    return EXIT_SUCCESS;
 
bad_argument:
    pg_log_error_hint("Try \"%s --help\" for more information.", progname);
    return EXIT_FAILURE;
}