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