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