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