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