LCOV - code coverage report
Current view: top level - src/bin/pg_rewind - libpq_source.c (source / functions) Hit Total Coverage
Test: PostgreSQL 18devel Lines: 194 228 85.1 %
Date: 2024-12-02 20:15:07 Functions: 13 13 100.0 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * libpq_source.c
       4             :  *    Functions for fetching files from a remote server via libpq.
       5             :  *
       6             :  * Copyright (c) 2013-2024, PostgreSQL Global Development Group
       7             :  *
       8             :  *-------------------------------------------------------------------------
       9             :  */
      10             : #include "postgres_fe.h"
      11             : 
      12             : #include "catalog/pg_type_d.h"
      13             : #include "common/connect.h"
      14             : #include "file_ops.h"
      15             : #include "filemap.h"
      16             : #include "lib/stringinfo.h"
      17             : #include "pg_rewind.h"
      18             : #include "port/pg_bswap.h"
      19             : #include "rewind_source.h"
      20             : 
      21             : /*
      22             :  * Files are fetched MAX_CHUNK_SIZE bytes at a time, and with a
      23             :  * maximum of MAX_CHUNKS_PER_QUERY chunks in a single query.
      24             :  */
      25             : #define MAX_CHUNK_SIZE (1024 * 1024)
      26             : #define MAX_CHUNKS_PER_QUERY 1000
      27             : 
      28             : /* represents a request to fetch a piece of a file from the source */
      29             : typedef struct
      30             : {
      31             :     const char *path;           /* path relative to data directory root */
      32             :     off_t       offset;
      33             :     size_t      length;
      34             : } fetch_range_request;
      35             : 
      36             : typedef struct
      37             : {
      38             :     rewind_source common;       /* common interface functions */
      39             : 
      40             :     PGconn     *conn;
      41             : 
      42             :     /*
      43             :      * Queue of chunks that have been requested with the queue_fetch_range()
      44             :      * function, but have not been fetched from the remote server yet.
      45             :      */
      46             :     int         num_requests;
      47             :     fetch_range_request request_queue[MAX_CHUNKS_PER_QUERY];
      48             : 
      49             :     /* temporary space for process_queued_fetch_requests() */
      50             :     StringInfoData paths;
      51             :     StringInfoData offsets;
      52             :     StringInfoData lengths;
      53             : } libpq_source;
      54             : 
      55             : static void init_libpq_conn(PGconn *conn);
      56             : static char *run_simple_query(PGconn *conn, const char *sql);
      57             : static void run_simple_command(PGconn *conn, const char *sql);
      58             : static void appendArrayEscapedString(StringInfo buf, const char *str);
      59             : 
      60             : static void process_queued_fetch_requests(libpq_source *src);
      61             : 
      62             : /* public interface functions */
      63             : static void libpq_traverse_files(rewind_source *source,
      64             :                                  process_file_callback_t callback);
      65             : static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
      66             : static void libpq_queue_fetch_range(rewind_source *source, const char *path,
      67             :                                     off_t off, size_t len);
      68             : static void libpq_finish_fetch(rewind_source *source);
      69             : static char *libpq_fetch_file(rewind_source *source, const char *path,
      70             :                               size_t *filesize);
      71             : static XLogRecPtr libpq_get_current_wal_insert_lsn(rewind_source *source);
      72             : static void libpq_destroy(rewind_source *source);
      73             : 
      74             : /*
      75             :  * Create a new libpq source.
      76             :  *
      77             :  * The caller has already established the connection, but should not try
      78             :  * to use it while the source is active.
      79             :  */
      80             : rewind_source *
      81          12 : init_libpq_source(PGconn *conn)
      82             : {
      83             :     libpq_source *src;
      84             : 
      85          12 :     init_libpq_conn(conn);
      86             : 
      87          12 :     src = pg_malloc0(sizeof(libpq_source));
      88             : 
      89          12 :     src->common.traverse_files = libpq_traverse_files;
      90          12 :     src->common.fetch_file = libpq_fetch_file;
      91          12 :     src->common.queue_fetch_file = libpq_queue_fetch_file;
      92          12 :     src->common.queue_fetch_range = libpq_queue_fetch_range;
      93          12 :     src->common.finish_fetch = libpq_finish_fetch;
      94          12 :     src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
      95          12 :     src->common.destroy = libpq_destroy;
      96             : 
      97          12 :     src->conn = conn;
      98             : 
      99          12 :     initStringInfo(&src->paths);
     100          12 :     initStringInfo(&src->offsets);
     101          12 :     initStringInfo(&src->lengths);
     102             : 
     103          12 :     return &src->common;
     104             : }
     105             : 
     106             : /*
     107             :  * Initialize a libpq connection for use.
     108             :  */
     109             : static void
     110          12 : init_libpq_conn(PGconn *conn)
     111             : {
     112             :     PGresult   *res;
     113             :     char       *str;
     114             : 
     115             :     /* disable all types of timeouts */
     116          12 :     run_simple_command(conn, "SET statement_timeout = 0");
     117          12 :     run_simple_command(conn, "SET lock_timeout = 0");
     118          12 :     run_simple_command(conn, "SET idle_in_transaction_session_timeout = 0");
     119          12 :     run_simple_command(conn, "SET transaction_timeout = 0");
     120             : 
     121             :     /*
     122             :      * we don't intend to do any updates, put the connection in read-only mode
     123             :      * to keep us honest
     124             :      */
     125          12 :     run_simple_command(conn, "SET default_transaction_read_only = on");
     126             : 
     127             :     /* secure search_path */
     128          12 :     res = PQexec(conn, ALWAYS_SECURE_SEARCH_PATH_SQL);
     129          12 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
     130           0 :         pg_fatal("could not clear \"search_path\": %s",
     131             :                  PQresultErrorMessage(res));
     132          12 :     PQclear(res);
     133             : 
     134             :     /*
     135             :      * Also check that full_page_writes is enabled.  We can get torn pages if
     136             :      * a page is modified while we read it with pg_read_binary_file(), and we
     137             :      * rely on full page images to fix them.
     138             :      */
     139          12 :     str = run_simple_query(conn, "SHOW full_page_writes");
     140          12 :     if (strcmp(str, "on") != 0)
     141           0 :         pg_fatal("\"full_page_writes\" must be enabled in the source server");
     142          12 :     pg_free(str);
     143             : 
     144             :     /* Prepare a statement we'll use to fetch files */
     145          12 :     res = PQprepare(conn, "fetch_chunks_stmt",
     146             :                     "SELECT path, begin,\n"
     147             :                     "  pg_read_binary_file(path, begin, len, true) AS chunk\n"
     148             :                     "FROM unnest ($1::text[], $2::int8[], $3::int4[]) as x(path, begin, len)",
     149             :                     3, NULL);
     150             : 
     151          12 :     if (PQresultStatus(res) != PGRES_COMMAND_OK)
     152           0 :         pg_fatal("could not prepare statement to fetch file contents: %s",
     153             :                  PQresultErrorMessage(res));
     154          12 :     PQclear(res);
     155          12 : }
     156             : 
     157             : /*
     158             :  * Run a query that returns a single value.
     159             :  *
     160             :  * The result should be pg_free'd after use.
     161             :  */
     162             : static char *
     163          22 : run_simple_query(PGconn *conn, const char *sql)
     164             : {
     165             :     PGresult   *res;
     166             :     char       *result;
     167             : 
     168          22 :     res = PQexec(conn, sql);
     169             : 
     170          22 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
     171           0 :         pg_fatal("error running query (%s) on source server: %s",
     172             :                  sql, PQresultErrorMessage(res));
     173             : 
     174             :     /* sanity check the result set */
     175          22 :     if (PQnfields(res) != 1 || PQntuples(res) != 1 || PQgetisnull(res, 0, 0))
     176           0 :         pg_fatal("unexpected result set from query");
     177             : 
     178          22 :     result = pg_strdup(PQgetvalue(res, 0, 0));
     179             : 
     180          22 :     PQclear(res);
     181             : 
     182          22 :     return result;
     183             : }
     184             : 
     185             : /*
     186             :  * Run a command.
     187             :  *
     188             :  * In the event of a failure, exit immediately.
     189             :  */
     190             : static void
     191          60 : run_simple_command(PGconn *conn, const char *sql)
     192             : {
     193             :     PGresult   *res;
     194             : 
     195          60 :     res = PQexec(conn, sql);
     196             : 
     197          60 :     if (PQresultStatus(res) != PGRES_COMMAND_OK)
     198           0 :         pg_fatal("error running query (%s) in source server: %s",
     199             :                  sql, PQresultErrorMessage(res));
     200             : 
     201          60 :     PQclear(res);
     202          60 : }
     203             : 
     204             : /*
     205             :  * Call the pg_current_wal_insert_lsn() function in the remote system.
     206             :  */
     207             : static XLogRecPtr
     208          10 : libpq_get_current_wal_insert_lsn(rewind_source *source)
     209             : {
     210          10 :     PGconn     *conn = ((libpq_source *) source)->conn;
     211             :     XLogRecPtr  result;
     212             :     uint32      hi;
     213             :     uint32      lo;
     214             :     char       *val;
     215             : 
     216          10 :     val = run_simple_query(conn, "SELECT pg_current_wal_insert_lsn()");
     217             : 
     218          10 :     if (sscanf(val, "%X/%X", &hi, &lo) != 2)
     219           0 :         pg_fatal("unrecognized result \"%s\" for current WAL insert location", val);
     220             : 
     221          10 :     result = ((uint64) hi) << 32 | lo;
     222             : 
     223          10 :     pg_free(val);
     224             : 
     225          10 :     return result;
     226             : }
     227             : 
     228             : /*
     229             :  * Get a list of all files in the data directory.
     230             :  */
     231             : static void
     232          12 : libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
     233             : {
     234          12 :     PGconn     *conn = ((libpq_source *) source)->conn;
     235             :     PGresult   *res;
     236             :     const char *sql;
     237             :     int         i;
     238             : 
     239             :     /*
     240             :      * Create a recursive directory listing of the whole data directory.
     241             :      *
     242             :      * The WITH RECURSIVE part does most of the work. The second part gets the
     243             :      * targets of the symlinks in pg_tblspc directory.
     244             :      *
     245             :      * XXX: There is no backend function to get a symbolic link's target in
     246             :      * general, so if the admin has put any custom symbolic links in the data
     247             :      * directory, they won't be copied correctly.
     248             :      */
     249          12 :     sql =
     250             :         "WITH RECURSIVE files (path, filename, size, isdir) AS (\n"
     251             :         "  SELECT '' AS path, filename, size, isdir FROM\n"
     252             :         "  (SELECT pg_ls_dir('.', true, false) AS filename) AS fn,\n"
     253             :         "        pg_stat_file(fn.filename, true) AS this\n"
     254             :         "  UNION ALL\n"
     255             :         "  SELECT parent.path || parent.filename || '/' AS path,\n"
     256             :         "         fn, this.size, this.isdir\n"
     257             :         "  FROM files AS parent,\n"
     258             :         "       pg_ls_dir(parent.path || parent.filename, true, false) AS fn,\n"
     259             :         "       pg_stat_file(parent.path || parent.filename || '/' || fn, true) AS this\n"
     260             :         "       WHERE parent.isdir = 't'\n"
     261             :         ")\n"
     262             :         "SELECT path || filename, size, isdir,\n"
     263             :         "       pg_tablespace_location(pg_tablespace.oid) AS link_target\n"
     264             :         "FROM files\n"
     265             :         "LEFT OUTER JOIN pg_tablespace ON files.path = 'pg_tblspc/'\n"
     266             :         "                             AND oid::text = files.filename\n";
     267          12 :     res = PQexec(conn, sql);
     268             : 
     269          12 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
     270           0 :         pg_fatal("could not fetch file list: %s",
     271             :                  PQresultErrorMessage(res));
     272             : 
     273             :     /* sanity check the result set */
     274          12 :     if (PQnfields(res) != 4)
     275           0 :         pg_fatal("unexpected result set while fetching file list");
     276             : 
     277             :     /* Read result to local variables */
     278       13990 :     for (i = 0; i < PQntuples(res); i++)
     279             :     {
     280             :         char       *path;
     281             :         int64       filesize;
     282             :         bool        isdir;
     283             :         char       *link_target;
     284             :         file_type_t type;
     285             : 
     286       13978 :         if (PQgetisnull(res, i, 1))
     287             :         {
     288             :             /*
     289             :              * The file was removed from the server while the query was
     290             :              * running. Ignore it.
     291             :              */
     292           0 :             continue;
     293             :         }
     294             : 
     295       13978 :         path = PQgetvalue(res, i, 0);
     296       13978 :         filesize = atoll(PQgetvalue(res, i, 1));
     297       13978 :         isdir = (strcmp(PQgetvalue(res, i, 2), "t") == 0);
     298       13978 :         link_target = PQgetvalue(res, i, 3);
     299             : 
     300       13978 :         if (link_target[0])
     301             :         {
     302             :             /*
     303             :              * In-place tablespaces are directories located in pg_tblspc/ with
     304             :              * relative paths.
     305             :              */
     306           2 :             if (is_absolute_path(link_target))
     307           0 :                 type = FILE_TYPE_SYMLINK;
     308             :             else
     309           2 :                 type = FILE_TYPE_DIRECTORY;
     310             :         }
     311       13976 :         else if (isdir)
     312         334 :             type = FILE_TYPE_DIRECTORY;
     313             :         else
     314       13642 :             type = FILE_TYPE_REGULAR;
     315             : 
     316       13978 :         callback(path, type, filesize, link_target);
     317             :     }
     318          12 :     PQclear(res);
     319          12 : }
     320             : 
     321             : /*
     322             :  * Queue up a request to fetch a file from remote system.
     323             :  */
     324             : static void
     325        3918 : libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
     326             : {
     327             :     /*
     328             :      * Truncate the target file immediately, and queue a request to fetch it
     329             :      * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
     330             :      * request fetching a full-sized chunk anyway, so that if the file has
     331             :      * become larger in the source system, after we scanned the source
     332             :      * directory, we still fetch the whole file. This only works for files up
     333             :      * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
     334             :      * and such that are changed every now and then, but not WAL-logged. For
     335             :      * larger files, we fetch up to the original size.
     336             :      *
     337             :      * Even with that mechanism, there is an inherent race condition if the
     338             :      * file is modified at the same instant that we're copying it, so that we
     339             :      * might copy a torn version of the file with one half from the old
     340             :      * version and another half from the new. But pg_basebackup has the same
     341             :      * problem, and it hasn't been a problem in practice.
     342             :      *
     343             :      * It might seem more natural to truncate the file later, when we receive
     344             :      * it from the source server, but then we'd need to track which
     345             :      * fetch-requests are for a whole file.
     346             :      */
     347        3918 :     open_target_file(path, true);
     348        3918 :     libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
     349        3918 : }
     350             : 
     351             : /*
     352             :  * Queue up a request to fetch a piece of a file from remote system.
     353             :  */
     354             : static void
     355        5546 : libpq_queue_fetch_range(rewind_source *source, const char *path, off_t off,
     356             :                         size_t len)
     357             : {
     358        5546 :     libpq_source *src = (libpq_source *) source;
     359             : 
     360             :     /*
     361             :      * Does this request happen to be a continuation of the previous chunk? If
     362             :      * so, merge it with the previous one.
     363             :      *
     364             :      * XXX: We use pointer equality to compare the path. That's good enough
     365             :      * for our purposes; the caller always passes the same pointer for the
     366             :      * same filename. If it didn't, we would fail to merge requests, but it
     367             :      * wouldn't affect correctness.
     368             :      */
     369        5546 :     if (src->num_requests > 0)
     370             :     {
     371        5534 :         fetch_range_request *prev = &src->request_queue[src->num_requests - 1];
     372             : 
     373        5534 :         if (prev->offset + prev->length == off &&
     374        1230 :             prev->length < MAX_CHUNK_SIZE &&
     375        1230 :             prev->path == path)
     376             :         {
     377             :             /*
     378             :              * Extend the previous request to cover as much of this new
     379             :              * request as possible, without exceeding MAX_CHUNK_SIZE.
     380             :              */
     381             :             size_t      thislen;
     382             : 
     383        1228 :             thislen = Min(len, MAX_CHUNK_SIZE - prev->length);
     384        1228 :             prev->length += thislen;
     385             : 
     386        1228 :             off += thislen;
     387        1228 :             len -= thislen;
     388             : 
     389             :             /*
     390             :              * Fall through to create new requests for any remaining 'len'
     391             :              * that didn't fit in the previous chunk.
     392             :              */
     393             :         }
     394             :     }
     395             : 
     396             :     /* Divide the request into pieces of MAX_CHUNK_SIZE bytes each */
     397       10434 :     while (len > 0)
     398             :     {
     399             :         int32       thislen;
     400             : 
     401             :         /* if the queue is full, perform all the work queued up so far */
     402        4888 :         if (src->num_requests == MAX_CHUNKS_PER_QUERY)
     403           0 :             process_queued_fetch_requests(src);
     404             : 
     405        4888 :         thislen = Min(len, MAX_CHUNK_SIZE);
     406        4888 :         src->request_queue[src->num_requests].path = path;
     407        4888 :         src->request_queue[src->num_requests].offset = off;
     408        4888 :         src->request_queue[src->num_requests].length = thislen;
     409        4888 :         src->num_requests++;
     410             : 
     411        4888 :         off += thislen;
     412        4888 :         len -= thislen;
     413             :     }
     414        5546 : }
     415             : 
     416             : /*
     417             :  * Fetch all the queued chunks and write them to the target data directory.
     418             :  */
     419             : static void
     420          12 : libpq_finish_fetch(rewind_source *source)
     421             : {
     422          12 :     process_queued_fetch_requests((libpq_source *) source);
     423          12 : }
     424             : 
     425             : static void
     426          12 : process_queued_fetch_requests(libpq_source *src)
     427             : {
     428             :     const char *params[3];
     429             :     PGresult   *res;
     430             :     int         chunkno;
     431             : 
     432          12 :     if (src->num_requests == 0)
     433           0 :         return;
     434             : 
     435          12 :     pg_log_debug("getting %d file chunks", src->num_requests);
     436             : 
     437             :     /*
     438             :      * The prepared statement, 'fetch_chunks_stmt', takes three arrays with
     439             :      * the same length as parameters: paths, offsets and lengths. Construct
     440             :      * the string representations of them.
     441             :      */
     442          12 :     resetStringInfo(&src->paths);
     443          12 :     resetStringInfo(&src->offsets);
     444          12 :     resetStringInfo(&src->lengths);
     445             : 
     446          12 :     appendStringInfoChar(&src->paths, '{');
     447          12 :     appendStringInfoChar(&src->offsets, '{');
     448          12 :     appendStringInfoChar(&src->lengths, '{');
     449        4900 :     for (int i = 0; i < src->num_requests; i++)
     450             :     {
     451        4888 :         fetch_range_request *rq = &src->request_queue[i];
     452             : 
     453        4888 :         if (i > 0)
     454             :         {
     455        4876 :             appendStringInfoChar(&src->paths, ',');
     456        4876 :             appendStringInfoChar(&src->offsets, ',');
     457        4876 :             appendStringInfoChar(&src->lengths, ',');
     458             :         }
     459             : 
     460        4888 :         appendArrayEscapedString(&src->paths, rq->path);
     461        4888 :         appendStringInfo(&src->offsets, INT64_FORMAT, (int64) rq->offset);
     462        4888 :         appendStringInfo(&src->lengths, INT64_FORMAT, (int64) rq->length);
     463             :     }
     464          12 :     appendStringInfoChar(&src->paths, '}');
     465          12 :     appendStringInfoChar(&src->offsets, '}');
     466          12 :     appendStringInfoChar(&src->lengths, '}');
     467             : 
     468             :     /*
     469             :      * Execute the prepared statement.
     470             :      */
     471          12 :     params[0] = src->paths.data;
     472          12 :     params[1] = src->offsets.data;
     473          12 :     params[2] = src->lengths.data;
     474             : 
     475          12 :     if (PQsendQueryPrepared(src->conn, "fetch_chunks_stmt", 3, params, NULL, NULL, 1) != 1)
     476           0 :         pg_fatal("could not send query: %s", PQerrorMessage(src->conn));
     477             : 
     478          12 :     if (PQsetSingleRowMode(src->conn) != 1)
     479           0 :         pg_fatal("could not set libpq connection to single row mode");
     480             : 
     481             :     /*----
     482             :      * The result set is of format:
     483             :      *
     484             :      * path     text    -- path in the data directory, e.g "base/1/123"
     485             :      * begin    int8    -- offset within the file
     486             :      * chunk    bytea   -- file content
     487             :      *----
     488             :      */
     489          12 :     chunkno = 0;
     490        4912 :     while ((res = PQgetResult(src->conn)) != NULL)
     491             :     {
     492        4900 :         fetch_range_request *rq = &src->request_queue[chunkno];
     493             :         char       *filename;
     494             :         int         filenamelen;
     495             :         int64       chunkoff;
     496             :         int         chunksize;
     497             :         char       *chunk;
     498             : 
     499        4900 :         switch (PQresultStatus(res))
     500             :         {
     501        4888 :             case PGRES_SINGLE_TUPLE:
     502        4888 :                 break;
     503             : 
     504          12 :             case PGRES_TUPLES_OK:
     505          12 :                 PQclear(res);
     506          12 :                 continue;       /* final zero-row result */
     507             : 
     508           0 :             default:
     509           0 :                 pg_fatal("unexpected result while fetching remote files: %s",
     510             :                          PQresultErrorMessage(res));
     511             :         }
     512             : 
     513        4888 :         if (chunkno > src->num_requests)
     514           0 :             pg_fatal("received more data chunks than requested");
     515             : 
     516             :         /* sanity check the result set */
     517        4888 :         if (PQnfields(res) != 3 || PQntuples(res) != 1)
     518           0 :             pg_fatal("unexpected result set size while fetching remote files");
     519             : 
     520        9776 :         if (PQftype(res, 0) != TEXTOID ||
     521        9776 :             PQftype(res, 1) != INT8OID ||
     522        4888 :             PQftype(res, 2) != BYTEAOID)
     523             :         {
     524           0 :             pg_fatal("unexpected data types in result set while fetching remote files: %u %u %u",
     525             :                      PQftype(res, 0), PQftype(res, 1), PQftype(res, 2));
     526             :         }
     527             : 
     528        4888 :         if (PQfformat(res, 0) != 1 &&
     529           0 :             PQfformat(res, 1) != 1 &&
     530           0 :             PQfformat(res, 2) != 1)
     531             :         {
     532           0 :             pg_fatal("unexpected result format while fetching remote files");
     533             :         }
     534             : 
     535        9776 :         if (PQgetisnull(res, 0, 0) ||
     536        4888 :             PQgetisnull(res, 0, 1))
     537             :         {
     538           0 :             pg_fatal("unexpected null values in result while fetching remote files");
     539             :         }
     540             : 
     541        4888 :         if (PQgetlength(res, 0, 1) != sizeof(int64))
     542           0 :             pg_fatal("unexpected result length while fetching remote files");
     543             : 
     544             :         /* Read result set to local variables */
     545        4888 :         memcpy(&chunkoff, PQgetvalue(res, 0, 1), sizeof(int64));
     546        4888 :         chunkoff = pg_ntoh64(chunkoff);
     547        4888 :         chunksize = PQgetlength(res, 0, 2);
     548             : 
     549        4888 :         filenamelen = PQgetlength(res, 0, 0);
     550        4888 :         filename = pg_malloc(filenamelen + 1);
     551        4888 :         memcpy(filename, PQgetvalue(res, 0, 0), filenamelen);
     552        4888 :         filename[filenamelen] = '\0';
     553             : 
     554        4888 :         chunk = PQgetvalue(res, 0, 2);
     555             : 
     556             :         /*
     557             :          * If a file has been deleted on the source, remove it on the target
     558             :          * as well.  Note that multiple unlink() calls may happen on the same
     559             :          * file if multiple data chunks are associated with it, hence ignore
     560             :          * unconditionally anything missing.
     561             :          */
     562        4888 :         if (PQgetisnull(res, 0, 2))
     563             :         {
     564           0 :             pg_log_debug("received null value for chunk for file \"%s\", file has been deleted",
     565             :                          filename);
     566           0 :             remove_target_file(filename, true);
     567             :         }
     568             :         else
     569             :         {
     570        4888 :             pg_log_debug("received chunk for file \"%s\", offset %lld, size %d",
     571             :                          filename, (long long int) chunkoff, chunksize);
     572             : 
     573        4888 :             if (strcmp(filename, rq->path) != 0)
     574             :             {
     575           0 :                 pg_fatal("received data for file \"%s\", when requested for \"%s\"",
     576             :                          filename, rq->path);
     577             :             }
     578        4888 :             if (chunkoff != rq->offset)
     579           0 :                 pg_fatal("received data at offset %lld of file \"%s\", when requested for offset %lld",
     580             :                          (long long int) chunkoff, rq->path, (long long int) rq->offset);
     581             : 
     582             :             /*
     583             :              * We should not receive more data than we requested, or
     584             :              * pg_read_binary_file() messed up.  We could receive less,
     585             :              * though, if the file was truncated in the source after we
     586             :              * checked its size. That's OK, there should be a WAL record of
     587             :              * the truncation, which will get replayed when you start the
     588             :              * target system for the first time after pg_rewind has completed.
     589             :              */
     590        4888 :             if (chunksize > rq->length)
     591           0 :                 pg_fatal("received more than requested for file \"%s\"", rq->path);
     592             : 
     593        4888 :             open_target_file(filename, false);
     594             : 
     595        4888 :             write_target_range(chunk, chunkoff, chunksize);
     596             :         }
     597             : 
     598        4888 :         pg_free(filename);
     599             : 
     600        4888 :         PQclear(res);
     601        4888 :         chunkno++;
     602             :     }
     603          12 :     if (chunkno != src->num_requests)
     604           0 :         pg_fatal("unexpected number of data chunks received");
     605             : 
     606          12 :     src->num_requests = 0;
     607             : }
     608             : 
     609             : /*
     610             :  * Escape a string to be used as element in a text array constant
     611             :  */
     612             : static void
     613        4888 : appendArrayEscapedString(StringInfo buf, const char *str)
     614             : {
     615        4888 :     appendStringInfoCharMacro(buf, '\"');
     616       89556 :     while (*str)
     617             :     {
     618       84668 :         char        ch = *str;
     619             : 
     620       84668 :         if (ch == '"' || ch == '\\')
     621           0 :             appendStringInfoCharMacro(buf, '\\');
     622             : 
     623       84668 :         appendStringInfoCharMacro(buf, ch);
     624             : 
     625       84668 :         str++;
     626             :     }
     627        4888 :     appendStringInfoCharMacro(buf, '\"');
     628        4888 : }
     629             : 
     630             : /*
     631             :  * Fetch a single file as a malloc'd buffer.
     632             :  */
     633             : static char *
     634          34 : libpq_fetch_file(rewind_source *source, const char *path, size_t *filesize)
     635             : {
     636          34 :     PGconn     *conn = ((libpq_source *) source)->conn;
     637             :     PGresult   *res;
     638             :     char       *result;
     639             :     int         len;
     640             :     const char *paramValues[1];
     641             : 
     642          34 :     paramValues[0] = path;
     643          34 :     res = PQexecParams(conn, "SELECT pg_read_binary_file($1)",
     644             :                        1, NULL, paramValues, NULL, NULL, 1);
     645             : 
     646          34 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
     647           0 :         pg_fatal("could not fetch remote file \"%s\": %s",
     648             :                  path, PQresultErrorMessage(res));
     649             : 
     650             :     /* sanity check the result set */
     651          34 :     if (PQntuples(res) != 1 || PQgetisnull(res, 0, 0))
     652           0 :         pg_fatal("unexpected result set while fetching remote file \"%s\"",
     653             :                  path);
     654             : 
     655             :     /* Read result to local variables */
     656          34 :     len = PQgetlength(res, 0, 0);
     657          34 :     result = pg_malloc(len + 1);
     658          34 :     memcpy(result, PQgetvalue(res, 0, 0), len);
     659          34 :     result[len] = '\0';
     660             : 
     661          34 :     PQclear(res);
     662             : 
     663          34 :     pg_log_debug("fetched file \"%s\", length %d", path, len);
     664             : 
     665          34 :     if (filesize)
     666          24 :         *filesize = len;
     667          34 :     return result;
     668             : }
     669             : 
     670             : /*
     671             :  * Close a libpq source.
     672             :  */
     673             : static void
     674          12 : libpq_destroy(rewind_source *source)
     675             : {
     676          12 :     libpq_source *src = (libpq_source *) source;
     677             : 
     678          12 :     pfree(src->paths.data);
     679          12 :     pfree(src->offsets.data);
     680          12 :     pfree(src->lengths.data);
     681          12 :     pfree(src);
     682             : 
     683             :     /* NOTE: we don't close the connection here, as it was not opened by us. */
     684          12 : }

Generated by: LCOV version 1.14