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