PostgreSQL Source Code  git master
fetch.h File Reference
#include "access/xlogdefs.h"
#include "filemap.h"
Include dependency graph for fetch.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Typedefs

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

Functions

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

Typedef Documentation

◆ process_file_callback_t

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

Definition at line 41 of file fetch.h.

Function Documentation

◆ copy_executeFileMap()

void copy_executeFileMap ( filemap_t map)

Definition at line 205 of file copy_fetch.c.

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

Referenced by executeFileMap().

206 {
207  file_entry_t *entry;
208  int i;
209 
210  for (i = 0; i < map->narray; i++)
211  {
212  entry = map->array[i];
213  execute_pagemap(&entry->pagemap, entry->path);
214 
215  switch (entry->action)
216  {
217  case FILE_ACTION_NONE:
218  /* ok, do nothing.. */
219  break;
220 
221  case FILE_ACTION_COPY:
222  rewind_copy_file_range(entry->path, 0, entry->newsize, true);
223  break;
224 
226  truncate_target_file(entry->path, entry->newsize);
227  break;
228 
230  rewind_copy_file_range(entry->path, entry->oldsize,
231  entry->newsize, false);
232  break;
233 
234  case FILE_ACTION_CREATE:
235  create_target(entry);
236  break;
237 
238  case FILE_ACTION_REMOVE:
239  remove_target(entry);
240  break;
241  }
242  }
243 
245 }
static void rewind_copy_file_range(const char *path, off_t begin, off_t end, bool trunc)
Definition: copy_fetch.c:156
file_entry_t ** array
Definition: filemap.h:78
size_t newsize
Definition: filemap.h:50
datapagemap_t pagemap
Definition: filemap.h:53
int narray
Definition: filemap.h:79
void truncate_target_file(const char *path, off_t newsize)
Definition: file_ops.c:191
static void execute_pagemap(datapagemap_t *pagemap, const char *path)
Definition: copy_fetch.c:248
file_action_t action
Definition: filemap.h:46
size_t oldsize
Definition: filemap.h:49
void remove_target(file_entry_t *entry)
Definition: file_ops.c:125
char * path
Definition: filemap.h:43
void close_target_file(void)
Definition: file_ops.c:70
int i
Definition: filemap.h:41
void create_target(file_entry_t *entry)
Definition: file_ops.c:146

◆ executeFileMap()

void executeFileMap ( void  )

Definition at line 40 of file fetch.c.

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

Referenced by main().

41 {
42  if (datadir_source)
44  else
46 }
filemap_t * filemap
Definition: filemap.c:23
void copy_executeFileMap(filemap_t *map)
Definition: copy_fetch.c:205
char * datadir_source
Definition: pg_rewind.c:54
void libpq_executeFileMap(filemap_t *map)
Definition: libpq_fetch.c:436

◆ fetchFile()

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

Definition at line 54 of file fetch.c.

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

Referenced by getTimelineHistory(), and main().

55 {
56  if (datadir_source)
57  return slurpFile(datadir_source, filename, filesize);
58  else
59  return libpqGetFile(filename, filesize);
60 }
char * datadir_source
Definition: pg_rewind.c:54
char * libpqGetFile(const char *filename, size_t *filesize)
Definition: libpq_fetch.c:364
static char * filename
Definition: pg_dumpall.c:90
char * slurpFile(const char *datadir, const char *path, size_t *filesize)
Definition: file_ops.c:284

◆ fetchSourceFileList()

void fetchSourceFileList ( void  )

Definition at line 28 of file fetch.c.

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

Referenced by main().

29 {
30  if (datadir_source)
32  else
34 }
void traverse_datadir(const char *datadir, process_file_callback_t callback)
Definition: copy_fetch.c:33
char * datadir_source
Definition: pg_rewind.c:54
void libpqProcessFileList(void)
Definition: libpq_fetch.c:170
void process_source_file(const char *path, file_type_t type, size_t newsize, const char *link_target)
Definition: filemap.c:135

◆ libpq_executeFileMap()

void libpq_executeFileMap ( filemap_t map)

Definition at line 436 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(), run_simple_command(), and truncate_target_file().

Referenced by executeFileMap().

437 {
438  file_entry_t *entry;
439  const char *sql;
440  PGresult *res;
441  int i;
442 
443  /*
444  * First create a temporary table, and load it with the blocks that we
445  * need to fetch.
446  */
447  sql = "CREATE TEMPORARY TABLE fetchchunks(path text, begin int8, len int4);";
448  run_simple_command(sql);
449 
450  sql = "COPY fetchchunks FROM STDIN";
451  res = PQexec(conn, sql);
452 
453  if (PQresultStatus(res) != PGRES_COPY_IN)
454  pg_fatal("could not send file list: %s",
455  PQresultErrorMessage(res));
456  PQclear(res);
457 
458  for (i = 0; i < map->narray; i++)
459  {
460  entry = map->array[i];
461 
462  /* If this is a relation file, copy the modified blocks */
463  execute_pagemap(&entry->pagemap, entry->path);
464 
465  switch (entry->action)
466  {
467  case FILE_ACTION_NONE:
468  /* nothing else to do */
469  break;
470 
471  case FILE_ACTION_COPY:
472  /* Truncate the old file out of the way, if any */
473  open_target_file(entry->path, true);
474  fetch_file_range(entry->path, 0, entry->newsize);
475  break;
476 
478  truncate_target_file(entry->path, entry->newsize);
479  break;
480 
482  fetch_file_range(entry->path, entry->oldsize, entry->newsize);
483  break;
484 
485  case FILE_ACTION_REMOVE:
486  remove_target(entry);
487  break;
488 
489  case FILE_ACTION_CREATE:
490  create_target(entry);
491  break;
492  }
493  }
494 
495  if (PQputCopyEnd(conn, NULL) != 1)
496  pg_fatal("could not send end-of-COPY: %s",
498 
499  while ((res = PQgetResult(conn)) != NULL)
500  {
501  if (PQresultStatus(res) != PGRES_COMMAND_OK)
502  pg_fatal("unexpected result while sending file list: %s",
503  PQresultErrorMessage(res));
504  PQclear(res);
505  }
506 
507  /*
508  * We've now copied the list of file ranges that we need to fetch to the
509  * temporary table. Now, actually fetch all of those ranges.
510  */
511  sql =
512  "SELECT path, begin,\n"
513  " pg_read_binary_file(path, begin, len, true) AS chunk\n"
514  "FROM fetchchunks\n";
515 
516  receiveFileChunks(sql);
517 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6623
void open_target_file(const char *path, bool trunc)
Definition: file_ops.c:42
file_entry_t ** array
Definition: filemap.h:78
size_t newsize
Definition: filemap.h:50
#define pg_fatal(...)
Definition: pg_rewind.h:41
static void execute_pagemap(datapagemap_t *pagemap, const char *path)
Definition: libpq_fetch.c:520
int PQputCopyEnd(PGconn *conn, const char *errormsg)
Definition: fe-exec.c:2384
datapagemap_t pagemap
Definition: filemap.h:53
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2692
int narray
Definition: filemap.h:79
void truncate_target_file(const char *path, off_t newsize)
Definition: file_ops.c:191
file_action_t action
Definition: filemap.h:46
static void receiveFileChunks(const char *sql)
Definition: libpq_fetch.c:255
size_t oldsize
Definition: filemap.h:49
PGconn * conn
Definition: libpq_fetch.c:26
static void run_simple_command(const char *sql)
Definition: libpq_fetch.c:130
void PQclear(PGresult *res)
Definition: fe-exec.c:694
void remove_target(file_entry_t *entry)
Definition: file_ops.c:125
char * path
Definition: filemap.h:43
char * PQresultErrorMessage(const PGresult *res)
Definition: fe-exec.c:2708
static void fetch_file_range(const char *path, uint64 begin, uint64 end)
Definition: libpq_fetch.c:407
int i
Definition: filemap.h:41
PGresult * PQexec(PGconn *conn, const char *query)
Definition: fe-exec.c:1939
void create_target(file_entry_t *entry)
Definition: file_ops.c:146
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1778

◆ libpqConnect()

void libpqConnect ( const char *  connstr)

Definition at line 43 of file libpq_fetch.c.

References ALWAYS_SECURE_SEARCH_PATH_SQL, CONNECTION_BAD, pg_fatal, pg_free(), pg_log_info, PGRES_TUPLES_OK, PQclear(), PQconnectdb(), PQerrorMessage(), PQexec(), PQresultErrorMessage(), PQresultStatus(), PQstatus(), run_simple_command(), run_simple_query(), showprogress, and generate_unaccent_rules::str.

Referenced by main().

44 {
45  char *str;
46  PGresult *res;
47 
50  pg_fatal("could not connect to server: %s",
52 
53  if (showprogress)
54  pg_log_info("connected to server");
55 
56  /* disable all types of timeouts */
57  run_simple_command("SET statement_timeout = 0");
58  run_simple_command("SET lock_timeout = 0");
59  run_simple_command("SET idle_in_transaction_session_timeout = 0");
60 
62  if (PQresultStatus(res) != PGRES_TUPLES_OK)
63  pg_fatal("could not clear search_path: %s",
65  PQclear(res);
66 
67  /*
68  * Check that the server is not in hot standby mode. There is no
69  * fundamental reason that couldn't be made to work, but it doesn't
70  * currently because we use a temporary table. Better to check for it
71  * explicitly than error out, for a better error message.
72  */
73  str = run_simple_query("SELECT pg_is_in_recovery()");
74  if (strcmp(str, "f") != 0)
75  pg_fatal("source server must not be in recovery mode");
76  pg_free(str);
77 
78  /*
79  * Also check that full_page_writes is enabled. We can get torn pages if
80  * a page is modified while we read it with pg_read_binary_file(), and we
81  * rely on full page images to fix them.
82  */
83  str = run_simple_query("SHOW full_page_writes");
84  if (strcmp(str, "on") != 0)
85  pg_fatal("full_page_writes must be enabled in the source server");
86  pg_free(str);
87 
88  /*
89  * Although we don't do any "real" updates, we do work with a temporary
90  * table. We don't care about synchronous commit for that. It doesn't
91  * otherwise matter much, but if the server is using synchronous
92  * replication, and replication isn't working for some reason, we don't
93  * want to get stuck, waiting for it to start working again.
94  */
95  run_simple_command("SET synchronous_commit = off");
96 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6623
#define pg_fatal(...)
Definition: pg_rewind.h:41
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2692
static char * run_simple_query(const char *sql)
Definition: libpq_fetch.c:103
PGconn * conn
Definition: libpq_fetch.c:26
static bool showprogress
static void run_simple_command(const char *sql)
Definition: libpq_fetch.c:130
void PQclear(PGresult *res)
Definition: fe-exec.c:694
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:2708
PGresult * PQexec(PGconn *conn, const char *query)
Definition: fe-exec.c:1939
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6570
#define pg_log_info(...)
Definition: logging.h:87
PGconn * PQconnectdb(const char *conninfo)
Definition: fe-connect.c:680
static char * connstr
Definition: pg_dumpall.c:61

◆ libpqGetCurrentXlogInsertLocation()

XLogRecPtr libpqGetCurrentXlogInsertLocation ( void  )

Definition at line 147 of file libpq_fetch.c.

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

Referenced by main().

148 {
149  XLogRecPtr result;
150  uint32 hi;
151  uint32 lo;
152  char *val;
153 
154  val = run_simple_query("SELECT pg_current_wal_insert_lsn()");
155 
156  if (sscanf(val, "%X/%X", &hi, &lo) != 2)
157  pg_fatal("unrecognized result \"%s\" for current WAL insert location", val);
158 
159  result = ((uint64) hi) << 32 | lo;
160 
161  pg_free(val);
162 
163  return result;
164 }
#define pg_fatal(...)
Definition: pg_rewind.h:41
static char * run_simple_query(const char *sql)
Definition: libpq_fetch.c:103
unsigned int uint32
Definition: c.h:359
uint64 XLogRecPtr
Definition: xlogdefs.h:21
void pg_free(void *ptr)
Definition: fe_memutils.c:105
long val
Definition: informix.c:664

◆ libpqGetFile()

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

Definition at line 364 of file libpq_fetch.c.

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

Referenced by fetchFile().

365 {
366  PGresult *res;
367  char *result;
368  int len;
369  const char *paramValues[1];
370 
371  paramValues[0] = filename;
372  res = PQexecParams(conn, "SELECT pg_read_binary_file($1)",
373  1, NULL, paramValues, NULL, NULL, 1);
374 
375  if (PQresultStatus(res) != PGRES_TUPLES_OK)
376  pg_fatal("could not fetch remote file \"%s\": %s",
378 
379  /* sanity check the result set */
380  if (PQntuples(res) != 1 || PQgetisnull(res, 0, 0))
381  pg_fatal("unexpected result set while fetching remote file \"%s\"",
382  filename);
383 
384  /* Read result to local variables */
385  len = PQgetlength(res, 0, 0);
386  result = pg_malloc(len + 1);
387  memcpy(result, PQgetvalue(res, 0, 0), len);
388  result[len] = '\0';
389 
390  PQclear(res);
391 
392  pg_log_debug("fetched file \"%s\", length %d", filename, len);
393 
394  if (filesize)
395  *filesize = len;
396  return result;
397 }
int PQgetlength(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3174
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3163
void * pg_malloc(size_t size)
Definition: fe_memutils.c:47
#define pg_fatal(...)
Definition: pg_rewind.h:41
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2769
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2692
#define pg_log_debug(...)
Definition: logging.h:91
PGconn * conn
Definition: libpq_fetch.c:26
void PQclear(PGresult *res)
Definition: fe-exec.c:694
static char * filename
Definition: pg_dumpall.c:90
char * PQresultErrorMessage(const PGresult *res)
Definition: fe-exec.c:2708
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:1953
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3188

◆ libpqProcessFileList()

void libpqProcessFileList ( void  )

Definition at line 170 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().

171 {
172  PGresult *res;
173  const char *sql;
174  int i;
175 
176  /*
177  * Create a recursive directory listing of the whole data directory.
178  *
179  * The WITH RECURSIVE part does most of the work. The second part gets the
180  * targets of the symlinks in pg_tblspc directory.
181  *
182  * XXX: There is no backend function to get a symbolic link's target in
183  * general, so if the admin has put any custom symbolic links in the data
184  * directory, they won't be copied correctly.
185  */
186  sql =
187  "WITH RECURSIVE files (path, filename, size, isdir) AS (\n"
188  " SELECT '' AS path, filename, size, isdir FROM\n"
189  " (SELECT pg_ls_dir('.', true, false) AS filename) AS fn,\n"
190  " pg_stat_file(fn.filename, true) AS this\n"
191  " UNION ALL\n"
192  " SELECT parent.path || parent.filename || '/' AS path,\n"
193  " fn, this.size, this.isdir\n"
194  " FROM files AS parent,\n"
195  " pg_ls_dir(parent.path || parent.filename, true, false) AS fn,\n"
196  " pg_stat_file(parent.path || parent.filename || '/' || fn, true) AS this\n"
197  " WHERE parent.isdir = 't'\n"
198  ")\n"
199  "SELECT path || filename, size, isdir,\n"
200  " pg_tablespace_location(pg_tablespace.oid) AS link_target\n"
201  "FROM files\n"
202  "LEFT OUTER JOIN pg_tablespace ON files.path = 'pg_tblspc/'\n"
203  " AND oid::text = files.filename\n";
204  res = PQexec(conn, sql);
205 
206  if (PQresultStatus(res) != PGRES_TUPLES_OK)
207  pg_fatal("could not fetch file list: %s",
208  PQresultErrorMessage(res));
209 
210  /* sanity check the result set */
211  if (PQnfields(res) != 4)
212  pg_fatal("unexpected result set while fetching file list");
213 
214  /* Read result to local variables */
215  for (i = 0; i < PQntuples(res); i++)
216  {
217  char *path = PQgetvalue(res, i, 0);
218  int64 filesize = atol(PQgetvalue(res, i, 1));
219  bool isdir = (strcmp(PQgetvalue(res, i, 2), "t") == 0);
220  char *link_target = PQgetvalue(res, i, 3);
222 
223  if (PQgetisnull(res, 0, 1))
224  {
225  /*
226  * The file was removed from the server while the query was
227  * running. Ignore it.
228  */
229  continue;
230  }
231 
232  if (link_target[0])
233  type = FILE_TYPE_SYMLINK;
234  else if (isdir)
235  type = FILE_TYPE_DIRECTORY;
236  else
237  type = FILE_TYPE_REGULAR;
238 
239  process_source_file(path, type, filesize, link_target);
240  }
241  PQclear(res);
242 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2777
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3163
#define pg_fatal(...)
Definition: pg_rewind.h:41
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2769
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2692
PGconn * conn
Definition: libpq_fetch.c:26
void PQclear(PGresult *res)
Definition: fe-exec.c:694
void process_source_file(const char *path, file_type_t type, size_t newsize, const char *link_target)
Definition: filemap.c:135
char * PQresultErrorMessage(const PGresult *res)
Definition: fe-exec.c:2708
int i
PGresult * PQexec(PGconn *conn, const char *query)
Definition: fe-exec.c:1939
file_type_t
Definition: filemap.h:34
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3188

◆ traverse_datadir()

void traverse_datadir ( const char *  datadir,
process_file_callback_t  callback 
)

Definition at line 33 of file copy_fetch.c.

References recurse_dir().

Referenced by fetchSourceFileList(), and main().

34 {
36 }
static void recurse_dir(const char *datadir, const char *path, process_file_callback_t callback)
Definition: copy_fetch.c:45
static void callback(struct sockaddr *addr, struct sockaddr *mask, void *unused)
Definition: test_ifaddrs.c:48
char * datadir