Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * libpq_source.c
4 : * Functions for fetching files from a remote server via libpq.
5 : *
6 : * Copyright (c) 2013-2024, PostgreSQL Global Development Group
7 : *
8 : *-------------------------------------------------------------------------
9 : */
10 : #include "postgres_fe.h"
11 :
12 : #include "catalog/pg_type_d.h"
13 : #include "common/connect.h"
14 : #include "datapagemap.h"
15 : #include "file_ops.h"
16 : #include "filemap.h"
17 : #include "lib/stringinfo.h"
18 : #include "pg_rewind.h"
19 : #include "port/pg_bswap.h"
20 : #include "rewind_source.h"
21 :
22 : /*
23 : * Files are fetched MAX_CHUNK_SIZE bytes at a time, and with a
24 : * maximum of MAX_CHUNKS_PER_QUERY chunks in a single query.
25 : */
26 : #define MAX_CHUNK_SIZE (1024 * 1024)
27 : #define MAX_CHUNKS_PER_QUERY 1000
28 :
29 : /* represents a request to fetch a piece of a file from the source */
30 : typedef struct
31 : {
32 : const char *path; /* path relative to data directory root */
33 : off_t offset;
34 : size_t length;
35 : } fetch_range_request;
36 :
37 : typedef struct
38 : {
39 : rewind_source common; /* common interface functions */
40 :
41 : PGconn *conn;
42 :
43 : /*
44 : * Queue of chunks that have been requested with the queue_fetch_range()
45 : * function, but have not been fetched from the remote server yet.
46 : */
47 : int num_requests;
48 : fetch_range_request request_queue[MAX_CHUNKS_PER_QUERY];
49 :
50 : /* temporary space for process_queued_fetch_requests() */
51 : StringInfoData paths;
52 : StringInfoData offsets;
53 : StringInfoData lengths;
54 : } libpq_source;
55 :
56 : static void init_libpq_conn(PGconn *conn);
57 : static char *run_simple_query(PGconn *conn, const char *sql);
58 : static void run_simple_command(PGconn *conn, const char *sql);
59 : static void appendArrayEscapedString(StringInfo buf, const char *str);
60 :
61 : static void process_queued_fetch_requests(libpq_source *src);
62 :
63 : /* public interface functions */
64 : static void libpq_traverse_files(rewind_source *source,
65 : process_file_callback_t callback);
66 : static void libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len);
67 : static void libpq_queue_fetch_range(rewind_source *source, const char *path,
68 : off_t off, size_t len);
69 : static void libpq_finish_fetch(rewind_source *source);
70 : static char *libpq_fetch_file(rewind_source *source, const char *path,
71 : size_t *filesize);
72 : static XLogRecPtr libpq_get_current_wal_insert_lsn(rewind_source *source);
73 : static void libpq_destroy(rewind_source *source);
74 :
75 : /*
76 : * Create a new libpq source.
77 : *
78 : * The caller has already established the connection, but should not try
79 : * to use it while the source is active.
80 : */
81 : rewind_source *
82 12 : init_libpq_source(PGconn *conn)
83 : {
84 : libpq_source *src;
85 :
86 12 : init_libpq_conn(conn);
87 :
88 12 : src = pg_malloc0(sizeof(libpq_source));
89 :
90 12 : src->common.traverse_files = libpq_traverse_files;
91 12 : src->common.fetch_file = libpq_fetch_file;
92 12 : src->common.queue_fetch_file = libpq_queue_fetch_file;
93 12 : src->common.queue_fetch_range = libpq_queue_fetch_range;
94 12 : src->common.finish_fetch = libpq_finish_fetch;
95 12 : src->common.get_current_wal_insert_lsn = libpq_get_current_wal_insert_lsn;
96 12 : src->common.destroy = libpq_destroy;
97 :
98 12 : src->conn = conn;
99 :
100 12 : initStringInfo(&src->paths);
101 12 : initStringInfo(&src->offsets);
102 12 : initStringInfo(&src->lengths);
103 :
104 12 : return &src->common;
105 : }
106 :
107 : /*
108 : * Initialize a libpq connection for use.
109 : */
110 : static void
111 12 : init_libpq_conn(PGconn *conn)
112 : {
113 : PGresult *res;
114 : char *str;
115 :
116 : /* disable all types of timeouts */
117 12 : run_simple_command(conn, "SET statement_timeout = 0");
118 12 : run_simple_command(conn, "SET lock_timeout = 0");
119 12 : run_simple_command(conn, "SET idle_in_transaction_session_timeout = 0");
120 12 : run_simple_command(conn, "SET transaction_timeout = 0");
121 :
122 : /*
123 : * we don't intend to do any updates, put the connection in read-only mode
124 : * to keep us honest
125 : */
126 12 : run_simple_command(conn, "SET default_transaction_read_only = on");
127 :
128 : /* secure search_path */
129 12 : res = PQexec(conn, ALWAYS_SECURE_SEARCH_PATH_SQL);
130 12 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
131 0 : pg_fatal("could not clear search_path: %s",
132 : PQresultErrorMessage(res));
133 12 : PQclear(res);
134 :
135 : /*
136 : * Also check that full_page_writes is enabled. We can get torn pages if
137 : * a page is modified while we read it with pg_read_binary_file(), and we
138 : * rely on full page images to fix them.
139 : */
140 12 : str = run_simple_query(conn, "SHOW full_page_writes");
141 12 : if (strcmp(str, "on") != 0)
142 0 : pg_fatal("full_page_writes must be enabled in the source server");
143 12 : pg_free(str);
144 :
145 : /* Prepare a statement we'll use to fetch files */
146 12 : res = PQprepare(conn, "fetch_chunks_stmt",
147 : "SELECT path, begin,\n"
148 : " pg_read_binary_file(path, begin, len, true) AS chunk\n"
149 : "FROM unnest ($1::text[], $2::int8[], $3::int4[]) as x(path, begin, len)",
150 : 3, NULL);
151 :
152 12 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
153 0 : pg_fatal("could not prepare statement to fetch file contents: %s",
154 : PQresultErrorMessage(res));
155 12 : PQclear(res);
156 12 : }
157 :
158 : /*
159 : * Run a query that returns a single value.
160 : *
161 : * The result should be pg_free'd after use.
162 : */
163 : static char *
164 22 : run_simple_query(PGconn *conn, const char *sql)
165 : {
166 : PGresult *res;
167 : char *result;
168 :
169 22 : res = PQexec(conn, sql);
170 :
171 22 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
172 0 : pg_fatal("error running query (%s) on source server: %s",
173 : sql, PQresultErrorMessage(res));
174 :
175 : /* sanity check the result set */
176 22 : if (PQnfields(res) != 1 || PQntuples(res) != 1 || PQgetisnull(res, 0, 0))
177 0 : pg_fatal("unexpected result set from query");
178 :
179 22 : result = pg_strdup(PQgetvalue(res, 0, 0));
180 :
181 22 : PQclear(res);
182 :
183 22 : return result;
184 : }
185 :
186 : /*
187 : * Run a command.
188 : *
189 : * In the event of a failure, exit immediately.
190 : */
191 : static void
192 60 : run_simple_command(PGconn *conn, const char *sql)
193 : {
194 : PGresult *res;
195 :
196 60 : res = PQexec(conn, sql);
197 :
198 60 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
199 0 : pg_fatal("error running query (%s) in source server: %s",
200 : sql, PQresultErrorMessage(res));
201 :
202 60 : PQclear(res);
203 60 : }
204 :
205 : /*
206 : * Call the pg_current_wal_insert_lsn() function in the remote system.
207 : */
208 : static XLogRecPtr
209 10 : libpq_get_current_wal_insert_lsn(rewind_source *source)
210 : {
211 10 : PGconn *conn = ((libpq_source *) source)->conn;
212 : XLogRecPtr result;
213 : uint32 hi;
214 : uint32 lo;
215 : char *val;
216 :
217 10 : val = run_simple_query(conn, "SELECT pg_current_wal_insert_lsn()");
218 :
219 10 : if (sscanf(val, "%X/%X", &hi, &lo) != 2)
220 0 : pg_fatal("unrecognized result \"%s\" for current WAL insert location", val);
221 :
222 10 : result = ((uint64) hi) << 32 | lo;
223 :
224 10 : pg_free(val);
225 :
226 10 : return result;
227 : }
228 :
229 : /*
230 : * Get a list of all files in the data directory.
231 : */
232 : static void
233 12 : libpq_traverse_files(rewind_source *source, process_file_callback_t callback)
234 : {
235 12 : PGconn *conn = ((libpq_source *) source)->conn;
236 : PGresult *res;
237 : const char *sql;
238 : int i;
239 :
240 : /*
241 : * Create a recursive directory listing of the whole data directory.
242 : *
243 : * The WITH RECURSIVE part does most of the work. The second part gets the
244 : * targets of the symlinks in pg_tblspc directory.
245 : *
246 : * XXX: There is no backend function to get a symbolic link's target in
247 : * general, so if the admin has put any custom symbolic links in the data
248 : * directory, they won't be copied correctly.
249 : */
250 12 : sql =
251 : "WITH RECURSIVE files (path, filename, size, isdir) AS (\n"
252 : " SELECT '' AS path, filename, size, isdir FROM\n"
253 : " (SELECT pg_ls_dir('.', true, false) AS filename) AS fn,\n"
254 : " pg_stat_file(fn.filename, true) AS this\n"
255 : " UNION ALL\n"
256 : " SELECT parent.path || parent.filename || '/' AS path,\n"
257 : " fn, this.size, this.isdir\n"
258 : " FROM files AS parent,\n"
259 : " pg_ls_dir(parent.path || parent.filename, true, false) AS fn,\n"
260 : " pg_stat_file(parent.path || parent.filename || '/' || fn, true) AS this\n"
261 : " WHERE parent.isdir = 't'\n"
262 : ")\n"
263 : "SELECT path || filename, size, isdir,\n"
264 : " pg_tablespace_location(pg_tablespace.oid) AS link_target\n"
265 : "FROM files\n"
266 : "LEFT OUTER JOIN pg_tablespace ON files.path = 'pg_tblspc/'\n"
267 : " AND oid::text = files.filename\n";
268 12 : res = PQexec(conn, sql);
269 :
270 12 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
271 0 : pg_fatal("could not fetch file list: %s",
272 : PQresultErrorMessage(res));
273 :
274 : /* sanity check the result set */
275 12 : if (PQnfields(res) != 4)
276 0 : pg_fatal("unexpected result set while fetching file list");
277 :
278 : /* Read result to local variables */
279 13932 : for (i = 0; i < PQntuples(res); i++)
280 : {
281 : char *path;
282 : int64 filesize;
283 : bool isdir;
284 : char *link_target;
285 : file_type_t type;
286 :
287 13920 : if (PQgetisnull(res, i, 1))
288 : {
289 : /*
290 : * The file was removed from the server while the query was
291 : * running. Ignore it.
292 : */
293 0 : continue;
294 : }
295 :
296 13920 : path = PQgetvalue(res, i, 0);
297 13920 : filesize = atol(PQgetvalue(res, i, 1));
298 13920 : isdir = (strcmp(PQgetvalue(res, i, 2), "t") == 0);
299 13920 : link_target = PQgetvalue(res, i, 3);
300 :
301 13920 : if (link_target[0])
302 : {
303 : /*
304 : * In-place tablespaces are directories located in pg_tblspc/ with
305 : * relative paths.
306 : */
307 2 : if (is_absolute_path(link_target))
308 0 : type = FILE_TYPE_SYMLINK;
309 : else
310 2 : type = FILE_TYPE_DIRECTORY;
311 : }
312 13918 : else if (isdir)
313 334 : type = FILE_TYPE_DIRECTORY;
314 : else
315 13584 : type = FILE_TYPE_REGULAR;
316 :
317 13920 : callback(path, type, filesize, link_target);
318 : }
319 12 : PQclear(res);
320 12 : }
321 :
322 : /*
323 : * Queue up a request to fetch a file from remote system.
324 : */
325 : static void
326 3916 : libpq_queue_fetch_file(rewind_source *source, const char *path, size_t len)
327 : {
328 : /*
329 : * Truncate the target file immediately, and queue a request to fetch it
330 : * from the source. If the file is small, smaller than MAX_CHUNK_SIZE,
331 : * request fetching a full-sized chunk anyway, so that if the file has
332 : * become larger in the source system, after we scanned the source
333 : * directory, we still fetch the whole file. This only works for files up
334 : * to MAX_CHUNK_SIZE, but that's good enough for small configuration files
335 : * and such that are changed every now and then, but not WAL-logged. For
336 : * larger files, we fetch up to the original size.
337 : *
338 : * Even with that mechanism, there is an inherent race condition if the
339 : * file is modified at the same instant that we're copying it, so that we
340 : * might copy a torn version of the file with one half from the old
341 : * version and another half from the new. But pg_basebackup has the same
342 : * problem, and it hasn't been a problem in practice.
343 : *
344 : * It might seem more natural to truncate the file later, when we receive
345 : * it from the source server, but then we'd need to track which
346 : * fetch-requests are for a whole file.
347 : */
348 3916 : open_target_file(path, true);
349 3916 : libpq_queue_fetch_range(source, path, 0, Max(len, MAX_CHUNK_SIZE));
350 3916 : }
351 :
352 : /*
353 : * Queue up a request to fetch a piece of a file from remote system.
354 : */
355 : static void
356 5512 : libpq_queue_fetch_range(rewind_source *source, const char *path, off_t off,
357 : size_t len)
358 : {
359 5512 : libpq_source *src = (libpq_source *) source;
360 :
361 : /*
362 : * Does this request happen to be a continuation of the previous chunk? If
363 : * so, merge it with the previous one.
364 : *
365 : * XXX: We use pointer equality to compare the path. That's good enough
366 : * for our purposes; the caller always passes the same pointer for the
367 : * same filename. If it didn't, we would fail to merge requests, but it
368 : * wouldn't affect correctness.
369 : */
370 5512 : if (src->num_requests > 0)
371 : {
372 5500 : fetch_range_request *prev = &src->request_queue[src->num_requests - 1];
373 :
374 5500 : if (prev->offset + prev->length == off &&
375 1200 : prev->length < MAX_CHUNK_SIZE &&
376 1200 : prev->path == path)
377 : {
378 : /*
379 : * Extend the previous request to cover as much of this new
380 : * request as possible, without exceeding MAX_CHUNK_SIZE.
381 : */
382 : size_t thislen;
383 :
384 1198 : thislen = Min(len, MAX_CHUNK_SIZE - prev->length);
385 1198 : prev->length += thislen;
386 :
387 1198 : off += thislen;
388 1198 : len -= thislen;
389 :
390 : /*
391 : * Fall through to create new requests for any remaining 'len'
392 : * that didn't fit in the previous chunk.
393 : */
394 : }
395 : }
396 :
397 : /* Divide the request into pieces of MAX_CHUNK_SIZE bytes each */
398 10396 : while (len > 0)
399 : {
400 : int32 thislen;
401 :
402 : /* if the queue is full, perform all the work queued up so far */
403 4884 : if (src->num_requests == MAX_CHUNKS_PER_QUERY)
404 0 : process_queued_fetch_requests(src);
405 :
406 4884 : thislen = Min(len, MAX_CHUNK_SIZE);
407 4884 : src->request_queue[src->num_requests].path = path;
408 4884 : src->request_queue[src->num_requests].offset = off;
409 4884 : src->request_queue[src->num_requests].length = thislen;
410 4884 : src->num_requests++;
411 :
412 4884 : off += thislen;
413 4884 : len -= thislen;
414 : }
415 5512 : }
416 :
417 : /*
418 : * Fetch all the queued chunks and write them to the target data directory.
419 : */
420 : static void
421 12 : libpq_finish_fetch(rewind_source *source)
422 : {
423 12 : process_queued_fetch_requests((libpq_source *) source);
424 12 : }
425 :
426 : static void
427 12 : process_queued_fetch_requests(libpq_source *src)
428 : {
429 : const char *params[3];
430 : PGresult *res;
431 : int chunkno;
432 :
433 12 : if (src->num_requests == 0)
434 0 : return;
435 :
436 12 : pg_log_debug("getting %d file chunks", src->num_requests);
437 :
438 : /*
439 : * The prepared statement, 'fetch_chunks_stmt', takes three arrays with
440 : * the same length as parameters: paths, offsets and lengths. Construct
441 : * the string representations of them.
442 : */
443 12 : resetStringInfo(&src->paths);
444 12 : resetStringInfo(&src->offsets);
445 12 : resetStringInfo(&src->lengths);
446 :
447 12 : appendStringInfoChar(&src->paths, '{');
448 12 : appendStringInfoChar(&src->offsets, '{');
449 12 : appendStringInfoChar(&src->lengths, '{');
450 4896 : for (int i = 0; i < src->num_requests; i++)
451 : {
452 4884 : fetch_range_request *rq = &src->request_queue[i];
453 :
454 4884 : if (i > 0)
455 : {
456 4872 : appendStringInfoChar(&src->paths, ',');
457 4872 : appendStringInfoChar(&src->offsets, ',');
458 4872 : appendStringInfoChar(&src->lengths, ',');
459 : }
460 :
461 4884 : appendArrayEscapedString(&src->paths, rq->path);
462 4884 : appendStringInfo(&src->offsets, INT64_FORMAT, (int64) rq->offset);
463 4884 : appendStringInfo(&src->lengths, INT64_FORMAT, (int64) rq->length);
464 : }
465 12 : appendStringInfoChar(&src->paths, '}');
466 12 : appendStringInfoChar(&src->offsets, '}');
467 12 : appendStringInfoChar(&src->lengths, '}');
468 :
469 : /*
470 : * Execute the prepared statement.
471 : */
472 12 : params[0] = src->paths.data;
473 12 : params[1] = src->offsets.data;
474 12 : params[2] = src->lengths.data;
475 :
476 12 : if (PQsendQueryPrepared(src->conn, "fetch_chunks_stmt", 3, params, NULL, NULL, 1) != 1)
477 0 : pg_fatal("could not send query: %s", PQerrorMessage(src->conn));
478 :
479 12 : if (PQsetSingleRowMode(src->conn) != 1)
480 0 : pg_fatal("could not set libpq connection to single row mode");
481 :
482 : /*----
483 : * The result set is of format:
484 : *
485 : * path text -- path in the data directory, e.g "base/1/123"
486 : * begin int8 -- offset within the file
487 : * chunk bytea -- file content
488 : *----
489 : */
490 12 : chunkno = 0;
491 4908 : while ((res = PQgetResult(src->conn)) != NULL)
492 : {
493 4896 : fetch_range_request *rq = &src->request_queue[chunkno];
494 : char *filename;
495 : int filenamelen;
496 : int64 chunkoff;
497 : int chunksize;
498 : char *chunk;
499 :
500 4896 : switch (PQresultStatus(res))
501 : {
502 4884 : case PGRES_SINGLE_TUPLE:
503 4884 : break;
504 :
505 12 : case PGRES_TUPLES_OK:
506 12 : PQclear(res);
507 12 : continue; /* final zero-row result */
508 :
509 0 : default:
510 0 : pg_fatal("unexpected result while fetching remote files: %s",
511 : PQresultErrorMessage(res));
512 : }
513 :
514 4884 : if (chunkno > src->num_requests)
515 0 : pg_fatal("received more data chunks than requested");
516 :
517 : /* sanity check the result set */
518 4884 : if (PQnfields(res) != 3 || PQntuples(res) != 1)
519 0 : pg_fatal("unexpected result set size while fetching remote files");
520 :
521 9768 : if (PQftype(res, 0) != TEXTOID ||
522 9768 : PQftype(res, 1) != INT8OID ||
523 4884 : PQftype(res, 2) != BYTEAOID)
524 : {
525 0 : pg_fatal("unexpected data types in result set while fetching remote files: %u %u %u",
526 : PQftype(res, 0), PQftype(res, 1), PQftype(res, 2));
527 : }
528 :
529 4884 : if (PQfformat(res, 0) != 1 &&
530 0 : PQfformat(res, 1) != 1 &&
531 0 : PQfformat(res, 2) != 1)
532 : {
533 0 : pg_fatal("unexpected result format while fetching remote files");
534 : }
535 :
536 9768 : if (PQgetisnull(res, 0, 0) ||
537 4884 : PQgetisnull(res, 0, 1))
538 : {
539 0 : pg_fatal("unexpected null values in result while fetching remote files");
540 : }
541 :
542 4884 : if (PQgetlength(res, 0, 1) != sizeof(int64))
543 0 : pg_fatal("unexpected result length while fetching remote files");
544 :
545 : /* Read result set to local variables */
546 4884 : memcpy(&chunkoff, PQgetvalue(res, 0, 1), sizeof(int64));
547 4884 : chunkoff = pg_ntoh64(chunkoff);
548 4884 : chunksize = PQgetlength(res, 0, 2);
549 :
550 4884 : filenamelen = PQgetlength(res, 0, 0);
551 4884 : filename = pg_malloc(filenamelen + 1);
552 4884 : memcpy(filename, PQgetvalue(res, 0, 0), filenamelen);
553 4884 : filename[filenamelen] = '\0';
554 :
555 4884 : chunk = PQgetvalue(res, 0, 2);
556 :
557 : /*
558 : * If a file has been deleted on the source, remove it on the target
559 : * as well. Note that multiple unlink() calls may happen on the same
560 : * file if multiple data chunks are associated with it, hence ignore
561 : * unconditionally anything missing.
562 : */
563 4884 : if (PQgetisnull(res, 0, 2))
564 : {
565 0 : pg_log_debug("received null value for chunk for file \"%s\", file has been deleted",
566 : filename);
567 0 : remove_target_file(filename, true);
568 : }
569 : else
570 : {
571 4884 : pg_log_debug("received chunk for file \"%s\", offset %lld, size %d",
572 : filename, (long long int) chunkoff, chunksize);
573 :
574 4884 : if (strcmp(filename, rq->path) != 0)
575 : {
576 0 : pg_fatal("received data for file \"%s\", when requested for \"%s\"",
577 : filename, rq->path);
578 : }
579 4884 : if (chunkoff != rq->offset)
580 0 : pg_fatal("received data at offset %lld of file \"%s\", when requested for offset %lld",
581 : (long long int) chunkoff, rq->path, (long long int) rq->offset);
582 :
583 : /*
584 : * We should not receive more data than we requested, or
585 : * pg_read_binary_file() messed up. We could receive less,
586 : * though, if the file was truncated in the source after we
587 : * checked its size. That's OK, there should be a WAL record of
588 : * the truncation, which will get replayed when you start the
589 : * target system for the first time after pg_rewind has completed.
590 : */
591 4884 : if (chunksize > rq->length)
592 0 : pg_fatal("received more than requested for file \"%s\"", rq->path);
593 :
594 4884 : open_target_file(filename, false);
595 :
596 4884 : write_target_range(chunk, chunkoff, chunksize);
597 : }
598 :
599 4884 : pg_free(filename);
600 :
601 4884 : PQclear(res);
602 4884 : chunkno++;
603 : }
604 12 : if (chunkno != src->num_requests)
605 0 : pg_fatal("unexpected number of data chunks received");
606 :
607 12 : src->num_requests = 0;
608 : }
609 :
610 : /*
611 : * Escape a string to be used as element in a text array constant
612 : */
613 : static void
614 4884 : appendArrayEscapedString(StringInfo buf, const char *str)
615 : {
616 4884 : appendStringInfoCharMacro(buf, '\"');
617 89564 : while (*str)
618 : {
619 84680 : char ch = *str;
620 :
621 84680 : if (ch == '"' || ch == '\\')
622 0 : appendStringInfoCharMacro(buf, '\\');
623 :
624 84680 : appendStringInfoCharMacro(buf, ch);
625 :
626 84680 : str++;
627 : }
628 4884 : appendStringInfoCharMacro(buf, '\"');
629 4884 : }
630 :
631 : /*
632 : * Fetch a single file as a malloc'd buffer.
633 : */
634 : static char *
635 34 : libpq_fetch_file(rewind_source *source, const char *path, size_t *filesize)
636 : {
637 34 : PGconn *conn = ((libpq_source *) source)->conn;
638 : PGresult *res;
639 : char *result;
640 : int len;
641 : const char *paramValues[1];
642 :
643 34 : paramValues[0] = path;
644 34 : res = PQexecParams(conn, "SELECT pg_read_binary_file($1)",
645 : 1, NULL, paramValues, NULL, NULL, 1);
646 :
647 34 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
648 0 : pg_fatal("could not fetch remote file \"%s\": %s",
649 : path, PQresultErrorMessage(res));
650 :
651 : /* sanity check the result set */
652 34 : if (PQntuples(res) != 1 || PQgetisnull(res, 0, 0))
653 0 : pg_fatal("unexpected result set while fetching remote file \"%s\"",
654 : path);
655 :
656 : /* Read result to local variables */
657 34 : len = PQgetlength(res, 0, 0);
658 34 : result = pg_malloc(len + 1);
659 34 : memcpy(result, PQgetvalue(res, 0, 0), len);
660 34 : result[len] = '\0';
661 :
662 34 : PQclear(res);
663 :
664 34 : pg_log_debug("fetched file \"%s\", length %d", path, len);
665 :
666 34 : if (filesize)
667 24 : *filesize = len;
668 34 : return result;
669 : }
670 :
671 : /*
672 : * Close a libpq source.
673 : */
674 : static void
675 12 : libpq_destroy(rewind_source *source)
676 : {
677 12 : libpq_source *src = (libpq_source *) source;
678 :
679 12 : pfree(src->paths.data);
680 12 : pfree(src->offsets.data);
681 12 : pfree(src->lengths.data);
682 12 : pfree(src);
683 :
684 : /* NOTE: we don't close the connection here, as it was not opened by us. */
685 12 : }
|
