PostgreSQL Source Code  git master
libpq_source.c File Reference
#include "postgres_fe.h"
#include "catalog/pg_type_d.h"
#include "common/connect.h"
#include "datapagemap.h"
#include "file_ops.h"
#include "filemap.h"
#include "lib/stringinfo.h"
#include "pg_rewind.h"
#include "port/pg_bswap.h"
#include "rewind_source.h"
Include dependency graph for libpq_source.c:

Go to the source code of this file.

Data Structures

struct  fetch_range_request
 
struct  libpq_source
 

Macros

#define MAX_CHUNK_SIZE   (1024 * 1024)
 
#define MAX_CHUNKS_PER_QUERY   1000
 

Functions

static void init_libpq_conn (PGconn *conn)
 
static char * run_simple_query (PGconn *conn, const char *sql)
 
static void run_simple_command (PGconn *conn, const char *sql)
 
static void appendArrayEscapedString (StringInfo buf, const char *str)
 
static void process_queued_fetch_requests (libpq_source *src)
 
static void libpq_traverse_files (rewind_source *source, process_file_callback_t callback)
 
static void libpq_queue_fetch_file (rewind_source *source, const char *path, size_t len)
 
static void libpq_queue_fetch_range (rewind_source *source, const char *path, off_t off, size_t len)
 
static void libpq_finish_fetch (rewind_source *source)
 
static char * libpq_fetch_file (rewind_source *source, const char *path, size_t *filesize)
 
static XLogRecPtr libpq_get_current_wal_insert_lsn (rewind_source *source)
 
static void libpq_destroy (rewind_source *source)
 
rewind_sourceinit_libpq_source (PGconn *conn)
 

Macro Definition Documentation

◆ MAX_CHUNK_SIZE

#define MAX_CHUNK_SIZE   (1024 * 1024)

Definition at line 26 of file libpq_source.c.

◆ MAX_CHUNKS_PER_QUERY

#define MAX_CHUNKS_PER_QUERY   1000

Definition at line 27 of file libpq_source.c.

Function Documentation

◆ appendArrayEscapedString()

static void appendArrayEscapedString ( StringInfo  buf,
const char *  str 
)
static

Definition at line 604 of file libpq_source.c.

605 {
607  while (*str)
608  {
609  char ch = *str;
610 
611  if (ch == '"' || ch == '\\')
613 
615 
616  str++;
617  }
619 }
static char * buf
Definition: pg_test_fsync.c:67
#define appendStringInfoCharMacro(str, ch)
Definition: stringinfo.h:128

References appendStringInfoCharMacro, buf, and generate_unaccent_rules::str.

Referenced by process_queued_fetch_requests().

◆ init_libpq_conn()

static void init_libpq_conn ( PGconn conn)
static

Definition at line 111 of file libpq_source.c.

112 {
113  PGresult *res;
114  char *str;
115 
116  /* disable all types of timeouts */
117  run_simple_command(conn, "SET statement_timeout = 0");
118  run_simple_command(conn, "SET lock_timeout = 0");
119  run_simple_command(conn, "SET idle_in_transaction_session_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  run_simple_command(conn, "SET default_transaction_read_only = on");
126 
127  /* secure search_path */
130  pg_fatal("could not clear search_path: %s",
132  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  str = run_simple_query(conn, "SHOW full_page_writes");
140  if (strcmp(str, "on") != 0)
141  pg_fatal("full_page_writes must be enabled in the source server");
142  pg_free(str);
143 
144  /* Prepare a statement we'll use to fetch files */
145  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 
152  pg_fatal("could not prepare statement to fetch file contents: %s",
154  PQclear(res);
155 }
#define ALWAYS_SECURE_SEARCH_PATH_SQL
Definition: connect.h:25
PGresult * PQprepare(PGconn *conn, const char *stmtName, const char *query, int nParams, const Oid *paramTypes)
Definition: fe-exec.c:2317
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:3270
void PQclear(PGresult *res)
Definition: fe-exec.c:718
char * PQresultErrorMessage(const PGresult *res)
Definition: fe-exec.c:3286
PGresult * PQexec(PGconn *conn, const char *query)
Definition: fe-exec.c:2273
void pg_free(void *ptr)
Definition: fe_memutils.c:105
@ PGRES_COMMAND_OK
Definition: libpq-fe.h:97
@ PGRES_TUPLES_OK
Definition: libpq-fe.h:100
static void run_simple_command(PGconn *conn, const char *sql)
Definition: libpq_source.c:191
static char * run_simple_query(PGconn *conn, const char *sql)
Definition: libpq_source.c:163
#define pg_fatal(...)
PGconn * conn
Definition: streamutil.c:54

References ALWAYS_SECURE_SEARCH_PATH_SQL, conn, pg_fatal, pg_free(), PGRES_COMMAND_OK, PGRES_TUPLES_OK, PQclear(), PQexec(), PQprepare(), PQresultErrorMessage(), PQresultStatus(), res, run_simple_command(), run_simple_query(), and generate_unaccent_rules::str.

Referenced by init_libpq_source().

◆ init_libpq_source()

rewind_source* init_libpq_source ( PGconn conn)

Definition at line 82 of file libpq_source.c.

83 {
84  libpq_source *src;
85 
87 
88  src = pg_malloc0(sizeof(libpq_source));
89 
97 
98  src->conn = conn;
99 
100  initStringInfo(&src->paths);
101  initStringInfo(&src->offsets);
102  initStringInfo(&src->lengths);
103 
104  return &src->common;
105 }
void * pg_malloc0(size_t size)
Definition: fe_memutils.c:53
static void init_libpq_conn(PGconn *conn)
Definition: libpq_source.c:111
static char * libpq_fetch_file(rewind_source *source, const char *path, size_t *filesize)
Definition: libpq_source.c:625
static XLogRecPtr libpq_get_current_wal_insert_lsn(rewind_source *source)
Definition: libpq_source.c:208
static void libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
Definition: libpq_source.c:232
static void libpq_destroy(rewind_source *source)
Definition: libpq_source.c:665
static void libpq_queue_fetch_range(rewind_source *source, const char *path, off_t off, size_t len)
Definition: libpq_source.c:346
static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
Definition: libpq_source.c:316
static void libpq_finish_fetch(rewind_source *source)
Definition: libpq_source.c:411
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
StringInfoData paths
Definition: libpq_source.c:51
StringInfoData lengths
Definition: libpq_source.c:53
PGconn * conn
Definition: libpq_source.c:41
StringInfoData offsets
Definition: libpq_source.c:52
rewind_source common
Definition: libpq_source.c:39
void(* queue_fetch_file)(struct rewind_source *, const char *path, size_t len)
Definition: rewind_source.h:60
void(* traverse_files)(struct rewind_source *, process_file_callback_t callback)
Definition: rewind_source.h:29
void(* finish_fetch)(struct rewind_source *)
Definition: rewind_source.h:66
XLogRecPtr(* get_current_wal_insert_lsn)(struct rewind_source *)
Definition: rewind_source.h:71
void(* queue_fetch_range)(struct rewind_source *, const char *path, off_t offset, size_t len)
Definition: rewind_source.h:47
char *(* fetch_file)(struct rewind_source *, const char *path, size_t *filesize)
Definition: rewind_source.h:37
void(* destroy)(struct rewind_source *)
Definition: rewind_source.h:76

References libpq_source::common, conn, libpq_source::conn, rewind_source::destroy, rewind_source::fetch_file, rewind_source::finish_fetch, rewind_source::get_current_wal_insert_lsn, init_libpq_conn(), initStringInfo(), libpq_source::lengths, libpq_destroy(), libpq_fetch_file(), libpq_finish_fetch(), libpq_get_current_wal_insert_lsn(), libpq_queue_fetch_file(), libpq_queue_fetch_range(), libpq_traverse_files(), libpq_source::offsets, libpq_source::paths, pg_malloc0(), rewind_source::queue_fetch_file, rewind_source::queue_fetch_range, and rewind_source::traverse_files.

Referenced by main().

◆ libpq_destroy()

static void libpq_destroy ( rewind_source source)
static

Definition at line 665 of file libpq_source.c.

666 {
667  libpq_source *src = (libpq_source *) source;
668 
669  pfree(src->paths.data);
670  pfree(src->offsets.data);
671  pfree(src->lengths.data);
672  pfree(src);
673 
674  /* NOTE: we don't close the connection here, as it was not opened by us. */
675 }
void pfree(void *pointer)
Definition: mcxt.c:1175
static rewind_source * source
Definition: pg_rewind.c:81

References StringInfoData::data, libpq_source::lengths, libpq_source::offsets, libpq_source::paths, pfree(), and source.

Referenced by init_libpq_source().

◆ libpq_fetch_file()

static char * libpq_fetch_file ( rewind_source source,
const char *  path,
size_t *  filesize 
)
static

Definition at line 625 of file libpq_source.c.

626 {
627  PGconn *conn = ((libpq_source *) source)->conn;
628  PGresult *res;
629  char *result;
630  int len;
631  const char *paramValues[1];
632 
633  paramValues[0] = path;
634  res = PQexecParams(conn, "SELECT pg_read_binary_file($1)",
635  1, NULL, paramValues, NULL, NULL, 1);
636 
638  pg_fatal("could not fetch remote file \"%s\": %s",
639  path, PQresultErrorMessage(res));
640 
641  /* sanity check the result set */
642  if (PQntuples(res) != 1 || PQgetisnull(res, 0, 0))
643  pg_fatal("unexpected result set while fetching remote file \"%s\"",
644  path);
645 
646  /* Read result to local variables */
647  len = PQgetlength(res, 0, 0);
648  result = pg_malloc(len + 1);
649  memcpy(result, PQgetvalue(res, 0, 0), len);
650  result[len] = '\0';
651 
652  PQclear(res);
653 
654  pg_log_debug("fetched file \"%s\", length %d", path, len);
655 
656  if (filesize)
657  *filesize = len;
658  return result;
659 }
int PQgetlength(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3746
PGresult * PQexecParams(PGconn *conn, const char *command, int nParams, const Oid *paramTypes, const char *const *paramValues, const int *paramLengths, const int *paramFormats, int resultFormat)
Definition: fe-exec.c:2287
int PQntuples(const PGresult *res)
Definition: fe-exec.c:3340
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3735
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3760
void * pg_malloc(size_t size)
Definition: fe_memutils.c:47
#define pg_log_debug(...)
Definition: logging.h:133
const void size_t len

References conn, len, pg_fatal, pg_log_debug, pg_malloc(), PGRES_TUPLES_OK, PQclear(), PQexecParams(), PQgetisnull(), PQgetlength(), PQgetvalue(), PQntuples(), PQresultErrorMessage(), PQresultStatus(), res, and source.

Referenced by init_libpq_source().

◆ libpq_finish_fetch()

static void libpq_finish_fetch ( rewind_source source)
static

Definition at line 411 of file libpq_source.c.

412 {
414 }
static void process_queued_fetch_requests(libpq_source *src)
Definition: libpq_source.c:417

References process_queued_fetch_requests(), and source.

Referenced by init_libpq_source().

◆ libpq_get_current_wal_insert_lsn()

static XLogRecPtr libpq_get_current_wal_insert_lsn ( rewind_source source)
static

Definition at line 208 of file libpq_source.c.

209 {
210  PGconn *conn = ((libpq_source *) source)->conn;
211  XLogRecPtr result;
212  uint32 hi;
213  uint32 lo;
214  char *val;
215 
216  val = run_simple_query(conn, "SELECT pg_current_wal_insert_lsn()");
217 
218  if (sscanf(val, "%X/%X", &hi, &lo) != 2)
219  pg_fatal("unrecognized result \"%s\" for current WAL insert location", val);
220 
221  result = ((uint64) hi) << 32 | lo;
222 
223  pg_free(val);
224 
225  return result;
226 }
unsigned int uint32
Definition: c.h:441
long val
Definition: informix.c:664
uint64 XLogRecPtr
Definition: xlogdefs.h:21

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

Referenced by init_libpq_source().

◆ libpq_queue_fetch_file()

static void libpq_queue_fetch_file ( rewind_source source,
const char *  path,
size_t  len 
)
static

Definition at line 316 of file libpq_source.c.

317 {
318  /*
319  * Truncate the target file immediately, and queue a request to fetch it
320  * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
321  * request fetching a full-sized chunk anyway, so that if the file has
322  * become larger in the source system, after we scanned the source
323  * directory, we still fetch the whole file. This only works for files up
324  * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
325  * and such that are changed every now and then, but not WAL-logged. For
326  * larger files, we fetch up to the original size.
327  *
328  * Even with that mechanism, there is an inherent race condition if the
329  * file is modified at the same instant that we're copying it, so that we
330  * might copy a torn version of the file with one half from the old
331  * version and another half from the new. But pg_basebackup has the same
332  * problem, and it hasn't been a problem in practice.
333  *
334  * It might seem more natural to truncate the file later, when we receive
335  * it from the source server, but then we'd need to track which
336  * fetch-requests are for a whole file.
337  */
338  open_target_file(path, true);
340 }
#define Max(x, y)
Definition: c.h:980
void open_target_file(const char *path, bool trunc)
Definition: file_ops.c:47
#define MAX_CHUNK_SIZE
Definition: libpq_source.c:26

References len, libpq_queue_fetch_range(), Max, MAX_CHUNK_SIZE, open_target_file(), and source.

Referenced by init_libpq_source().

◆ libpq_queue_fetch_range()

static void libpq_queue_fetch_range ( rewind_source source,
const char *  path,
off_t  off,
size_t  len 
)
static

Definition at line 346 of file libpq_source.c.

348 {
349  libpq_source *src = (libpq_source *) source;
350 
351  /*
352  * Does this request happen to be a continuation of the previous chunk? If
353  * so, merge it with the previous one.
354  *
355  * XXX: We use pointer equality to compare the path. That's good enough
356  * for our purposes; the caller always passes the same pointer for the
357  * same filename. If it didn't, we would fail to merge requests, but it
358  * wouldn't affect correctness.
359  */
360  if (src->num_requests > 0)
361  {
362  fetch_range_request *prev = &src->request_queue[src->num_requests - 1];
363 
364  if (prev->offset + prev->length == off &&
365  prev->length < MAX_CHUNK_SIZE &&
366  prev->path == path)
367  {
368  /*
369  * Extend the previous request to cover as much of this new
370  * request as possible, without exceeding MAX_CHUNK_SIZE.
371  */
372  size_t thislen;
373 
374  thislen = Min(len, MAX_CHUNK_SIZE - prev->length);
375  prev->length += thislen;
376 
377  off += thislen;
378  len -= thislen;
379 
380  /*
381  * Fall through to create new requests for any remaining 'len'
382  * that didn't fit in the previous chunk.
383  */
384  }
385  }
386 
387  /* Divide the request into pieces of MAX_CHUNK_SIZE bytes each */
388  while (len > 0)
389  {
390  int32 thislen;
391 
392  /* if the queue is full, perform all the work queued up so far */
395 
396  thislen = Min(len, MAX_CHUNK_SIZE);
397  src->request_queue[src->num_requests].path = path;
398  src->request_queue[src->num_requests].offset = off;
399  src->request_queue[src->num_requests].length = thislen;
400  src->num_requests++;
401 
402  off += thislen;
403  len -= thislen;
404  }
405 }
#define Min(x, y)
Definition: c.h:986
signed int int32
Definition: c.h:429
#define MAX_CHUNKS_PER_QUERY
Definition: libpq_source.c:27
const char * path
Definition: libpq_source.c:32
fetch_range_request request_queue[MAX_CHUNKS_PER_QUERY]
Definition: libpq_source.c:48

References len, fetch_range_request::length, MAX_CHUNK_SIZE, MAX_CHUNKS_PER_QUERY, Min, libpq_source::num_requests, fetch_range_request::offset, fetch_range_request::path, process_queued_fetch_requests(), libpq_source::request_queue, and source.

Referenced by init_libpq_source(), and libpq_queue_fetch_file().

◆ libpq_traverse_files()

static void libpq_traverse_files ( rewind_source source,
process_file_callback_t  callback 
)
static

Definition at line 232 of file libpq_source.c.

233 {
234  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  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  res = PQexec(conn, sql);
268 
270  pg_fatal("could not fetch file list: %s",
272 
273  /* sanity check the result set */
274  if (PQnfields(res) != 4)
275  pg_fatal("unexpected result set while fetching file list");
276 
277  /* Read result to local variables */
278  for (i = 0; i < PQntuples(res); i++)
279  {
280  char *path;
281  int64 filesize;
282  bool isdir;
283  char *link_target;
285 
286  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  continue;
293  }
294 
295  path = PQgetvalue(res, i, 0);
296  filesize = atol(PQgetvalue(res, i, 1));
297  isdir = (strcmp(PQgetvalue(res, i, 2), "t") == 0);
298  link_target = PQgetvalue(res, i, 3);
299 
300  if (link_target[0])
302  else if (isdir)
304  else
306 
307  process_source_file(path, type, filesize, link_target);
308  }
309  PQclear(res);
310 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:3348
void process_source_file(const char *path, file_type_t type, size_t size, const char *link_target)
Definition: filemap.c:218
file_type_t
Definition: filemap.h:31
@ FILE_TYPE_REGULAR
Definition: filemap.h:34
@ FILE_TYPE_SYMLINK
Definition: filemap.h:36
@ FILE_TYPE_DIRECTORY
Definition: filemap.h:35
int i
Definition: isn.c:73

References conn, 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(), res, source, and generate_unaccent_rules::type.

Referenced by init_libpq_source().

◆ process_queued_fetch_requests()

static void process_queued_fetch_requests ( libpq_source src)
static

Definition at line 417 of file libpq_source.c.

418 {
419  const char *params[3];
420  PGresult *res;
421  int chunkno;
422 
423  if (src->num_requests == 0)
424  return;
425 
426  pg_log_debug("getting %d file chunks", src->num_requests);
427 
428  /*
429  * The prepared statement, 'fetch_chunks_stmt', takes three arrays with
430  * the same length as parameters: paths, offsets and lengths. Construct
431  * the string representations of them.
432  */
433  resetStringInfo(&src->paths);
434  resetStringInfo(&src->offsets);
435  resetStringInfo(&src->lengths);
436 
437  appendStringInfoChar(&src->paths, '{');
438  appendStringInfoChar(&src->offsets, '{');
439  appendStringInfoChar(&src->lengths, '{');
440  for (int i = 0; i < src->num_requests; i++)
441  {
442  fetch_range_request *rq = &src->request_queue[i];
443 
444  if (i > 0)
445  {
446  appendStringInfoChar(&src->paths, ',');
447  appendStringInfoChar(&src->offsets, ',');
448  appendStringInfoChar(&src->lengths, ',');
449  }
450 
451  appendArrayEscapedString(&src->paths, rq->path);
452  appendStringInfo(&src->offsets, INT64_FORMAT, (int64) rq->offset);
453  appendStringInfo(&src->lengths, INT64_FORMAT, (int64) rq->length);
454  }
455  appendStringInfoChar(&src->paths, '}');
456  appendStringInfoChar(&src->offsets, '}');
457  appendStringInfoChar(&src->lengths, '}');
458 
459  /*
460  * Execute the prepared statement.
461  */
462  params[0] = src->paths.data;
463  params[1] = src->offsets.data;
464  params[2] = src->lengths.data;
465 
466  if (PQsendQueryPrepared(src->conn, "fetch_chunks_stmt", 3, params, NULL, NULL, 1) != 1)
467  pg_fatal("could not send query: %s", PQerrorMessage(src->conn));
468 
469  if (PQsetSingleRowMode(src->conn) != 1)
470  pg_fatal("could not set libpq connection to single row mode");
471 
472  /*----
473  * The result set is of format:
474  *
475  * path text -- path in the data directory, e.g "base/1/123"
476  * begin int8 -- offset within the file
477  * chunk bytea -- file content
478  *----
479  */
480  chunkno = 0;
481  while ((res = PQgetResult(src->conn)) != NULL)
482  {
483  fetch_range_request *rq = &src->request_queue[chunkno];
484  char *filename;
485  int filenamelen;
486  int64 chunkoff;
487  int chunksize;
488  char *chunk;
489 
490  switch (PQresultStatus(res))
491  {
492  case PGRES_SINGLE_TUPLE:
493  break;
494 
495  case PGRES_TUPLES_OK:
496  PQclear(res);
497  continue; /* final zero-row result */
498 
499  default:
500  pg_fatal("unexpected result while fetching remote files: %s",
502  }
503 
504  if (chunkno > src->num_requests)
505  pg_fatal("received more data chunks than requested");
506 
507  /* sanity check the result set */
508  if (PQnfields(res) != 3 || PQntuples(res) != 1)
509  pg_fatal("unexpected result set size while fetching remote files");
510 
511  if (PQftype(res, 0) != TEXTOID ||
512  PQftype(res, 1) != INT8OID ||
513  PQftype(res, 2) != BYTEAOID)
514  {
515  pg_fatal("unexpected data types in result set while fetching remote files: %u %u %u",
516  PQftype(res, 0), PQftype(res, 1), PQftype(res, 2));
517  }
518 
519  if (PQfformat(res, 0) != 1 &&
520  PQfformat(res, 1) != 1 &&
521  PQfformat(res, 2) != 1)
522  {
523  pg_fatal("unexpected result format while fetching remote files");
524  }
525 
526  if (PQgetisnull(res, 0, 0) ||
527  PQgetisnull(res, 0, 1))
528  {
529  pg_fatal("unexpected null values in result while fetching remote files");
530  }
531 
532  if (PQgetlength(res, 0, 1) != sizeof(int64))
533  pg_fatal("unexpected result length while fetching remote files");
534 
535  /* Read result set to local variables */
536  memcpy(&chunkoff, PQgetvalue(res, 0, 1), sizeof(int64));
537  chunkoff = pg_ntoh64(chunkoff);
538  chunksize = PQgetlength(res, 0, 2);
539 
540  filenamelen = PQgetlength(res, 0, 0);
541  filename = pg_malloc(filenamelen + 1);
542  memcpy(filename, PQgetvalue(res, 0, 0), filenamelen);
543  filename[filenamelen] = '\0';
544 
545  chunk = PQgetvalue(res, 0, 2);
546 
547  /*
548  * If a file has been deleted on the source, remove it on the target
549  * as well. Note that multiple unlink() calls may happen on the same
550  * file if multiple data chunks are associated with it, hence ignore
551  * unconditionally anything missing.
552  */
553  if (PQgetisnull(res, 0, 2))
554  {
555  pg_log_debug("received null value for chunk for file \"%s\", file has been deleted",
556  filename);
558  }
559  else
560  {
561  pg_log_debug("received chunk for file \"%s\", offset %lld, size %d",
562  filename, (long long int) chunkoff, chunksize);
563 
564  if (strcmp(filename, rq->path) != 0)
565  {
566  pg_fatal("received data for file \"%s\", when requested for \"%s\"",
567  filename, rq->path);
568  }
569  if (chunkoff != rq->offset)
570  pg_fatal("received data at offset %lld of file \"%s\", when requested for offset %lld",
571  (long long int) chunkoff, rq->path, (long long int) rq->offset);
572 
573  /*
574  * We should not receive more data than we requested, or
575  * pg_read_binary_file() messed up. We could receive less,
576  * though, if the file was truncated in the source after we
577  * checked its size. That's OK, there should be a WAL record of
578  * the truncation, which will get replayed when you start the
579  * target system for the first time after pg_rewind has completed.
580  */
581  if (chunksize > rq->length)
582  pg_fatal("received more than requested for file \"%s\"", rq->path);
583 
584  open_target_file(filename, false);
585 
586  write_target_range(chunk, chunkoff, chunksize);
587  }
588 
589  pg_free(filename);
590 
591  PQclear(res);
592  chunkno++;
593  }
594  if (chunkno != src->num_requests)
595  pg_fatal("unexpected number of data chunks received");
596 
597  src->num_requests = 0;
598 }
#define INT64_FORMAT
Definition: c.h:483
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6908
int PQsetSingleRowMode(PGconn *conn)
Definition: fe-exec.c:1976
Oid PQftype(const PGresult *res, int field_num)
Definition: fe-exec.c:3578
int PQfformat(const PGresult *res, int field_num)
Definition: fe-exec.c:3567
int PQsendQueryPrepared(PGconn *conn, const char *stmtName, int nParams, const char *const *paramValues, const int *paramLengths, const int *paramFormats, int resultFormat)
Definition: fe-exec.c:1682
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:2082
void remove_target_file(const char *path, bool missing_ok)
Definition: file_ops.c:187
void write_target_range(char *buf, off_t begin, size_t size)
Definition: file_ops.c:88
@ PGRES_SINGLE_TUPLE
Definition: libpq-fe.h:110
static void appendArrayEscapedString(StringInfo buf, const char *str)
Definition: libpq_source.c:604
#define pg_ntoh64(x)
Definition: pg_bswap.h:126
static char * filename
Definition: pg_dumpall.c:94
void resetStringInfo(StringInfo str)
Definition: stringinfo.c:75
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:91
void appendStringInfoChar(StringInfo str, char ch)
Definition: stringinfo.c:188

References appendArrayEscapedString(), appendStringInfo(), appendStringInfoChar(), libpq_source::conn, StringInfoData::data, filename, i, INT64_FORMAT, fetch_range_request::length, libpq_source::lengths, libpq_source::num_requests, fetch_range_request::offset, libpq_source::offsets, open_target_file(), fetch_range_request::path, libpq_source::paths, pg_fatal, pg_free(), pg_log_debug, pg_malloc(), pg_ntoh64, PGRES_SINGLE_TUPLE, PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQfformat(), PQftype(), PQgetisnull(), PQgetlength(), PQgetResult(), PQgetvalue(), PQnfields(), PQntuples(), PQresultErrorMessage(), PQresultStatus(), PQsendQueryPrepared(), PQsetSingleRowMode(), remove_target_file(), libpq_source::request_queue, res, resetStringInfo(), and write_target_range().

Referenced by libpq_finish_fetch(), and libpq_queue_fetch_range().

◆ run_simple_command()

static void run_simple_command ( PGconn conn,
const char *  sql 
)
static

Definition at line 191 of file libpq_source.c.

192 {
193  PGresult *res;
194 
195  res = PQexec(conn, sql);
196 
198  pg_fatal("error running query (%s) in source server: %s",
199  sql, PQresultErrorMessage(res));
200 
201  PQclear(res);
202 }

References conn, pg_fatal, PGRES_COMMAND_OK, PQclear(), PQexec(), PQresultErrorMessage(), PQresultStatus(), and res.

Referenced by init_libpq_conn().

◆ run_simple_query()

static char * run_simple_query ( PGconn conn,
const char *  sql 
)
static

Definition at line 163 of file libpq_source.c.

164 {
165  PGresult *res;
166  char *result;
167 
168  res = PQexec(conn, sql);
169 
171  pg_fatal("error running query (%s) on source server: %s",
172  sql, PQresultErrorMessage(res));
173 
174  /* sanity check the result set */
175  if (PQnfields(res) != 1 || PQntuples(res) != 1 || PQgetisnull(res, 0, 0))
176  pg_fatal("unexpected result set from query");
177 
178  result = pg_strdup(PQgetvalue(res, 0, 0));
179 
180  PQclear(res);
181 
182  return result;
183 }
char * pg_strdup(const char *in)
Definition: fe_memutils.c:85

References conn, pg_fatal, pg_strdup(), PGRES_TUPLES_OK, PQclear(), PQexec(), PQgetisnull(), PQgetvalue(), PQnfields(), PQntuples(), PQresultErrorMessage(), PQresultStatus(), and res.

Referenced by init_libpq_conn(), and libpq_get_current_wal_insert_lsn().