Age Owner Branch data TLA 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 : :
486 tgl@sss.pgh.pa.us 40 :CBC 1114 : 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, size_t *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(List *strings);
120 : :
121 : : /*
122 : : * Module initialization function
123 : : */
124 : : void
6030 heikki.linnakangas@i 125 : 1114 : _PG_init(void)
126 : : {
3524 peter_e@gmx.net 127 [ - + ]: 1114 : if (WalReceiverFunctions != NULL)
6030 heikki.linnakangas@i 128 [ # # ]:UBC 0 : elog(ERROR, "libpqwalreceiver already loaded");
3524 peter_e@gmx.net 129 :CBC 1114 : WalReceiverFunctions = &PQWalReceiverFunctions;
6030 heikki.linnakangas@i 130 : 1114 : }
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 *
901 akapila@postgresql.o 148 : 1038 : 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];
3524 peter_e@gmx.net 154 : 1038 : int i = 0;
200 fujii@postgresql.org 155 : 1038 : 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 : : */
925 jdavis@postgresql.or 164 : 1038 : 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 : : */
3524 peter_e@gmx.net 170 : 1030 : keys[i] = "dbname";
171 : 1030 : vals[i] = conninfo;
172 : :
173 : : /* We can not have logical without replication */
901 akapila@postgresql.o 174 [ + + - + ]: 1030 : Assert(replication || !logical);
175 : :
176 [ + + ]: 1030 : if (replication)
177 : : {
178 : 1015 : keys[++i] = "replication";
179 [ + + ]: 1015 : vals[i] = logical ? "database" : "true";
180 : :
181 [ + + ]: 1015 : if (logical)
182 : : {
200 fujii@postgresql.org 183 : 761 : char *opt = NULL;
184 : :
185 : : /* Tell the publisher to translate to our encoding */
901 akapila@postgresql.o 186 : 761 : keys[++i] = "client_encoding";
187 : 761 : 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 : : */
200 fujii@postgresql.org 197 : 761 : opt = libpqrcv_get_option_from_conninfo(conninfo, "options");
198 [ + + ]: 761 : options_val = psprintf("%s -c datestyle=ISO -c intervalstyle=postgres -c extra_float_digits=3",
199 : : (opt == NULL) ? "" : opt);
901 akapila@postgresql.o 200 : 761 : keys[++i] = "options";
200 fujii@postgresql.org 201 : 761 : vals[i] = options_val;
202 [ + + ]: 761 : if (opt != NULL)
203 : 9 : 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 : : */
901 akapila@postgresql.o 211 : 254 : keys[++i] = "dbname";
212 : 254 : vals[i] = "replication";
213 : : }
214 : : }
215 : :
3524 peter_e@gmx.net 216 : 1030 : keys[++i] = "fallback_application_name";
217 : 1030 : vals[i] = appname;
218 : :
219 : 1030 : keys[++i] = NULL;
220 : 1030 : vals[i] = NULL;
221 : :
503 heikki.linnakangas@i 222 [ - + ]: 1030 : Assert(i < lengthof(keys));
223 : :
227 michael@paquier.xyz 224 : 1030 : conn = palloc0_object(WalReceiverConn);
477 heikki.linnakangas@i 225 : 1030 : conn->streamConn =
63 fujii@postgresql.org 226 : 1030 : libpqsrv_connect_params_start(keys, vals,
227 : : /* expand_dbname = */ true);
228 : 1030 : PQsetNoticeReceiver(conn->streamConn, libpqsrv_notice_receiver,
229 : : "received message via replication");
230 : 1030 : libpqsrv_connect_complete(conn->streamConn,
231 : : WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);
232 : :
200 233 [ + + ]: 1028 : if (options_val != NULL)
234 : 759 : pfree(options_val);
235 : :
3524 peter_e@gmx.net 236 [ + + ]: 1028 : if (PQstatus(conn->streamConn) != CONNECTION_OK)
1279 andres@anarazel.de 237 : 129 : goto bad_connection_errmsg;
238 : :
1213 rhaas@postgresql.org 239 [ + + - + ]: 899 : if (must_use_password && !PQconnectionUsedPassword(conn->streamConn))
240 : : {
477 heikki.linnakangas@i 241 :UBC 0 : libpqsrv_disconnect(conn->streamConn);
1213 rhaas@postgresql.org 242 : 0 : pfree(conn);
243 : :
244 [ # # ]: 0 : ereport(ERROR,
245 : : (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
246 : : errmsg("password is required"),
247 : : errdetail("Non-superuser cannot connect if the server does not request a password."),
248 : : errhint("Target server's authentication method must be changed, or set password_required=false in the subscription parameters.")));
249 : : }
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 : : */
877 akapila@postgresql.o 255 [ + + + + ]:CBC 899 : if (!replication || logical)
256 : : {
257 : : PGresult *res;
258 : :
477 heikki.linnakangas@i 259 : 742 : res = libpqsrv_exec(conn->streamConn,
260 : : ALWAYS_SECURE_SEARCH_PATH_SQL,
261 : : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
2175 noah@leadboat.com 262 [ - + ]: 742 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
263 : : {
2175 noah@leadboat.com 264 :UBC 0 : PQclear(res);
1279 andres@anarazel.de 265 : 0 : *err = psprintf(_("could not clear search path: %s"),
266 : 0 : pchomp(PQerrorMessage(conn->streamConn)));
267 : 0 : goto bad_connection;
268 : : }
2175 noah@leadboat.com 269 :CBC 742 : PQclear(res);
270 : : }
271 : :
3524 peter_e@gmx.net 272 : 899 : conn->logical = logical;
273 : :
274 : 899 : return conn;
275 : :
276 : : /* error path, using libpq's error message */
1279 andres@anarazel.de 277 : 129 : bad_connection_errmsg:
278 : 129 : *err = pchomp(PQerrorMessage(conn->streamConn));
279 : :
280 : : /* error path, error already set */
281 : 129 : bad_connection:
477 heikki.linnakangas@i 282 : 129 : libpqsrv_disconnect(conn->streamConn);
1279 andres@anarazel.de 283 : 129 : pfree(conn);
284 : 129 : 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
1213 rhaas@postgresql.org 296 : 1299 : libpqrcv_check_conninfo(const char *conninfo, bool must_use_password)
297 : : {
3356 bruce@momjian.us 298 : 1299 : PQconninfoOption *opts = NULL;
299 : : PQconninfoOption *opt;
300 : 1299 : char *err = NULL;
301 : :
3474 peter_e@gmx.net 302 : 1299 : opts = PQconninfoParse(conninfo, &err);
303 [ + + ]: 1299 : if (opts == NULL)
304 : : {
305 : : /* The error string is malloc'd, so we must free it explicitly */
1955 tgl@sss.pgh.pa.us 306 [ + - ]: 12 : char *errcopy = err ? pstrdup(err) : "out of memory";
307 : :
308 : 12 : PQfreemem(err);
3474 peter_e@gmx.net 309 [ + - ]: 12 : ereport(ERROR,
310 : : (errcode(ERRCODE_SYNTAX_ERROR),
311 : : errmsg("invalid connection string syntax: %s", errcopy)));
312 : : }
313 : :
1213 rhaas@postgresql.org 314 [ + + ]: 1287 : if (must_use_password)
315 : : {
1163 tgl@sss.pgh.pa.us 316 : 41 : bool uses_password = false;
317 : :
1213 rhaas@postgresql.org 318 [ + + ]: 948 : for (opt = opts; opt->keyword != NULL; ++opt)
319 : : {
320 : : /* Ignore connection options that are not present. */
321 [ + + ]: 932 : if (opt->val == NULL)
322 : 856 : continue;
323 : :
324 [ + + + - ]: 76 : if (strcmp(opt->keyword, "password") == 0 && opt->val[0] != '\0')
325 : : {
326 : 25 : uses_password = true;
327 : 25 : break;
328 : : }
329 : : }
330 : :
331 [ + + ]: 41 : if (!uses_password)
332 : : {
333 : : /* malloc'd, so we must free it explicitly */
925 jdavis@postgresql.or 334 : 16 : PQconninfoFree(opts);
335 : :
1213 rhaas@postgresql.org 336 [ + - ]: 16 : 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 : :
3474 peter_e@gmx.net 343 : 1271 : PQconninfoFree(opts);
344 : 1271 : }
345 : :
346 : : /*
347 : : * Return a user-displayable conninfo string. Any security-sensitive fields
348 : : * are obfuscated.
349 : : */
350 : : static char *
3524 351 : 157 : libpqrcv_get_conninfo(WalReceiverConn *conn)
352 : : {
353 : : PQconninfoOption *conn_opts;
354 : : PQconninfoOption *conn_opt;
355 : : PQExpBufferData buf;
356 : : char *retval;
357 : :
358 [ - + ]: 157 : Assert(conn->streamConn != NULL);
359 : :
3678 alvherre@alvh.no-ip. 360 : 157 : initPQExpBuffer(&buf);
3524 peter_e@gmx.net 361 : 157 : conn_opts = PQconninfo(conn->streamConn);
362 : :
3678 alvherre@alvh.no-ip. 363 [ - + ]: 157 : if (conn_opts == NULL)
3678 alvherre@alvh.no-ip. 364 [ # # ]:UBC 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 */
3678 alvherre@alvh.no-ip. 370 [ + + ]:CBC 8321 : for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
371 : : {
372 : : bool obfuscate;
373 : :
374 : : /* Skip debug and empty options */
375 [ + + ]: 8164 : if (strchr(conn_opt->dispchar, 'D') ||
376 [ + + ]: 7536 : conn_opt->val == NULL ||
377 [ + + ]: 2996 : conn_opt->val[0] == '\0')
378 : 5325 : continue;
379 : :
380 : : /* Obfuscate security-sensitive options */
381 : 2839 : obfuscate = strchr(conn_opt->dispchar, '*') != NULL;
382 : :
383 [ + + ]: 5678 : appendPQExpBuffer(&buf, "%s%s=%s",
384 [ + + ]: 2839 : buf.len == 0 ? "" : " ",
385 : : conn_opt->keyword,
386 : : obfuscate ? "********" : conn_opt->val);
387 : : }
388 : :
389 : 157 : PQconninfoFree(conn_opts);
390 : :
391 [ + - ]: 157 : retval = PQExpBufferDataBroken(buf) ? NULL : pstrdup(buf.data);
392 : 157 : termPQExpBuffer(&buf);
393 : 157 : return retval;
394 : : }
395 : :
396 : : /*
397 : : * Provides information of sender this WAL receiver is connected to.
398 : : */
399 : : static void
3038 fujii@postgresql.org 400 : 157 : libpqrcv_get_senderinfo(WalReceiverConn *conn, char **sender_host,
401 : : int *sender_port)
402 : : {
3012 tgl@sss.pgh.pa.us 403 : 157 : char *ret = NULL;
404 : :
3038 fujii@postgresql.org 405 : 157 : *sender_host = NULL;
406 : 157 : *sender_port = 0;
407 : :
408 [ - + ]: 157 : Assert(conn->streamConn != NULL);
409 : :
410 : 157 : ret = PQhost(conn->streamConn);
411 [ + - + - ]: 157 : if (ret && strlen(ret) != 0)
412 : 157 : *sender_host = pstrdup(ret);
413 : :
414 : 157 : ret = PQport(conn->streamConn);
415 [ + - + - ]: 157 : if (ret && strlen(ret) != 0)
416 : 157 : *sender_port = atoi(ret);
417 : 157 : }
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 *
2689 peter@eisentraut.org 424 : 410 : 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 : : */
477 heikki.linnakangas@i 433 : 410 : res = libpqsrv_exec(conn->streamConn,
434 : : "IDENTIFY_SYSTEM",
435 : : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
6030 436 [ - + ]: 410 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
6030 heikki.linnakangas@i 437 [ # # ]:UBC 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 : : */
4358 fujii@postgresql.org 447 [ + - - + ]:CBC 410 : if (PQnfields(res) < 3 || PQntuples(res) != 1)
6030 heikki.linnakangas@i 448 [ # # ]:UBC 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)));
3524 peter_e@gmx.net 453 :CBC 410 : primary_sysid = pstrdup(PQgetvalue(res, 0, 0));
2925 andres@anarazel.de 454 : 410 : *primary_tli = pg_strtoint32(PQgetvalue(res, 0, 1));
6030 heikki.linnakangas@i 455 : 410 : PQclear(res);
456 : :
3524 peter_e@gmx.net 457 : 410 : return primary_sysid;
458 : : }
459 : :
460 : : /*
461 : : * Thin wrapper around libpq to obtain server version.
462 : : */
463 : : static int
2689 peter@eisentraut.org 464 : 1062 : libpqrcv_server_version(WalReceiverConn *conn)
465 : : {
466 : 1062 : 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 *
901 akapila@postgresql.o 475 : 16 : libpqrcv_get_dbname_from_conninfo(const char *connInfo)
476 : : {
200 fujii@postgresql.org 477 : 16 : 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 : 777 : libpqrcv_get_option_from_conninfo(const char *connInfo, const char *keyword)
488 : : {
489 : : PQconninfoOption *opts;
490 : 777 : char *option = NULL;
901 akapila@postgresql.o 491 : 777 : char *err = NULL;
492 : :
493 : 777 : opts = PQconninfoParse(connInfo, &err);
494 [ - + ]: 777 : if (opts == NULL)
495 : : {
496 : : /* The error string is malloc'd, so we must free it explicitly */
901 akapila@postgresql.o 497 [ # # ]:UBC 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 : :
901 akapila@postgresql.o 505 [ + + ]:CBC 41181 : 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 : : */
200 fujii@postgresql.org 511 [ + + + + ]: 40404 : if (strcmp(opt->keyword, keyword) == 0 && opt->val &&
901 akapila@postgresql.o 512 [ + - ]: 24 : *opt->val)
513 : : {
200 fujii@postgresql.org 514 [ - + ]: 24 : if (option)
200 fujii@postgresql.org 515 :UBC 0 : pfree(option);
516 : :
200 fujii@postgresql.org 517 :CBC 24 : option = pstrdup(opt->val);
518 : : }
519 : : }
520 : :
901 akapila@postgresql.o 521 : 777 : PQconninfoFree(opts);
200 fujii@postgresql.org 522 : 777 : return option;
523 : : }
524 : :
525 : : /*
526 : : * Append a suitably-quoted identifier or string literal to buf.
527 : : * "quote" should be either a double-quote or single-quote character.
528 : : *
529 : : * Caution: this quoting logic is sufficient for identifiers and literals
530 : : * in the replication grammar, but not always in regular SQL. Specifically,
531 : : * it'd fail for a string literal if standard_conforming_strings is off.
532 : : */
533 : : static void
40 tgl@sss.pgh.pa.us 534 : 2827 : appendQuotedString(StringInfo buf, const char *str, char quote)
535 : : {
536 : 2827 : appendStringInfoChar(buf, quote);
537 [ + + ]: 42516 : while (*str)
538 : : {
539 : 39689 : char c = *str++;
540 : :
541 [ - + ]: 39689 : if (c == quote)
40 tgl@sss.pgh.pa.us 542 :UBC 0 : appendStringInfoChar(buf, c);
40 tgl@sss.pgh.pa.us 543 :CBC 39689 : appendStringInfoChar(buf, c);
544 : : }
545 : 2827 : appendStringInfoChar(buf, quote);
546 : 2827 : }
547 : :
548 : : #define appendQuotedIdentifier(b, s) appendQuotedString(b, s, '"')
549 : : #define appendQuotedLiteral(b, s) appendQuotedString(b, s, '\'')
550 : :
551 : : /*
552 : : * Start streaming WAL data from given streaming options.
553 : : *
554 : : * Returns true if we switched successfully to copy-both mode. False
555 : : * means the server received the command and executed it successfully, but
556 : : * didn't switch to copy-mode. That means that there was no WAL on the
557 : : * requested timeline and starting point, because the server switched to
558 : : * another timeline at or before the requested starting point. On failure,
559 : : * throws an ERROR.
560 : : */
561 : : static bool
3524 peter_e@gmx.net 562 : 608 : libpqrcv_startstreaming(WalReceiverConn *conn,
563 : : const WalRcvStreamOptions *options)
564 : : {
565 : : StringInfoData cmd;
566 : : PGresult *res;
567 : :
3474 568 [ - + ]: 608 : Assert(options->logical == conn->logical);
569 [ + + - + ]: 608 : Assert(options->slotname || !options->logical);
570 : :
3524 571 : 608 : initStringInfo(&cmd);
572 : :
573 : : /* Build the command. */
3474 574 : 608 : appendStringInfoString(&cmd, "START_REPLICATION");
575 [ + + ]: 608 : if (options->slotname != NULL)
576 : : {
40 tgl@sss.pgh.pa.us 577 : 489 : appendStringInfoString(&cmd, " SLOT ");
578 : 489 : appendQuotedIdentifier(&cmd, options->slotname);
579 : : }
580 : :
3474 peter_e@gmx.net 581 [ + + ]: 608 : if (options->logical)
3266 582 : 438 : appendStringInfoString(&cmd, " LOGICAL");
583 : :
383 alvherre@kurilemu.de 584 : 608 : appendStringInfo(&cmd, " %X/%08X", LSN_FORMAT_ARGS(options->startpoint));
585 : :
586 : : /*
587 : : * Additional options are different depending on if we are doing logical
588 : : * or physical replication.
589 : : */
3474 peter_e@gmx.net 590 [ + + ]: 608 : if (options->logical)
591 : : {
592 : : char *pubnames_str;
593 : : List *pubnames;
594 : :
595 : 438 : appendStringInfoString(&cmd, " (");
596 : :
597 : 438 : appendStringInfo(&cmd, "proto_version '%u'",
598 : 438 : options->proto.logical.proto_version);
599 : :
1293 akapila@postgresql.o 600 [ + + ]: 438 : if (options->proto.logical.streaming_str)
601 : : {
40 tgl@sss.pgh.pa.us 602 : 430 : appendStringInfoString(&cmd, ", streaming ");
603 : 430 : appendQuotedLiteral(&cmd, options->proto.logical.streaming_str);
604 : : }
605 : :
1837 akapila@postgresql.o 606 [ + + + - ]: 445 : if (options->proto.logical.twophase &&
607 : 7 : PQserverVersion(conn->streamConn) >= 150000)
608 : 7 : appendStringInfoString(&cmd, ", two_phase 'on'");
609 : :
1465 610 [ + - + - ]: 876 : if (options->proto.logical.origin &&
611 : 438 : PQserverVersion(conn->streamConn) >= 160000)
612 : : {
40 tgl@sss.pgh.pa.us 613 : 438 : appendStringInfoString(&cmd, ", origin ");
614 : 438 : appendQuotedLiteral(&cmd, options->proto.logical.origin);
615 : : }
616 : :
3474 peter_e@gmx.net 617 : 438 : pubnames = options->proto.logical.publication_names;
40 tgl@sss.pgh.pa.us 618 : 438 : pubnames_str = stringlist_to_identifierstr(pubnames);
619 : 438 : appendStringInfoString(&cmd, ", publication_names ");
620 : 438 : appendQuotedLiteral(&cmd, pubnames_str);
3474 peter_e@gmx.net 621 : 438 : pfree(pubnames_str);
622 : :
2198 tgl@sss.pgh.pa.us 623 [ + + + - ]: 449 : if (options->proto.logical.binary &&
624 : 11 : PQserverVersion(conn->streamConn) >= 140000)
625 : 11 : appendStringInfoString(&cmd, ", binary 'true'");
626 : :
3470 peter_e@gmx.net 627 : 438 : appendStringInfoChar(&cmd, ')');
628 : : }
629 : : else
3474 630 : 170 : appendStringInfo(&cmd, " TIMELINE %u",
631 : 170 : options->proto.physical.startpointTLI);
632 : :
633 : : /* Start streaming. */
477 heikki.linnakangas@i 634 : 608 : res = libpqsrv_exec(conn->streamConn,
635 : 608 : cmd.data,
636 : : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
3524 peter_e@gmx.net 637 : 608 : pfree(cmd.data);
638 : :
4972 heikki.linnakangas@i 639 [ - + ]: 608 : if (PQresultStatus(res) == PGRES_COMMAND_OK)
640 : : {
4972 heikki.linnakangas@i 641 :UBC 0 : PQclear(res);
642 : 0 : return false;
643 : : }
4972 heikki.linnakangas@i 644 [ + + ]:CBC 608 : else if (PQresultStatus(res) != PGRES_COPY_BOTH)
6030 645 [ + - ]: 1 : ereport(ERROR,
646 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
647 : : errmsg("could not start WAL streaming: %s",
648 : : pchomp(PQerrorMessage(conn->streamConn)))));
649 : 607 : PQclear(res);
4972 650 : 607 : return true;
651 : : }
652 : :
653 : : /*
654 : : * Stop streaming WAL data. Returns the next timeline's ID in *next_tli, as
655 : : * reported by the server, or 0 if it did not report it.
656 : : */
657 : : static void
3524 peter_e@gmx.net 658 : 255 : libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
659 : : {
660 : : PGresult *res;
661 : :
662 : : /*
663 : : * Send copy-end message. As in libpqsrv_exec, this could theoretically
664 : : * block, but the risk seems small.
665 : : */
666 [ + + - + ]: 466 : if (PQputCopyEnd(conn->streamConn, NULL) <= 0 ||
667 : 211 : PQflush(conn->streamConn))
4972 heikki.linnakangas@i 668 [ + - ]: 44 : ereport(ERROR,
669 : : (errcode(ERRCODE_CONNECTION_FAILURE),
670 : : errmsg("could not send end-of-streaming message to primary: %s",
671 : : pchomp(PQerrorMessage(conn->streamConn)))));
672 : :
3524 peter_e@gmx.net 673 : 211 : *next_tli = 0;
674 : :
675 : : /*
676 : : * After COPY is finished, we should receive a result set indicating the
677 : : * next timeline's ID, or just CommandComplete if the server was shut
678 : : * down.
679 : : *
680 : : * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
681 : : * also possible in case we aborted the copy in mid-stream.
682 : : */
477 heikki.linnakangas@i 683 : 211 : res = libpqsrv_get_result(conn->streamConn,
684 : : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
4936 685 [ + + ]: 211 : if (PQresultStatus(res) == PGRES_TUPLES_OK)
686 : : {
687 : : /*
688 : : * Read the next timeline's ID. The server also sends the timeline's
689 : : * starting point, but it is ignored.
690 : : */
4826 691 [ + - - + ]: 13 : if (PQnfields(res) < 2 || PQntuples(res) != 1)
4936 heikki.linnakangas@i 692 [ # # ]:UBC 0 : ereport(ERROR,
693 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
694 : : errmsg("unexpected result set after end-of-streaming")));
2925 andres@anarazel.de 695 :CBC 13 : *next_tli = pg_strtoint32(PQgetvalue(res, 0, 0));
4972 heikki.linnakangas@i 696 : 13 : PQclear(res);
697 : :
698 : : /* the result set should be followed by CommandComplete */
477 699 : 13 : res = libpqsrv_get_result(conn->streamConn,
700 : : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
701 : : }
3524 peter_e@gmx.net 702 [ + - ]: 198 : else if (PQresultStatus(res) == PGRES_COPY_OUT)
703 : : {
704 : 198 : PQclear(res);
705 : :
706 : : /* End the copy */
3312 tgl@sss.pgh.pa.us 707 [ - + ]: 198 : if (PQendcopy(conn->streamConn))
3312 tgl@sss.pgh.pa.us 708 [ # # ]:UBC 0 : ereport(ERROR,
709 : : (errcode(ERRCODE_CONNECTION_FAILURE),
710 : : errmsg("error while shutting down streaming COPY: %s",
711 : : pchomp(PQerrorMessage(conn->streamConn)))));
712 : :
713 : : /* CommandComplete should follow */
477 heikki.linnakangas@i 714 :CBC 198 : res = libpqsrv_get_result(conn->streamConn,
715 : : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
716 : : }
717 : :
4936 718 [ - + ]: 211 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
4936 heikki.linnakangas@i 719 [ # # ]:UBC 0 : ereport(ERROR,
720 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
721 : : errmsg("error reading result of streaming command: %s",
722 : : pchomp(PQerrorMessage(conn->streamConn)))));
4182 tgl@sss.pgh.pa.us 723 :CBC 211 : PQclear(res);
724 : :
725 : : /* Verify that there are no more results */
477 heikki.linnakangas@i 726 : 211 : res = libpqsrv_get_result(conn->streamConn,
727 : : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
4936 728 [ - + ]: 211 : if (res != NULL)
4936 heikki.linnakangas@i 729 [ # # ]:UBC 0 : ereport(ERROR,
730 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
731 : : errmsg("unexpected result after CommandComplete: %s",
732 : : pchomp(PQerrorMessage(conn->streamConn)))));
4972 heikki.linnakangas@i 733 :CBC 211 : }
734 : :
735 : : /*
736 : : * Fetch the timeline history file for 'tli' from primary.
737 : : */
738 : : static void
3524 peter_e@gmx.net 739 : 12 : libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
740 : : TimeLineID tli, char **filename,
741 : : char **content, size_t *len)
742 : : {
743 : : PGresult *res;
744 : : char cmd[64];
745 : :
746 [ - + ]: 12 : Assert(!conn->logical);
747 : :
748 : : /*
749 : : * Request the primary to send over the history file for given timeline.
750 : : */
4972 heikki.linnakangas@i 751 : 12 : snprintf(cmd, sizeof(cmd), "TIMELINE_HISTORY %u", tli);
477 752 : 12 : res = libpqsrv_exec(conn->streamConn,
753 : : cmd,
754 : : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
4972 755 [ - + ]: 12 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
4972 heikki.linnakangas@i 756 [ # # ]:UBC 0 : ereport(ERROR,
757 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
758 : : errmsg("could not receive timeline history file from "
759 : : "the primary server: %s",
760 : : pchomp(PQerrorMessage(conn->streamConn)))));
4972 heikki.linnakangas@i 761 [ + - - + ]:CBC 12 : if (PQnfields(res) != 2 || PQntuples(res) != 1)
4972 heikki.linnakangas@i 762 [ # # ]:UBC 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 : : PQntuples(res), PQnfields(res))));
4972 heikki.linnakangas@i 767 :CBC 12 : *filename = pstrdup(PQgetvalue(res, 0, 0));
768 : :
769 : 12 : *len = PQgetlength(res, 0, 1);
770 : 12 : *content = palloc(*len);
771 : 12 : memcpy(*content, PQgetvalue(res, 0, 1), *len);
772 : 12 : PQclear(res);
6030 773 : 12 : }
774 : :
775 : : /*
776 : : * Disconnect connection to primary, if any.
777 : : */
778 : : static void
3524 peter_e@gmx.net 779 : 899 : libpqrcv_disconnect(WalReceiverConn *conn)
780 : : {
477 heikki.linnakangas@i 781 : 899 : libpqsrv_disconnect(conn->streamConn);
1429 peter@eisentraut.org 782 : 899 : PQfreemem(conn->recvBuf);
3524 peter_e@gmx.net 783 : 899 : pfree(conn);
6030 heikki.linnakangas@i 784 : 899 : }
785 : :
786 : : /*
787 : : * Receive a message available from XLOG stream.
788 : : *
789 : : * Returns:
790 : : *
791 : : * If data was received, returns the length of the data. *buffer is set to
792 : : * point to a buffer holding the received message. The buffer is only valid
793 : : * until the next libpqrcv_* call.
794 : : *
795 : : * If no data was available immediately, returns 0, and *wait_fd is set to a
796 : : * socket descriptor which can be waited on before trying again.
797 : : *
798 : : * -1 if the server ended the COPY.
799 : : *
800 : : * ereports on error.
801 : : */
802 : : static int
3524 peter_e@gmx.net 803 : 452818 : libpqrcv_receive(WalReceiverConn *conn, char **buffer,
804 : : pgsocket *wait_fd)
805 : : {
806 : : int rawlen;
807 : :
1429 peter@eisentraut.org 808 : 452818 : PQfreemem(conn->recvBuf);
3524 peter_e@gmx.net 809 : 452818 : conn->recvBuf = NULL;
810 : :
811 : : /* Try to receive a CopyData message */
812 : 452818 : rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
5672 heikki.linnakangas@i 813 [ + + ]: 452818 : if (rawlen == 0)
814 : : {
815 : : /* Try consuming some data. */
3524 peter_e@gmx.net 816 [ + + ]: 254062 : if (PQconsumeInput(conn->streamConn) == 0)
6030 heikki.linnakangas@i 817 [ + - ]: 43 : ereport(ERROR,
818 : : (errcode(ERRCODE_CONNECTION_FAILURE),
819 : : errmsg("could not receive data from WAL stream: %s",
820 : : pchomp(PQerrorMessage(conn->streamConn)))));
821 : :
822 : : /* Now that we've consumed some input, try again */
3524 peter_e@gmx.net 823 : 254019 : rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
5672 heikki.linnakangas@i 824 [ + + ]: 254019 : if (rawlen == 0)
825 : : {
826 : : /* Tell caller to try again when our socket is ready. */
3524 peter_e@gmx.net 827 : 111264 : *wait_fd = PQsocket(conn->streamConn);
4972 heikki.linnakangas@i 828 : 111264 : return 0;
829 : : }
830 : : }
5993 bruce@momjian.us 831 [ + + ]: 341511 : if (rawlen == -1) /* end-of-streaming or error */
832 : : {
833 : : PGresult *res;
834 : :
477 heikki.linnakangas@i 835 : 268 : res = libpqsrv_get_result(conn->streamConn,
836 : : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
3411 peter_e@gmx.net 837 [ + + ]: 268 : if (PQresultStatus(res) == PGRES_COMMAND_OK)
838 : : {
839 : 251 : PQclear(res);
840 : :
841 : : /* Verify that there are no more results. */
477 heikki.linnakangas@i 842 : 251 : res = libpqsrv_get_result(conn->streamConn,
843 : : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
3411 peter_e@gmx.net 844 [ + + ]: 251 : if (res != NULL)
845 : : {
3334 andres@anarazel.de 846 : 45 : PQclear(res);
847 : :
848 : : /*
849 : : * If the other side closed the connection orderly (otherwise
850 : : * we'd seen an error, or PGRES_COPY_IN) don't report an error
851 : : * here, but let callers deal with it.
852 : : */
853 [ + - ]: 45 : if (PQstatus(conn->streamConn) == CONNECTION_BAD)
854 : 45 : return -1;
855 : :
3411 peter_e@gmx.net 856 [ # # ]:UBC 0 : ereport(ERROR,
857 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
858 : : errmsg("unexpected result after CommandComplete: %s",
859 : : PQerrorMessage(conn->streamConn))));
860 : : }
861 : :
3411 peter_e@gmx.net 862 :CBC 206 : return -1;
863 : : }
864 [ + + ]: 17 : else if (PQresultStatus(res) == PGRES_COPY_IN)
865 : : {
4972 heikki.linnakangas@i 866 : 13 : PQclear(res);
867 : 13 : return -1;
868 : : }
869 : : else
6030 870 [ + - ]: 4 : ereport(ERROR,
871 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
872 : : errmsg("could not receive data from WAL stream: %s",
873 : : pchomp(PQerrorMessage(conn->streamConn)))));
874 : : }
875 [ - + ]: 341243 : if (rawlen < -1)
6030 heikki.linnakangas@i 876 [ # # ]:UBC 0 : ereport(ERROR,
877 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
878 : : errmsg("could not receive data from WAL stream: %s",
879 : : pchomp(PQerrorMessage(conn->streamConn)))));
880 : :
881 : : /* Return received messages to caller */
3524 peter_e@gmx.net 882 :CBC 341243 : *buffer = conn->recvBuf;
4972 heikki.linnakangas@i 883 : 341243 : return rawlen;
884 : : }
885 : :
886 : : /*
887 : : * Send a message to XLOG stream.
888 : : *
889 : : * ereports on error.
890 : : */
891 : : static void
3524 peter_e@gmx.net 892 : 158472 : libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes)
893 : : {
894 [ + + + + ]: 316943 : if (PQputCopyData(conn->streamConn, buffer, nbytes) <= 0 ||
895 : 158471 : PQflush(conn->streamConn))
5705 rhaas@postgresql.org 896 [ + - ]: 2 : ereport(ERROR,
897 : : (errcode(ERRCODE_CONNECTION_FAILURE),
898 : : errmsg("could not send data to WAL stream: %s",
899 : : pchomp(PQerrorMessage(conn->streamConn)))));
900 : 158470 : }
901 : :
902 : : /*
903 : : * Create new replication slot.
904 : : * Returns the name of the exported snapshot for logical slot or NULL for
905 : : * physical slot.
906 : : */
907 : : static char *
3474 peter_e@gmx.net 908 : 331 : libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
909 : : bool temporary, bool two_phase, bool failover,
910 : : CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
911 : : {
912 : : PGresult *res;
913 : : StringInfoData cmd;
914 : : char *snapshot;
915 : : int use_new_options_syntax;
916 : :
1754 rhaas@postgresql.org 917 : 331 : use_new_options_syntax = (PQserverVersion(conn->streamConn) >= 150000);
918 : :
3474 peter_e@gmx.net 919 : 331 : initStringInfo(&cmd);
920 : :
40 tgl@sss.pgh.pa.us 921 : 331 : appendStringInfoString(&cmd, "CREATE_REPLICATION_SLOT ");
922 : 331 : appendQuotedIdentifier(&cmd, slotname);
923 : :
3474 peter_e@gmx.net 924 [ - + ]: 331 : if (temporary)
3266 peter_e@gmx.net 925 :UBC 0 : appendStringInfoString(&cmd, " TEMPORARY");
926 : :
3474 peter_e@gmx.net 927 [ + - ]:CBC 331 : if (conn->logical)
928 : : {
1754 rhaas@postgresql.org 929 : 331 : appendStringInfoString(&cmd, " LOGICAL pgoutput ");
930 [ + - ]: 331 : if (use_new_options_syntax)
931 : 331 : appendStringInfoChar(&cmd, '(');
1837 akapila@postgresql.o 932 [ + + ]: 331 : if (two_phase)
933 : : {
1754 rhaas@postgresql.org 934 : 1 : appendStringInfoString(&cmd, "TWO_PHASE");
935 [ + - ]: 1 : if (use_new_options_syntax)
936 : 1 : appendStringInfoString(&cmd, ", ");
937 : : else
1754 rhaas@postgresql.org 938 :UBC 0 : appendStringInfoChar(&cmd, ' ');
939 : : }
940 : :
908 akapila@postgresql.o 941 [ + + ]:CBC 331 : if (failover)
942 : : {
943 : 11 : appendStringInfoString(&cmd, "FAILOVER");
944 [ + - ]: 11 : if (use_new_options_syntax)
945 : 11 : appendStringInfoString(&cmd, ", ");
946 : : else
908 akapila@postgresql.o 947 :UBC 0 : appendStringInfoChar(&cmd, ' ');
948 : : }
949 : :
1754 rhaas@postgresql.org 950 [ + - ]:CBC 331 : if (use_new_options_syntax)
951 : : {
952 [ - + + - ]: 331 : switch (snapshot_action)
953 : : {
1754 rhaas@postgresql.org 954 :UBC 0 : case CRS_EXPORT_SNAPSHOT:
955 : 0 : appendStringInfoString(&cmd, "SNAPSHOT 'export'");
956 : 0 : break;
1754 rhaas@postgresql.org 957 :CBC 120 : case CRS_NOEXPORT_SNAPSHOT:
958 : 120 : appendStringInfoString(&cmd, "SNAPSHOT 'nothing'");
959 : 120 : break;
960 : 211 : case CRS_USE_SNAPSHOT:
961 : 211 : appendStringInfoString(&cmd, "SNAPSHOT 'use'");
962 : 211 : break;
963 : : }
964 : : }
965 : : else
966 : : {
1754 rhaas@postgresql.org 967 [ # # # # ]:UBC 0 : switch (snapshot_action)
968 : : {
969 : 0 : case CRS_EXPORT_SNAPSHOT:
970 : 0 : appendStringInfoString(&cmd, "EXPORT_SNAPSHOT");
971 : 0 : break;
972 : 0 : case CRS_NOEXPORT_SNAPSHOT:
973 : 0 : appendStringInfoString(&cmd, "NOEXPORT_SNAPSHOT");
974 : 0 : break;
975 : 0 : case CRS_USE_SNAPSHOT:
976 : 0 : appendStringInfoString(&cmd, "USE_SNAPSHOT");
977 : 0 : break;
978 : : }
979 : : }
980 : :
1754 rhaas@postgresql.org 981 [ + - ]:CBC 331 : if (use_new_options_syntax)
982 : 331 : appendStringInfoChar(&cmd, ')');
983 : : }
984 : : else
985 : : {
1754 rhaas@postgresql.org 986 [ # # ]:UBC 0 : if (use_new_options_syntax)
987 : 0 : appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
988 : : else
989 : 0 : appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
990 : : }
991 : :
477 heikki.linnakangas@i 992 :CBC 331 : res = libpqsrv_exec(conn->streamConn,
993 : 331 : cmd.data,
994 : : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
3474 peter_e@gmx.net 995 : 331 : pfree(cmd.data);
996 : :
997 [ - + ]: 331 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
3474 peter_e@gmx.net 998 [ # # ]:UBC 0 : ereport(ERROR,
999 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
1000 : : errmsg("could not create replication slot \"%s\": %s",
1001 : : slotname, pchomp(PQerrorMessage(conn->streamConn)))));
1002 : :
1003 : : /* CREATE_REPLICATION_SLOT returns a single row with four columns */
9 fujii@postgresql.org 1004 [ + - - + ]:CBC 331 : if (PQnfields(res) != 4 || PQntuples(res) != 1)
9 fujii@postgresql.org 1005 [ # # ]:UBC 0 : ereport(ERROR,
1006 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
1007 : : errmsg("invalid response from primary server"),
1008 : : errdetail("Could not create replication slot \"%s\": got %d rows and %d fields, expected %d rows and %d fields.",
1009 : : slotname, PQntuples(res), PQnfields(res), 1, 4)));
1010 : :
2387 peter@eisentraut.org 1011 [ + + ]:CBC 331 : if (lsn)
1012 : 211 : *lsn = DatumGetLSN(DirectFunctionCall1Coll(pg_lsn_in, InvalidOid,
1013 : 211 : CStringGetDatum(PQgetvalue(res, 0, 1))));
1014 : :
3474 peter_e@gmx.net 1015 [ - + ]: 331 : if (!PQgetisnull(res, 0, 2))
3474 peter_e@gmx.net 1016 :UBC 0 : snapshot = pstrdup(PQgetvalue(res, 0, 2));
1017 : : else
3474 peter_e@gmx.net 1018 :CBC 331 : snapshot = NULL;
1019 : :
1020 : 331 : PQclear(res);
1021 : :
1022 : 331 : return snapshot;
1023 : : }
1024 : :
1025 : : /*
1026 : : * Change the definition of the replication slot.
1027 : : */
1028 : : static void
908 akapila@postgresql.o 1029 : 5 : libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
1030 : : const bool *failover, const bool *two_phase)
1031 : : {
1032 : : StringInfoData cmd;
1033 : : PGresult *res;
1034 : :
1035 : 5 : initStringInfo(&cmd);
40 tgl@sss.pgh.pa.us 1036 : 5 : appendStringInfoString(&cmd, "ALTER_REPLICATION_SLOT ");
1037 : 5 : appendQuotedIdentifier(&cmd, slotname);
1038 : 5 : appendStringInfoString(&cmd, " ( ");
1039 : :
731 akapila@postgresql.o 1040 [ + + ]: 5 : if (failover)
1041 : 4 : appendStringInfo(&cmd, "FAILOVER %s",
1042 [ + + ]: 4 : *failover ? "true" : "false");
1043 : :
1044 [ + + - + ]: 5 : if (failover && two_phase)
470 drowley@postgresql.o 1045 :UBC 0 : appendStringInfoString(&cmd, ", ");
1046 : :
731 akapila@postgresql.o 1047 [ + + ]:CBC 5 : if (two_phase)
1048 : 1 : appendStringInfo(&cmd, "TWO_PHASE %s",
1049 [ - + ]: 1 : *two_phase ? "true" : "false");
1050 : :
1051 : 5 : appendStringInfoString(&cmd, " );");
1052 : :
477 heikki.linnakangas@i 1053 : 5 : res = libpqsrv_exec(conn->streamConn, cmd.data,
1054 : : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
908 akapila@postgresql.o 1055 : 5 : pfree(cmd.data);
1056 : :
1057 [ - + ]: 5 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
908 akapila@postgresql.o 1058 [ # # ]:UBC 0 : ereport(ERROR,
1059 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
1060 : : errmsg("could not alter replication slot \"%s\": %s",
1061 : : slotname, pchomp(PQerrorMessage(conn->streamConn)))));
1062 : :
908 akapila@postgresql.o 1063 :CBC 5 : PQclear(res);
1064 : 5 : }
1065 : :
1066 : : /*
1067 : : * Return PID of remote backend process.
1068 : : */
1069 : : static pid_t
2384 peter@eisentraut.org 1070 :UBC 0 : libpqrcv_get_backend_pid(WalReceiverConn *conn)
1071 : : {
1072 : 0 : return PQbackendPID(conn->streamConn);
1073 : : }
1074 : :
1075 : : /*
1076 : : * Convert tuple query result to tuplestore.
1077 : : */
1078 : : static void
3411 peter_e@gmx.net 1079 :CBC 1233 : libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
1080 : : const int nRetTypes, const Oid *retTypes)
1081 : : {
1082 : : int tupn;
1083 : : int coln;
3356 bruce@momjian.us 1084 : 1233 : int nfields = PQnfields(pgres);
1085 : : HeapTuple tuple;
1086 : : AttInMetadata *attinmeta;
1087 : : MemoryContext rowcontext;
1088 : : MemoryContext oldcontext;
1089 : :
1090 : : /* Make sure we got expected number of fields. */
3411 peter_e@gmx.net 1091 [ - + ]: 1233 : if (nfields != nRetTypes)
3411 peter_e@gmx.net 1092 [ # # ]:UBC 0 : ereport(ERROR,
1093 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
1094 : : errmsg("invalid query response"),
1095 : : errdetail("Expected %d fields, got %d fields.",
1096 : : nRetTypes, nfields)));
1097 : :
3411 peter_e@gmx.net 1098 :CBC 1233 : walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
1099 : :
1100 : : /* Create tuple descriptor corresponding to expected result. */
2804 andres@anarazel.de 1101 : 1233 : walres->tupledesc = CreateTemplateTupleDesc(nRetTypes);
3411 peter_e@gmx.net 1102 [ + + ]: 4634 : for (coln = 0; coln < nRetTypes; coln++)
1103 : 3401 : TupleDescInitEntry(walres->tupledesc, (AttrNumber) coln + 1,
1104 : 3401 : PQfname(pgres, coln), retTypes[coln], -1, 0);
131 drowley@postgresql.o 1105 : 1233 : TupleDescFinalize(walres->tupledesc);
3411 peter_e@gmx.net 1106 : 1233 : attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
1107 : :
1108 : : /* No point in doing more here if there were no tuples returned. */
3410 1109 [ + + ]: 1233 : if (PQntuples(pgres) == 0)
1110 : 34 : return;
1111 : :
1112 : : /* Create temporary context for local allocations. */
3411 1113 : 1199 : rowcontext = AllocSetContextCreate(CurrentMemoryContext,
1114 : : "libpqrcv query result context",
1115 : : ALLOCSET_DEFAULT_SIZES);
1116 : :
1117 : : /* Process returned rows. */
1118 [ + + ]: 2791 : for (tupn = 0; tupn < PQntuples(pgres); tupn++)
1119 : : {
1120 : : char *cstrs[MaxTupleAttributeNumber];
1121 : :
477 heikki.linnakangas@i 1122 [ - + ]: 1592 : CHECK_FOR_INTERRUPTS();
1123 : :
1124 : : /* Do the allocations in temporary context. */
3411 peter_e@gmx.net 1125 : 1592 : oldcontext = MemoryContextSwitchTo(rowcontext);
1126 : :
1127 : : /*
1128 : : * Fill cstrs with null-terminated strings of column values.
1129 : : */
1130 [ + + ]: 6774 : for (coln = 0; coln < nfields; coln++)
1131 : : {
1132 [ + + ]: 5182 : if (PQgetisnull(pgres, tupn, coln))
1133 : 670 : cstrs[coln] = NULL;
1134 : : else
1135 : 4512 : cstrs[coln] = PQgetvalue(pgres, tupn, coln);
1136 : : }
1137 : :
1138 : : /* Convert row to a tuple, and add it to the tuplestore */
1139 : 1592 : tuple = BuildTupleFromCStrings(attinmeta, cstrs);
1140 : 1592 : tuplestore_puttuple(walres->tuplestore, tuple);
1141 : :
1142 : : /* Clean up */
1143 : 1592 : MemoryContextSwitchTo(oldcontext);
1144 : 1592 : MemoryContextReset(rowcontext);
1145 : : }
1146 : :
1147 : 1199 : MemoryContextDelete(rowcontext);
1148 : : }
1149 : :
1150 : : /*
1151 : : * Public interface for sending generic queries (and commands).
1152 : : *
1153 : : * This can only be called from process connected to database.
1154 : : */
1155 : : static WalRcvExecResult *
1156 : 2143 : libpqrcv_exec(WalReceiverConn *conn, const char *query,
1157 : : const int nRetTypes, const Oid *retTypes)
1158 : : {
1159 : 2143 : PGresult *pgres = NULL;
227 michael@paquier.xyz 1160 : 2143 : WalRcvExecResult *walres = palloc0_object(WalRcvExecResult);
1161 : : char *diag_sqlstate;
1162 : :
3411 peter_e@gmx.net 1163 [ - + ]: 2143 : if (MyDatabaseId == InvalidOid)
3411 peter_e@gmx.net 1164 [ # # ]:UBC 0 : ereport(ERROR,
1165 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1166 : : errmsg("the query interface requires a database connection")));
1167 : :
477 heikki.linnakangas@i 1168 :CBC 2143 : pgres = libpqsrv_exec(conn->streamConn,
1169 : : query,
1170 : : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
1171 : :
3411 peter_e@gmx.net 1172 [ + - + - : 2142 : switch (PQresultStatus(pgres))
+ - - +
- ]
1173 : : {
1174 : 1233 : case PGRES_TUPLES_OK:
1175 : : case PGRES_SINGLE_TUPLE:
1176 : : case PGRES_TUPLES_CHUNK:
1177 : 1233 : walres->status = WALRCV_OK_TUPLES;
1178 : 1233 : libpqrcv_processTuples(pgres, walres, nRetTypes, retTypes);
1179 : 1233 : break;
1180 : :
3411 peter_e@gmx.net 1181 :UBC 0 : case PGRES_COPY_IN:
1182 : 0 : walres->status = WALRCV_OK_COPY_IN;
1183 : 0 : break;
1184 : :
3411 peter_e@gmx.net 1185 :CBC 208 : case PGRES_COPY_OUT:
1186 : 208 : walres->status = WALRCV_OK_COPY_OUT;
1187 : 208 : break;
1188 : :
3411 peter_e@gmx.net 1189 :UBC 0 : case PGRES_COPY_BOTH:
1190 : 0 : walres->status = WALRCV_OK_COPY_BOTH;
1191 : 0 : break;
1192 : :
3411 peter_e@gmx.net 1193 :CBC 696 : case PGRES_COMMAND_OK:
1194 : 696 : walres->status = WALRCV_OK_COMMAND;
1195 : 696 : break;
1196 : :
1197 : : /* Empty query is considered error. */
3411 peter_e@gmx.net 1198 :UBC 0 : case PGRES_EMPTY_QUERY:
1199 : 0 : walres->status = WALRCV_ERROR;
1200 : 0 : walres->err = _("empty query");
1201 : 0 : break;
1202 : :
1958 alvherre@alvh.no-ip. 1203 : 0 : case PGRES_PIPELINE_SYNC:
1204 : : case PGRES_PIPELINE_ABORTED:
1205 : 0 : walres->status = WALRCV_ERROR;
1206 : 0 : walres->err = _("unexpected pipeline mode");
1207 : 0 : break;
1208 : :
3411 peter_e@gmx.net 1209 :CBC 5 : case PGRES_NONFATAL_ERROR:
1210 : : case PGRES_FATAL_ERROR:
1211 : : case PGRES_BAD_RESPONSE:
1212 : 5 : walres->status = WALRCV_ERROR;
1213 : 5 : walres->err = pchomp(PQerrorMessage(conn->streamConn));
1989 akapila@postgresql.o 1214 : 5 : diag_sqlstate = PQresultErrorField(pgres, PG_DIAG_SQLSTATE);
1215 [ + + ]: 5 : if (diag_sqlstate)
1216 : 4 : walres->sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
1217 : : diag_sqlstate[1],
1218 : : diag_sqlstate[2],
1219 : : diag_sqlstate[3],
1220 : : diag_sqlstate[4]);
3411 peter_e@gmx.net 1221 : 5 : break;
1222 : : }
1223 : :
1224 : 2142 : PQclear(pgres);
1225 : :
1226 : 2142 : return walres;
1227 : : }
1228 : :
1229 : : /*
1230 : : * Given a List of strings, return it as single comma separated
1231 : : * string, quoting identifiers as needed.
1232 : : *
1233 : : * This is essentially the reverse of SplitIdentifierString.
1234 : : *
1235 : : * The caller should pfree the result.
1236 : : */
1237 : : static char *
40 tgl@sss.pgh.pa.us 1238 : 438 : stringlist_to_identifierstr(List *strings)
1239 : : {
1240 : : ListCell *lc;
1241 : : StringInfoData res;
3356 bruce@momjian.us 1242 : 438 : bool first = true;
1243 : :
3474 peter_e@gmx.net 1244 : 438 : initStringInfo(&res);
1245 : :
3356 bruce@momjian.us 1246 [ + - + + : 1134 : foreach(lc, strings)
+ + ]
1247 : : {
1248 : 696 : char *val = strVal(lfirst(lc));
1249 : :
3474 peter_e@gmx.net 1250 [ + + ]: 696 : if (first)
1251 : 438 : first = false;
1252 : : else
1253 : 258 : appendStringInfoChar(&res, ',');
40 tgl@sss.pgh.pa.us 1254 : 696 : appendQuotedIdentifier(&res, val);
1255 : : }
1256 : :
3474 peter_e@gmx.net 1257 : 438 : return res.data;
1258 : : }
|
