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