PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
libpq_fetch.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  *
3  * libpq_fetch.c
4  * Functions for fetching files from a remote server.
5  *
6  * Copyright (c) 2013-2017, PostgreSQL Global Development Group
7  *
8  *-------------------------------------------------------------------------
9  */
10 #include "postgres_fe.h"
11 
12 #include <sys/stat.h>
13 #include <dirent.h>
14 #include <fcntl.h>
15 #include <unistd.h>
16 
17 #include "pg_rewind.h"
18 #include "datapagemap.h"
19 #include "fetch.h"
20 #include "file_ops.h"
21 #include "filemap.h"
22 #include "logging.h"
23 
24 #include "libpq-fe.h"
25 #include "catalog/catalog.h"
26 #include "catalog/pg_type.h"
27 #include "port/pg_bswap.h"
28 
29 static PGconn *conn = NULL;
30 
31 /*
32  * Files are fetched max CHUNKSIZE bytes at a time.
33  *
34  * (This only applies to files that are copied in whole, or for truncated
35  * files where we copy the tail. Relation files, where we know the individual
36  * blocks that need to be fetched, are fetched in BLCKSZ chunks.)
37  */
38 #define CHUNKSIZE 1000000
39 
40 static void receiveFileChunks(const char *sql);
41 static void execute_pagemap(datapagemap_t *pagemap, const char *path);
42 static char *run_simple_query(const char *sql);
43 
44 void
45 libpqConnect(const char *connstr)
46 {
47  char *str;
48  PGresult *res;
49 
50  conn = PQconnectdb(connstr);
51  if (PQstatus(conn) == CONNECTION_BAD)
52  pg_fatal("could not connect to server: %s",
53  PQerrorMessage(conn));
54 
55  pg_log(PG_PROGRESS, "connected to server\n");
56 
57  /*
58  * Check that the server is not in hot standby mode. There is no
59  * fundamental reason that couldn't be made to work, but it doesn't
60  * currently because we use a temporary table. Better to check for it
61  * explicitly than error out, for a better error message.
62  */
63  str = run_simple_query("SELECT pg_is_in_recovery()");
64  if (strcmp(str, "f") != 0)
65  pg_fatal("source server must not be in recovery mode\n");
66  pg_free(str);
67 
68  /*
69  * Also check that full_page_writes is enabled. We can get torn pages if
70  * a page is modified while we read it with pg_read_binary_file(), and we
71  * rely on full page images to fix them.
72  */
73  str = run_simple_query("SHOW full_page_writes");
74  if (strcmp(str, "on") != 0)
75  pg_fatal("full_page_writes must be enabled in the source server\n");
76  pg_free(str);
77 
78  /*
79  * Although we don't do any "real" updates, we do work with a temporary
80  * table. We don't care about synchronous commit for that. It doesn't
81  * otherwise matter much, but if the server is using synchronous
82  * replication, and replication isn't working for some reason, we don't
83  * want to get stuck, waiting for it to start working again.
84  */
85  res = PQexec(conn, "SET synchronous_commit = off");
86  if (PQresultStatus(res) != PGRES_COMMAND_OK)
87  pg_fatal("could not set up connection context: %s",
89  PQclear(res);
90 }
91 
92 /*
93  * Runs a query that returns a single value.
94  * The result should be pg_free'd after use.
95  */
96 static char *
97 run_simple_query(const char *sql)
98 {
99  PGresult *res;
100  char *result;
101 
102  res = PQexec(conn, sql);
103 
104  if (PQresultStatus(res) != PGRES_TUPLES_OK)
105  pg_fatal("error running query (%s) in source server: %s",
106  sql, PQresultErrorMessage(res));
107 
108  /* sanity check the result set */
109  if (PQnfields(res) != 1 || PQntuples(res) != 1 || PQgetisnull(res, 0, 0))
110  pg_fatal("unexpected result set from query\n");
111 
112  result = pg_strdup(PQgetvalue(res, 0, 0));
113 
114  PQclear(res);
115 
116  return result;
117 }
118 
119 /*
120  * Calls pg_current_wal_insert_lsn() function
121  */
124 {
125  XLogRecPtr result;
126  uint32 hi;
127  uint32 lo;
128  char *val;
129 
130  val = run_simple_query("SELECT pg_current_wal_insert_lsn()");
131 
132  if (sscanf(val, "%X/%X", &hi, &lo) != 2)
133  pg_fatal("unrecognized result \"%s\" for current WAL insert location\n", val);
134 
135  result = ((uint64) hi) << 32 | lo;
136 
137  pg_free(val);
138 
139  return result;
140 }
141 
142 /*
143  * Get a list of all files in the data directory.
144  */
145 void
147 {
148  PGresult *res;
149  const char *sql;
150  int i;
151 
152  /*
153  * Create a recursive directory listing of the whole data directory.
154  *
155  * The WITH RECURSIVE part does most of the work. The second part gets the
156  * targets of the symlinks in pg_tblspc directory.
157  *
158  * XXX: There is no backend function to get a symbolic link's target in
159  * general, so if the admin has put any custom symbolic links in the data
160  * directory, they won't be copied correctly.
161  */
162  sql =
163  "WITH RECURSIVE files (path, filename, size, isdir) AS (\n"
164  " SELECT '' AS path, filename, size, isdir FROM\n"
165  " (SELECT pg_ls_dir('.', true, false) AS filename) AS fn,\n"
166  " pg_stat_file(fn.filename, true) AS this\n"
167  " UNION ALL\n"
168  " SELECT parent.path || parent.filename || '/' AS path,\n"
169  " fn, this.size, this.isdir\n"
170  " FROM files AS parent,\n"
171  " pg_ls_dir(parent.path || parent.filename, true, false) AS fn,\n"
172  " pg_stat_file(parent.path || parent.filename || '/' || fn, true) AS this\n"
173  " WHERE parent.isdir = 't'\n"
174  ")\n"
175  "SELECT path || filename, size, isdir,\n"
176  " pg_tablespace_location(pg_tablespace.oid) AS link_target\n"
177  "FROM files\n"
178  "LEFT OUTER JOIN pg_tablespace ON files.path = 'pg_tblspc/'\n"
179  " AND oid::text = files.filename\n";
180  res = PQexec(conn, sql);
181 
182  if (PQresultStatus(res) != PGRES_TUPLES_OK)
183  pg_fatal("could not fetch file list: %s",
184  PQresultErrorMessage(res));
185 
186  /* sanity check the result set */
187  if (PQnfields(res) != 4)
188  pg_fatal("unexpected result set while fetching file list\n");
189 
190  /* Read result to local variables */
191  for (i = 0; i < PQntuples(res); i++)
192  {
193  char *path = PQgetvalue(res, i, 0);
194  int64 filesize = atol(PQgetvalue(res, i, 1));
195  bool isdir = (strcmp(PQgetvalue(res, i, 2), "t") == 0);
196  char *link_target = PQgetvalue(res, i, 3);
197  file_type_t type;
198 
199  if (PQgetisnull(res, 0, 1))
200  {
201  /*
202  * The file was removed from the server while the query was
203  * running. Ignore it.
204  */
205  continue;
206  }
207 
208  if (link_target[0])
209  type = FILE_TYPE_SYMLINK;
210  else if (isdir)
211  type = FILE_TYPE_DIRECTORY;
212  else
213  type = FILE_TYPE_REGULAR;
214 
215  process_source_file(path, type, filesize, link_target);
216  }
217  PQclear(res);
218 }
219 
220 /*----
221  * Runs a query, which returns pieces of files from the remote source data
222  * directory, and overwrites the corresponding parts of target files with
223  * the received parts. The result set is expected to be of format:
224  *
225  * path text -- path in the data directory, e.g "base/1/123"
226  * begin int8 -- offset within the file
227  * chunk bytea -- file content
228  *----
229  */
230 static void
231 receiveFileChunks(const char *sql)
232 {
233  PGresult *res;
234 
235  if (PQsendQueryParams(conn, sql, 0, NULL, NULL, NULL, NULL, 1) != 1)
236  pg_fatal("could not send query: %s", PQerrorMessage(conn));
237 
238  pg_log(PG_DEBUG, "getting file chunks\n");
239 
240  if (PQsetSingleRowMode(conn) != 1)
241  pg_fatal("could not set libpq connection to single row mode\n");
242 
243  while ((res = PQgetResult(conn)) != NULL)
244  {
245  char *filename;
246  int filenamelen;
247  int64 chunkoff;
248  char chunkoff_str[32];
249  int chunksize;
250  char *chunk;
251 
252  switch (PQresultStatus(res))
253  {
254  case PGRES_SINGLE_TUPLE:
255  break;
256 
257  case PGRES_TUPLES_OK:
258  PQclear(res);
259  continue; /* final zero-row result */
260 
261  default:
262  pg_fatal("unexpected result while fetching remote files: %s",
263  PQresultErrorMessage(res));
264  }
265 
266  /* sanity check the result set */
267  if (PQnfields(res) != 3 || PQntuples(res) != 1)
268  pg_fatal("unexpected result set size while fetching remote files\n");
269 
270  if (PQftype(res, 0) != TEXTOID ||
271  PQftype(res, 1) != INT8OID ||
272  PQftype(res, 2) != BYTEAOID)
273  {
274  pg_fatal("unexpected data types in result set while fetching remote files: %u %u %u\n",
275  PQftype(res, 0), PQftype(res, 1), PQftype(res, 2));
276  }
277 
278  if (PQfformat(res, 0) != 1 &&
279  PQfformat(res, 1) != 1 &&
280  PQfformat(res, 2) != 1)
281  {
282  pg_fatal("unexpected result format while fetching remote files\n");
283  }
284 
285  if (PQgetisnull(res, 0, 0) ||
286  PQgetisnull(res, 0, 1))
287  {
288  pg_fatal("unexpected null values in result while fetching remote files\n");
289  }
290 
291  if (PQgetlength(res, 0, 1) != sizeof(int64))
292  pg_fatal("unexpected result length while fetching remote files\n");
293 
294  /* Read result set to local variables */
295  memcpy(&chunkoff, PQgetvalue(res, 0, 1), sizeof(int64));
296  chunkoff = pg_ntoh64(chunkoff);
297  chunksize = PQgetlength(res, 0, 2);
298 
299  filenamelen = PQgetlength(res, 0, 0);
300  filename = pg_malloc(filenamelen + 1);
301  memcpy(filename, PQgetvalue(res, 0, 0), filenamelen);
302  filename[filenamelen] = '\0';
303 
304  chunk = PQgetvalue(res, 0, 2);
305 
306  /*
307  * It's possible that the file was deleted on remote side after we
308  * created the file map. In this case simply ignore it, as if it was
309  * not there in the first place, and move on.
310  */
311  if (PQgetisnull(res, 0, 2))
312  {
314  "received null value for chunk for file \"%s\", file has been deleted\n",
315  filename);
316  pg_free(filename);
317  PQclear(res);
318  continue;
319  }
320 
321  /*
322  * Separate step to keep platform-dependent format code out of
323  * translatable strings.
324  */
325  snprintf(chunkoff_str, sizeof(chunkoff_str), INT64_FORMAT, chunkoff);
326  pg_log(PG_DEBUG, "received chunk for file \"%s\", offset %s, size %d\n",
327  filename, chunkoff_str, chunksize);
328 
329  open_target_file(filename, false);
330 
331  write_target_range(chunk, chunkoff, chunksize);
332 
333  pg_free(filename);
334 
335  PQclear(res);
336  }
337 }
338 
339 /*
340  * Receive a single file as a malloc'd buffer.
341  */
342 char *
343 libpqGetFile(const char *filename, size_t *filesize)
344 {
345  PGresult *res;
346  char *result;
347  int len;
348  const char *paramValues[1];
349 
350  paramValues[0] = filename;
351  res = PQexecParams(conn, "SELECT pg_read_binary_file($1)",
352  1, NULL, paramValues, NULL, NULL, 1);
353 
354  if (PQresultStatus(res) != PGRES_TUPLES_OK)
355  pg_fatal("could not fetch remote file \"%s\": %s",
356  filename, PQresultErrorMessage(res));
357 
358  /* sanity check the result set */
359  if (PQntuples(res) != 1 || PQgetisnull(res, 0, 0))
360  pg_fatal("unexpected result set while fetching remote file \"%s\"\n",
361  filename);
362 
363  /* Read result to local variables */
364  len = PQgetlength(res, 0, 0);
365  result = pg_malloc(len + 1);
366  memcpy(result, PQgetvalue(res, 0, 0), len);
367  result[len] = '\0';
368 
369  PQclear(res);
370 
371  pg_log(PG_DEBUG, "fetched file \"%s\", length %d\n", filename, len);
372 
373  if (filesize)
374  *filesize = len;
375  return result;
376 }
377 
378 /*
379  * Write a file range to a temporary table in the server.
380  *
381  * The range is sent to the server as a COPY formatted line, to be inserted
382  * into the 'fetchchunks' temporary table. It is used in receiveFileChunks()
383  * function to actually fetch the data.
384  */
385 static void
386 fetch_file_range(const char *path, uint64 begin, uint64 end)
387 {
388  char linebuf[MAXPGPATH + 23];
389 
390  /* Split the range into CHUNKSIZE chunks */
391  while (end - begin > 0)
392  {
393  unsigned int len;
394 
395  /* Fine as long as CHUNKSIZE is not bigger than UINT32_MAX */
396  if (end - begin > CHUNKSIZE)
397  len = CHUNKSIZE;
398  else
399  len = (unsigned int) (end - begin);
400 
401  snprintf(linebuf, sizeof(linebuf), "%s\t" UINT64_FORMAT "\t%u\n", path, begin, len);
402 
403  if (PQputCopyData(conn, linebuf, strlen(linebuf)) != 1)
404  pg_fatal("could not send COPY data: %s",
405  PQerrorMessage(conn));
406 
407  begin += len;
408  }
409 }
410 
411 /*
412  * Fetch all changed blocks from remote source data directory.
413  */
414 void
416 {
417  file_entry_t *entry;
418  const char *sql;
419  PGresult *res;
420  int i;
421 
422  /*
423  * First create a temporary table, and load it with the blocks that we
424  * need to fetch.
425  */
426  sql = "CREATE TEMPORARY TABLE fetchchunks(path text, begin int8, len int4);";
427  res = PQexec(conn, sql);
428 
429  if (PQresultStatus(res) != PGRES_COMMAND_OK)
430  pg_fatal("could not create temporary table: %s",
431  PQresultErrorMessage(res));
432  PQclear(res);
433 
434  sql = "COPY fetchchunks FROM STDIN";
435  res = PQexec(conn, sql);
436 
437  if (PQresultStatus(res) != PGRES_COPY_IN)
438  pg_fatal("could not send file list: %s",
439  PQresultErrorMessage(res));
440  PQclear(res);
441 
442  for (i = 0; i < map->narray; i++)
443  {
444  entry = map->array[i];
445 
446  /* If this is a relation file, copy the modified blocks */
447  execute_pagemap(&entry->pagemap, entry->path);
448 
449  switch (entry->action)
450  {
451  case FILE_ACTION_NONE:
452  /* nothing else to do */
453  break;
454 
455  case FILE_ACTION_COPY:
456  /* Truncate the old file out of the way, if any */
457  open_target_file(entry->path, true);
458  fetch_file_range(entry->path, 0, entry->newsize);
459  break;
460 
462  truncate_target_file(entry->path, entry->newsize);
463  break;
464 
466  fetch_file_range(entry->path, entry->oldsize, entry->newsize);
467  break;
468 
469  case FILE_ACTION_REMOVE:
470  remove_target(entry);
471  break;
472 
473  case FILE_ACTION_CREATE:
474  create_target(entry);
475  break;
476  }
477  }
478 
479  if (PQputCopyEnd(conn, NULL) != 1)
480  pg_fatal("could not send end-of-COPY: %s",
481  PQerrorMessage(conn));
482 
483  while ((res = PQgetResult(conn)) != NULL)
484  {
485  if (PQresultStatus(res) != PGRES_COMMAND_OK)
486  pg_fatal("unexpected result while sending file list: %s",
487  PQresultErrorMessage(res));
488  PQclear(res);
489  }
490 
491  /*
492  * We've now copied the list of file ranges that we need to fetch to the
493  * temporary table. Now, actually fetch all of those ranges.
494  */
495  sql =
496  "SELECT path, begin,\n"
497  " pg_read_binary_file(path, begin, len, true) AS chunk\n"
498  "FROM fetchchunks\n";
499 
500  receiveFileChunks(sql);
501 }
502 
503 static void
504 execute_pagemap(datapagemap_t *pagemap, const char *path)
505 {
507  BlockNumber blkno;
508  off_t offset;
509 
510  iter = datapagemap_iterate(pagemap);
511  while (datapagemap_next(iter, &blkno))
512  {
513  offset = blkno * BLCKSZ;
514 
515  fetch_file_range(path, offset, offset + BLCKSZ);
516  }
517  pg_free(iter);
518 }
int PQputCopyData(PGconn *conn, const char *buffer, int nbytes)
Definition: fe-exec.c:2272
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:6097
void libpqProcessFileList(void)
Definition: libpq_fetch.c:146
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
file_entry_t ** array
Definition: filemap.h:79
void * pg_malloc(size_t size)
Definition: fe_memutils.c:47
#define TEXTOID
Definition: pg_type.h:324
size_t newsize
Definition: filemap.h:51
static void execute_pagemap(datapagemap_t *pagemap, const char *path)
Definition: libpq_fetch.c:504
int PQputCopyEnd(PGconn *conn, const char *errormsg)
Definition: fe-exec.c:2339
int snprintf(char *str, size_t count, const char *fmt,...) pg_attribute_printf(3
uint32 BlockNumber
Definition: block.h:31
#define CHUNKSIZE
Definition: libpq_fetch.c:38
void pg_fatal(const char *fmt,...)
Definition: logging.c:83
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2724
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:183
static char * run_simple_query(const char *sql)
Definition: libpq_fetch.c:97
bool datapagemap_next(datapagemap_iterator_t *iter, BlockNumber *blkno)
Definition: datapagemap.c:87
#define MAXPGPATH
Oid PQftype(const PGresult *res, int field_num)
Definition: fe-exec.c:2962
file_action_t action
Definition: filemap.h:47
static void receiveFileChunks(const char *sql)
Definition: libpq_fetch.c:231
int PQsetSingleRowMode(PGconn *conn)
Definition: fe-exec.c:1655
size_t oldsize
Definition: filemap.h:50
char * pg_strdup(const char *in)
Definition: fe_memutils.c:85
unsigned int uint32
Definition: c.h:258
static PGconn * conn
Definition: libpq_fetch.c:29
void libpq_executeFileMap(filemap_t *map)
Definition: libpq_fetch.c:415
void pg_log(eLogType type, const char *fmt,...)
Definition: logging.c:69
void libpqConnect(const char *connstr)
Definition: libpq_fetch.c:45
void PQclear(PGresult *res)
Definition: fe-exec.c:671
#define INT8OID
Definition: pg_type.h:304
void remove_target(file_entry_t *entry)
Definition: file_ops.c:126
uint64 XLogRecPtr
Definition: xlogdefs.h:21
datapagemap_iterator_t * datapagemap_iterate(datapagemap_t *map)
Definition: datapagemap.c:75
char * path
Definition: filemap.h:44
void pg_free(void *ptr)
Definition: fe_memutils.c:105
void process_source_file(const char *path, file_type_t type, size_t newsize, const char *link_target)
Definition: filemap.c:60
#define INT64_FORMAT
Definition: c.h:300
#define BYTEAOID
Definition: pg_type.h:292
static char * filename
Definition: pg_dumpall.c:90
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
static void fetch_file_range(const char *path, uint64 begin, uint64 end)
Definition: libpq_fetch.c:386
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
XLogRecPtr libpqGetCurrentXlogInsertLocation(void)
Definition: libpq_fetch.c:123
file_type_t
Definition: filemap.h:35
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3143
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6044
int PQfformat(const PGresult *res, int field_num)
Definition: fe-exec.c:2951
char * libpqGetFile(const char *filename, size_t *filesize)
Definition: libpq_fetch.c:343
#define UINT64_FORMAT
Definition: c.h:301
#define pg_ntoh64(x)
Definition: pg_bswap.h:126
long val
Definition: informix.c:689
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1753
PGconn * PQconnectdb(const char *conninfo)
Definition: fe-connect.c:585
static char * connstr
Definition: pg_dumpall.c:64