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