pg_waldump.h File Reference

#include "access/xlogdefs.h"
#include "fe_utils/astreamer.h"

Include dependency graph for pg_waldump.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	XLogDumpPrivate

Typedefs
typedef struct XLogDumpPrivate	XLogDumpPrivate

Functions
int	open_file_in_directory (const char *directory, const char *fname)
void	init_archive_reader (XLogDumpPrivate *privateInfo, pg_compress_algorithm compression)
void	free_archive_reader (XLogDumpPrivate *privateInfo)
int	read_archive_wal_page (XLogDumpPrivate *privateInfo, XLogRecPtr targetPagePtr, Size count, char *readBuff)
void	free_archive_wal_entry (const char *fname, XLogDumpPrivate *privateInfo)

Variables
char *	TmpWalSegDir

Typedef Documentation

XLogDumpPrivate

typedef struct XLogDumpPrivate XLogDumpPrivate

Function Documentation

free_archive_reader()

void free_archive_reader ( XLogDumpPrivate *  privateInfo )

extern

Definition at line 234 of file archive_waldump.c.

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

References XLogDumpPrivate::archive_fd, XLogDumpPrivate::archive_name, XLogDumpPrivate::archive_read_buf, XLogDumpPrivate::archive_streamer, XLogDumpPrivate::archive_wal_htab, astreamer_free(), ArchivedWALFile::buf, close, destroyStringInfo(), fb(), pg_free(), and pg_log_error.

Referenced by main().

free_archive_wal_entry()

void free_archive_wal_entry ( const char *  fname, XLogDumpPrivate *  privateInfo  )

extern

Definition at line 415 of file archive_waldump.c.

{
    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);
}

References XLogDumpPrivate::archive_wal_htab, ArchivedWALFile::buf, XLogDumpPrivate::cur_file, destroyStringInfo(), fb(), MAXPGPATH, pg_log_debug, snprintf, ArchivedWALFile::spilled, and TmpWalSegDir.

Referenced by init_archive_reader(), and TarWALDumpReadPage().

init_archive_reader()

void init_archive_reader ( XLogDumpPrivate *  privateInfo, pg_compress_algorithm  compression  )

extern

Definition at line 124 of file archive_waldump.c.

{
    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);
}

References XLogDumpPrivate::archive_dir, XLogDumpPrivate::archive_fd, XLogDumpPrivate::archive_name, XLogDumpPrivate::archive_read_buf, XLogDumpPrivate::archive_streamer, XLogDumpPrivate::archive_wal_htab, Assert, astreamer_gzip_decompressor_new(), astreamer_lz4_decompressor_new(), astreamer_tar_parser_new(), astreamer_waldump_new(), astreamer_zstd_decompressor_new(), ArchivedWALFile::buf, XLogDumpPrivate::cur_file, StringInfoData::data, XLogDumpPrivate::end_segno, XLogDumpPrivate::endptr, fb(), fd(), ArchivedWALFile::fname, free_archive_wal_entry(), IsValidWalSegSize, StringInfoData::len, ngettext, open_file_in_directory(), PG_COMPRESSION_GZIP, PG_COMPRESSION_LZ4, PG_COMPRESSION_ZSTD, pg_fatal, pg_log_error, pg_log_error_detail, pg_malloc(), read_archive_file(), READ_CHUNK_SIZE, XLogDumpPrivate::segsize, XLogDumpPrivate::start_segno, XLogDumpPrivate::startptr, XLogDumpPrivate::timeline, XLByteToSeg, XLogFromFileName(), and XLogRecPtrIsInvalid.

Referenced by main().

open_file_in_directory()

int open_file_in_directory ( const char *  directory, const char *  fname  )

extern

Definition at line 180 of file pg_waldump.c.

{
    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;
}

References Assert, directory, fb(), fd(), MAXPGPATH, PG_BINARY, pg_fatal, and snprintf.

Referenced by init_archive_reader(), main(), search_directory(), TarWALDumpReadPage(), and WALDumpOpenSegment().

read_archive_wal_page()

int read_archive_wal_page ( XLogDumpPrivate *  privateInfo, XLogRecPtr  targetPagePtr, Size  count, char *  readBuff  )

extern

Definition at line 281 of file archive_waldump.c.

{
    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;
}

References appendBinaryStringInfo(), XLogDumpPrivate::archive_name, Assert, buf, ArchivedWALFile::buf, XLogDumpPrivate::cur_file, StringInfoData::data, XLogDumpPrivate::decoding_started, fb(), get_archive_wal_entry(), StringInfoData::len, MAXFNAMELEN, Min, pg_fatal, read_archive_file(), READ_CHUNK_SIZE, ArchivedWALFile::read_len, resetStringInfo(), XLogDumpPrivate::segsize, XLogDumpPrivate::timeline, XLByteToSeg, XLogFileName(), and XLogSegNoOffsetToRecPtr.

Referenced by TarWALDumpReadPage().

Variable Documentation

TmpWalSegDir

char* TmpWalSegDir

extern

Definition at line 32 of file archive_waldump.c.

Referenced by cleanup_tmpwal_dir_atexit(), free_archive_wal_entry(), perform_tmp_write(), prepare_tmp_write(), setup_tmpwal_dir(), and TarWALDumpReadPage().