PostgreSQL Source Code  git master
libpq_fetch.c File Reference
#include "postgres_fe.h"
#include <sys/stat.h>
#include <dirent.h>
#include <fcntl.h>
#include <unistd.h>
#include "pg_rewind.h"
#include "datapagemap.h"
#include "fetch.h"
#include "file_ops.h"
#include "filemap.h"
#include "logging.h"
#include "libpq-fe.h"
#include "catalog/pg_type_d.h"
#include "fe_utils/connect.h"
#include "port/pg_bswap.h"
Include dependency graph for libpq_fetch.c:

Go to the source code of this file.

Macros

#define CHUNKSIZE   1000000
 

Functions

static void receiveFileChunks (const char *sql)
 
static void execute_pagemap (datapagemap_t *pagemap, const char *path)
 
static char * run_simple_query (const char *sql)
 
void libpqConnect (const char *connstr)
 
XLogRecPtr libpqGetCurrentXlogInsertLocation (void)
 
void libpqProcessFileList (void)
 
char * libpqGetFile (const char *filename, size_t *filesize)
 
static void fetch_file_range (const char *path, uint64 begin, uint64 end)
 
void libpq_executeFileMap (filemap_t *map)
 

Variables

static PGconnconn = NULL
 

Macro Definition Documentation

◆ CHUNKSIZE

#define CHUNKSIZE   1000000

Definition at line 38 of file libpq_fetch.c.

Referenced by fetch_file_range().

Function Documentation

◆ execute_pagemap()

static void execute_pagemap ( datapagemap_t pagemap,
const char *  path 
)
static

Definition at line 514 of file libpq_fetch.c.

References datapagemap_iterate(), datapagemap_next(), fetch_file_range(), and pg_free().

Referenced by libpq_executeFileMap().

515 {
517  BlockNumber blkno;
518  off_t offset;
519 
520  iter = datapagemap_iterate(pagemap);
521  while (datapagemap_next(iter, &blkno))
522  {
523  offset = blkno * BLCKSZ;
524 
525  fetch_file_range(path, offset, offset + BLCKSZ);
526  }
527  pg_free(iter);
528 }
uint32 BlockNumber
Definition: block.h:31
bool datapagemap_next(datapagemap_iterator_t *iter, BlockNumber *blkno)
Definition: datapagemap.c:87
datapagemap_iterator_t * datapagemap_iterate(datapagemap_t *map)
Definition: datapagemap.c:75
void pg_free(void *ptr)
Definition: fe_memutils.c:105
static void fetch_file_range(const char *path, uint64 begin, uint64 end)
Definition: libpq_fetch.c:396

◆ fetch_file_range()

static void fetch_file_range ( const char *  path,
uint64  begin,
uint64  end 
)
static

Definition at line 396 of file libpq_fetch.c.

References CHUNKSIZE, MAXPGPATH, pg_fatal(), PQerrorMessage(), PQputCopyData(), snprintf(), and UINT64_FORMAT.

Referenced by execute_pagemap(), and libpq_executeFileMap().

397 {
398  char linebuf[MAXPGPATH + 23];
399 
400  /* Split the range into CHUNKSIZE chunks */
401  while (end - begin > 0)
402  {
403  unsigned int len;
404 
405  /* Fine as long as CHUNKSIZE is not bigger than UINT32_MAX */
406  if (end - begin > CHUNKSIZE)
407  len = CHUNKSIZE;
408  else
409  len = (unsigned int) (end - begin);
410 
411  snprintf(linebuf, sizeof(linebuf), "%s\t" UINT64_FORMAT "\t%u\n", path, begin, len);
412 
413  if (PQputCopyData(conn, linebuf, strlen(linebuf)) != 1)
414  pg_fatal("could not send COPY data: %s",
416 
417  begin += len;
418  }
419 }
int PQputCopyData(PGconn *conn, const char *buffer, int nbytes)
Definition: fe-exec.c:2272
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6116
int snprintf(char *str, size_t count, const char *fmt,...) pg_attribute_printf(3
#define CHUNKSIZE
Definition: libpq_fetch.c:38
void pg_fatal(const char *fmt,...)
Definition: logging.c:83
#define MAXPGPATH
static PGconn * conn
Definition: libpq_fetch.c:29
#define UINT64_FORMAT
Definition: c.h:368

◆ 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().

426 {
427  file_entry_t *entry;
428  const char *sql;
429  PGresult *res;
430  int i;
431 
432  /*
433  * First create a temporary table, and load it with the blocks that we
434  * need to fetch.
435  */
436  sql = "CREATE TEMPORARY TABLE fetchchunks(path text, begin int8, len int4);";
437  res = PQexec(conn, sql);
438 
439  if (PQresultStatus(res) != PGRES_COMMAND_OK)
440  pg_fatal("could not create temporary table: %s",
441  PQresultErrorMessage(res));
442  PQclear(res);
443 
444  sql = "COPY fetchchunks FROM STDIN";
445  res = PQexec(conn, sql);
446 
447  if (PQresultStatus(res) != PGRES_COPY_IN)
448  pg_fatal("could not send file list: %s",
449  PQresultErrorMessage(res));
450  PQclear(res);
451 
452  for (i = 0; i < map->narray; i++)
453  {
454  entry = map->array[i];
455 
456  /* If this is a relation file, copy the modified blocks */
457  execute_pagemap(&entry->pagemap, entry->path);
458 
459  switch (entry->action)
460  {
461  case FILE_ACTION_NONE:
462  /* nothing else to do */
463  break;
464 
465  case FILE_ACTION_COPY:
466  /* Truncate the old file out of the way, if any */
467  open_target_file(entry->path, true);
468  fetch_file_range(entry->path, 0, entry->newsize);
469  break;
470 
472  truncate_target_file(entry->path, entry->newsize);
473  break;
474 
476  fetch_file_range(entry->path, entry->oldsize, entry->newsize);
477  break;
478 
479  case FILE_ACTION_REMOVE:
480  remove_target(entry);
481  break;
482 
483  case FILE_ACTION_CREATE:
484  create_target(entry);
485  break;
486  }
487  }
488 
489  if (PQputCopyEnd(conn, NULL) != 1)
490  pg_fatal("could not send end-of-COPY: %s",
492 
493  while ((res = PQgetResult(conn)) != NULL)
494  {
495  if (PQresultStatus(res) != PGRES_COMMAND_OK)
496  pg_fatal("unexpected result while sending file list: %s",
497  PQresultErrorMessage(res));
498  PQclear(res);
499  }
500 
501  /*
502  * We've now copied the list of file ranges that we need to fetch to the
503  * temporary table. Now, actually fetch all of those ranges.
504  */
505  sql =
506  "SELECT path, begin,\n"
507  " pg_read_binary_file(path, begin, len, true) AS chunk\n"
508  "FROM fetchchunks\n";
509 
510  receiveFileChunks(sql);
511 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6116
void open_target_file(const char *path, bool trunc)
Definition: file_ops.c:43
file_entry_t ** array
Definition: filemap.h:79
size_t newsize
Definition: filemap.h:51
static void execute_pagemap(datapagemap_t *pagemap, const char *path)
Definition: libpq_fetch.c:514
int PQputCopyEnd(PGconn *conn, const char *errormsg)
Definition: fe-exec.c:2339
void pg_fatal(const char *fmt,...)
Definition: logging.c:83
datapagemap_t pagemap
Definition: filemap.h:54
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
int narray
Definition: filemap.h:80
void truncate_target_file(const char *path, off_t newsize)
Definition: file_ops.c:192
file_action_t action
Definition: filemap.h:47
static void receiveFileChunks(const char *sql)
Definition: libpq_fetch.c:237
size_t oldsize
Definition: filemap.h:50
static PGconn * conn
Definition: libpq_fetch.c:29
void PQclear(PGresult *res)
Definition: fe-exec.c:671
void remove_target(file_entry_t *entry)
Definition: file_ops.c:126
char * path
Definition: filemap.h:44
char * PQresultErrorMessage(const PGresult *res)
Definition: fe-exec.c:2663
static void fetch_file_range(const char *path, uint64 begin, uint64 end)
Definition: libpq_fetch.c:396
int i
Definition: filemap.h:42
PGresult * PQexec(PGconn *conn, const char *query)
Definition: fe-exec.c:1897
void create_target(file_entry_t *entry)
Definition: file_ops.c:147
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1753

◆ 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().

46 {
47  char *str;
48  PGresult *res;
49 
52  pg_fatal("could not connect to server: %s",
54 
55  pg_log(PG_PROGRESS, "connected to server\n");
56 
58  if (PQresultStatus(res) != PGRES_TUPLES_OK)
59  pg_fatal("could not clear search_path: %s",
61  PQclear(res);
62 
63  /*
64  * Check that the server is not in hot standby mode. There is no
65  * fundamental reason that couldn't be made to work, but it doesn't
66  * currently because we use a temporary table. Better to check for it
67  * explicitly than error out, for a better error message.
68  */
69  str = run_simple_query("SELECT pg_is_in_recovery()");
70  if (strcmp(str, "f") != 0)
71  pg_fatal("source server must not be in recovery mode\n");
72  pg_free(str);
73 
74  /*
75  * Also check that full_page_writes is enabled. We can get torn pages if
76  * a page is modified while we read it with pg_read_binary_file(), and we
77  * rely on full page images to fix them.
78  */
79  str = run_simple_query("SHOW full_page_writes");
80  if (strcmp(str, "on") != 0)
81  pg_fatal("full_page_writes must be enabled in the source server\n");
82  pg_free(str);
83 
84  /*
85  * Although we don't do any "real" updates, we do work with a temporary
86  * table. We don't care about synchronous commit for that. It doesn't
87  * otherwise matter much, but if the server is using synchronous
88  * replication, and replication isn't working for some reason, we don't
89  * want to get stuck, waiting for it to start working again.
90  */
91  res = PQexec(conn, "SET synchronous_commit = off");
92  if (PQresultStatus(res) != PGRES_COMMAND_OK)
93  pg_fatal("could not set up connection context: %s",
95  PQclear(res);
96 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6116
void pg_fatal(const char *fmt,...)
Definition: logging.c:83
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
static char * run_simple_query(const char *sql)
Definition: libpq_fetch.c:103
static PGconn * conn
Definition: libpq_fetch.c:29
void pg_log(eLogType type, const char *fmt,...)
Definition: logging.c:69
void PQclear(PGresult *res)
Definition: fe-exec.c:671
void pg_free(void *ptr)
Definition: fe_memutils.c:105
#define ALWAYS_SECURE_SEARCH_PATH_SQL
Definition: connect.h:25
char * PQresultErrorMessage(const PGresult *res)
Definition: fe-exec.c:2663
PGresult * PQexec(PGconn *conn, const char *query)
Definition: fe-exec.c:1897
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6063
PGconn * PQconnectdb(const char *conninfo)
Definition: fe-connect.c:592
static char * connstr
Definition: pg_dumpall.c:59

◆ 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().

130 {
131  XLogRecPtr result;
132  uint32 hi;
133  uint32 lo;
134  char *val;
135 
136  val = run_simple_query("SELECT pg_current_wal_insert_lsn()");
137 
138  if (sscanf(val, "%X/%X", &hi, &lo) != 2)
139  pg_fatal("unrecognized result \"%s\" for current WAL insert location\n", val);
140 
141  result = ((uint64) hi) << 32 | lo;
142 
143  pg_free(val);
144 
145  return result;
146 }
void pg_fatal(const char *fmt,...)
Definition: logging.c:83
static char * run_simple_query(const char *sql)
Definition: libpq_fetch.c:103
unsigned int uint32
Definition: c.h:325
uint64 XLogRecPtr
Definition: xlogdefs.h:21
void pg_free(void *ptr)
Definition: fe_memutils.c:105
long val
Definition: informix.c:689

◆ 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().

354 {
355  PGresult *res;
356  char *result;
357  int len;
358  const char *paramValues[1];
359 
360  paramValues[0] = filename;
361  res = PQexecParams(conn, "SELECT pg_read_binary_file($1)",
362  1, NULL, paramValues, NULL, NULL, 1);
363 
364  if (PQresultStatus(res) != PGRES_TUPLES_OK)
365  pg_fatal("could not fetch remote file \"%s\": %s",
367 
368  /* sanity check the result set */
369  if (PQntuples(res) != 1 || PQgetisnull(res, 0, 0))
370  pg_fatal("unexpected result set while fetching remote file \"%s\"\n",
371  filename);
372 
373  /* Read result to local variables */
374  len = PQgetlength(res, 0, 0);
375  result = pg_malloc(len + 1);
376  memcpy(result, PQgetvalue(res, 0, 0), len);
377  result[len] = '\0';
378 
379  PQclear(res);
380 
381  pg_log(PG_DEBUG, "fetched file \"%s\", length %d\n", filename, len);
382 
383  if (filesize)
384  *filesize = len;
385  return result;
386 }
int PQgetlength(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3129
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3118
void * pg_malloc(size_t size)
Definition: fe_memutils.c:47
void pg_fatal(const char *fmt,...)
Definition: logging.c:83
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2724
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
static PGconn * conn
Definition: libpq_fetch.c:29
void pg_log(eLogType type, const char *fmt,...)
Definition: logging.c:69
void PQclear(PGresult *res)
Definition: fe-exec.c:671
static char * filename
Definition: pg_dumpall.c:87
char * PQresultErrorMessage(const PGresult *res)
Definition: fe-exec.c:2663
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:1911
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3143

◆ 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().

153 {
154  PGresult *res;
155  const char *sql;
156  int i;
157 
158  /*
159  * Create a recursive directory listing of the whole data directory.
160  *
161  * The WITH RECURSIVE part does most of the work. The second part gets the
162  * targets of the symlinks in pg_tblspc directory.
163  *
164  * XXX: There is no backend function to get a symbolic link's target in
165  * general, so if the admin has put any custom symbolic links in the data
166  * directory, they won't be copied correctly.
167  */
168  sql =
169  "WITH RECURSIVE files (path, filename, size, isdir) AS (\n"
170  " SELECT '' AS path, filename, size, isdir FROM\n"
171  " (SELECT pg_ls_dir('.', true, false) AS filename) AS fn,\n"
172  " pg_stat_file(fn.filename, true) AS this\n"
173  " UNION ALL\n"
174  " SELECT parent.path || parent.filename || '/' AS path,\n"
175  " fn, this.size, this.isdir\n"
176  " FROM files AS parent,\n"
177  " pg_ls_dir(parent.path || parent.filename, true, false) AS fn,\n"
178  " pg_stat_file(parent.path || parent.filename || '/' || fn, true) AS this\n"
179  " WHERE parent.isdir = 't'\n"
180  ")\n"
181  "SELECT path || filename, size, isdir,\n"
182  " pg_tablespace_location(pg_tablespace.oid) AS link_target\n"
183  "FROM files\n"
184  "LEFT OUTER JOIN pg_tablespace ON files.path = 'pg_tblspc/'\n"
185  " AND oid::text = files.filename\n";
186  res = PQexec(conn, sql);
187 
188  if (PQresultStatus(res) != PGRES_TUPLES_OK)
189  pg_fatal("could not fetch file list: %s",
190  PQresultErrorMessage(res));
191 
192  /* sanity check the result set */
193  if (PQnfields(res) != 4)
194  pg_fatal("unexpected result set while fetching file list\n");
195 
196  /* Read result to local variables */
197  for (i = 0; i < PQntuples(res); i++)
198  {
199  char *path = PQgetvalue(res, i, 0);
200  int64 filesize = atol(PQgetvalue(res, i, 1));
201  bool isdir = (strcmp(PQgetvalue(res, i, 2), "t") == 0);
202  char *link_target = PQgetvalue(res, i, 3);
204 
205  if (PQgetisnull(res, 0, 1))
206  {
207  /*
208  * The file was removed from the server while the query was
209  * running. Ignore it.
210  */
211  continue;
212  }
213 
214  if (link_target[0])
215  type = FILE_TYPE_SYMLINK;
216  else if (isdir)
217  type = FILE_TYPE_DIRECTORY;
218  else
219  type = FILE_TYPE_REGULAR;
220 
221  process_source_file(path, type, filesize, link_target);
222  }
223  PQclear(res);
224 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2732
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3118
void pg_fatal(const char *fmt,...)
Definition: logging.c:83
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2724
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
static PGconn * conn
Definition: libpq_fetch.c:29
void PQclear(PGresult *res)
Definition: fe-exec.c:671
void process_source_file(const char *path, file_type_t type, size_t newsize, const char *link_target)
Definition: filemap.c:137
char * PQresultErrorMessage(const PGresult *res)
Definition: fe-exec.c:2663
int i
PGresult * PQexec(PGconn *conn, const char *query)
Definition: fe-exec.c:1897
file_type_t
Definition: filemap.h:35
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3143

◆ receiveFileChunks()

static void receiveFileChunks ( const char *  sql)
static

Definition at line 237 of file libpq_fetch.c.

References filename, INT64_FORMAT, open_target_file(), PG_DEBUG, pg_fatal(), pg_free(), pg_log(), pg_malloc(), pg_ntoh64, PGRES_SINGLE_TUPLE, PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQfformat(), PQftype(), PQgetisnull(), PQgetlength(), PQgetResult(), PQgetvalue(), PQnfields(), PQntuples(), PQresultErrorMessage(), PQresultStatus(), PQsendQueryParams(), PQsetSingleRowMode(), remove_target_file(), snprintf(), and write_target_range().

Referenced by libpq_executeFileMap().

238 {
239  PGresult *res;
240 
241  if (PQsendQueryParams(conn, sql, 0, NULL, NULL, NULL, NULL, 1) != 1)
242  pg_fatal("could not send query: %s", PQerrorMessage(conn));
243 
244  pg_log(PG_DEBUG, "getting file chunks\n");
245 
246  if (PQsetSingleRowMode(conn) != 1)
247  pg_fatal("could not set libpq connection to single row mode\n");
248 
249  while ((res = PQgetResult(conn)) != NULL)
250  {
251  char *filename;
252  int filenamelen;
253  int64 chunkoff;
254  char chunkoff_str[32];
255  int chunksize;
256  char *chunk;
257 
258  switch (PQresultStatus(res))
259  {
260  case PGRES_SINGLE_TUPLE:
261  break;
262 
263  case PGRES_TUPLES_OK:
264  PQclear(res);
265  continue; /* final zero-row result */
266 
267  default:
268  pg_fatal("unexpected result while fetching remote files: %s",
269  PQresultErrorMessage(res));
270  }
271 
272  /* sanity check the result set */
273  if (PQnfields(res) != 3 || PQntuples(res) != 1)
274  pg_fatal("unexpected result set size while fetching remote files\n");
275 
276  if (PQftype(res, 0) != TEXTOID ||
277  PQftype(res, 1) != INT8OID ||
278  PQftype(res, 2) != BYTEAOID)
279  {
280  pg_fatal("unexpected data types in result set while fetching remote files: %u %u %u\n",
281  PQftype(res, 0), PQftype(res, 1), PQftype(res, 2));
282  }
283 
284  if (PQfformat(res, 0) != 1 &&
285  PQfformat(res, 1) != 1 &&
286  PQfformat(res, 2) != 1)
287  {
288  pg_fatal("unexpected result format while fetching remote files\n");
289  }
290 
291  if (PQgetisnull(res, 0, 0) ||
292  PQgetisnull(res, 0, 1))
293  {
294  pg_fatal("unexpected null values in result while fetching remote files\n");
295  }
296 
297  if (PQgetlength(res, 0, 1) != sizeof(int64))
298  pg_fatal("unexpected result length while fetching remote files\n");
299 
300  /* Read result set to local variables */
301  memcpy(&chunkoff, PQgetvalue(res, 0, 1), sizeof(int64));
302  chunkoff = pg_ntoh64(chunkoff);
303  chunksize = PQgetlength(res, 0, 2);
304 
305  filenamelen = PQgetlength(res, 0, 0);
306  filename = pg_malloc(filenamelen + 1);
307  memcpy(filename, PQgetvalue(res, 0, 0), filenamelen);
308  filename[filenamelen] = '\0';
309 
310  chunk = PQgetvalue(res, 0, 2);
311 
312  /*
313  * If a file has been deleted on the source, remove it on the target
314  * as well. Note that multiple unlink() calls may happen on the same
315  * file if multiple data chunks are associated with it, hence ignore
316  * unconditionally anything missing. If this file is not a relation
317  * data file, then it has been already truncated when creating the
318  * file chunk list at the previous execution of the filemap.
319  */
320  if (PQgetisnull(res, 0, 2))
321  {
323  "received null value for chunk for file \"%s\", file has been deleted\n",
324  filename);
325  remove_target_file(filename, true);
326  pg_free(filename);
327  PQclear(res);
328  continue;
329  }
330 
331  /*
332  * Separate step to keep platform-dependent format code out of
333  * translatable strings.
334  */
335  snprintf(chunkoff_str, sizeof(chunkoff_str), INT64_FORMAT, chunkoff);
336  pg_log(PG_DEBUG, "received chunk for file \"%s\", offset %s, size %d\n",
337  filename, chunkoff_str, chunksize);
338 
339  open_target_file(filename, false);
340 
341  write_target_range(chunk, chunkoff, chunksize);
342 
343  pg_free(filename);
344 
345  PQclear(res);
346  }
347 }
int PQgetlength(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3129
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2732
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6116
void open_target_file(const char *path, bool trunc)
Definition: file_ops.c:43
void write_target_range(char *buf, off_t begin, size_t size)
Definition: file_ops.c:84
int PQsendQueryParams(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:1234
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3118
void * pg_malloc(size_t size)
Definition: fe_memutils.c:47
void remove_target_file(const char *path, bool missing_ok)
Definition: file_ops.c:173
int snprintf(char *str, size_t count, const char *fmt,...) pg_attribute_printf(3
void pg_fatal(const char *fmt,...)
Definition: logging.c:83
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2724
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
Oid PQftype(const PGresult *res, int field_num)
Definition: fe-exec.c:2962
int PQsetSingleRowMode(PGconn *conn)
Definition: fe-exec.c:1655
static PGconn * conn
Definition: libpq_fetch.c:29
void pg_log(eLogType type, const char *fmt,...)
Definition: logging.c:69
void PQclear(PGresult *res)
Definition: fe-exec.c:671
void pg_free(void *ptr)
Definition: fe_memutils.c:105
#define INT64_FORMAT
Definition: c.h:367
static char * filename
Definition: pg_dumpall.c:87
char * PQresultErrorMessage(const PGresult *res)
Definition: fe-exec.c:2663
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3143
int PQfformat(const PGresult *res, int field_num)
Definition: fe-exec.c:2951
#define pg_ntoh64(x)
Definition: pg_bswap.h:126
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1753

◆ run_simple_query()

static char * run_simple_query ( const char *  sql)
static

Definition at line 103 of file libpq_fetch.c.

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

Referenced by libpqConnect(), and libpqGetCurrentXlogInsertLocation().

104 {
105  PGresult *res;
106  char *result;
107 
108  res = PQexec(conn, sql);
109 
110  if (PQresultStatus(res) != PGRES_TUPLES_OK)
111  pg_fatal("error running query (%s) in source server: %s",
112  sql, PQresultErrorMessage(res));
113 
114  /* sanity check the result set */
115  if (PQnfields(res) != 1 || PQntuples(res) != 1 || PQgetisnull(res, 0, 0))
116  pg_fatal("unexpected result set from query\n");
117 
118  result = pg_strdup(PQgetvalue(res, 0, 0));
119 
120  PQclear(res);
121 
122  return result;
123 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2732
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3118
void pg_fatal(const char *fmt,...)
Definition: logging.c:83
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2724
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
char * pg_strdup(const char *in)
Definition: fe_memutils.c:85
static PGconn * conn
Definition: libpq_fetch.c:29
void PQclear(PGresult *res)
Definition: fe-exec.c:671
char * PQresultErrorMessage(const PGresult *res)
Definition: fe-exec.c:2663
PGresult * PQexec(PGconn *conn, const char *query)
Definition: fe-exec.c:1897
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3143

Variable Documentation

◆ conn

PGconn* conn = NULL
static

Definition at line 29 of file libpq_fetch.c.