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 : * Portions Copyright (c) 2010-2020, PostgreSQL Global Development Group
10 : *
11 : *
12 : * IDENTIFICATION
13 : * src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
14 : *
15 : *-------------------------------------------------------------------------
16 : */
17 : #include "postgres.h"
18 :
19 : #include <unistd.h>
20 : #include <sys/time.h>
21 :
22 : #include "access/xlog.h"
23 : #include "catalog/pg_type.h"
24 : #include "funcapi.h"
25 : #include "libpq-fe.h"
26 : #include "mb/pg_wchar.h"
27 : #include "miscadmin.h"
28 : #include "pgstat.h"
29 : #include "pqexpbuffer.h"
30 : #include "replication/walreceiver.h"
31 : #include "utils/builtins.h"
32 : #include "utils/memutils.h"
33 : #include "utils/pg_lsn.h"
34 : #include "utils/tuplestore.h"
35 :
36 438 : PG_MODULE_MAGIC;
37 :
38 : void _PG_init(void);
39 :
40 : struct WalReceiverConn
41 : {
42 : /* Current connection to the primary, if any */
43 : PGconn *streamConn;
44 : /* Used to remember if the connection is logical or physical */
45 : bool logical;
46 : /* Buffer for currently read records */
47 : char *recvBuf;
48 : };
49 :
50 : /* Prototypes for interface functions */
51 : static WalReceiverConn *libpqrcv_connect(const char *conninfo,
52 : bool logical, const char *appname,
53 : char **err);
54 : static void libpqrcv_check_conninfo(const char *conninfo);
55 : static char *libpqrcv_get_conninfo(WalReceiverConn *conn);
56 : static void libpqrcv_get_senderinfo(WalReceiverConn *conn,
57 : char **sender_host, int *sender_port);
58 : static char *libpqrcv_identify_system(WalReceiverConn *conn,
59 : TimeLineID *primary_tli);
60 : static int libpqrcv_server_version(WalReceiverConn *conn);
61 : static void libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
62 : TimeLineID tli, char **filename,
63 : char **content, int *len);
64 : static bool libpqrcv_startstreaming(WalReceiverConn *conn,
65 : const WalRcvStreamOptions *options);
66 : static void libpqrcv_endstreaming(WalReceiverConn *conn,
67 : TimeLineID *next_tli);
68 : static int libpqrcv_receive(WalReceiverConn *conn, char **buffer,
69 : pgsocket *wait_fd);
70 : static void libpqrcv_send(WalReceiverConn *conn, const char *buffer,
71 : int nbytes);
72 : static char *libpqrcv_create_slot(WalReceiverConn *conn,
73 : const char *slotname,
74 : bool temporary,
75 : CRSSnapshotAction snapshot_action,
76 : XLogRecPtr *lsn);
77 : static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
78 : static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
79 : const char *query,
80 : const int nRetTypes,
81 : const Oid *retTypes);
82 : static void libpqrcv_disconnect(WalReceiverConn *conn);
83 :
84 : static WalReceiverFunctionsType PQWalReceiverFunctions = {
85 : libpqrcv_connect,
86 : libpqrcv_check_conninfo,
87 : libpqrcv_get_conninfo,
88 : libpqrcv_get_senderinfo,
89 : libpqrcv_identify_system,
90 : libpqrcv_server_version,
91 : libpqrcv_readtimelinehistoryfile,
92 : libpqrcv_startstreaming,
93 : libpqrcv_endstreaming,
94 : libpqrcv_receive,
95 : libpqrcv_send,
96 : libpqrcv_create_slot,
97 : libpqrcv_get_backend_pid,
98 : libpqrcv_exec,
99 : libpqrcv_disconnect
100 : };
101 :
102 : /* Prototypes for private functions */
103 : static PGresult *libpqrcv_PQexec(PGconn *streamConn, const char *query);
104 : static PGresult *libpqrcv_PQgetResult(PGconn *streamConn);
105 : static char *stringlist_to_identifierstr(PGconn *conn, List *strings);
106 :
107 : /*
108 : * Module initialization function
109 : */
110 : void
111 438 : _PG_init(void)
112 : {
113 438 : if (WalReceiverFunctions != NULL)
114 0 : elog(ERROR, "libpqwalreceiver already loaded");
115 438 : WalReceiverFunctions = &PQWalReceiverFunctions;
116 438 : }
117 :
118 : /*
119 : * Establish the connection to the primary server for XLOG streaming
120 : *
121 : * Returns NULL on error and fills the err with palloc'ed error message.
122 : */
123 : static WalReceiverConn *
124 424 : libpqrcv_connect(const char *conninfo, bool logical, const char *appname,
125 : char **err)
126 : {
127 : WalReceiverConn *conn;
128 : PostgresPollingStatusType status;
129 : const char *keys[5];
130 : const char *vals[5];
131 424 : int i = 0;
132 :
133 : /*
134 : * We use the expand_dbname parameter to process the connection string (or
135 : * URI), and pass some extra options.
136 : */
137 424 : keys[i] = "dbname";
138 424 : vals[i] = conninfo;
139 424 : keys[++i] = "replication";
140 424 : vals[i] = logical ? "database" : "true";
141 424 : if (!logical)
142 : {
143 : /*
144 : * The database name is ignored by the server in replication mode, but
145 : * specify "replication" for .pgpass lookup.
146 : */
147 198 : keys[++i] = "dbname";
148 198 : vals[i] = "replication";
149 : }
150 424 : keys[++i] = "fallback_application_name";
151 424 : vals[i] = appname;
152 424 : if (logical)
153 : {
154 226 : keys[++i] = "client_encoding";
155 226 : vals[i] = GetDatabaseEncodingName();
156 : }
157 424 : keys[++i] = NULL;
158 424 : vals[i] = NULL;
159 :
160 : Assert(i < sizeof(keys));
161 :
162 424 : conn = palloc0(sizeof(WalReceiverConn));
163 424 : conn->streamConn = PQconnectStartParams(keys, vals,
164 : /* expand_dbname = */ true);
165 424 : if (PQstatus(conn->streamConn) == CONNECTION_BAD)
166 : {
167 68 : *err = pchomp(PQerrorMessage(conn->streamConn));
168 68 : return NULL;
169 : }
170 :
171 : /*
172 : * Poll connection until we have OK or FAILED status.
173 : *
174 : * Per spec for PQconnectPoll, first wait till socket is write-ready.
175 : */
176 356 : status = PGRES_POLLING_WRITING;
177 : do
178 : {
179 : int io_flag;
180 : int rc;
181 :
182 1006 : if (status == PGRES_POLLING_READING)
183 354 : io_flag = WL_SOCKET_READABLE;
184 : #ifdef WIN32
185 : /* Windows needs a different test while waiting for connection-made */
186 : else if (PQstatus(conn->streamConn) == CONNECTION_STARTED)
187 : io_flag = WL_SOCKET_CONNECTED;
188 : #endif
189 : else
190 652 : io_flag = WL_SOCKET_WRITEABLE;
191 :
192 1006 : rc = WaitLatchOrSocket(MyLatch,
193 : WL_EXIT_ON_PM_DEATH | WL_LATCH_SET | io_flag,
194 1006 : PQsocket(conn->streamConn),
195 : 0,
196 : WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);
197 :
198 : /* Interrupted? */
199 1006 : if (rc & WL_LATCH_SET)
200 : {
201 298 : ResetLatch(MyLatch);
202 298 : ProcessWalRcvInterrupts();
203 : }
204 :
205 : /* If socket is ready, advance the libpq state machine */
206 1004 : if (rc & io_flag)
207 708 : status = PQconnectPoll(conn->streamConn);
208 1004 : } while (status != PGRES_POLLING_OK && status != PGRES_POLLING_FAILED);
209 :
210 354 : if (PQstatus(conn->streamConn) != CONNECTION_OK)
211 : {
212 16 : *err = pchomp(PQerrorMessage(conn->streamConn));
213 16 : return NULL;
214 : }
215 :
216 338 : conn->logical = logical;
217 :
218 338 : return conn;
219 : }
220 :
221 : /*
222 : * Validate connection info string (just try to parse it)
223 : */
224 : static void
225 74 : libpqrcv_check_conninfo(const char *conninfo)
226 : {
227 74 : PQconninfoOption *opts = NULL;
228 74 : char *err = NULL;
229 :
230 74 : opts = PQconninfoParse(conninfo, &err);
231 74 : if (opts == NULL)
232 8 : ereport(ERROR,
233 : (errcode(ERRCODE_SYNTAX_ERROR),
234 : errmsg("invalid connection string syntax: %s", err)));
235 :
236 66 : PQconninfoFree(opts);
237 66 : }
238 :
239 : /*
240 : * Return a user-displayable conninfo string. Any security-sensitive fields
241 : * are obfuscated.
242 : */
243 : static char *
244 122 : libpqrcv_get_conninfo(WalReceiverConn *conn)
245 : {
246 : PQconninfoOption *conn_opts;
247 : PQconninfoOption *conn_opt;
248 : PQExpBufferData buf;
249 : char *retval;
250 :
251 : Assert(conn->streamConn != NULL);
252 :
253 122 : initPQExpBuffer(&buf);
254 122 : conn_opts = PQconninfo(conn->streamConn);
255 :
256 122 : if (conn_opts == NULL)
257 0 : ereport(ERROR,
258 : (errmsg("could not parse connection string: %s",
259 : _("out of memory"))));
260 :
261 : /* build a clean connection string from pieces */
262 4514 : for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
263 : {
264 : bool obfuscate;
265 :
266 : /* Skip debug and empty options */
267 4392 : if (strchr(conn_opt->dispchar, 'D') ||
268 4026 : conn_opt->val == NULL ||
269 1586 : conn_opt->val[0] == '\0')
270 2928 : continue;
271 :
272 : /* Obfuscate security-sensitive options */
273 1464 : obfuscate = strchr(conn_opt->dispchar, '*') != NULL;
274 :
275 2928 : appendPQExpBuffer(&buf, "%s%s=%s",
276 1464 : buf.len == 0 ? "" : " ",
277 : conn_opt->keyword,
278 : obfuscate ? "********" : conn_opt->val);
279 : }
280 :
281 122 : PQconninfoFree(conn_opts);
282 :
283 122 : retval = PQExpBufferDataBroken(buf) ? NULL : pstrdup(buf.data);
284 122 : termPQExpBuffer(&buf);
285 122 : return retval;
286 : }
287 :
288 : /*
289 : * Provides information of sender this WAL receiver is connected to.
290 : */
291 : static void
292 122 : libpqrcv_get_senderinfo(WalReceiverConn *conn, char **sender_host,
293 : int *sender_port)
294 : {
295 122 : char *ret = NULL;
296 :
297 122 : *sender_host = NULL;
298 122 : *sender_port = 0;
299 :
300 : Assert(conn->streamConn != NULL);
301 :
302 122 : ret = PQhost(conn->streamConn);
303 122 : if (ret && strlen(ret) != 0)
304 122 : *sender_host = pstrdup(ret);
305 :
306 122 : ret = PQport(conn->streamConn);
307 122 : if (ret && strlen(ret) != 0)
308 122 : *sender_port = atoi(ret);
309 122 : }
310 :
311 : /*
312 : * Check that primary's system identifier matches ours, and fetch the current
313 : * timeline ID of the primary.
314 : */
315 : static char *
316 184 : libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli)
317 : {
318 : PGresult *res;
319 : char *primary_sysid;
320 :
321 : /*
322 : * Get the system identifier and timeline ID as a DataRow message from the
323 : * primary server.
324 : */
325 184 : res = libpqrcv_PQexec(conn->streamConn, "IDENTIFY_SYSTEM");
326 184 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
327 : {
328 0 : PQclear(res);
329 0 : ereport(ERROR,
330 : (errmsg("could not receive database system identifier and timeline ID from "
331 : "the primary server: %s",
332 : pchomp(PQerrorMessage(conn->streamConn)))));
333 : }
334 184 : if (PQnfields(res) < 3 || PQntuples(res) != 1)
335 : {
336 0 : int ntuples = PQntuples(res);
337 0 : int nfields = PQnfields(res);
338 :
339 0 : PQclear(res);
340 0 : ereport(ERROR,
341 : (errmsg("invalid response from primary server"),
342 : errdetail("Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
343 : ntuples, nfields, 3, 1)));
344 : }
345 184 : primary_sysid = pstrdup(PQgetvalue(res, 0, 0));
346 184 : *primary_tli = pg_strtoint32(PQgetvalue(res, 0, 1));
347 184 : PQclear(res);
348 :
349 184 : return primary_sysid;
350 : }
351 :
352 : /*
353 : * Thin wrapper around libpq to obtain server version.
354 : */
355 : static int
356 102 : libpqrcv_server_version(WalReceiverConn *conn)
357 : {
358 102 : return PQserverVersion(conn->streamConn);
359 : }
360 :
361 : /*
362 : * Start streaming WAL data from given streaming options.
363 : *
364 : * Returns true if we switched successfully to copy-both mode. False
365 : * means the server received the command and executed it successfully, but
366 : * didn't switch to copy-mode. That means that there was no WAL on the
367 : * requested timeline and starting point, because the server switched to
368 : * another timeline at or before the requested starting point. On failure,
369 : * throws an ERROR.
370 : */
371 : static bool
372 184 : libpqrcv_startstreaming(WalReceiverConn *conn,
373 : const WalRcvStreamOptions *options)
374 : {
375 : StringInfoData cmd;
376 : PGresult *res;
377 :
378 : Assert(options->logical == conn->logical);
379 : Assert(options->slotname || !options->logical);
380 :
381 184 : initStringInfo(&cmd);
382 :
383 : /* Build the command. */
384 184 : appendStringInfoString(&cmd, "START_REPLICATION");
385 184 : if (options->slotname != NULL)
386 72 : appendStringInfo(&cmd, " SLOT \"%s\"",
387 : options->slotname);
388 :
389 184 : if (options->logical)
390 50 : appendStringInfoString(&cmd, " LOGICAL");
391 :
392 368 : appendStringInfo(&cmd, " %X/%X",
393 184 : (uint32) (options->startpoint >> 32),
394 184 : (uint32) options->startpoint);
395 :
396 : /*
397 : * Additional options are different depending on if we are doing logical
398 : * or physical replication.
399 : */
400 184 : if (options->logical)
401 : {
402 : char *pubnames_str;
403 : List *pubnames;
404 : char *pubnames_literal;
405 :
406 50 : appendStringInfoString(&cmd, " (");
407 :
408 50 : appendStringInfo(&cmd, "proto_version '%u'",
409 : options->proto.logical.proto_version);
410 :
411 50 : pubnames = options->proto.logical.publication_names;
412 50 : pubnames_str = stringlist_to_identifierstr(conn->streamConn, pubnames);
413 50 : if (!pubnames_str)
414 0 : ereport(ERROR,
415 : (errmsg("could not start WAL streaming: %s",
416 : pchomp(PQerrorMessage(conn->streamConn)))));
417 50 : pubnames_literal = PQescapeLiteral(conn->streamConn, pubnames_str,
418 : strlen(pubnames_str));
419 50 : if (!pubnames_literal)
420 0 : ereport(ERROR,
421 : (errmsg("could not start WAL streaming: %s",
422 : pchomp(PQerrorMessage(conn->streamConn)))));
423 50 : appendStringInfo(&cmd, ", publication_names %s", pubnames_literal);
424 50 : PQfreemem(pubnames_literal);
425 50 : pfree(pubnames_str);
426 :
427 50 : appendStringInfoChar(&cmd, ')');
428 : }
429 : else
430 134 : appendStringInfo(&cmd, " TIMELINE %u",
431 : options->proto.physical.startpointTLI);
432 :
433 : /* Start streaming. */
434 184 : res = libpqrcv_PQexec(conn->streamConn, cmd.data);
435 184 : pfree(cmd.data);
436 :
437 184 : if (PQresultStatus(res) == PGRES_COMMAND_OK)
438 : {
439 0 : PQclear(res);
440 0 : return false;
441 : }
442 184 : else if (PQresultStatus(res) != PGRES_COPY_BOTH)
443 : {
444 0 : PQclear(res);
445 0 : ereport(ERROR,
446 : (errmsg("could not start WAL streaming: %s",
447 : pchomp(PQerrorMessage(conn->streamConn)))));
448 : }
449 184 : PQclear(res);
450 184 : return true;
451 : }
452 :
453 : /*
454 : * Stop streaming WAL data. Returns the next timeline's ID in *next_tli, as
455 : * reported by the server, or 0 if it did not report it.
456 : */
457 : static void
458 46 : libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
459 : {
460 : PGresult *res;
461 :
462 : /*
463 : * Send copy-end message. As in libpqrcv_PQexec, this could theoretically
464 : * block, but the risk seems small.
465 : */
466 58 : if (PQputCopyEnd(conn->streamConn, NULL) <= 0 ||
467 12 : PQflush(conn->streamConn))
468 34 : ereport(ERROR,
469 : (errmsg("could not send end-of-streaming message to primary: %s",
470 : pchomp(PQerrorMessage(conn->streamConn)))));
471 :
472 12 : *next_tli = 0;
473 :
474 : /*
475 : * After COPY is finished, we should receive a result set indicating the
476 : * next timeline's ID, or just CommandComplete if the server was shut
477 : * down.
478 : *
479 : * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
480 : * also possible in case we aborted the copy in mid-stream.
481 : */
482 12 : res = libpqrcv_PQgetResult(conn->streamConn);
483 12 : if (PQresultStatus(res) == PGRES_TUPLES_OK)
484 : {
485 : /*
486 : * Read the next timeline's ID. The server also sends the timeline's
487 : * starting point, but it is ignored.
488 : */
489 12 : if (PQnfields(res) < 2 || PQntuples(res) != 1)
490 0 : ereport(ERROR,
491 : (errmsg("unexpected result set after end-of-streaming")));
492 12 : *next_tli = pg_strtoint32(PQgetvalue(res, 0, 0));
493 12 : PQclear(res);
494 :
495 : /* the result set should be followed by CommandComplete */
496 12 : res = libpqrcv_PQgetResult(conn->streamConn);
497 : }
498 0 : else if (PQresultStatus(res) == PGRES_COPY_OUT)
499 : {
500 0 : PQclear(res);
501 :
502 : /* End the copy */
503 0 : if (PQendcopy(conn->streamConn))
504 0 : ereport(ERROR,
505 : (errmsg("error while shutting down streaming COPY: %s",
506 : pchomp(PQerrorMessage(conn->streamConn)))));
507 :
508 : /* CommandComplete should follow */
509 0 : res = libpqrcv_PQgetResult(conn->streamConn);
510 : }
511 :
512 12 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
513 0 : ereport(ERROR,
514 : (errmsg("error reading result of streaming command: %s",
515 : pchomp(PQerrorMessage(conn->streamConn)))));
516 12 : PQclear(res);
517 :
518 : /* Verify that there are no more results */
519 12 : res = libpqrcv_PQgetResult(conn->streamConn);
520 12 : if (res != NULL)
521 0 : ereport(ERROR,
522 : (errmsg("unexpected result after CommandComplete: %s",
523 : pchomp(PQerrorMessage(conn->streamConn)))));
524 12 : }
525 :
526 : /*
527 : * Fetch the timeline history file for 'tli' from primary.
528 : */
529 : static void
530 12 : libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
531 : TimeLineID tli, char **filename,
532 : char **content, int *len)
533 : {
534 : PGresult *res;
535 : char cmd[64];
536 :
537 : Assert(!conn->logical);
538 :
539 : /*
540 : * Request the primary to send over the history file for given timeline.
541 : */
542 12 : snprintf(cmd, sizeof(cmd), "TIMELINE_HISTORY %u", tli);
543 12 : res = libpqrcv_PQexec(conn->streamConn, cmd);
544 12 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
545 : {
546 0 : PQclear(res);
547 0 : ereport(ERROR,
548 : (errmsg("could not receive timeline history file from "
549 : "the primary server: %s",
550 : pchomp(PQerrorMessage(conn->streamConn)))));
551 : }
552 12 : if (PQnfields(res) != 2 || PQntuples(res) != 1)
553 : {
554 0 : int ntuples = PQntuples(res);
555 0 : int nfields = PQnfields(res);
556 :
557 0 : PQclear(res);
558 0 : ereport(ERROR,
559 : (errmsg("invalid response from primary server"),
560 : errdetail("Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
561 : ntuples, nfields)));
562 : }
563 12 : *filename = pstrdup(PQgetvalue(res, 0, 0));
564 :
565 12 : *len = PQgetlength(res, 0, 1);
566 12 : *content = palloc(*len);
567 12 : memcpy(*content, PQgetvalue(res, 0, 1), *len);
568 12 : PQclear(res);
569 12 : }
570 :
571 : /*
572 : * Send a query and wait for the results by using the asynchronous libpq
573 : * functions and socket readiness events.
574 : *
575 : * We must not use the regular blocking libpq functions like PQexec()
576 : * since they are uninterruptible by signals on some platforms, such as
577 : * Windows.
578 : *
579 : * The function is modeled on PQexec() in libpq, but only implements
580 : * those parts that are in use in the walreceiver api.
581 : *
582 : * May return NULL, rather than an error result, on failure.
583 : */
584 : static PGresult *
585 1092 : libpqrcv_PQexec(PGconn *streamConn, const char *query)
586 : {
587 1092 : PGresult *lastResult = NULL;
588 :
589 : /*
590 : * PQexec() silently discards any prior query results on the connection.
591 : * This is not required for this function as it's expected that the caller
592 : * (which is this library in all cases) will behave correctly and we don't
593 : * have to be backwards compatible with old libpq.
594 : */
595 :
596 : /*
597 : * Submit the query. Since we don't use non-blocking mode, this could
598 : * theoretically block. In practice, since we don't send very long query
599 : * strings, the risk seems negligible.
600 : */
601 1092 : if (!PQsendQuery(streamConn, query))
602 0 : return NULL;
603 :
604 : for (;;)
605 820 : {
606 : /* Wait for, and collect, the next PGresult. */
607 : PGresult *result;
608 :
609 1912 : result = libpqrcv_PQgetResult(streamConn);
610 1912 : if (result == NULL)
611 806 : break; /* query is complete, or failure */
612 :
613 : /*
614 : * Emulate PQexec()'s behavior of returning the last result when there
615 : * are many. We are fine with returning just last error message.
616 : */
617 1106 : PQclear(lastResult);
618 1106 : lastResult = result;
619 :
620 2212 : if (PQresultStatus(lastResult) == PGRES_COPY_IN ||
621 2110 : PQresultStatus(lastResult) == PGRES_COPY_OUT ||
622 1824 : PQresultStatus(lastResult) == PGRES_COPY_BOTH ||
623 820 : PQstatus(streamConn) == CONNECTION_BAD)
624 : break;
625 : }
626 :
627 1092 : return lastResult;
628 : }
629 :
630 : /*
631 : * Perform the equivalent of PQgetResult(), but watch for interrupts.
632 : */
633 : static PGresult *
634 2238 : libpqrcv_PQgetResult(PGconn *streamConn)
635 : {
636 : /*
637 : * Collect data until PQgetResult is ready to get the result without
638 : * blocking.
639 : */
640 3342 : while (PQisBusy(streamConn))
641 : {
642 : int rc;
643 :
644 : /*
645 : * We don't need to break down the sleep into smaller increments,
646 : * since we'll get interrupted by signals and can handle any
647 : * interrupts here.
648 : */
649 1138 : rc = WaitLatchOrSocket(MyLatch,
650 : WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
651 : WL_LATCH_SET,
652 : PQsocket(streamConn),
653 : 0,
654 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
655 :
656 : /* Interrupted? */
657 1138 : if (rc & WL_LATCH_SET)
658 : {
659 0 : ResetLatch(MyLatch);
660 0 : ProcessWalRcvInterrupts();
661 : }
662 :
663 : /* Consume whatever data is available from the socket */
664 1138 : if (PQconsumeInput(streamConn) == 0)
665 : {
666 : /* trouble; return NULL */
667 34 : return NULL;
668 : }
669 : }
670 :
671 : /* Now we can collect and return the next PGresult */
672 2204 : return PQgetResult(streamConn);
673 : }
674 :
675 : /*
676 : * Disconnect connection to primary, if any.
677 : */
678 : static void
679 338 : libpqrcv_disconnect(WalReceiverConn *conn)
680 : {
681 338 : PQfinish(conn->streamConn);
682 338 : if (conn->recvBuf != NULL)
683 2 : PQfreemem(conn->recvBuf);
684 338 : pfree(conn);
685 338 : }
686 :
687 : /*
688 : * Receive a message available from XLOG stream.
689 : *
690 : * Returns:
691 : *
692 : * If data was received, returns the length of the data. *buffer is set to
693 : * point to a buffer holding the received message. The buffer is only valid
694 : * until the next libpqrcv_* call.
695 : *
696 : * If no data was available immediately, returns 0, and *wait_fd is set to a
697 : * socket descriptor which can be waited on before trying again.
698 : *
699 : * -1 if the server ended the COPY.
700 : *
701 : * ereports on error.
702 : */
703 : static int
704 20828 : libpqrcv_receive(WalReceiverConn *conn, char **buffer,
705 : pgsocket *wait_fd)
706 : {
707 : int rawlen;
708 :
709 20828 : if (conn->recvBuf != NULL)
710 11910 : PQfreemem(conn->recvBuf);
711 20828 : conn->recvBuf = NULL;
712 :
713 : /* Try to receive a CopyData message */
714 20828 : rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
715 20828 : if (rawlen == 0)
716 : {
717 : /* Try consuming some data. */
718 16304 : if (PQconsumeInput(conn->streamConn) == 0)
719 44 : ereport(ERROR,
720 : (errmsg("could not receive data from WAL stream: %s",
721 : pchomp(PQerrorMessage(conn->streamConn)))));
722 :
723 : /* Now that we've consumed some input, try again */
724 16260 : rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
725 16260 : if (rawlen == 0)
726 : {
727 : /* Tell caller to try again when our socket is ready. */
728 8718 : *wait_fd = PQsocket(conn->streamConn);
729 8718 : return 0;
730 : }
731 : }
732 12066 : if (rawlen == -1) /* end-of-streaming or error */
733 : {
734 : PGresult *res;
735 :
736 154 : res = libpqrcv_PQgetResult(conn->streamConn);
737 154 : if (PQresultStatus(res) == PGRES_COMMAND_OK)
738 : {
739 136 : PQclear(res);
740 :
741 : /* Verify that there are no more results. */
742 136 : res = libpqrcv_PQgetResult(conn->streamConn);
743 136 : if (res != NULL)
744 : {
745 0 : PQclear(res);
746 :
747 : /*
748 : * If the other side closed the connection orderly (otherwise
749 : * we'd seen an error, or PGRES_COPY_IN) don't report an error
750 : * here, but let callers deal with it.
751 : */
752 0 : if (PQstatus(conn->streamConn) == CONNECTION_BAD)
753 0 : return -1;
754 :
755 0 : ereport(ERROR,
756 : (errmsg("unexpected result after CommandComplete: %s",
757 : PQerrorMessage(conn->streamConn))));
758 : }
759 :
760 136 : return -1;
761 : }
762 18 : else if (PQresultStatus(res) == PGRES_COPY_IN)
763 : {
764 12 : PQclear(res);
765 12 : return -1;
766 : }
767 : else
768 : {
769 6 : PQclear(res);
770 6 : ereport(ERROR,
771 : (errmsg("could not receive data from WAL stream: %s",
772 : pchomp(PQerrorMessage(conn->streamConn)))));
773 : }
774 : }
775 11912 : if (rawlen < -1)
776 0 : ereport(ERROR,
777 : (errmsg("could not receive data from WAL stream: %s",
778 : pchomp(PQerrorMessage(conn->streamConn)))));
779 :
780 : /* Return received messages to caller */
781 11912 : *buffer = conn->recvBuf;
782 11912 : return rawlen;
783 : }
784 :
785 : /*
786 : * Send a message to XLOG stream.
787 : *
788 : * ereports on error.
789 : */
790 : static void
791 7934 : libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes)
792 : {
793 15868 : if (PQputCopyData(conn->streamConn, buffer, nbytes) <= 0 ||
794 7934 : PQflush(conn->streamConn))
795 0 : ereport(ERROR,
796 : (errmsg("could not send data to WAL stream: %s",
797 : pchomp(PQerrorMessage(conn->streamConn)))));
798 7934 : }
799 :
800 : /*
801 : * Create new replication slot.
802 : * Returns the name of the exported snapshot for logical slot or NULL for
803 : * physical slot.
804 : */
805 : static char *
806 142 : libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
807 : bool temporary, CRSSnapshotAction snapshot_action,
808 : XLogRecPtr *lsn)
809 : {
810 : PGresult *res;
811 : StringInfoData cmd;
812 : char *snapshot;
813 :
814 142 : initStringInfo(&cmd);
815 :
816 142 : appendStringInfo(&cmd, "CREATE_REPLICATION_SLOT \"%s\"", slotname);
817 :
818 142 : if (temporary)
819 102 : appendStringInfoString(&cmd, " TEMPORARY");
820 :
821 142 : if (conn->logical)
822 : {
823 142 : appendStringInfoString(&cmd, " LOGICAL pgoutput");
824 142 : switch (snapshot_action)
825 : {
826 0 : case CRS_EXPORT_SNAPSHOT:
827 0 : appendStringInfoString(&cmd, " EXPORT_SNAPSHOT");
828 0 : break;
829 40 : case CRS_NOEXPORT_SNAPSHOT:
830 40 : appendStringInfoString(&cmd, " NOEXPORT_SNAPSHOT");
831 40 : break;
832 102 : case CRS_USE_SNAPSHOT:
833 102 : appendStringInfoString(&cmd, " USE_SNAPSHOT");
834 102 : break;
835 : }
836 142 : }
837 : else
838 : {
839 0 : appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
840 : }
841 :
842 142 : res = libpqrcv_PQexec(conn->streamConn, cmd.data);
843 142 : pfree(cmd.data);
844 :
845 142 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
846 : {
847 0 : PQclear(res);
848 0 : ereport(ERROR,
849 : (errmsg("could not create replication slot \"%s\": %s",
850 : slotname, pchomp(PQerrorMessage(conn->streamConn)))));
851 : }
852 :
853 142 : if (lsn)
854 102 : *lsn = DatumGetLSN(DirectFunctionCall1Coll(pg_lsn_in, InvalidOid,
855 : CStringGetDatum(PQgetvalue(res, 0, 1))));
856 :
857 142 : if (!PQgetisnull(res, 0, 2))
858 0 : snapshot = pstrdup(PQgetvalue(res, 0, 2));
859 : else
860 142 : snapshot = NULL;
861 :
862 142 : PQclear(res);
863 :
864 142 : return snapshot;
865 : }
866 :
867 : /*
868 : * Return PID of remote backend process.
869 : */
870 : static pid_t
871 0 : libpqrcv_get_backend_pid(WalReceiverConn *conn)
872 : {
873 0 : return PQbackendPID(conn->streamConn);
874 : }
875 :
876 : /*
877 : * Convert tuple query result to tuplestore.
878 : */
879 : static void
880 254 : libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
881 : const int nRetTypes, const Oid *retTypes)
882 : {
883 : int tupn;
884 : int coln;
885 254 : int nfields = PQnfields(pgres);
886 : HeapTuple tuple;
887 : AttInMetadata *attinmeta;
888 : MemoryContext rowcontext;
889 : MemoryContext oldcontext;
890 :
891 : /* Make sure we got expected number of fields. */
892 254 : if (nfields != nRetTypes)
893 0 : ereport(ERROR,
894 : (errmsg("invalid query response"),
895 : errdetail("Expected %d fields, got %d fields.",
896 : nRetTypes, nfields)));
897 :
898 254 : walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
899 :
900 : /* Create tuple descriptor corresponding to expected result. */
901 254 : walres->tupledesc = CreateTemplateTupleDesc(nRetTypes);
902 1068 : for (coln = 0; coln < nRetTypes; coln++)
903 814 : TupleDescInitEntry(walres->tupledesc, (AttrNumber) coln + 1,
904 814 : PQfname(pgres, coln), retTypes[coln], -1, 0);
905 254 : attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
906 :
907 : /* No point in doing more here if there were no tuples returned. */
908 254 : if (PQntuples(pgres) == 0)
909 2 : return;
910 :
911 : /* Create temporary context for local allocations. */
912 252 : rowcontext = AllocSetContextCreate(CurrentMemoryContext,
913 : "libpqrcv query result context",
914 : ALLOCSET_DEFAULT_SIZES);
915 :
916 : /* Process returned rows. */
917 646 : for (tupn = 0; tupn < PQntuples(pgres); tupn++)
918 : {
919 : char *cstrs[MaxTupleAttributeNumber];
920 :
921 394 : ProcessWalRcvInterrupts();
922 :
923 : /* Do the allocations in temporary context. */
924 394 : oldcontext = MemoryContextSwitchTo(rowcontext);
925 :
926 : /*
927 : * Fill cstrs with null-terminated strings of column values.
928 : */
929 1640 : for (coln = 0; coln < nfields; coln++)
930 : {
931 1246 : if (PQgetisnull(pgres, tupn, coln))
932 30 : cstrs[coln] = NULL;
933 : else
934 1216 : cstrs[coln] = PQgetvalue(pgres, tupn, coln);
935 : }
936 :
937 : /* Convert row to a tuple, and add it to the tuplestore */
938 394 : tuple = BuildTupleFromCStrings(attinmeta, cstrs);
939 394 : tuplestore_puttuple(walres->tuplestore, tuple);
940 :
941 : /* Clean up */
942 394 : MemoryContextSwitchTo(oldcontext);
943 394 : MemoryContextReset(rowcontext);
944 : }
945 :
946 252 : MemoryContextDelete(rowcontext);
947 : }
948 :
949 : /*
950 : * Public interface for sending generic queries (and commands).
951 : *
952 : * This can only be called from process connected to database.
953 : */
954 : static WalRcvExecResult *
955 570 : libpqrcv_exec(WalReceiverConn *conn, const char *query,
956 : const int nRetTypes, const Oid *retTypes)
957 : {
958 570 : PGresult *pgres = NULL;
959 570 : WalRcvExecResult *walres = palloc0(sizeof(WalRcvExecResult));
960 :
961 570 : if (MyDatabaseId == InvalidOid)
962 0 : ereport(ERROR,
963 : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
964 : errmsg("the query interface requires a database connection")));
965 :
966 570 : pgres = libpqrcv_PQexec(conn->streamConn, query);
967 :
968 570 : switch (PQresultStatus(pgres))
969 : {
970 254 : case PGRES_SINGLE_TUPLE:
971 : case PGRES_TUPLES_OK:
972 254 : walres->status = WALRCV_OK_TUPLES;
973 254 : libpqrcv_processTuples(pgres, walres, nRetTypes, retTypes);
974 254 : break;
975 :
976 0 : case PGRES_COPY_IN:
977 0 : walres->status = WALRCV_OK_COPY_IN;
978 0 : break;
979 :
980 102 : case PGRES_COPY_OUT:
981 102 : walres->status = WALRCV_OK_COPY_OUT;
982 102 : break;
983 :
984 0 : case PGRES_COPY_BOTH:
985 0 : walres->status = WALRCV_OK_COPY_BOTH;
986 0 : break;
987 :
988 214 : case PGRES_COMMAND_OK:
989 214 : walres->status = WALRCV_OK_COMMAND;
990 214 : break;
991 :
992 : /* Empty query is considered error. */
993 0 : case PGRES_EMPTY_QUERY:
994 0 : walres->status = WALRCV_ERROR;
995 0 : walres->err = _("empty query");
996 0 : break;
997 :
998 0 : case PGRES_NONFATAL_ERROR:
999 : case PGRES_FATAL_ERROR:
1000 : case PGRES_BAD_RESPONSE:
1001 0 : walres->status = WALRCV_ERROR;
1002 0 : walres->err = pchomp(PQerrorMessage(conn->streamConn));
1003 0 : break;
1004 : }
1005 :
1006 570 : PQclear(pgres);
1007 :
1008 570 : return walres;
1009 : }
1010 :
1011 : /*
1012 : * Given a List of strings, return it as single comma separated
1013 : * string, quoting identifiers as needed.
1014 : *
1015 : * This is essentially the reverse of SplitIdentifierString.
1016 : *
1017 : * The caller should free the result.
1018 : */
1019 : static char *
1020 50 : stringlist_to_identifierstr(PGconn *conn, List *strings)
1021 : {
1022 : ListCell *lc;
1023 : StringInfoData res;
1024 50 : bool first = true;
1025 :
1026 50 : initStringInfo(&res);
1027 :
1028 104 : foreach(lc, strings)
1029 : {
1030 54 : char *val = strVal(lfirst(lc));
1031 : char *val_escaped;
1032 :
1033 54 : if (first)
1034 50 : first = false;
1035 : else
1036 4 : appendStringInfoChar(&res, ',');
1037 :
1038 54 : val_escaped = PQescapeIdentifier(conn, val, strlen(val));
1039 54 : if (!val_escaped)
1040 : {
1041 0 : free(res.data);
1042 0 : return NULL;
1043 : }
1044 54 : appendStringInfoString(&res, val_escaped);
1045 54 : PQfreemem(val_escaped);
1046 : }
1047 :
1048 50 : return res.data;
1049 : }
|
