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

Include dependency graph for fetch.h:

This graph shows which files directly or indirectly include 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 (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

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

void copy_executeFileMap ( filemap_t * map )

Definition at line 208 of file copy_fetch.c.

References file_entry_t::action, filemap_t::array, close_target_file(), copy_file_range(), 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(), 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:
                 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:
                 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();
 }

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

char* fetchFile	(	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);
 }

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

void libpq_executeFileMap ( filemap_t * map )

Definition at line 434 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, NULL, 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);
 }

void libpqConnect ( const char * connstr )

Definition at line 48 of file libpq_fetch.c.

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

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

XLogRecPtr libpqGetCurrentXlogInsertLocation ( void )

Definition at line 126 of file libpq_fetch.c.

References pg_fatal(), pg_free(), result, 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;
 }

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

Definition at line 362 of file libpq_fetch.c.

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

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

void libpqProcessFileList ( void )

Definition at line 149 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(), and process_source_file().

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

void traverse_datadir	(	const char *	datadir,
		process_file_callback_t	callback
	)

Definition at line 36 of file copy_fetch.c.

References NULL, and recurse_dir().

Referenced by fetchSourceFileList(), and main().

 {
     recurse_dir(datadir, NULL, callback);
 }

Typedefs

Functions

Typedef Documentation

Function Documentation