#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"
#include "lib/simplehash.h"

Include dependency graph for archive_waldump.c:

Data Structures
struct	ArchivedWALFile

struct	astreamer_waldump

Macros
#define	READ_CHUNK_SIZE (128 * 1024)

#define	READ_ANY_WAL(privateInfo) ((privateInfo)->start_segno == 0)

#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

Typedefs
typedef struct ArchivedWALFile	ArchivedWALFile

typedef struct astreamer_waldump	astreamer_waldump

Functions
static uint32	hash_string_pointer (const char *s)

static ArchivedWALFile *	get_archive_wal_entry (const char fname, XLogDumpPrivate privateInfo)

static bool	read_archive_file (XLogDumpPrivate *privateInfo)

static void	setup_tmpwal_dir (const char *waldir)

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)

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_t count, char readBuff)

void	free_archive_wal_entry (const char fname, XLogDumpPrivate privateInfo)

Variables
char *	TmpWalSegDir = NULL

static const astreamer_ops	astreamer_waldump_ops

Macro Definition Documentation

◆ READ_ANY_WAL

#define READ_ANY_WAL ( privateInfo ) ((privateInfo)->start_segno == 0)

Definition at line 38 of file archive_waldump.c.

◆ READ_CHUNK_SIZE

#define READ_CHUNK_SIZE (128 * 1024)

Definition at line 29 of file archive_waldump.c.

◆ SH_DECLARE

#define SH_DECLARE

Definition at line 86 of file archive_waldump.c.

◆ SH_DEFINE

#define SH_DEFINE

Definition at line 87 of file archive_waldump.c.

◆ SH_ELEMENT_TYPE

#define SH_ELEMENT_TYPE ArchivedWALFile

Definition at line 79 of file archive_waldump.c.

◆ SH_EQUAL

#define SH_EQUAL	(	tb,
		a,
		b
	)	(strcmp(a, b) == 0)

Definition at line 83 of file archive_waldump.c.

◆ SH_HASH_KEY

#define SH_HASH_KEY	(	tb,
		key
	)	hash_string_pointer(key)

Definition at line 82 of file archive_waldump.c.

◆ SH_KEY

#define SH_KEY fname

Definition at line 81 of file archive_waldump.c.

◆ SH_KEY_TYPE

#define SH_KEY_TYPE const char *

Definition at line 80 of file archive_waldump.c.

◆ SH_PREFIX

#define SH_PREFIX ArchivedWAL

Definition at line 78 of file archive_waldump.c.

◆ SH_RAW_ALLOCATOR

#define SH_RAW_ALLOCATOR pg_malloc0

Definition at line 85 of file archive_waldump.c.

◆ SH_SCOPE

#define SH_SCOPE static inline

Definition at line 84 of file archive_waldump.c.

Typedef Documentation

◆ ArchivedWALFile

typedef struct ArchivedWALFile ArchivedWALFile

◆ astreamer_waldump

typedef struct astreamer_waldump astreamer_waldump

Function Documentation

◆ astreamer_waldump_content()

static void astreamer_waldump_content	(	astreamer *	streamer,
		astreamer_member *	member,
		const char *	data,
		int	len,
		astreamer_archive_context	context
	)

static

Definition at line 679 of file archive_waldump.c.

{
    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;
 
                /* Shouldn't see MEMBER_HEADER in the middle of a file */
                Assert(privateInfo->cur_file == NULL);
 
                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;
                    }
                }
 
                /*
                 * Note: ArchivedWAL_insert may cause existing hash table
                 * entries to move.  While cur_file is known to be NULL right
                 * now, read_archive_wal_page may have a live hash entry
                 * pointer, which it needs to take care to update after
                 * read_archive_file completes.
                 */
                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");
    }
}

References appendBinaryStringInfo(), XLogDumpPrivate::archive_name, XLogDumpPrivate::archive_wal_htab, Assert, ASTREAMER_ARCHIVE_TRAILER, ASTREAMER_MEMBER_CONTENTS, ASTREAMER_MEMBER_HEADER, ASTREAMER_MEMBER_TRAILER, ASTREAMER_UNKNOWN, ArchivedWALFile::buf, XLogDumpPrivate::cur_file, data, XLogDumpPrivate::end_segno, fb(), len, makeStringInfo(), member_is_wal_file(), astreamer_member::pathname, pfree(), pg_fatal, pg_log_debug, pg_log_warning, READ_ANY_WAL, ArchivedWALFile::read_len, XLogDumpPrivate::segsize, ArchivedWALFile::spilled, XLogDumpPrivate::start_segno, XLogDumpPrivate::timeline, and XLogFromFileName().

◆ astreamer_waldump_finalize()

static void astreamer_waldump_finalize ( astreamer * streamer )

static

Definition at line 788 of file archive_waldump.c.

{
    Assert(streamer->bbs_next == NULL);
}

References Assert, astreamer::bbs_next, and fb().

◆ astreamer_waldump_free()

static void astreamer_waldump_free ( astreamer * streamer )

static

Definition at line 797 of file archive_waldump.c.

{
    Assert(streamer->bbs_next == NULL);
    pfree(streamer);
}

References Assert, astreamer::bbs_next, fb(), and pfree().

◆ astreamer_waldump_new()

static astreamer * astreamer_waldump_new ( XLogDumpPrivate * privateInfo )

static

Definition at line 659 of file archive_waldump.c.

{
    astreamer_waldump *streamer;
 
    streamer = palloc0_object(astreamer_waldump);
    *((const astreamer_ops **) &streamer->base.bbs_ops) =
        &astreamer_waldump_ops;
 
    streamer->privateInfo = privateInfo;
 
    return &streamer->base;
}

References astreamer_waldump_ops, astreamer_waldump::base, astreamer::bbs_ops, palloc0_object, and astreamer_waldump::privateInfo.

Referenced by init_archive_reader().

◆ free_archive_reader()

void free_archive_reader ( XLogDumpPrivate * privateInfo )

Definition at line 253 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.  read_archive_file() may have done so.  However,
     * when decoding WAL we can stop once we hit the end LSN, so we may never
     * have read all of the input file.  In that case 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
	)

Definition at line 405 of file archive_waldump.c.

{
    ArchivedWALFile *entry;
    const char *oldfname;
 
    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 may cause other hash table entries to move.
     * Therefore, if cur_file isn't NULL now, we have to be prepared to look
     * that entry up again after the deletion.  Fortunately, the entry's fname
     * string won't move.
     */
    oldfname = privateInfo->cur_file ? privateInfo->cur_file->fname : NULL;
 
    ArchivedWAL_delete_item(privateInfo->archive_wal_htab, entry);
 
    if (oldfname)
    {
        privateInfo->cur_file = ArchivedWAL_lookup(privateInfo->archive_wal_htab,
                                                   oldfname);
        /* ... it had better still be there */
        Assert(privateInfo->cur_file != NULL);
    }
}

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

Referenced by init_archive_reader(), and TarWALDumpReadPage().

◆ get_archive_wal_entry()

static ArchivedWALFile * get_archive_wal_entry	(	const char *	fname,
		XLogDumpPrivate *	privateInfo
	)

static

Definition at line 464 of file archive_waldump.c.

{
    while (1)
    {
        ArchivedWALFile *entry;
        ArchivedWAL_iterator iter;
 
        /*
         * 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; we must invoke the archive streamer to fetch it.
         */
        entry = ArchivedWAL_lookup(privateInfo->archive_wal_htab, fname);
 
        if (entry != NULL)
            return entry;
 
        /*
         * Before loading more data, scan the hash table to see if we have
         * loaded any files we don't need yet.  If so, spill their data to
         * disk to conserve memory space.  But don't try to spill a
         * partially-read file; it's not worth the complication.
         */
        ArchivedWAL_start_iterate(privateInfo->archive_wal_htab, &iter);
        while ((entry = ArchivedWAL_iterate(privateInfo->archive_wal_htab,
                                            &iter)) != NULL)
        {
            FILE       *write_fp;
 
            /* OK to spill? */
            if (entry->spilled)
                continue;       /* already spilled */
            if (entry == privateInfo->cur_file)
                continue;       /* still being read */
 
            /* Write out the completed WAL file contents to a temp file. */
            write_fp = prepare_tmp_write(entry->fname, privateInfo);
            perform_tmp_write(entry->fname, entry->buf, write_fp);
            if (fclose(write_fp) != 0)
                pg_fatal("could not close file \"%s/%s\": %m",
                         TmpWalSegDir, entry->fname);
 
            /* resetStringInfo won't release storage, so delete/recreate. */
            destroyStringInfo(entry->buf);
            entry->buf = makeStringInfo();
            entry->spilled = true;
        }
 
        /*
         * Read more data.  If we reach EOF, the desired file is not present.
         */
        if (!read_archive_file(privateInfo))
            pg_fatal("could not find WAL \"%s\" in archive \"%s\"",
                     fname, privateInfo->archive_name);
    }
}

References XLogDumpPrivate::archive_name, XLogDumpPrivate::archive_wal_htab, ArchivedWALFile::buf, XLogDumpPrivate::cur_file, destroyStringInfo(), fb(), ArchivedWALFile::fname, makeStringInfo(), perform_tmp_write(), pg_fatal, prepare_tmp_write(), read_archive_file(), ArchivedWALFile::spilled, and TmpWalSegDir.

Referenced by read_archive_wal_page().

◆ hash_string_pointer()

static uint32 hash_string_pointer ( const char * s )

static

Definition at line 851 of file archive_waldump.c.

{
    const unsigned char *ss = (const unsigned char *) s;
 
    return hash_bytes(ss, strlen(s));
}

References fb(), and hash_bytes().

◆ init_archive_reader()

void init_archive_reader	(	XLogDumpPrivate *	privateInfo,
		pg_compress_algorithm	compression
	)

Definition at line 129 of file archive_waldump.c.

{
    int         fd;
    astreamer  *streamer;
    ArchivedWALFile *entry = NULL;
    XLogLongPageHeader longhdr;
    ArchivedWAL_iterator iter;
 
    /* 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;
    privateInfo->archive_fd_eof = false;
 
    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 begin content
     * decoding.
     */
    privateInfo->archive_read_buf = pg_malloc(READ_CHUNK_SIZE);
    privateInfo->archive_read_buf_size = READ_CHUNK_SIZE;
 
    /*
     * 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 WAL segment with enough data to extract
     * the WAL segment size from the long page header.
     *
     * We must not rely on cur_file here, because it can become NULL if a
     * member trailer is processed during a read_archive_file() call. Instead,
     * scan the hash table after each read to find any entry with sufficient
     * data.
     */
    while (entry == NULL)
    {
        if (!read_archive_file(privateInfo))
            pg_fatal("could not find WAL in archive \"%s\"",
                     privateInfo->archive_name);
 
        ArchivedWAL_start_iterate(privateInfo->archive_wal_htab, &iter);
        while ((entry = ArchivedWAL_iterate(privateInfo->archive_wal_htab,
                                            &iter)) != NULL)
        {
            if (entry->read_len >= sizeof(XLogLongPageHeaderData))
                break;
        }
    }
 
    /* 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);
 
    /*
     * Now that we have initialized the filtering parameters (start_segno and
     * end_segno), we can discard any already-loaded WAL hash table entries
     * for segments we don't actually need.  Subsequent WAL will be filtered
     * automatically by the archive streamer using the updated start_segno and
     * end_segno values.
     */
    ArchivedWAL_start_iterate(privateInfo->archive_wal_htab, &iter);
    while ((entry = ArchivedWAL_iterate(privateInfo->archive_wal_htab,
                                        &iter)) != NULL)
    {
        XLogSegNo   segno;
        TimeLineID  timeline;
 
        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);
    }
}

Referenced by main().

◆ member_is_wal_file()

static bool member_is_wal_file	(	astreamer_waldump *	mystreamer,
		astreamer_member *	member,
		char **	fname
	)

static

Definition at line 808 of file archive_waldump.c.

{
    int         pathlen;
    char        pathname[MAXPGPATH];
    char       *filename;
 
    /* We are only interested in normal files */
    if (!member->is_regular)
        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;
}

References canonicalize_path(), fb(), filename, astreamer_member::is_regular, IsXLogFileName(), MAXPGPATH, astreamer_member::pathname, pnstrdup(), snprintf, XLOG_FNAME_LEN, and XLOGDIR.

Referenced by astreamer_waldump_content().

◆ perform_tmp_write()

static void perform_tmp_write	(	const char *	fname,
		StringInfo	buf,
		FILE *	file
	)

static

Definition at line 639 of file archive_waldump.c.

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

References Assert, buf, fb(), pg_fatal, and TmpWalSegDir.

Referenced by get_archive_wal_entry().

◆ prepare_tmp_write()

static FILE * prepare_tmp_write	(	const char *	fname,
		XLogDumpPrivate *	privateInfo
	)

static

Definition at line 608 of file archive_waldump.c.

{
    char        fpath[MAXPGPATH];
    FILE       *file;
 
    /* Setup temporary directory to store WAL segments, if we didn't already */
    if (unlikely(TmpWalSegDir == NULL))
        setup_tmpwal_dir(privateInfo->archive_dir);
 
    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;
}

References XLogDumpPrivate::archive_dir, fb(), MAXPGPATH, PG_BINARY_W, pg_fatal, pg_file_create_mode, pg_log_debug, setup_tmpwal_dir(), snprintf, TmpWalSegDir, and unlikely.

Referenced by get_archive_wal_entry().

◆ read_archive_file()

static bool read_archive_file ( XLogDumpPrivate * privateInfo )

static

Definition at line 536 of file archive_waldump.c.

{
    int         rc;
 
    /* Fail if we already reached EOF in a prior call. */
    if (privateInfo->archive_fd_eof)
        return false;
 
    /* Try to read some more data. */
    rc = read(privateInfo->archive_fd, privateInfo->archive_read_buf,
              privateInfo->archive_read_buf_size);
    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);
    else
    {
        /*
         * We reached EOF, but there is probably still data queued in the
         * astreamer pipeline's buffers.  Flush it out to ensure that we
         * process everything.
         */
        astreamer_finalize(privateInfo->archive_streamer);
        /* Set flag to ensure we don't finalize more than once. */
        privateInfo->archive_fd_eof = true;
    }
 
    return true;
}

References XLogDumpPrivate::archive_fd, XLogDumpPrivate::archive_fd_eof, XLogDumpPrivate::archive_name, XLogDumpPrivate::archive_read_buf, XLogDumpPrivate::archive_read_buf_size, XLogDumpPrivate::archive_streamer, astreamer_content(), astreamer_finalize(), ASTREAMER_UNKNOWN, fb(), pg_fatal, and read.

Referenced by get_archive_wal_entry(), init_archive_reader(), and read_archive_wal_page().

◆ read_archive_wal_page()

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

Definition at line 301 of file archive_waldump.c.

{
    char       *p = readBuff;
    size_t      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);
    Assert(!entry->spilled);
 
    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, 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
        {
            /*
             * We evidently need to 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 %zu of %zu bytes",
                         fname, privateInfo->archive_name,
                         (count - nbytes), count);
            if (!read_archive_file(privateInfo))
                pg_fatal("unexpected end of archive \"%s\" while reading \"%s\": read %zu of %zu bytes",
                         privateInfo->archive_name, fname,
                         (count - nbytes), count);
 
            /*
             * Loading more data may have moved hash table entries, so we must
             * re-look-up the one we are reading from.
             */
            entry = ArchivedWAL_lookup(privateInfo->archive_wal_htab, fname);
            /* ... it had better still be there */
            Assert(entry != NULL);
        }
    }
 
    /*
     * 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 XLogDumpPrivate::archive_name, XLogDumpPrivate::archive_wal_htab, Assert, buf, ArchivedWALFile::buf, XLogDumpPrivate::cur_file, StringInfoData::data, fb(), get_archive_wal_entry(), StringInfoData::len, MAXFNAMELEN, memcpy(), Min, pg_fatal, read_archive_file(), ArchivedWALFile::read_len, XLogDumpPrivate::segsize, ArchivedWALFile::spilled, XLogDumpPrivate::timeline, XLByteToSeg, XLogFileName(), and XLogSegNoOffsetToRecPtr.

Referenced by TarWALDumpReadPage().

◆ setup_tmpwal_dir()

static void setup_tmpwal_dir ( const char * waldir )

static

Definition at line 578 of file archive_waldump.c.

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

References Assert, canonicalize_path(), fb(), mkdtemp(), pg_fatal, pg_log_debug, psprintf(), and TmpWalSegDir.

Referenced by prepare_tmp_write().

Variable Documentation

◆ astreamer_waldump_ops

const astreamer_ops astreamer_waldump_ops

static

Initial value:

= {
    .content = astreamer_waldump_content,
    .finalize = astreamer_waldump_finalize,
    .free = astreamer_waldump_free
}

Definition at line 116 of file archive_waldump.c.

                                                   {
    .content = astreamer_waldump_content,
    .finalize = astreamer_waldump_finalize,
    .free = astreamer_waldump_free
};

Referenced by astreamer_waldump_new().

◆ TmpWalSegDir

char* TmpWalSegDir = NULL

Definition at line 32 of file archive_waldump.c.

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

Data Structures

Macros

Typedefs

Functions

Variables

Macro Definition Documentation

◆ READ_ANY_WAL

◆ READ_CHUNK_SIZE

◆ SH_DECLARE

◆ SH_DEFINE

◆ SH_ELEMENT_TYPE

◆ SH_EQUAL

◆ SH_HASH_KEY

◆ SH_KEY

◆ SH_KEY_TYPE

◆ SH_PREFIX

◆ SH_RAW_ALLOCATOR

◆ SH_SCOPE

Typedef Documentation

◆ ArchivedWALFile

◆ astreamer_waldump

Function Documentation

◆ astreamer_waldump_content()

◆ astreamer_waldump_finalize()

◆ astreamer_waldump_free()

◆ astreamer_waldump_new()

◆ free_archive_reader()

◆ free_archive_wal_entry()

◆ get_archive_wal_entry()

◆ hash_string_pointer()

◆ init_archive_reader()

◆ member_is_wal_file()

◆ perform_tmp_write()

◆ prepare_tmp_write()

◆ read_archive_file()

◆ read_archive_wal_page()

◆ setup_tmpwal_dir()

Variable Documentation

◆ astreamer_waldump_ops

◆ TmpWalSegDir