Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * libpqwalreceiver.c
4 : *
5 : * This file contains the libpq-specific parts of walreceiver. It's
6 : * loaded as a dynamic module to avoid linking the main server binary with
7 : * libpq.
8 : *
9 : * Apart from walreceiver, the libpq-specific routines are now being used by
10 : * logical replication workers and slot synchronization.
11 : *
12 : * Portions Copyright (c) 2010-2026, PostgreSQL Global Development Group
13 : *
14 : *
15 : * IDENTIFICATION
16 : * src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
17 : *
18 : *-------------------------------------------------------------------------
19 : */
20 : #include "postgres.h"
21 :
22 : #include <unistd.h>
23 : #include <sys/time.h>
24 :
25 : #include "common/connect.h"
26 : #include "funcapi.h"
27 : #include "libpq-fe.h"
28 : #include "libpq/libpq-be-fe-helpers.h"
29 : #include "mb/pg_wchar.h"
30 : #include "miscadmin.h"
31 : #include "pgstat.h"
32 : #include "pqexpbuffer.h"
33 : #include "replication/walreceiver.h"
34 : #include "storage/latch.h"
35 : #include "utils/builtins.h"
36 : #include "utils/memutils.h"
37 : #include "utils/pg_lsn.h"
38 : #include "utils/tuplestore.h"
39 :
40 2260 : PG_MODULE_MAGIC_EXT(
41 : .name = "libpqwalreceiver",
42 : .version = PG_VERSION
43 : );
44 :
45 : struct WalReceiverConn
46 : {
47 : /* Current connection to the primary, if any */
48 : PGconn *streamConn;
49 : /* Used to remember if the connection is logical or physical */
50 : bool logical;
51 : /* Buffer for currently read records */
52 : char *recvBuf;
53 : };
54 :
55 : /* Prototypes for interface functions */
56 : static WalReceiverConn *libpqrcv_connect(const char *conninfo,
57 : bool replication, bool logical,
58 : bool must_use_password,
59 : const char *appname, char **err);
60 : static void libpqrcv_check_conninfo(const char *conninfo,
61 : bool must_use_password);
62 : static char *libpqrcv_get_conninfo(WalReceiverConn *conn);
63 : static void libpqrcv_get_senderinfo(WalReceiverConn *conn,
64 : char **sender_host, int *sender_port);
65 : static char *libpqrcv_identify_system(WalReceiverConn *conn,
66 : TimeLineID *primary_tli);
67 : static char *libpqrcv_get_dbname_from_conninfo(const char *connInfo);
68 : static char *libpqrcv_get_option_from_conninfo(const char *connInfo,
69 : const char *keyword);
70 : static int libpqrcv_server_version(WalReceiverConn *conn);
71 : static void libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
72 : TimeLineID tli, char **filename,
73 : char **content, int *len);
74 : static bool libpqrcv_startstreaming(WalReceiverConn *conn,
75 : const WalRcvStreamOptions *options);
76 : static void libpqrcv_endstreaming(WalReceiverConn *conn,
77 : TimeLineID *next_tli);
78 : static int libpqrcv_receive(WalReceiverConn *conn, char **buffer,
79 : pgsocket *wait_fd);
80 : static void libpqrcv_send(WalReceiverConn *conn, const char *buffer,
81 : int nbytes);
82 : static char *libpqrcv_create_slot(WalReceiverConn *conn,
83 : const char *slotname,
84 : bool temporary,
85 : bool two_phase,
86 : bool failover,
87 : CRSSnapshotAction snapshot_action,
88 : XLogRecPtr *lsn);
89 : static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
90 : const bool *failover, const bool *two_phase);
91 : static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
92 : static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
93 : const char *query,
94 : const int nRetTypes,
95 : const Oid *retTypes);
96 : static void libpqrcv_disconnect(WalReceiverConn *conn);
97 :
98 : static WalReceiverFunctionsType PQWalReceiverFunctions = {
99 : .walrcv_connect = libpqrcv_connect,
100 : .walrcv_check_conninfo = libpqrcv_check_conninfo,
101 : .walrcv_get_conninfo = libpqrcv_get_conninfo,
102 : .walrcv_get_senderinfo = libpqrcv_get_senderinfo,
103 : .walrcv_identify_system = libpqrcv_identify_system,
104 : .walrcv_server_version = libpqrcv_server_version,
105 : .walrcv_readtimelinehistoryfile = libpqrcv_readtimelinehistoryfile,
106 : .walrcv_startstreaming = libpqrcv_startstreaming,
107 : .walrcv_endstreaming = libpqrcv_endstreaming,
108 : .walrcv_receive = libpqrcv_receive,
109 : .walrcv_send = libpqrcv_send,
110 : .walrcv_create_slot = libpqrcv_create_slot,
111 : .walrcv_alter_slot = libpqrcv_alter_slot,
112 : .walrcv_get_dbname_from_conninfo = libpqrcv_get_dbname_from_conninfo,
113 : .walrcv_get_backend_pid = libpqrcv_get_backend_pid,
114 : .walrcv_exec = libpqrcv_exec,
115 : .walrcv_disconnect = libpqrcv_disconnect
116 : };
117 :
118 : /* Prototypes for private functions */
119 : static char *stringlist_to_identifierstr(PGconn *conn, List *strings);
120 :
121 : /*
122 : * Module initialization function
123 : */
124 : void
125 2260 : _PG_init(void)
126 : {
127 2260 : if (WalReceiverFunctions != NULL)
128 0 : elog(ERROR, "libpqwalreceiver already loaded");
129 2260 : WalReceiverFunctions = &PQWalReceiverFunctions;
130 2260 : }
131 :
132 : /*
133 : * Establish the connection to the primary server.
134 : *
135 : * This function can be used for both replication and regular connections.
136 : * If it is a replication connection, it could be either logical or physical
137 : * based on input argument 'logical'.
138 : *
139 : * If an error occurs, this function will normally return NULL and set *err
140 : * to a palloc'ed error message. However, if must_use_password is true and
141 : * the connection fails to use the password, this function will ereport(ERROR).
142 : * We do this because in that case the error includes a detail and a hint for
143 : * consistency with other parts of the system, and it's not worth adding the
144 : * machinery to pass all of those back to the caller just to cover this one
145 : * case.
146 : */
147 : static WalReceiverConn *
148 2142 : libpqrcv_connect(const char *conninfo, bool replication, bool logical,
149 : bool must_use_password, const char *appname, char **err)
150 : {
151 : WalReceiverConn *conn;
152 : const char *keys[6];
153 : const char *vals[6];
154 2142 : int i = 0;
155 2142 : char *options_val = NULL;
156 :
157 : /*
158 : * Re-validate connection string. The validation already happened at DDL
159 : * time, but the subscription owner may have changed. If we don't recheck
160 : * with the correct must_use_password, it's possible that the connection
161 : * will obtain the password from a different source, such as PGPASSFILE or
162 : * PGPASSWORD.
163 : */
164 2142 : libpqrcv_check_conninfo(conninfo, must_use_password);
165 :
166 : /*
167 : * We use the expand_dbname parameter to process the connection string (or
168 : * URI), and pass some extra options.
169 : */
170 2120 : keys[i] = "dbname";
171 2120 : vals[i] = conninfo;
172 :
173 : /* We can not have logical without replication */
174 : Assert(replication || !logical);
175 :
176 2120 : if (replication)
177 : {
178 2094 : keys[++i] = "replication";
179 2094 : vals[i] = logical ? "database" : "true";
180 :
181 2094 : if (logical)
182 : {
183 1394 : char *opt = NULL;
184 :
185 : /* Tell the publisher to translate to our encoding */
186 1394 : keys[++i] = "client_encoding";
187 1394 : vals[i] = GetDatabaseEncodingName();
188 :
189 : /*
190 : * Force assorted GUC parameters to settings that ensure that the
191 : * publisher will output data values in a form that is unambiguous
192 : * to the subscriber. (We don't want to modify the subscriber's
193 : * GUC settings, since that might surprise user-defined code
194 : * running in the subscriber, such as triggers.) This should
195 : * match what pg_dump does.
196 : */
197 1394 : opt = libpqrcv_get_option_from_conninfo(conninfo, "options");
198 1394 : options_val = psprintf("%s -c datestyle=ISO -c intervalstyle=postgres -c extra_float_digits=3",
199 : (opt == NULL) ? "" : opt);
200 1394 : keys[++i] = "options";
201 1394 : vals[i] = options_val;
202 1394 : if (opt != NULL)
203 20 : pfree(opt);
204 : }
205 : else
206 : {
207 : /*
208 : * The database name is ignored by the server in replication mode,
209 : * but specify "replication" for .pgpass lookup.
210 : */
211 700 : keys[++i] = "dbname";
212 700 : vals[i] = "replication";
213 : }
214 : }
215 :
216 2120 : keys[++i] = "fallback_application_name";
217 2120 : vals[i] = appname;
218 :
219 2120 : keys[++i] = NULL;
220 2120 : vals[i] = NULL;
221 :
222 : Assert(i < lengthof(keys));
223 :
224 2120 : conn = palloc0_object(WalReceiverConn);
225 2118 : conn->streamConn =
226 2120 : libpqsrv_connect_params(keys, vals,
227 : /* expand_dbname = */ true,
228 : WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);
229 :
230 2118 : if (options_val != NULL)
231 1394 : pfree(options_val);
232 :
233 2118 : if (PQstatus(conn->streamConn) != CONNECTION_OK)
234 468 : goto bad_connection_errmsg;
235 :
236 1650 : if (must_use_password && !PQconnectionUsedPassword(conn->streamConn))
237 : {
238 0 : libpqsrv_disconnect(conn->streamConn);
239 0 : pfree(conn);
240 :
241 0 : ereport(ERROR,
242 : (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
243 : errmsg("password is required"),
244 : errdetail("Non-superuser cannot connect if the server does not request a password."),
245 : errhint("Target server's authentication method must be changed, or set password_required=false in the subscription parameters.")));
246 : }
247 :
248 1650 : PQsetNoticeReceiver(conn->streamConn, libpqsrv_notice_receiver,
249 : "received message via replication");
250 :
251 : /*
252 : * Set always-secure search path for the cases where the connection is
253 : * used to run SQL queries, so malicious users can't get control.
254 : */
255 1650 : if (!replication || logical)
256 : {
257 : PGresult *res;
258 :
259 1370 : res = libpqsrv_exec(conn->streamConn,
260 : ALWAYS_SECURE_SEARCH_PATH_SQL,
261 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
262 1370 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
263 : {
264 0 : PQclear(res);
265 0 : *err = psprintf(_("could not clear search path: %s"),
266 0 : pchomp(PQerrorMessage(conn->streamConn)));
267 0 : goto bad_connection;
268 : }
269 1370 : PQclear(res);
270 : }
271 :
272 1650 : conn->logical = logical;
273 :
274 1650 : return conn;
275 :
276 : /* error path, using libpq's error message */
277 468 : bad_connection_errmsg:
278 468 : *err = pchomp(PQerrorMessage(conn->streamConn));
279 :
280 : /* error path, error already set */
281 468 : bad_connection:
282 468 : libpqsrv_disconnect(conn->streamConn);
283 468 : pfree(conn);
284 468 : return NULL;
285 : }
286 :
287 : /*
288 : * Validate connection info string.
289 : *
290 : * If the connection string can't be parsed, this function will raise
291 : * an error. If must_use_password is true, the function raises an error
292 : * if no password is provided in the connection string. In any other case
293 : * it successfully completes.
294 : */
295 : static void
296 2526 : libpqrcv_check_conninfo(const char *conninfo, bool must_use_password)
297 : {
298 2526 : PQconninfoOption *opts = NULL;
299 : PQconninfoOption *opt;
300 2526 : char *err = NULL;
301 :
302 2526 : opts = PQconninfoParse(conninfo, &err);
303 2526 : if (opts == NULL)
304 : {
305 : /* The error string is malloc'd, so we must free it explicitly */
306 18 : char *errcopy = err ? pstrdup(err) : "out of memory";
307 :
308 18 : PQfreemem(err);
309 18 : ereport(ERROR,
310 : (errcode(ERRCODE_SYNTAX_ERROR),
311 : errmsg("invalid connection string syntax: %s", errcopy)));
312 : }
313 :
314 2508 : if (must_use_password)
315 : {
316 40 : bool uses_password = false;
317 :
318 1504 : for (opt = opts; opt->keyword != NULL; ++opt)
319 : {
320 : /* Ignore connection options that are not present. */
321 1476 : if (opt->val == NULL)
322 1382 : continue;
323 :
324 94 : if (strcmp(opt->keyword, "password") == 0 && opt->val[0] != '\0')
325 : {
326 12 : uses_password = true;
327 12 : break;
328 : }
329 : }
330 :
331 40 : if (!uses_password)
332 : {
333 : /* malloc'd, so we must free it explicitly */
334 28 : PQconninfoFree(opts);
335 :
336 28 : ereport(ERROR,
337 : (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
338 : errmsg("password is required"),
339 : errdetail("Non-superusers must provide a password in the connection string.")));
340 : }
341 : }
342 :
343 2480 : PQconninfoFree(opts);
344 2480 : }
345 :
346 : /*
347 : * Return a user-displayable conninfo string. Any security-sensitive fields
348 : * are obfuscated.
349 : */
350 : static char *
351 280 : libpqrcv_get_conninfo(WalReceiverConn *conn)
352 : {
353 : PQconninfoOption *conn_opts;
354 : PQconninfoOption *conn_opt;
355 : PQExpBufferData buf;
356 : char *retval;
357 :
358 : Assert(conn->streamConn != NULL);
359 :
360 280 : initPQExpBuffer(&buf);
361 280 : conn_opts = PQconninfo(conn->streamConn);
362 :
363 280 : if (conn_opts == NULL)
364 0 : ereport(ERROR,
365 : (errcode(ERRCODE_OUT_OF_MEMORY),
366 : errmsg("could not parse connection string: %s",
367 : _("out of memory"))));
368 :
369 : /* build a clean connection string from pieces */
370 14560 : for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
371 : {
372 : bool obfuscate;
373 :
374 : /* Skip debug and empty options */
375 14280 : if (strchr(conn_opt->dispchar, 'D') ||
376 13160 : conn_opt->val == NULL ||
377 5344 : conn_opt->val[0] == '\0')
378 9216 : continue;
379 :
380 : /* Obfuscate security-sensitive options */
381 5064 : obfuscate = strchr(conn_opt->dispchar, '*') != NULL;
382 :
383 10128 : appendPQExpBuffer(&buf, "%s%s=%s",
384 5064 : buf.len == 0 ? "" : " ",
385 : conn_opt->keyword,
386 : obfuscate ? "********" : conn_opt->val);
387 : }
388 :
389 280 : PQconninfoFree(conn_opts);
390 :
391 280 : retval = PQExpBufferDataBroken(buf) ? NULL : pstrdup(buf.data);
392 280 : termPQExpBuffer(&buf);
393 280 : return retval;
394 : }
395 :
396 : /*
397 : * Provides information of sender this WAL receiver is connected to.
398 : */
399 : static void
400 280 : libpqrcv_get_senderinfo(WalReceiverConn *conn, char **sender_host,
401 : int *sender_port)
402 : {
403 280 : char *ret = NULL;
404 :
405 280 : *sender_host = NULL;
406 280 : *sender_port = 0;
407 :
408 : Assert(conn->streamConn != NULL);
409 :
410 280 : ret = PQhost(conn->streamConn);
411 280 : if (ret && strlen(ret) != 0)
412 280 : *sender_host = pstrdup(ret);
413 :
414 280 : ret = PQport(conn->streamConn);
415 280 : if (ret && strlen(ret) != 0)
416 280 : *sender_port = atoi(ret);
417 280 : }
418 :
419 : /*
420 : * Check that primary's system identifier matches ours, and fetch the current
421 : * timeline ID of the primary.
422 : */
423 : static char *
424 738 : libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli)
425 : {
426 : PGresult *res;
427 : char *primary_sysid;
428 :
429 : /*
430 : * Get the system identifier and timeline ID as a DataRow message from the
431 : * primary server.
432 : */
433 738 : res = libpqsrv_exec(conn->streamConn,
434 : "IDENTIFY_SYSTEM",
435 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
436 736 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
437 0 : ereport(ERROR,
438 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
439 : errmsg("could not receive database system identifier and timeline ID from "
440 : "the primary server: %s",
441 : pchomp(PQerrorMessage(conn->streamConn)))));
442 :
443 : /*
444 : * IDENTIFY_SYSTEM returns 3 columns in 9.3 and earlier, and 4 columns in
445 : * 9.4 and onwards.
446 : */
447 736 : if (PQnfields(res) < 3 || PQntuples(res) != 1)
448 0 : ereport(ERROR,
449 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
450 : errmsg("invalid response from primary server"),
451 : errdetail("Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
452 : PQntuples(res), PQnfields(res), 1, 3)));
453 736 : primary_sysid = pstrdup(PQgetvalue(res, 0, 0));
454 736 : *primary_tli = pg_strtoint32(PQgetvalue(res, 0, 1));
455 736 : PQclear(res);
456 :
457 736 : return primary_sysid;
458 : }
459 :
460 : /*
461 : * Thin wrapper around libpq to obtain server version.
462 : */
463 : static int
464 1952 : libpqrcv_server_version(WalReceiverConn *conn)
465 : {
466 1952 : return PQserverVersion(conn->streamConn);
467 : }
468 :
469 : /*
470 : * Get database name from the primary server's conninfo.
471 : *
472 : * If dbname is not found in connInfo, return NULL value.
473 : */
474 : static char *
475 28 : libpqrcv_get_dbname_from_conninfo(const char *connInfo)
476 : {
477 28 : return libpqrcv_get_option_from_conninfo(connInfo, "dbname");
478 : }
479 :
480 : /*
481 : * Get the value of the option with the given keyword from the primary
482 : * server's conninfo.
483 : *
484 : * If the option is not found in connInfo, return NULL value.
485 : */
486 : static char *
487 1422 : libpqrcv_get_option_from_conninfo(const char *connInfo, const char *keyword)
488 : {
489 : PQconninfoOption *opts;
490 1422 : char *option = NULL;
491 1422 : char *err = NULL;
492 :
493 1422 : opts = PQconninfoParse(connInfo, &err);
494 1422 : if (opts == NULL)
495 : {
496 : /* The error string is malloc'd, so we must free it explicitly */
497 0 : char *errcopy = err ? pstrdup(err) : "out of memory";
498 :
499 0 : PQfreemem(err);
500 0 : ereport(ERROR,
501 : (errcode(ERRCODE_SYNTAX_ERROR),
502 : errmsg("invalid connection string syntax: %s", errcopy)));
503 : }
504 :
505 73944 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
506 : {
507 : /*
508 : * If the same option appears multiple times, then the last one will
509 : * be returned
510 : */
511 72522 : if (strcmp(opt->keyword, keyword) == 0 && opt->val &&
512 46 : *opt->val)
513 : {
514 46 : if (option)
515 0 : pfree(option);
516 :
517 46 : option = pstrdup(opt->val);
518 : }
519 : }
520 :
521 1422 : PQconninfoFree(opts);
522 1422 : return option;
523 : }
524 :
525 : /*
526 : * Start streaming WAL data from given streaming options.
527 : *
528 : * Returns true if we switched successfully to copy-both mode. False
529 : * means the server received the command and executed it successfully, but
530 : * didn't switch to copy-mode. That means that there was no WAL on the
531 : * requested timeline and starting point, because the server switched to
532 : * another timeline at or before the requested starting point. On failure,
533 : * throws an ERROR.
534 : */
535 : static bool
536 1108 : libpqrcv_startstreaming(WalReceiverConn *conn,
537 : const WalRcvStreamOptions *options)
538 : {
539 : StringInfoData cmd;
540 : PGresult *res;
541 :
542 : Assert(options->logical == conn->logical);
543 : Assert(options->slotname || !options->logical);
544 :
545 1108 : initStringInfo(&cmd);
546 :
547 : /* Build the command. */
548 1108 : appendStringInfoString(&cmd, "START_REPLICATION");
549 1108 : if (options->slotname != NULL)
550 900 : appendStringInfo(&cmd, " SLOT \"%s\"",
551 900 : options->slotname);
552 :
553 1108 : if (options->logical)
554 806 : appendStringInfoString(&cmd, " LOGICAL");
555 :
556 1108 : appendStringInfo(&cmd, " %X/%08X", LSN_FORMAT_ARGS(options->startpoint));
557 :
558 : /*
559 : * Additional options are different depending on if we are doing logical
560 : * or physical replication.
561 : */
562 1108 : if (options->logical)
563 : {
564 : char *pubnames_str;
565 : List *pubnames;
566 : char *pubnames_literal;
567 :
568 806 : appendStringInfoString(&cmd, " (");
569 :
570 806 : appendStringInfo(&cmd, "proto_version '%u'",
571 806 : options->proto.logical.proto_version);
572 :
573 806 : if (options->proto.logical.streaming_str)
574 790 : appendStringInfo(&cmd, ", streaming '%s'",
575 790 : options->proto.logical.streaming_str);
576 :
577 824 : if (options->proto.logical.twophase &&
578 18 : PQserverVersion(conn->streamConn) >= 150000)
579 18 : appendStringInfoString(&cmd, ", two_phase 'on'");
580 :
581 1612 : if (options->proto.logical.origin &&
582 806 : PQserverVersion(conn->streamConn) >= 160000)
583 806 : appendStringInfo(&cmd, ", origin '%s'",
584 806 : options->proto.logical.origin);
585 :
586 806 : pubnames = options->proto.logical.publication_names;
587 806 : pubnames_str = stringlist_to_identifierstr(conn->streamConn, pubnames);
588 806 : if (!pubnames_str)
589 0 : ereport(ERROR,
590 : (errcode(ERRCODE_OUT_OF_MEMORY), /* likely guess */
591 : errmsg("could not start WAL streaming: %s",
592 : pchomp(PQerrorMessage(conn->streamConn)))));
593 806 : pubnames_literal = PQescapeLiteral(conn->streamConn, pubnames_str,
594 : strlen(pubnames_str));
595 806 : if (!pubnames_literal)
596 0 : ereport(ERROR,
597 : (errcode(ERRCODE_OUT_OF_MEMORY), /* likely guess */
598 : errmsg("could not start WAL streaming: %s",
599 : pchomp(PQerrorMessage(conn->streamConn)))));
600 806 : appendStringInfo(&cmd, ", publication_names %s", pubnames_literal);
601 806 : PQfreemem(pubnames_literal);
602 806 : pfree(pubnames_str);
603 :
604 828 : if (options->proto.logical.binary &&
605 22 : PQserverVersion(conn->streamConn) >= 140000)
606 22 : appendStringInfoString(&cmd, ", binary 'true'");
607 :
608 806 : appendStringInfoChar(&cmd, ')');
609 : }
610 : else
611 302 : appendStringInfo(&cmd, " TIMELINE %u",
612 302 : options->proto.physical.startpointTLI);
613 :
614 : /* Start streaming. */
615 1108 : res = libpqsrv_exec(conn->streamConn,
616 1108 : cmd.data,
617 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
618 1108 : pfree(cmd.data);
619 :
620 1108 : if (PQresultStatus(res) == PGRES_COMMAND_OK)
621 : {
622 0 : PQclear(res);
623 0 : return false;
624 : }
625 1108 : else if (PQresultStatus(res) != PGRES_COPY_BOTH)
626 2 : ereport(ERROR,
627 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
628 : errmsg("could not start WAL streaming: %s",
629 : pchomp(PQerrorMessage(conn->streamConn)))));
630 1106 : PQclear(res);
631 1106 : return true;
632 : }
633 :
634 : /*
635 : * Stop streaming WAL data. Returns the next timeline's ID in *next_tli, as
636 : * reported by the server, or 0 if it did not report it.
637 : */
638 : static void
639 472 : libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
640 : {
641 : PGresult *res;
642 :
643 : /*
644 : * Send copy-end message. As in libpqsrv_exec, this could theoretically
645 : * block, but the risk seems small.
646 : */
647 866 : if (PQputCopyEnd(conn->streamConn, NULL) <= 0 ||
648 394 : PQflush(conn->streamConn))
649 78 : ereport(ERROR,
650 : (errcode(ERRCODE_CONNECTION_FAILURE),
651 : errmsg("could not send end-of-streaming message to primary: %s",
652 : pchomp(PQerrorMessage(conn->streamConn)))));
653 :
654 394 : *next_tli = 0;
655 :
656 : /*
657 : * After COPY is finished, we should receive a result set indicating the
658 : * next timeline's ID, or just CommandComplete if the server was shut
659 : * down.
660 : *
661 : * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
662 : * also possible in case we aborted the copy in mid-stream.
663 : */
664 394 : res = libpqsrv_get_result(conn->streamConn,
665 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
666 394 : if (PQresultStatus(res) == PGRES_TUPLES_OK)
667 : {
668 : /*
669 : * Read the next timeline's ID. The server also sends the timeline's
670 : * starting point, but it is ignored.
671 : */
672 22 : if (PQnfields(res) < 2 || PQntuples(res) != 1)
673 0 : ereport(ERROR,
674 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
675 : errmsg("unexpected result set after end-of-streaming")));
676 22 : *next_tli = pg_strtoint32(PQgetvalue(res, 0, 0));
677 22 : PQclear(res);
678 :
679 : /* the result set should be followed by CommandComplete */
680 22 : res = libpqsrv_get_result(conn->streamConn,
681 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
682 : }
683 372 : else if (PQresultStatus(res) == PGRES_COPY_OUT)
684 : {
685 372 : PQclear(res);
686 :
687 : /* End the copy */
688 372 : if (PQendcopy(conn->streamConn))
689 0 : ereport(ERROR,
690 : (errcode(ERRCODE_CONNECTION_FAILURE),
691 : errmsg("error while shutting down streaming COPY: %s",
692 : pchomp(PQerrorMessage(conn->streamConn)))));
693 :
694 : /* CommandComplete should follow */
695 372 : res = libpqsrv_get_result(conn->streamConn,
696 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
697 : }
698 :
699 394 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
700 0 : ereport(ERROR,
701 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
702 : errmsg("error reading result of streaming command: %s",
703 : pchomp(PQerrorMessage(conn->streamConn)))));
704 394 : PQclear(res);
705 :
706 : /* Verify that there are no more results */
707 394 : res = libpqsrv_get_result(conn->streamConn,
708 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
709 394 : if (res != NULL)
710 0 : ereport(ERROR,
711 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
712 : errmsg("unexpected result after CommandComplete: %s",
713 : pchomp(PQerrorMessage(conn->streamConn)))));
714 394 : }
715 :
716 : /*
717 : * Fetch the timeline history file for 'tli' from primary.
718 : */
719 : static void
720 20 : libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
721 : TimeLineID tli, char **filename,
722 : char **content, int *len)
723 : {
724 : PGresult *res;
725 : char cmd[64];
726 :
727 : Assert(!conn->logical);
728 :
729 : /*
730 : * Request the primary to send over the history file for given timeline.
731 : */
732 20 : snprintf(cmd, sizeof(cmd), "TIMELINE_HISTORY %u", tli);
733 20 : res = libpqsrv_exec(conn->streamConn,
734 : cmd,
735 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
736 20 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
737 0 : ereport(ERROR,
738 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
739 : errmsg("could not receive timeline history file from "
740 : "the primary server: %s",
741 : pchomp(PQerrorMessage(conn->streamConn)))));
742 20 : if (PQnfields(res) != 2 || PQntuples(res) != 1)
743 0 : ereport(ERROR,
744 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
745 : errmsg("invalid response from primary server"),
746 : errdetail("Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
747 : PQntuples(res), PQnfields(res))));
748 20 : *filename = pstrdup(PQgetvalue(res, 0, 0));
749 :
750 20 : *len = PQgetlength(res, 0, 1);
751 20 : *content = palloc(*len);
752 20 : memcpy(*content, PQgetvalue(res, 0, 1), *len);
753 20 : PQclear(res);
754 20 : }
755 :
756 : /*
757 : * Disconnect connection to primary, if any.
758 : */
759 : static void
760 1650 : libpqrcv_disconnect(WalReceiverConn *conn)
761 : {
762 1650 : libpqsrv_disconnect(conn->streamConn);
763 1650 : PQfreemem(conn->recvBuf);
764 1650 : pfree(conn);
765 1650 : }
766 :
767 : /*
768 : * Receive a message available from XLOG stream.
769 : *
770 : * Returns:
771 : *
772 : * If data was received, returns the length of the data. *buffer is set to
773 : * point to a buffer holding the received message. The buffer is only valid
774 : * until the next libpqrcv_* call.
775 : *
776 : * If no data was available immediately, returns 0, and *wait_fd is set to a
777 : * socket descriptor which can be waited on before trying again.
778 : *
779 : * -1 if the server ended the COPY.
780 : *
781 : * ereports on error.
782 : */
783 : static int
784 817502 : libpqrcv_receive(WalReceiverConn *conn, char **buffer,
785 : pgsocket *wait_fd)
786 : {
787 : int rawlen;
788 :
789 817502 : PQfreemem(conn->recvBuf);
790 817502 : conn->recvBuf = NULL;
791 :
792 : /* Try to receive a CopyData message */
793 817502 : rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
794 817502 : if (rawlen == 0)
795 : {
796 : /* Try consuming some data. */
797 558738 : if (PQconsumeInput(conn->streamConn) == 0)
798 70 : ereport(ERROR,
799 : (errcode(ERRCODE_CONNECTION_FAILURE),
800 : errmsg("could not receive data from WAL stream: %s",
801 : pchomp(PQerrorMessage(conn->streamConn)))));
802 :
803 : /* Now that we've consumed some input, try again */
804 558668 : rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
805 558668 : if (rawlen == 0)
806 : {
807 : /* Tell caller to try again when our socket is ready. */
808 202228 : *wait_fd = PQsocket(conn->streamConn);
809 202228 : return 0;
810 : }
811 : }
812 615204 : if (rawlen == -1) /* end-of-streaming or error */
813 : {
814 : PGresult *res;
815 :
816 494 : res = libpqsrv_get_result(conn->streamConn,
817 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
818 494 : if (PQresultStatus(res) == PGRES_COMMAND_OK)
819 : {
820 466 : PQclear(res);
821 :
822 : /* Verify that there are no more results. */
823 466 : res = libpqsrv_get_result(conn->streamConn,
824 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
825 466 : if (res != NULL)
826 : {
827 78 : PQclear(res);
828 :
829 : /*
830 : * If the other side closed the connection orderly (otherwise
831 : * we'd seen an error, or PGRES_COPY_IN) don't report an error
832 : * here, but let callers deal with it.
833 : */
834 78 : if (PQstatus(conn->streamConn) == CONNECTION_BAD)
835 78 : return -1;
836 :
837 0 : ereport(ERROR,
838 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
839 : errmsg("unexpected result after CommandComplete: %s",
840 : PQerrorMessage(conn->streamConn))));
841 : }
842 :
843 388 : return -1;
844 : }
845 28 : else if (PQresultStatus(res) == PGRES_COPY_IN)
846 : {
847 22 : PQclear(res);
848 22 : return -1;
849 : }
850 : else
851 6 : ereport(ERROR,
852 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
853 : errmsg("could not receive data from WAL stream: %s",
854 : pchomp(PQerrorMessage(conn->streamConn)))));
855 : }
856 614710 : if (rawlen < -1)
857 0 : ereport(ERROR,
858 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
859 : errmsg("could not receive data from WAL stream: %s",
860 : pchomp(PQerrorMessage(conn->streamConn)))));
861 :
862 : /* Return received messages to caller */
863 614710 : *buffer = conn->recvBuf;
864 614710 : return rawlen;
865 : }
866 :
867 : /*
868 : * Send a message to XLOG stream.
869 : *
870 : * ereports on error.
871 : */
872 : static void
873 224882 : libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes)
874 : {
875 449764 : if (PQputCopyData(conn->streamConn, buffer, nbytes) <= 0 ||
876 224882 : PQflush(conn->streamConn))
877 0 : ereport(ERROR,
878 : (errcode(ERRCODE_CONNECTION_FAILURE),
879 : errmsg("could not send data to WAL stream: %s",
880 : pchomp(PQerrorMessage(conn->streamConn)))));
881 224882 : }
882 :
883 : /*
884 : * Create new replication slot.
885 : * Returns the name of the exported snapshot for logical slot or NULL for
886 : * physical slot.
887 : */
888 : static char *
889 624 : libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
890 : bool temporary, bool two_phase, bool failover,
891 : CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
892 : {
893 : PGresult *res;
894 : StringInfoData cmd;
895 : char *snapshot;
896 : int use_new_options_syntax;
897 :
898 624 : use_new_options_syntax = (PQserverVersion(conn->streamConn) >= 150000);
899 :
900 624 : initStringInfo(&cmd);
901 :
902 624 : appendStringInfo(&cmd, "CREATE_REPLICATION_SLOT \"%s\"", slotname);
903 :
904 624 : if (temporary)
905 0 : appendStringInfoString(&cmd, " TEMPORARY");
906 :
907 624 : if (conn->logical)
908 : {
909 624 : appendStringInfoString(&cmd, " LOGICAL pgoutput ");
910 624 : if (use_new_options_syntax)
911 624 : appendStringInfoChar(&cmd, '(');
912 624 : if (two_phase)
913 : {
914 2 : appendStringInfoString(&cmd, "TWO_PHASE");
915 2 : if (use_new_options_syntax)
916 2 : appendStringInfoString(&cmd, ", ");
917 : else
918 0 : appendStringInfoChar(&cmd, ' ');
919 : }
920 :
921 624 : if (failover)
922 : {
923 18 : appendStringInfoString(&cmd, "FAILOVER");
924 18 : if (use_new_options_syntax)
925 18 : appendStringInfoString(&cmd, ", ");
926 : else
927 0 : appendStringInfoChar(&cmd, ' ');
928 : }
929 :
930 624 : if (use_new_options_syntax)
931 : {
932 624 : switch (snapshot_action)
933 : {
934 0 : case CRS_EXPORT_SNAPSHOT:
935 0 : appendStringInfoString(&cmd, "SNAPSHOT 'export'");
936 0 : break;
937 228 : case CRS_NOEXPORT_SNAPSHOT:
938 228 : appendStringInfoString(&cmd, "SNAPSHOT 'nothing'");
939 228 : break;
940 396 : case CRS_USE_SNAPSHOT:
941 396 : appendStringInfoString(&cmd, "SNAPSHOT 'use'");
942 396 : break;
943 : }
944 : }
945 : else
946 : {
947 0 : switch (snapshot_action)
948 : {
949 0 : case CRS_EXPORT_SNAPSHOT:
950 0 : appendStringInfoString(&cmd, "EXPORT_SNAPSHOT");
951 0 : break;
952 0 : case CRS_NOEXPORT_SNAPSHOT:
953 0 : appendStringInfoString(&cmd, "NOEXPORT_SNAPSHOT");
954 0 : break;
955 0 : case CRS_USE_SNAPSHOT:
956 0 : appendStringInfoString(&cmd, "USE_SNAPSHOT");
957 0 : break;
958 : }
959 : }
960 :
961 624 : if (use_new_options_syntax)
962 624 : appendStringInfoChar(&cmd, ')');
963 : }
964 : else
965 : {
966 0 : if (use_new_options_syntax)
967 0 : appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
968 : else
969 0 : appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
970 : }
971 :
972 624 : res = libpqsrv_exec(conn->streamConn,
973 624 : cmd.data,
974 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
975 624 : pfree(cmd.data);
976 :
977 624 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
978 0 : ereport(ERROR,
979 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
980 : errmsg("could not create replication slot \"%s\": %s",
981 : slotname, pchomp(PQerrorMessage(conn->streamConn)))));
982 :
983 624 : if (lsn)
984 396 : *lsn = DatumGetLSN(DirectFunctionCall1Coll(pg_lsn_in, InvalidOid,
985 396 : CStringGetDatum(PQgetvalue(res, 0, 1))));
986 :
987 624 : if (!PQgetisnull(res, 0, 2))
988 0 : snapshot = pstrdup(PQgetvalue(res, 0, 2));
989 : else
990 624 : snapshot = NULL;
991 :
992 624 : PQclear(res);
993 :
994 624 : return snapshot;
995 : }
996 :
997 : /*
998 : * Change the definition of the replication slot.
999 : */
1000 : static void
1001 10 : libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
1002 : const bool *failover, const bool *two_phase)
1003 : {
1004 : StringInfoData cmd;
1005 : PGresult *res;
1006 :
1007 10 : initStringInfo(&cmd);
1008 10 : appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( ",
1009 : quote_identifier(slotname));
1010 :
1011 10 : if (failover)
1012 8 : appendStringInfo(&cmd, "FAILOVER %s",
1013 8 : *failover ? "true" : "false");
1014 :
1015 10 : if (failover && two_phase)
1016 0 : appendStringInfoString(&cmd, ", ");
1017 :
1018 10 : if (two_phase)
1019 2 : appendStringInfo(&cmd, "TWO_PHASE %s",
1020 2 : *two_phase ? "true" : "false");
1021 :
1022 10 : appendStringInfoString(&cmd, " );");
1023 :
1024 10 : res = libpqsrv_exec(conn->streamConn, cmd.data,
1025 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
1026 10 : pfree(cmd.data);
1027 :
1028 10 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1029 0 : ereport(ERROR,
1030 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
1031 : errmsg("could not alter replication slot \"%s\": %s",
1032 : slotname, pchomp(PQerrorMessage(conn->streamConn)))));
1033 :
1034 10 : PQclear(res);
1035 10 : }
1036 :
1037 : /*
1038 : * Return PID of remote backend process.
1039 : */
1040 : static pid_t
1041 0 : libpqrcv_get_backend_pid(WalReceiverConn *conn)
1042 : {
1043 0 : return PQbackendPID(conn->streamConn);
1044 : }
1045 :
1046 : /*
1047 : * Convert tuple query result to tuplestore.
1048 : */
1049 : static void
1050 2320 : libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
1051 : const int nRetTypes, const Oid *retTypes)
1052 : {
1053 : int tupn;
1054 : int coln;
1055 2320 : int nfields = PQnfields(pgres);
1056 : HeapTuple tuple;
1057 : AttInMetadata *attinmeta;
1058 : MemoryContext rowcontext;
1059 : MemoryContext oldcontext;
1060 :
1061 : /* Make sure we got expected number of fields. */
1062 2320 : if (nfields != nRetTypes)
1063 0 : ereport(ERROR,
1064 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
1065 : errmsg("invalid query response"),
1066 : errdetail("Expected %d fields, got %d fields.",
1067 : nRetTypes, nfields)));
1068 :
1069 2320 : walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
1070 :
1071 : /* Create tuple descriptor corresponding to expected result. */
1072 2320 : walres->tupledesc = CreateTemplateTupleDesc(nRetTypes);
1073 8578 : for (coln = 0; coln < nRetTypes; coln++)
1074 6258 : TupleDescInitEntry(walres->tupledesc, (AttrNumber) coln + 1,
1075 6258 : PQfname(pgres, coln), retTypes[coln], -1, 0);
1076 2320 : attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
1077 :
1078 : /* No point in doing more here if there were no tuples returned. */
1079 2320 : if (PQntuples(pgres) == 0)
1080 68 : return;
1081 :
1082 : /* Create temporary context for local allocations. */
1083 2252 : rowcontext = AllocSetContextCreate(CurrentMemoryContext,
1084 : "libpqrcv query result context",
1085 : ALLOCSET_DEFAULT_SIZES);
1086 :
1087 : /* Process returned rows. */
1088 5244 : for (tupn = 0; tupn < PQntuples(pgres); tupn++)
1089 : {
1090 : char *cstrs[MaxTupleAttributeNumber];
1091 :
1092 2992 : CHECK_FOR_INTERRUPTS();
1093 :
1094 : /* Do the allocations in temporary context. */
1095 2992 : oldcontext = MemoryContextSwitchTo(rowcontext);
1096 :
1097 : /*
1098 : * Fill cstrs with null-terminated strings of column values.
1099 : */
1100 12440 : for (coln = 0; coln < nfields; coln++)
1101 : {
1102 9448 : if (PQgetisnull(pgres, tupn, coln))
1103 1232 : cstrs[coln] = NULL;
1104 : else
1105 8216 : cstrs[coln] = PQgetvalue(pgres, tupn, coln);
1106 : }
1107 :
1108 : /* Convert row to a tuple, and add it to the tuplestore */
1109 2992 : tuple = BuildTupleFromCStrings(attinmeta, cstrs);
1110 2992 : tuplestore_puttuple(walres->tuplestore, tuple);
1111 :
1112 : /* Clean up */
1113 2992 : MemoryContextSwitchTo(oldcontext);
1114 2992 : MemoryContextReset(rowcontext);
1115 : }
1116 :
1117 2252 : MemoryContextDelete(rowcontext);
1118 : }
1119 :
1120 : /*
1121 : * Public interface for sending generic queries (and commands).
1122 : *
1123 : * This can only be called from process connected to database.
1124 : */
1125 : static WalRcvExecResult *
1126 4028 : libpqrcv_exec(WalReceiverConn *conn, const char *query,
1127 : const int nRetTypes, const Oid *retTypes)
1128 : {
1129 4028 : PGresult *pgres = NULL;
1130 4028 : WalRcvExecResult *walres = palloc0_object(WalRcvExecResult);
1131 : char *diag_sqlstate;
1132 :
1133 4028 : if (MyDatabaseId == InvalidOid)
1134 0 : ereport(ERROR,
1135 : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1136 : errmsg("the query interface requires a database connection")));
1137 :
1138 4028 : pgres = libpqsrv_exec(conn->streamConn,
1139 : query,
1140 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
1141 :
1142 4028 : switch (PQresultStatus(pgres))
1143 : {
1144 2320 : case PGRES_TUPLES_OK:
1145 : case PGRES_SINGLE_TUPLE:
1146 : case PGRES_TUPLES_CHUNK:
1147 2320 : walres->status = WALRCV_OK_TUPLES;
1148 2320 : libpqrcv_processTuples(pgres, walres, nRetTypes, retTypes);
1149 2320 : break;
1150 :
1151 0 : case PGRES_COPY_IN:
1152 0 : walres->status = WALRCV_OK_COPY_IN;
1153 0 : break;
1154 :
1155 392 : case PGRES_COPY_OUT:
1156 392 : walres->status = WALRCV_OK_COPY_OUT;
1157 392 : break;
1158 :
1159 0 : case PGRES_COPY_BOTH:
1160 0 : walres->status = WALRCV_OK_COPY_BOTH;
1161 0 : break;
1162 :
1163 1308 : case PGRES_COMMAND_OK:
1164 1308 : walres->status = WALRCV_OK_COMMAND;
1165 1308 : break;
1166 :
1167 : /* Empty query is considered error. */
1168 0 : case PGRES_EMPTY_QUERY:
1169 0 : walres->status = WALRCV_ERROR;
1170 0 : walres->err = _("empty query");
1171 0 : break;
1172 :
1173 0 : case PGRES_PIPELINE_SYNC:
1174 : case PGRES_PIPELINE_ABORTED:
1175 0 : walres->status = WALRCV_ERROR;
1176 0 : walres->err = _("unexpected pipeline mode");
1177 0 : break;
1178 :
1179 8 : case PGRES_NONFATAL_ERROR:
1180 : case PGRES_FATAL_ERROR:
1181 : case PGRES_BAD_RESPONSE:
1182 8 : walres->status = WALRCV_ERROR;
1183 8 : walres->err = pchomp(PQerrorMessage(conn->streamConn));
1184 8 : diag_sqlstate = PQresultErrorField(pgres, PG_DIAG_SQLSTATE);
1185 8 : if (diag_sqlstate)
1186 8 : walres->sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
1187 : diag_sqlstate[1],
1188 : diag_sqlstate[2],
1189 : diag_sqlstate[3],
1190 : diag_sqlstate[4]);
1191 8 : break;
1192 : }
1193 :
1194 4028 : PQclear(pgres);
1195 :
1196 4028 : return walres;
1197 : }
1198 :
1199 : /*
1200 : * Given a List of strings, return it as single comma separated
1201 : * string, quoting identifiers as needed.
1202 : *
1203 : * This is essentially the reverse of SplitIdentifierString.
1204 : *
1205 : * The caller should free the result.
1206 : */
1207 : static char *
1208 806 : stringlist_to_identifierstr(PGconn *conn, List *strings)
1209 : {
1210 : ListCell *lc;
1211 : StringInfoData res;
1212 806 : bool first = true;
1213 :
1214 806 : initStringInfo(&res);
1215 :
1216 2120 : foreach(lc, strings)
1217 : {
1218 1314 : char *val = strVal(lfirst(lc));
1219 : char *val_escaped;
1220 :
1221 1314 : if (first)
1222 806 : first = false;
1223 : else
1224 508 : appendStringInfoChar(&res, ',');
1225 :
1226 1314 : val_escaped = PQescapeIdentifier(conn, val, strlen(val));
1227 1314 : if (!val_escaped)
1228 : {
1229 0 : free(res.data);
1230 0 : return NULL;
1231 : }
1232 1314 : appendStringInfoString(&res, val_escaped);
1233 1314 : PQfreemem(val_escaped);
1234 : }
1235 :
1236 806 : return res.data;
1237 : }
|
