archive_waldump.c

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * archive_waldump.c
 *      A generic facility for reading WAL data from tar archives via archive
 *      streamer.
 *
 * Portions Copyright (c) 2026, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *      src/bin/pg_waldump/archive_waldump.c
 *
 *-------------------------------------------------------------------------
 */
 
#include "postgres_fe.h"
 
#include <unistd.h>
 
#include "access/xlog_internal.h"
#include "common/file_perm.h"
#include "common/hashfn.h"
#include "common/logging.h"
#include "fe_utils/simple_list.h"
#include "pg_waldump.h"
 
/*
 * How many bytes should we try to read from a file at once?
 */
#define READ_CHUNK_SIZE             (128 * 1024)
 
/* Temporary directory for spilled WAL segment files */
char       *TmpWalSegDir = NULL;
 
/*
 * Check if the start segment number is zero; this indicates a request to read
 * any WAL file.
 */
#define READ_ANY_WAL(privateInfo)   ((privateInfo)->start_segno == 0)
 
/*
 * Hash entry representing a WAL segment retrieved from the archive.
 *
 * While WAL segments are typically read sequentially, individual entries
 * maintain their own buffers for the following reasons:
 *
 * 1. Boundary Handling: The archive streamer provides a continuous byte
 * stream. A single streaming chunk may contain the end of one WAL segment
 * and the start of the next. Separate buffers allow us to easily
 * partition and track these bytes by their respective segments.
 *
 * 2. Out-of-Order Support: Dedicated buffers simplify logic when segments
 * are archived or retrieved out of sequence.
 *
 * To minimize the memory footprint, entries and their associated buffers are
 * freed immediately once consumed. Since pg_waldump does not request the same
 * bytes twice, a segment is discarded as soon as pg_waldump moves past it.
 */
typedef struct ArchivedWALFile
{
    uint32      status;         /* hash status */
    const char *fname;          /* hash key: WAL segment name */
 
    StringInfo  buf;            /* holds WAL bytes read from archive */
    bool        spilled;        /* true if the WAL data was spilled to a
                                 * temporary file */
 
    int         read_len;       /* total bytes received from archive for this
                                 * segment, including already-consumed data */
} ArchivedWALFile;
 
static uint32 hash_string_pointer(const char *s);
#define SH_PREFIX               ArchivedWAL
#define SH_ELEMENT_TYPE         ArchivedWALFile
#define SH_KEY_TYPE             const char *
#define SH_KEY                  fname
#define SH_HASH_KEY(tb, key)    hash_string_pointer(key)
#define SH_EQUAL(tb, a, b)      (strcmp(a, b) == 0)
#define SH_SCOPE                static inline
#define SH_RAW_ALLOCATOR        pg_malloc0
#define SH_DECLARE
#define SH_DEFINE
#include "lib/simplehash.h"
 
typedef struct astreamer_waldump
{
    astreamer   base;
    XLogDumpPrivate *privateInfo;
} astreamer_waldump;
 
static ArchivedWALFile *get_archive_wal_entry(const char *fname,
                                              XLogDumpPrivate *privateInfo);
static int  read_archive_file(XLogDumpPrivate *privateInfo, Size count);
static void setup_tmpwal_dir(const char *waldir);
static void cleanup_tmpwal_dir_atexit(void);
 
static FILE *prepare_tmp_write(const char *fname, XLogDumpPrivate *privateInfo);
static void perform_tmp_write(const char *fname, StringInfo buf, FILE *file);
 
static astreamer *astreamer_waldump_new(XLogDumpPrivate *privateInfo);
static void astreamer_waldump_content(astreamer *streamer,
                                      astreamer_member *member,
                                      const char *data, int len,
                                      astreamer_archive_context context);
static void astreamer_waldump_finalize(astreamer *streamer);
static void astreamer_waldump_free(astreamer *streamer);
 
static bool member_is_wal_file(astreamer_waldump *mystreamer,
                               astreamer_member *member,
                               char **fname);
 
static const astreamer_ops astreamer_waldump_ops = {
    .content = astreamer_waldump_content,
    .finalize = astreamer_waldump_finalize,
    .free = astreamer_waldump_free
};
 
/*
 * Initializes the tar archive reader: opens the archive, builds a hash table
 * for WAL entries, reads ahead until a full WAL page header is available to
 * determine the WAL segment size, and computes start/end segment numbers for
 * filtering.
 */
void
init_archive_reader(XLogDumpPrivate *privateInfo,
                    pg_compress_algorithm compression)
{
    int         fd;
    astreamer  *streamer;
    ArchivedWALFile *entry = NULL;
    XLogLongPageHeader longhdr;
    XLogSegNo   segno;
    TimeLineID  timeline;
 
    /* Open tar archive and store its file descriptor */
    fd = open_file_in_directory(privateInfo->archive_dir,
                                privateInfo->archive_name);
 
    if (fd < 0)
        pg_fatal("could not open file \"%s\"", privateInfo->archive_name);
 
    privateInfo->archive_fd = fd;
 
    streamer = astreamer_waldump_new(privateInfo);
 
    /* We must first parse the tar archive. */
    streamer = astreamer_tar_parser_new(streamer);
 
    /* If the archive is compressed, decompress before parsing. */
    if (compression == PG_COMPRESSION_GZIP)
        streamer = astreamer_gzip_decompressor_new(streamer);
    else if (compression == PG_COMPRESSION_LZ4)
        streamer = astreamer_lz4_decompressor_new(streamer);
    else if (compression == PG_COMPRESSION_ZSTD)
        streamer = astreamer_zstd_decompressor_new(streamer);
 
    privateInfo->archive_streamer = streamer;
 
    /*
     * Allocate a buffer for reading the archive file to facilitate content
     * decoding; read requests must not exceed the allocated buffer size.
     */
    privateInfo->archive_read_buf = pg_malloc(READ_CHUNK_SIZE);
 
#ifdef USE_ASSERT_CHECKING
    privateInfo->archive_read_buf_size = READ_CHUNK_SIZE;
#endif
 
    /*
     * Hash table storing WAL entries read from the archive with an arbitrary
     * initial size.
     */
    privateInfo->archive_wal_htab = ArchivedWAL_create(8, NULL);
 
    /*
     * Read until we have at least one full WAL page (XLOG_BLCKSZ bytes) from
     * the first WAL segment in the archive so we can extract the WAL segment
     * size from the long page header.
     */
    while (entry == NULL || entry->buf->len < XLOG_BLCKSZ)
    {
        if (read_archive_file(privateInfo, XLOG_BLCKSZ) == 0)
            pg_fatal("could not find WAL in archive \"%s\"",
                     privateInfo->archive_name);
 
        entry = privateInfo->cur_file;
    }
 
    /* Extract the WAL segment size from the long page header */
    longhdr = (XLogLongPageHeader) entry->buf->data;
 
    if (!IsValidWalSegSize(longhdr->xlp_seg_size))
    {
        pg_log_error(ngettext("invalid WAL segment size in WAL file from archive \"%s\" (%d byte)",
                              "invalid WAL segment size in WAL file from archive \"%s\" (%d bytes)",
                              longhdr->xlp_seg_size),
                     privateInfo->archive_name, longhdr->xlp_seg_size);
        pg_log_error_detail("The WAL segment size must be a power of two between 1 MB and 1 GB.");
        exit(1);
    }
 
    privateInfo->segsize = longhdr->xlp_seg_size;
 
    /*
     * With the WAL segment size available, we can now initialize the
     * dependent start and end segment numbers.
     */
    Assert(!XLogRecPtrIsInvalid(privateInfo->startptr));
    XLByteToSeg(privateInfo->startptr, privateInfo->start_segno,
                privateInfo->segsize);
 
    if (!XLogRecPtrIsInvalid(privateInfo->endptr))
        XLByteToSeg(privateInfo->endptr, privateInfo->end_segno,
                    privateInfo->segsize);
 
    /*
     * This WAL record was fetched before the filtering parameters
     * (start_segno and end_segno) were fully initialized. Perform the
     * relevance check against the user-provided range now; if the WAL falls
     * outside this range, remove it from the hash table. Subsequent WAL will
     * be filtered automatically by the archive streamer using the updated
     * start_segno and end_segno values.
     */
    XLogFromFileName(entry->fname, &timeline, &segno, privateInfo->segsize);
    if (privateInfo->timeline != timeline ||
        privateInfo->start_segno > segno ||
        privateInfo->end_segno < segno)
        free_archive_wal_entry(entry->fname, privateInfo);
}
 
/*
 * Release the archive streamer chain and close the archive file.
 */
void
free_archive_reader(XLogDumpPrivate *privateInfo)
{
    /*
     * NB: Normally, astreamer_finalize() is called before astreamer_free() to
     * flush any remaining buffered data or to ensure the end of the tar
     * archive is reached.  However, when decoding WAL, once we hit the end
     * LSN, any remaining buffered data or unread portion of the archive can
     * be safely ignored.
     */
    astreamer_free(privateInfo->archive_streamer);
 
    /* Free any remaining hash table entries and their buffers. */
    if (privateInfo->archive_wal_htab != NULL)
    {
        ArchivedWAL_iterator iter;
        ArchivedWALFile *entry;
 
        ArchivedWAL_start_iterate(privateInfo->archive_wal_htab, &iter);
        while ((entry = ArchivedWAL_iterate(privateInfo->archive_wal_htab,
                                            &iter)) != NULL)
        {
            if (entry->buf != NULL)
                destroyStringInfo(entry->buf);
        }
        ArchivedWAL_destroy(privateInfo->archive_wal_htab);
        privateInfo->archive_wal_htab = NULL;
    }
 
    /* Free the reusable read buffer. */
    if (privateInfo->archive_read_buf != NULL)
    {
        pg_free(privateInfo->archive_read_buf);
        privateInfo->archive_read_buf = NULL;
    }
 
    /* Close the file. */
    if (close(privateInfo->archive_fd) != 0)
        pg_log_error("could not close file \"%s\": %m",
                     privateInfo->archive_name);
}
 
/*
 * Copies the requested WAL data from the hash entry's buffer into readBuff.
 * If the buffer does not yet contain the needed bytes, fetches more data from
 * the tar archive via the archive streamer.
 */
int
read_archive_wal_page(XLogDumpPrivate *privateInfo, XLogRecPtr targetPagePtr,
                      Size count, char *readBuff)
{
    char       *p = readBuff;
    Size        nbytes = count;
    XLogRecPtr  recptr = targetPagePtr;
    int         segsize = privateInfo->segsize;
    XLogSegNo   segno;
    char        fname[MAXFNAMELEN];
    ArchivedWALFile *entry;
 
    /* Identify the segment and locate its entry in the archive hash */
    XLByteToSeg(targetPagePtr, segno, segsize);
    XLogFileName(fname, privateInfo->timeline, segno, segsize);
    entry = get_archive_wal_entry(fname, privateInfo);
 
    while (nbytes > 0)
    {
        char       *buf = entry->buf->data;
        int         bufLen = entry->buf->len;
        XLogRecPtr  endPtr;
        XLogRecPtr  startPtr;
 
        /*
         * Calculate the LSN range currently residing in the buffer.
         *
         * read_len tracks total bytes received for this segment (including
         * already-discarded data), so endPtr is the LSN just past the last
         * buffered byte, and startPtr is the LSN of the first buffered byte.
         */
        XLogSegNoOffsetToRecPtr(segno, entry->read_len, segsize, endPtr);
        startPtr = endPtr - bufLen;
 
        /*
         * Copy the requested WAL record if it exists in the buffer.
         */
        if (bufLen > 0 && startPtr <= recptr && recptr < endPtr)
        {
            int         copyBytes;
            int         offset = recptr - startPtr;
 
            /*
             * Given startPtr <= recptr < endPtr and a total buffer size
             * 'bufLen', the offset (recptr - startPtr) will always be less
             * than 'bufLen'.
             */
            Assert(offset < bufLen);
 
            copyBytes = Min(nbytes, bufLen - offset);
            memcpy(p, buf + offset, copyBytes);
 
            /* Update state for read */
            recptr += copyBytes;
            nbytes -= copyBytes;
            p += copyBytes;
        }
        else
        {
            /*
             * Before starting the actual decoding loop, pg_waldump tries to
             * locate the first valid record from the user-specified start
             * position, which might not be the start of a WAL record and
             * could fall in the middle of a record that spans multiple pages.
             * Consequently, the valid start position the decoder is looking
             * for could be far away from that initial position.
             *
             * This may involve reading across multiple pages, and this
             * pre-reading fetches data in multiple rounds from the archive
             * streamer; normally, we would throw away existing buffer
             * contents to fetch the next set of data, but that existing data
             * might be needed once the main loop starts. Because previously
             * read data cannot be re-read by the archive streamer, we delay
             * resetting the buffer until the main decoding loop is entered.
             *
             * Once pg_waldump has entered the main loop, it may re-read the
             * currently active page, but never an older one; therefore, any
             * fully consumed WAL data preceding the current page can then be
             * safely discarded.
             */
            if (privateInfo->decoding_started)
            {
                resetStringInfo(entry->buf);
 
                /*
                 * Push back the partial page data for the current page to the
                 * buffer, ensuring a full page remains available for
                 * re-reading if requested.
                 */
                if (p > readBuff)
                {
                    Assert((count - nbytes) > 0);
                    appendBinaryStringInfo(entry->buf, readBuff, count - nbytes);
                }
            }
 
            /*
             * Now, fetch more data.  Raise an error if the archive streamer
             * has moved past our segment (meaning the WAL file in the archive
             * is shorter than expected) or if reading the archive reached
             * EOF.
             */
            if (privateInfo->cur_file != entry)
                pg_fatal("WAL segment \"%s\" in archive \"%s\" is too short: read %lld of %lld bytes",
                         fname, privateInfo->archive_name,
                         (long long int) (count - nbytes),
                         (long long int) count);
            if (read_archive_file(privateInfo, READ_CHUNK_SIZE) == 0)
                pg_fatal("unexpected end of archive \"%s\" while reading \"%s\": read %lld of %lld bytes",
                         privateInfo->archive_name, fname,
                         (long long int) (count - nbytes),
                         (long long int) count);
        }
    }
 
    /*
     * Should have successfully read all the requested bytes or reported a
     * failure before this point.
     */
    Assert(nbytes == 0);
 
    /*
     * Return count unchanged; the caller expects this convention, matching
     * the routine that reads WAL pages from physical files.
     */
    return count;
}
 
/*
 * Releases the buffer of a WAL entry that is no longer needed, preventing the
 * accumulation of irrelevant WAL data.  Also removes any associated temporary
 * file and clears privateInfo->cur_file if it points to this entry, so the
 * archive streamer skips subsequent data for it.
 */
void
free_archive_wal_entry(const char *fname, XLogDumpPrivate *privateInfo)
{
    ArchivedWALFile *entry;
 
    entry = ArchivedWAL_lookup(privateInfo->archive_wal_htab, fname);
 
    if (entry == NULL)
        return;
 
    /* Destroy the buffer */
    destroyStringInfo(entry->buf);
    entry->buf = NULL;
 
    /* Remove temporary file if any */
    if (entry->spilled)
    {
        char        fpath[MAXPGPATH];
 
        snprintf(fpath, MAXPGPATH, "%s/%s", TmpWalSegDir, fname);
 
        if (unlink(fpath) == 0)
            pg_log_debug("removed file \"%s\"", fpath);
    }
 
    /* Clear cur_file if it points to the entry being freed */
    if (privateInfo->cur_file == entry)
        privateInfo->cur_file = NULL;
 
    ArchivedWAL_delete_item(privateInfo->archive_wal_htab, entry);
}
 
/*
 * Returns the archived WAL entry from the hash table if it already exists.
 * Otherwise, reads more data from the archive until the requested entry is
 * found.  If the archive streamer is reading a WAL file from the archive that
 * is not currently needed, that data is spilled to a temporary file for later
 * retrieval.
 */
static ArchivedWALFile *
get_archive_wal_entry(const char *fname, XLogDumpPrivate *privateInfo)
{
    ArchivedWALFile *entry = NULL;
    FILE       *write_fp = NULL;
 
    /*
     * Search the hash table first. If the entry is found, return it.
     * Otherwise, the requested WAL entry hasn't been read from the archive
     * yet; invoke the archive streamer to fetch it.
     */
    while (1)
    {
        /*
         * Search hash table.
         *
         * We perform the search inside the loop because a single iteration of
         * the archive reader may decompress and extract multiple files into
         * the hash table. One of these newly added files could be the one we
         * are seeking.
         */
        entry = ArchivedWAL_lookup(privateInfo->archive_wal_htab, fname);
 
        if (entry != NULL)
            return entry;
 
        /*
         * Capture the current entry before calling read_archive_file(),
         * because cur_file may advance to a new segment during streaming. We
         * hold this reference so we can flush any remaining buffer data and
         * close the write handle once we detect that cur_file has moved on.
         */
        entry = privateInfo->cur_file;
 
        /*
         * Fetch more data either when no current file is being tracked or
         * when its buffer has been fully flushed to the temporary file.
         */
        if (entry == NULL || entry->buf->len == 0)
        {
            if (read_archive_file(privateInfo, READ_CHUNK_SIZE) == 0)
                break;          /* archive file ended */
        }
 
        /*
         * Archive streamer is reading a non-WAL file or an irrelevant WAL
         * file.
         */
        if (entry == NULL)
            continue;
 
        /*
         * The streamer is producing a WAL segment that isn't the one asked
         * for; it must be arriving out of order.  Spill its data to disk so
         * it can be read back when needed.
         */
        Assert(strcmp(fname, entry->fname) != 0);
 
        /* Create a temporary file if one does not already exist */
        if (!entry->spilled)
        {
            write_fp = prepare_tmp_write(entry->fname, privateInfo);
            entry->spilled = true;
        }
 
        /* Flush data from the buffer to the file */
        perform_tmp_write(entry->fname, entry->buf, write_fp);
        resetStringInfo(entry->buf);
 
        /*
         * If cur_file changed since we captured entry above, the archive
         * streamer has finished this segment and moved on.  Close its spill
         * file handle so data is flushed to disk before the next segment
         * starts writing to a different handle.
         */
        if (entry != privateInfo->cur_file && write_fp != NULL)
        {
            fclose(write_fp);
            write_fp = NULL;
        }
    }
 
    /* Requested WAL segment not found */
    pg_fatal("could not find WAL \"%s\" in archive \"%s\"",
             fname, privateInfo->archive_name);
}
 
/*
 * Reads a chunk from the archive file and passes it through the streamer
 * pipeline for decompression (if needed) and tar member extraction.
 */
static int
read_archive_file(XLogDumpPrivate *privateInfo, Size count)
{
    int         rc;
 
    /* The read request must not exceed the allocated buffer size. */
    Assert(privateInfo->archive_read_buf_size >= count);
 
    rc = read(privateInfo->archive_fd, privateInfo->archive_read_buf, count);
    if (rc < 0)
        pg_fatal("could not read file \"%s\": %m",
                 privateInfo->archive_name);
 
    /*
     * Decompress (if required), and then parse the previously read contents
     * of the tar file.
     */
    if (rc > 0)
        astreamer_content(privateInfo->archive_streamer, NULL,
                          privateInfo->archive_read_buf, rc,
                          ASTREAMER_UNKNOWN);
 
    return rc;
}
 
/*
 * Set up a temporary directory to temporarily store WAL segments.
 */
static void
setup_tmpwal_dir(const char *waldir)
{
    const char *tmpdir = getenv("TMPDIR");
    char       *template;
 
    Assert(TmpWalSegDir == NULL);
 
    /*
     * Use the directory specified by the TMPDIR environment variable. If it's
     * not set, fall back to the provided WAL directory to store WAL files
     * temporarily.
     */
    template = psprintf("%s/waldump_tmp-XXXXXX",
                        tmpdir ? tmpdir : waldir);
    TmpWalSegDir = mkdtemp(template);
 
    if (TmpWalSegDir == NULL)
        pg_fatal("could not create directory \"%s\": %m", template);
 
    canonicalize_path(TmpWalSegDir);
 
    pg_log_debug("created directory \"%s\"", TmpWalSegDir);
}
 
/*
 * Remove temporary directory at exit, if any.
 */
static void
cleanup_tmpwal_dir_atexit(void)
{
    Assert(TmpWalSegDir != NULL);
 
    rmtree(TmpWalSegDir, true);
 
    TmpWalSegDir = NULL;
}
 
/*
 * Open a file in the temporary spill directory for writing an out-of-order
 * WAL segment, creating the directory and registering the cleanup callback
 * if not already done.  Returns the open file handle.
 */
static FILE *
prepare_tmp_write(const char *fname, XLogDumpPrivate *privateInfo)
{
    char        fpath[MAXPGPATH];
    FILE       *file;
 
    /*
     * Setup temporary directory to store WAL segments and set up an exit
     * callback to remove it upon completion if not already.
     */
    if (unlikely(TmpWalSegDir == NULL))
    {
        setup_tmpwal_dir(privateInfo->archive_dir);
        atexit(cleanup_tmpwal_dir_atexit);
    }
 
    snprintf(fpath, MAXPGPATH, "%s/%s", TmpWalSegDir, fname);
 
    /* Open the spill file for writing */
    file = fopen(fpath, PG_BINARY_W);
    if (file == NULL)
        pg_fatal("could not create file \"%s\": %m", fpath);
 
#ifndef WIN32
    if (chmod(fpath, pg_file_create_mode))
        pg_fatal("could not set permissions on file \"%s\": %m",
                 fpath);
#endif
 
    pg_log_debug("spilling to temporary file \"%s\"", fpath);
 
    return file;
}
 
/*
 * Write buffer data to the given file handle.
 */
static void
perform_tmp_write(const char *fname, StringInfo buf, FILE *file)
{
    Assert(file);
 
    errno = 0;
    if (buf->len > 0 && fwrite(buf->data, buf->len, 1, file) != 1)
    {
        /*
         * If write didn't set errno, assume problem is no disk space
         */
        if (errno == 0)
            errno = ENOSPC;
        pg_fatal("could not write to file \"%s/%s\": %m", TmpWalSegDir, fname);
    }
}
 
/*
 * Create an astreamer that can read WAL from tar file.
 */
static astreamer *
astreamer_waldump_new(XLogDumpPrivate *privateInfo)
{
    astreamer_waldump *streamer;
 
    streamer = palloc0_object(astreamer_waldump);
    *((const astreamer_ops **) &streamer->base.bbs_ops) =
        &astreamer_waldump_ops;
 
    streamer->privateInfo = privateInfo;
 
    return &streamer->base;
}
 
/*
 * Main entry point of the archive streamer for reading WAL data from a tar
 * file. If a member is identified as a valid WAL file, a hash entry is created
 * for it, and its contents are copied into that entry's buffer, making them
 * accessible to the decoding routine.
 */
static void
astreamer_waldump_content(astreamer *streamer, astreamer_member *member,
                          const char *data, int len,
                          astreamer_archive_context context)
{
    astreamer_waldump *mystreamer = (astreamer_waldump *) streamer;
    XLogDumpPrivate *privateInfo = mystreamer->privateInfo;
 
    Assert(context != ASTREAMER_UNKNOWN);
 
    switch (context)
    {
        case ASTREAMER_MEMBER_HEADER:
            {
                char       *fname = NULL;
                ArchivedWALFile *entry;
                bool        found;
 
                pg_log_debug("reading \"%s\"", member->pathname);
 
                if (!member_is_wal_file(mystreamer, member, &fname))
                    break;
 
                /*
                 * Skip range filtering during initial startup, before the WAL
                 * segment size and segment number bounds are known.
                 */
                if (!READ_ANY_WAL(privateInfo))
                {
                    XLogSegNo   segno;
                    TimeLineID  timeline;
 
                    /*
                     * Skip the segment if the timeline does not match, if it
                     * falls outside the caller-specified range.
                     */
                    XLogFromFileName(fname, &timeline, &segno, privateInfo->segsize);
                    if (privateInfo->timeline != timeline ||
                        privateInfo->start_segno > segno ||
                        privateInfo->end_segno < segno)
                    {
                        pfree(fname);
                        break;
                    }
                }
 
                entry = ArchivedWAL_insert(privateInfo->archive_wal_htab,
                                           fname, &found);
 
                /*
                 * Shouldn't happen, but if it does, simply ignore the
                 * duplicate WAL file.
                 */
                if (found)
                {
                    pg_log_warning("ignoring duplicate WAL \"%s\" found in archive \"%s\"",
                                   member->pathname, privateInfo->archive_name);
                    pfree(fname);
                    break;
                }
 
                entry->buf = makeStringInfo();
                entry->spilled = false;
                entry->read_len = 0;
                privateInfo->cur_file = entry;
            }
            break;
 
        case ASTREAMER_MEMBER_CONTENTS:
            if (privateInfo->cur_file)
            {
                appendBinaryStringInfo(privateInfo->cur_file->buf, data, len);
                privateInfo->cur_file->read_len += len;
            }
            break;
 
        case ASTREAMER_MEMBER_TRAILER:
 
            /*
             * End of this tar member; mark cur_file NULL so subsequent
             * content callbacks (if any) know no WAL file is currently
             * active.
             */
            privateInfo->cur_file = NULL;
            break;
 
        case ASTREAMER_ARCHIVE_TRAILER:
            break;
 
        default:
            /* Shouldn't happen. */
            pg_fatal("unexpected state while parsing tar file");
    }
}
 
/*
 * End-of-stream processing for an astreamer_waldump stream.  This is a
 * terminal streamer so it must have no successor.
 */
static void
astreamer_waldump_finalize(astreamer *streamer)
{
    Assert(streamer->bbs_next == NULL);
}
 
/*
 * Free memory associated with an astreamer_waldump stream.
 */
static void
astreamer_waldump_free(astreamer *streamer)
{
    Assert(streamer->bbs_next == NULL);
    pfree(streamer);
}
 
/*
 * Returns true if the archive member name matches the WAL naming format. If
 * successful, it also outputs the WAL segment name.
 */
static bool
member_is_wal_file(astreamer_waldump *mystreamer, astreamer_member *member,
                   char **fname)
{
    int         pathlen;
    char        pathname[MAXPGPATH];
    char       *filename;
 
    /* We are only interested in normal files */
    if (member->is_directory || member->is_link)
        return false;
 
    if (strlen(member->pathname) < XLOG_FNAME_LEN)
        return false;
 
    /*
     * For a correct comparison, we must remove any '.' or '..' components
     * from the member pathname. Similar to member_verify_header(), we prepend
     * './' to the path so that canonicalize_path() can properly resolve and
     * strip these references from the tar member name.
     */
    snprintf(pathname, MAXPGPATH, "./%s", member->pathname);
    canonicalize_path(pathname);
    pathlen = strlen(pathname);
 
    /* Skip files in subdirectories other than pg_wal/ */
    if (pathlen > XLOG_FNAME_LEN &&
        strncmp(pathname, XLOGDIR, strlen(XLOGDIR)) != 0)
        return false;
 
    /* WAL file may appear with a full path (e.g., pg_wal/<name>) */
    filename = pathname + (pathlen - XLOG_FNAME_LEN);
    if (!IsXLogFileName(filename))
        return false;
 
    *fname = pnstrdup(filename, XLOG_FNAME_LEN);
 
    return true;
}
 
/*
 * Helper function for WAL file hash table.
 */
static uint32
hash_string_pointer(const char *s)
{
    unsigned char *ss = (unsigned char *) s;
 
    return hash_bytes(ss, strlen(s));
}