fetch.h File Reference

#include "access/xlogdefs.h"
#include "filemap.h"

Include dependency graph for fetch.h:

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

Go to the source code of this file.

Typedefs
typedef void(*	process_file_callback_t) (const char *path, file_type_t type, size_t size, const char *link_target)

Functions
void	fetchSourceFileList (void)
char *	fetchFile (const char *filename, size_t *filesize)
void	executeFileMap (void)
void	libpqProcessFileList (void)
char *	libpqGetFile (const char *filename, size_t *filesize)
void	libpq_executeFileMap (filemap_t *map)
void	libpqConnect (const char *connstr)
XLogRecPtr	libpqGetCurrentXlogInsertLocation (void)
void	copy_executeFileMap (filemap_t *map)
void	traverse_datadir (const char *datadir, process_file_callback_t callback)

Typedef Documentation

process_file_callback_t

typedef void(* process_file_callback_t) (const char *path, file_type_t type, size_t size, const char *link_target)

Definition at line 41 of file fetch.h.

Function Documentation

copy_executeFileMap()

void copy_executeFileMap

(

filemap_t *

map

)

Definition at line 206 of file copy_fetch.c.

References file_entry_t::action, filemap_t::array, close_target_file(), create_target(), execute_pagemap(), FILE_ACTION_COPY, FILE_ACTION_COPY_TAIL, FILE_ACTION_CREATE, FILE_ACTION_NONE, FILE_ACTION_REMOVE, FILE_ACTION_TRUNCATE, i, filemap_t::narray, file_entry_t::newsize, file_entry_t::oldsize, file_entry_t::pagemap, file_entry_t::path, remove_target(), rewind_copy_file_range(), and truncate_target_file().

Referenced by executeFileMap().

{
    file_entry_t *entry;
    int         i;

    for (i = 0; i < map->narray; i++)
    {
        entry = map->array[i];
        execute_pagemap(&entry->pagemap, entry->path);

        switch (entry->action)
        {
            case FILE_ACTION_NONE:
                /* ok, do nothing.. */
                break;

            case FILE_ACTION_COPY:
                rewind_copy_file_range(entry->path, 0, entry->newsize, true);
                break;

            case FILE_ACTION_TRUNCATE:
                truncate_target_file(entry->path, entry->newsize);
                break;

            case FILE_ACTION_COPY_TAIL:
                rewind_copy_file_range(entry->path, entry->oldsize,
                                       entry->newsize, false);
                break;

            case FILE_ACTION_CREATE:
                create_target(entry);
                break;

            case FILE_ACTION_REMOVE:
                remove_target(entry);
                break;
        }
    }

    close_target_file();
}

executeFileMap()

void executeFileMap

(

void

)

Definition at line 40 of file fetch.c.

References copy_executeFileMap(), datadir_source, filemap, and libpq_executeFileMap().

Referenced by main().

{
    if (datadir_source)
        copy_executeFileMap(filemap);
    else
        libpq_executeFileMap(filemap);
}

fetchFile()

char* fetchFile	(	const char *	filename,
		size_t *	filesize
	)

Definition at line 54 of file fetch.c.

References datadir_source, libpqGetFile(), and slurpFile().

Referenced by getTimelineHistory(), and main().

{
    if (datadir_source)
        return slurpFile(datadir_source, filename, filesize);
    else
        return libpqGetFile(filename, filesize);
}

fetchSourceFileList()

void fetchSourceFileList

(

void

)

Definition at line 28 of file fetch.c.

References datadir_source, libpqProcessFileList(), process_source_file(), and traverse_datadir().

Referenced by main().

{
    if (datadir_source)
        traverse_datadir(datadir_source, &process_source_file);
    else
        libpqProcessFileList();
}

libpq_executeFileMap()

void libpq_executeFileMap

(

filemap_t *

map

)

Definition at line 425 of file libpq_fetch.c.

References file_entry_t::action, filemap_t::array, create_target(), execute_pagemap(), fetch_file_range(), FILE_ACTION_COPY, FILE_ACTION_COPY_TAIL, FILE_ACTION_CREATE, FILE_ACTION_NONE, FILE_ACTION_REMOVE, FILE_ACTION_TRUNCATE, i, filemap_t::narray, file_entry_t::newsize, file_entry_t::oldsize, open_target_file(), file_entry_t::pagemap, file_entry_t::path, pg_fatal(), PGRES_COMMAND_OK, PGRES_COPY_IN, PQclear(), PQerrorMessage(), PQexec(), PQgetResult(), PQputCopyEnd(), PQresultErrorMessage(), PQresultStatus(), receiveFileChunks(), remove_target(), and truncate_target_file().

Referenced by executeFileMap().

{
    file_entry_t *entry;
    const char *sql;
    PGresult   *res;
    int         i;

    /*
     * First create a temporary table, and load it with the blocks that we
     * need to fetch.
     */
    sql = "CREATE TEMPORARY TABLE fetchchunks(path text, begin int8, len int4);";
    res = PQexec(conn, sql);

    if (PQresultStatus(res) != PGRES_COMMAND_OK)
        pg_fatal("could not create temporary table: %s",
                 PQresultErrorMessage(res));
    PQclear(res);

    sql = "COPY fetchchunks FROM STDIN";
    res = PQexec(conn, sql);

    if (PQresultStatus(res) != PGRES_COPY_IN)
        pg_fatal("could not send file list: %s",
                 PQresultErrorMessage(res));
    PQclear(res);

    for (i = 0; i < map->narray; i++)
    {
        entry = map->array[i];

        /* If this is a relation file, copy the modified blocks */
        execute_pagemap(&entry->pagemap, entry->path);

        switch (entry->action)
        {
            case FILE_ACTION_NONE:
                /* nothing else to do */
                break;

            case FILE_ACTION_COPY:
                /* Truncate the old file out of the way, if any */
                open_target_file(entry->path, true);
                fetch_file_range(entry->path, 0, entry->newsize);
                break;

            case FILE_ACTION_TRUNCATE:
                truncate_target_file(entry->path, entry->newsize);
                break;

            case FILE_ACTION_COPY_TAIL:
                fetch_file_range(entry->path, entry->oldsize, entry->newsize);
                break;

            case FILE_ACTION_REMOVE:
                remove_target(entry);
                break;

            case FILE_ACTION_CREATE:
                create_target(entry);
                break;
        }
    }

    if (PQputCopyEnd(conn, NULL) != 1)
        pg_fatal("could not send end-of-COPY: %s",
                 PQerrorMessage(conn));

    while ((res = PQgetResult(conn)) != NULL)
    {
        if (PQresultStatus(res) != PGRES_COMMAND_OK)
            pg_fatal("unexpected result while sending file list: %s",
                     PQresultErrorMessage(res));
        PQclear(res);
    }

    /*
     * We've now copied the list of file ranges that we need to fetch to the
     * temporary table. Now, actually fetch all of those ranges.
     */
    sql =
        "SELECT path, begin,\n"
        "  pg_read_binary_file(path, begin, len, true) AS chunk\n"
        "FROM fetchchunks\n";

    receiveFileChunks(sql);
}

libpqConnect()

void libpqConnect

(

const char *

connstr

)

Definition at line 45 of file libpq_fetch.c.

References ALWAYS_SECURE_SEARCH_PATH_SQL, CONNECTION_BAD, pg_fatal(), pg_free(), pg_log(), PG_PROGRESS, PGRES_COMMAND_OK, PGRES_TUPLES_OK, PQclear(), PQconnectdb(), PQerrorMessage(), PQexec(), PQresultErrorMessage(), PQresultStatus(), PQstatus(), run_simple_query(), and generate_unaccent_rules::str.

Referenced by main().

{
    char       *str;
    PGresult   *res;

    conn = PQconnectdb(connstr);
    if (PQstatus(conn) == CONNECTION_BAD)
        pg_fatal("could not connect to server: %s",
                 PQerrorMessage(conn));

    pg_log(PG_PROGRESS, "connected to server\n");

    res = PQexec(conn, ALWAYS_SECURE_SEARCH_PATH_SQL);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
        pg_fatal("could not clear search_path: %s",
                 PQresultErrorMessage(res));
    PQclear(res);

    /*
     * Check that the server is not in hot standby mode. There is no
     * fundamental reason that couldn't be made to work, but it doesn't
     * currently because we use a temporary table. Better to check for it
     * explicitly than error out, for a better error message.
     */
    str = run_simple_query("SELECT pg_is_in_recovery()");
    if (strcmp(str, "f") != 0)
        pg_fatal("source server must not be in recovery mode\n");
    pg_free(str);

    /*
     * Also check that full_page_writes is enabled.  We can get torn pages if
     * a page is modified while we read it with pg_read_binary_file(), and we
     * rely on full page images to fix them.
     */
    str = run_simple_query("SHOW full_page_writes");
    if (strcmp(str, "on") != 0)
        pg_fatal("full_page_writes must be enabled in the source server\n");
    pg_free(str);

    /*
     * Although we don't do any "real" updates, we do work with a temporary
     * table. We don't care about synchronous commit for that. It doesn't
     * otherwise matter much, but if the server is using synchronous
     * replication, and replication isn't working for some reason, we don't
     * want to get stuck, waiting for it to start working again.
     */
    res = PQexec(conn, "SET synchronous_commit = off");
    if (PQresultStatus(res) != PGRES_COMMAND_OK)
        pg_fatal("could not set up connection context: %s",
                 PQresultErrorMessage(res));
    PQclear(res);
}

libpqGetCurrentXlogInsertLocation()

XLogRecPtr libpqGetCurrentXlogInsertLocation

(

void

)

Definition at line 129 of file libpq_fetch.c.

References pg_fatal(), pg_free(), run_simple_query(), and val.

Referenced by main().

{
    XLogRecPtr  result;
    uint32      hi;
    uint32      lo;
    char       *val;

    val = run_simple_query("SELECT pg_current_wal_insert_lsn()");

    if (sscanf(val, "%X/%X", &hi, &lo) != 2)
        pg_fatal("unrecognized result \"%s\" for current WAL insert location\n", val);

    result = ((uint64) hi) << 32 | lo;

    pg_free(val);

    return result;
}

libpqGetFile()

char* libpqGetFile	(	const char *	filename,
		size_t *	filesize
	)

Definition at line 353 of file libpq_fetch.c.

References filename, PG_DEBUG, pg_fatal(), pg_log(), pg_malloc(), PGRES_TUPLES_OK, PQclear(), PQexecParams(), PQgetisnull(), PQgetlength(), PQgetvalue(), PQntuples(), PQresultErrorMessage(), and PQresultStatus().

Referenced by fetchFile().

{
    PGresult   *res;
    char       *result;
    int         len;
    const char *paramValues[1];

    paramValues[0] = filename;
    res = PQexecParams(conn, "SELECT pg_read_binary_file($1)",
                       1, NULL, paramValues, NULL, NULL, 1);

    if (PQresultStatus(res) != PGRES_TUPLES_OK)
        pg_fatal("could not fetch remote file \"%s\": %s",
                 filename, PQresultErrorMessage(res));

    /* sanity check the result set */
    if (PQntuples(res) != 1 || PQgetisnull(res, 0, 0))
        pg_fatal("unexpected result set while fetching remote file \"%s\"\n",
                 filename);

    /* Read result to local variables */
    len = PQgetlength(res, 0, 0);
    result = pg_malloc(len + 1);
    memcpy(result, PQgetvalue(res, 0, 0), len);
    result[len] = '\0';

    PQclear(res);

    pg_log(PG_DEBUG, "fetched file \"%s\", length %d\n", filename, len);

    if (filesize)
        *filesize = len;
    return result;
}

libpqProcessFileList()

void libpqProcessFileList

(

void

)

Definition at line 152 of file libpq_fetch.c.

References FILE_TYPE_DIRECTORY, FILE_TYPE_REGULAR, FILE_TYPE_SYMLINK, i, pg_fatal(), PGRES_TUPLES_OK, PQclear(), PQexec(), PQgetisnull(), PQgetvalue(), PQnfields(), PQntuples(), PQresultErrorMessage(), PQresultStatus(), process_source_file(), and generate_unaccent_rules::type.

Referenced by fetchSourceFileList().

{
    PGresult   *res;
    const char *sql;
    int         i;

    /*
     * Create a recursive directory listing of the whole data directory.
     *
     * The WITH RECURSIVE part does most of the work. The second part gets the
     * targets of the symlinks in pg_tblspc directory.
     *
     * XXX: There is no backend function to get a symbolic link's target in
     * general, so if the admin has put any custom symbolic links in the data
     * directory, they won't be copied correctly.
     */
    sql =
        "WITH RECURSIVE files (path, filename, size, isdir) AS (\n"
        "  SELECT '' AS path, filename, size, isdir FROM\n"
        "  (SELECT pg_ls_dir('.', true, false) AS filename) AS fn,\n"
        "        pg_stat_file(fn.filename, true) AS this\n"
        "  UNION ALL\n"
        "  SELECT parent.path || parent.filename || '/' AS path,\n"
        "         fn, this.size, this.isdir\n"
        "  FROM files AS parent,\n"
        "       pg_ls_dir(parent.path || parent.filename, true, false) AS fn,\n"
        "       pg_stat_file(parent.path || parent.filename || '/' || fn, true) AS this\n"
        "       WHERE parent.isdir = 't'\n"
        ")\n"
        "SELECT path || filename, size, isdir,\n"
        "       pg_tablespace_location(pg_tablespace.oid) AS link_target\n"
        "FROM files\n"
        "LEFT OUTER JOIN pg_tablespace ON files.path = 'pg_tblspc/'\n"
        "                             AND oid::text = files.filename\n";
    res = PQexec(conn, sql);

    if (PQresultStatus(res) != PGRES_TUPLES_OK)
        pg_fatal("could not fetch file list: %s",
                 PQresultErrorMessage(res));

    /* sanity check the result set */
    if (PQnfields(res) != 4)
        pg_fatal("unexpected result set while fetching file list\n");

    /* Read result to local variables */
    for (i = 0; i < PQntuples(res); i++)
    {
        char       *path = PQgetvalue(res, i, 0);
        int64       filesize = atol(PQgetvalue(res, i, 1));
        bool        isdir = (strcmp(PQgetvalue(res, i, 2), "t") == 0);
        char       *link_target = PQgetvalue(res, i, 3);
        file_type_t type;

        if (PQgetisnull(res, 0, 1))
        {
            /*
             * The file was removed from the server while the query was
             * running. Ignore it.
             */
            continue;
        }

        if (link_target[0])
            type = FILE_TYPE_SYMLINK;
        else if (isdir)
            type = FILE_TYPE_DIRECTORY;
        else
            type = FILE_TYPE_REGULAR;

        process_source_file(path, type, filesize, link_target);
    }
    PQclear(res);
}

traverse_datadir()

void traverse_datadir	(	const char *	datadir,
		process_file_callback_t	callback
	)

Definition at line 34 of file copy_fetch.c.

References recurse_dir().

Referenced by fetchSourceFileList(), and main().

{
    recurse_dir(datadir, NULL, callback);
}