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