PostgreSQL Source Code  git master
libpqwalreceiver.c
Go to the documentation of this file.
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-2021, 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 "common/connect.h"
25 #include "funcapi.h"
26 #include "libpq-fe.h"
27 #include "mb/pg_wchar.h"
28 #include "miscadmin.h"
29 #include "pgstat.h"
30 #include "pqexpbuffer.h"
32 #include "utils/builtins.h"
33 #include "utils/memutils.h"
34 #include "utils/pg_lsn.h"
35 #include "utils/tuplestore.h"
36 
38 
39 void _PG_init(void);
40 
42 {
43  /* Current connection to the primary, if any */
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);
58  char **sender_host, int *sender_port);
60  TimeLineID *primary_tli);
63  TimeLineID tli, char **filename,
64  char **content, int *len);
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);
74  const char *slotname,
75  bool temporary,
76  bool two_phase,
77  CRSSnapshotAction snapshot_action,
78  XLogRecPtr *lsn);
81  const char *query,
82  const int nRetTypes,
83  const Oid *retTypes);
85 
102 };
103 
104 /* Prototypes for private functions */
105 static PGresult *libpqrcv_PQexec(PGconn *streamConn, const char *query);
106 static PGresult *libpqrcv_PQgetResult(PGconn *streamConn);
107 static char *stringlist_to_identifierstr(PGconn *conn, List *strings);
108 
109 /*
110  * Module initialization function
111  */
112 void
113 _PG_init(void)
114 {
115  if (WalReceiverFunctions != NULL)
116  elog(ERROR, "libpqwalreceiver already loaded");
118 }
119 
120 /*
121  * Establish the connection to the primary server for XLOG streaming
122  *
123  * Returns NULL on error and fills the err with palloc'ed error message.
124  */
125 static WalReceiverConn *
126 libpqrcv_connect(const char *conninfo, bool logical, const char *appname,
127  char **err)
128 {
131  const char *keys[5];
132  const char *vals[5];
133  int i = 0;
134 
135  /*
136  * We use the expand_dbname parameter to process the connection string (or
137  * URI), and pass some extra options.
138  */
139  keys[i] = "dbname";
140  vals[i] = conninfo;
141  keys[++i] = "replication";
142  vals[i] = logical ? "database" : "true";
143  if (!logical)
144  {
145  /*
146  * The database name is ignored by the server in replication mode, but
147  * specify "replication" for .pgpass lookup.
148  */
149  keys[++i] = "dbname";
150  vals[i] = "replication";
151  }
152  keys[++i] = "fallback_application_name";
153  vals[i] = appname;
154  if (logical)
155  {
156  keys[++i] = "client_encoding";
157  vals[i] = GetDatabaseEncodingName();
158  }
159  keys[++i] = NULL;
160  vals[i] = NULL;
161 
162  Assert(i < sizeof(keys));
163 
164  conn = palloc0(sizeof(WalReceiverConn));
165  conn->streamConn = PQconnectStartParams(keys, vals,
166  /* expand_dbname = */ true);
167  if (PQstatus(conn->streamConn) == CONNECTION_BAD)
168  {
169  *err = pchomp(PQerrorMessage(conn->streamConn));
170  return NULL;
171  }
172 
173  /*
174  * Poll connection until we have OK or FAILED status.
175  *
176  * Per spec for PQconnectPoll, first wait till socket is write-ready.
177  */
178  status = PGRES_POLLING_WRITING;
179  do
180  {
181  int io_flag;
182  int rc;
183 
184  if (status == PGRES_POLLING_READING)
185  io_flag = WL_SOCKET_READABLE;
186 #ifdef WIN32
187  /* Windows needs a different test while waiting for connection-made */
188  else if (PQstatus(conn->streamConn) == CONNECTION_STARTED)
189  io_flag = WL_SOCKET_CONNECTED;
190 #endif
191  else
192  io_flag = WL_SOCKET_WRITEABLE;
193 
195  WL_EXIT_ON_PM_DEATH | WL_LATCH_SET | io_flag,
196  PQsocket(conn->streamConn),
197  0,
199 
200  /* Interrupted? */
201  if (rc & WL_LATCH_SET)
202  {
205  }
206 
207  /* If socket is ready, advance the libpq state machine */
208  if (rc & io_flag)
209  status = PQconnectPoll(conn->streamConn);
210  } while (status != PGRES_POLLING_OK && status != PGRES_POLLING_FAILED);
211 
212  if (PQstatus(conn->streamConn) != CONNECTION_OK)
213  {
214  *err = pchomp(PQerrorMessage(conn->streamConn));
215  return NULL;
216  }
217 
218  if (logical)
219  {
220  PGresult *res;
221 
222  res = libpqrcv_PQexec(conn->streamConn,
224  if (PQresultStatus(res) != PGRES_TUPLES_OK)
225  {
226  PQclear(res);
227  ereport(ERROR,
228  (errmsg("could not clear search path: %s",
229  pchomp(PQerrorMessage(conn->streamConn)))));
230  }
231  PQclear(res);
232  }
233 
234  conn->logical = logical;
235 
236  return conn;
237 }
238 
239 /*
240  * Validate connection info string (just try to parse it)
241  */
242 static void
243 libpqrcv_check_conninfo(const char *conninfo)
244 {
245  PQconninfoOption *opts = NULL;
246  char *err = NULL;
247 
248  opts = PQconninfoParse(conninfo, &err);
249  if (opts == NULL)
250  {
251  /* The error string is malloc'd, so we must free it explicitly */
252  char *errcopy = err ? pstrdup(err) : "out of memory";
253 
254  PQfreemem(err);
255  ereport(ERROR,
256  (errcode(ERRCODE_SYNTAX_ERROR),
257  errmsg("invalid connection string syntax: %s", errcopy)));
258  }
259 
260  PQconninfoFree(opts);
261 }
262 
263 /*
264  * Return a user-displayable conninfo string. Any security-sensitive fields
265  * are obfuscated.
266  */
267 static char *
269 {
270  PQconninfoOption *conn_opts;
271  PQconninfoOption *conn_opt;
273  char *retval;
274 
275  Assert(conn->streamConn != NULL);
276 
277  initPQExpBuffer(&buf);
278  conn_opts = PQconninfo(conn->streamConn);
279 
280  if (conn_opts == NULL)
281  ereport(ERROR,
282  (errcode(ERRCODE_OUT_OF_MEMORY),
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
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 *
342 {
343  PGresult *res;
344  char *primary_sysid;
345 
346  /*
347  * Get the system identifier and timeline ID as a DataRow message from the
348  * primary server.
349  */
350  res = libpqrcv_PQexec(conn->streamConn, "IDENTIFY_SYSTEM");
351  if (PQresultStatus(res) != PGRES_TUPLES_OK)
352  {
353  PQclear(res);
354  ereport(ERROR,
355  (errcode(ERRCODE_PROTOCOL_VIOLATION),
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  (errcode(ERRCODE_PROTOCOL_VIOLATION),
368  errmsg("invalid response from primary server"),
369  errdetail("Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
370  ntuples, nfields, 3, 1)));
371  }
372  primary_sysid = pstrdup(PQgetvalue(res, 0, 0));
373  *primary_tli = pg_strtoint32(PQgetvalue(res, 0, 1));
374  PQclear(res);
375 
376  return primary_sysid;
377 }
378 
379 /*
380  * Thin wrapper around libpq to obtain server version.
381  */
382 static int
384 {
385  return PQserverVersion(conn->streamConn);
386 }
387 
388 /*
389  * Start streaming WAL data from given streaming options.
390  *
391  * Returns true if we switched successfully to copy-both mode. False
392  * means the server received the command and executed it successfully, but
393  * didn't switch to copy-mode. That means that there was no WAL on the
394  * requested timeline and starting point, because the server switched to
395  * another timeline at or before the requested starting point. On failure,
396  * throws an ERROR.
397  */
398 static bool
401 {
402  StringInfoData cmd;
403  PGresult *res;
404 
405  Assert(options->logical == conn->logical);
406  Assert(options->slotname || !options->logical);
407 
408  initStringInfo(&cmd);
409 
410  /* Build the command. */
411  appendStringInfoString(&cmd, "START_REPLICATION");
412  if (options->slotname != NULL)
413  appendStringInfo(&cmd, " SLOT \"%s\"",
414  options->slotname);
415 
416  if (options->logical)
417  appendStringInfoString(&cmd, " LOGICAL");
418 
419  appendStringInfo(&cmd, " %X/%X", LSN_FORMAT_ARGS(options->startpoint));
420 
421  /*
422  * Additional options are different depending on if we are doing logical
423  * or physical replication.
424  */
425  if (options->logical)
426  {
427  char *pubnames_str;
428  List *pubnames;
429  char *pubnames_literal;
430 
431  appendStringInfoString(&cmd, " (");
432 
433  appendStringInfo(&cmd, "proto_version '%u'",
434  options->proto.logical.proto_version);
435 
436  if (options->proto.logical.streaming &&
437  PQserverVersion(conn->streamConn) >= 140000)
438  appendStringInfoString(&cmd, ", streaming 'on'");
439 
440  if (options->proto.logical.twophase &&
441  PQserverVersion(conn->streamConn) >= 150000)
442  appendStringInfoString(&cmd, ", two_phase 'on'");
443 
444  pubnames = options->proto.logical.publication_names;
445  pubnames_str = stringlist_to_identifierstr(conn->streamConn, pubnames);
446  if (!pubnames_str)
447  ereport(ERROR,
448  (errcode(ERRCODE_OUT_OF_MEMORY), /* likely guess */
449  errmsg("could not start WAL streaming: %s",
450  pchomp(PQerrorMessage(conn->streamConn)))));
451  pubnames_literal = PQescapeLiteral(conn->streamConn, pubnames_str,
452  strlen(pubnames_str));
453  if (!pubnames_literal)
454  ereport(ERROR,
455  (errcode(ERRCODE_OUT_OF_MEMORY), /* likely guess */
456  errmsg("could not start WAL streaming: %s",
457  pchomp(PQerrorMessage(conn->streamConn)))));
458  appendStringInfo(&cmd, ", publication_names %s", pubnames_literal);
459  PQfreemem(pubnames_literal);
460  pfree(pubnames_str);
461 
462  if (options->proto.logical.binary &&
463  PQserverVersion(conn->streamConn) >= 140000)
464  appendStringInfoString(&cmd, ", binary 'true'");
465 
466  appendStringInfoChar(&cmd, ')');
467  }
468  else
469  appendStringInfo(&cmd, " TIMELINE %u",
470  options->proto.physical.startpointTLI);
471 
472  /* Start streaming. */
473  res = libpqrcv_PQexec(conn->streamConn, cmd.data);
474  pfree(cmd.data);
475 
476  if (PQresultStatus(res) == PGRES_COMMAND_OK)
477  {
478  PQclear(res);
479  return false;
480  }
481  else if (PQresultStatus(res) != PGRES_COPY_BOTH)
482  {
483  PQclear(res);
484  ereport(ERROR,
485  (errcode(ERRCODE_PROTOCOL_VIOLATION),
486  errmsg("could not start WAL streaming: %s",
487  pchomp(PQerrorMessage(conn->streamConn)))));
488  }
489  PQclear(res);
490  return true;
491 }
492 
493 /*
494  * Stop streaming WAL data. Returns the next timeline's ID in *next_tli, as
495  * reported by the server, or 0 if it did not report it.
496  */
497 static void
499 {
500  PGresult *res;
501 
502  /*
503  * Send copy-end message. As in libpqrcv_PQexec, this could theoretically
504  * block, but the risk seems small.
505  */
506  if (PQputCopyEnd(conn->streamConn, NULL) <= 0 ||
507  PQflush(conn->streamConn))
508  ereport(ERROR,
509  (errcode(ERRCODE_CONNECTION_FAILURE),
510  errmsg("could not send end-of-streaming message to primary: %s",
511  pchomp(PQerrorMessage(conn->streamConn)))));
512 
513  *next_tli = 0;
514 
515  /*
516  * After COPY is finished, we should receive a result set indicating the
517  * next timeline's ID, or just CommandComplete if the server was shut
518  * down.
519  *
520  * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
521  * also possible in case we aborted the copy in mid-stream.
522  */
523  res = libpqrcv_PQgetResult(conn->streamConn);
524  if (PQresultStatus(res) == PGRES_TUPLES_OK)
525  {
526  /*
527  * Read the next timeline's ID. The server also sends the timeline's
528  * starting point, but it is ignored.
529  */
530  if (PQnfields(res) < 2 || PQntuples(res) != 1)
531  ereport(ERROR,
532  (errcode(ERRCODE_PROTOCOL_VIOLATION),
533  errmsg("unexpected result set after end-of-streaming")));
534  *next_tli = pg_strtoint32(PQgetvalue(res, 0, 0));
535  PQclear(res);
536 
537  /* the result set should be followed by CommandComplete */
538  res = libpqrcv_PQgetResult(conn->streamConn);
539  }
540  else if (PQresultStatus(res) == PGRES_COPY_OUT)
541  {
542  PQclear(res);
543 
544  /* End the copy */
545  if (PQendcopy(conn->streamConn))
546  ereport(ERROR,
547  (errcode(ERRCODE_CONNECTION_FAILURE),
548  errmsg("error while shutting down streaming COPY: %s",
549  pchomp(PQerrorMessage(conn->streamConn)))));
550 
551  /* CommandComplete should follow */
552  res = libpqrcv_PQgetResult(conn->streamConn);
553  }
554 
555  if (PQresultStatus(res) != PGRES_COMMAND_OK)
556  ereport(ERROR,
557  (errcode(ERRCODE_PROTOCOL_VIOLATION),
558  errmsg("error reading result of streaming command: %s",
559  pchomp(PQerrorMessage(conn->streamConn)))));
560  PQclear(res);
561 
562  /* Verify that there are no more results */
563  res = libpqrcv_PQgetResult(conn->streamConn);
564  if (res != NULL)
565  ereport(ERROR,
566  (errcode(ERRCODE_PROTOCOL_VIOLATION),
567  errmsg("unexpected result after CommandComplete: %s",
568  pchomp(PQerrorMessage(conn->streamConn)))));
569 }
570 
571 /*
572  * Fetch the timeline history file for 'tli' from primary.
573  */
574 static void
576  TimeLineID tli, char **filename,
577  char **content, int *len)
578 {
579  PGresult *res;
580  char cmd[64];
581 
582  Assert(!conn->logical);
583 
584  /*
585  * Request the primary to send over the history file for given timeline.
586  */
587  snprintf(cmd, sizeof(cmd), "TIMELINE_HISTORY %u", tli);
588  res = libpqrcv_PQexec(conn->streamConn, cmd);
589  if (PQresultStatus(res) != PGRES_TUPLES_OK)
590  {
591  PQclear(res);
592  ereport(ERROR,
593  (errcode(ERRCODE_PROTOCOL_VIOLATION),
594  errmsg("could not receive timeline history file from "
595  "the primary server: %s",
596  pchomp(PQerrorMessage(conn->streamConn)))));
597  }
598  if (PQnfields(res) != 2 || PQntuples(res) != 1)
599  {
600  int ntuples = PQntuples(res);
601  int nfields = PQnfields(res);
602 
603  PQclear(res);
604  ereport(ERROR,
605  (errcode(ERRCODE_PROTOCOL_VIOLATION),
606  errmsg("invalid response from primary server"),
607  errdetail("Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
608  ntuples, nfields)));
609  }
610  *filename = pstrdup(PQgetvalue(res, 0, 0));
611 
612  *len = PQgetlength(res, 0, 1);
613  *content = palloc(*len);
614  memcpy(*content, PQgetvalue(res, 0, 1), *len);
615  PQclear(res);
616 }
617 
618 /*
619  * Send a query and wait for the results by using the asynchronous libpq
620  * functions and socket readiness events.
621  *
622  * We must not use the regular blocking libpq functions like PQexec()
623  * since they are uninterruptible by signals on some platforms, such as
624  * Windows.
625  *
626  * The function is modeled on PQexec() in libpq, but only implements
627  * those parts that are in use in the walreceiver api.
628  *
629  * May return NULL, rather than an error result, on failure.
630  */
631 static PGresult *
632 libpqrcv_PQexec(PGconn *streamConn, const char *query)
633 {
634  PGresult *lastResult = NULL;
635 
636  /*
637  * PQexec() silently discards any prior query results on the connection.
638  * This is not required for this function as it's expected that the caller
639  * (which is this library in all cases) will behave correctly and we don't
640  * have to be backwards compatible with old libpq.
641  */
642 
643  /*
644  * Submit the query. Since we don't use non-blocking mode, this could
645  * theoretically block. In practice, since we don't send very long query
646  * strings, the risk seems negligible.
647  */
648  if (!PQsendQuery(streamConn, query))
649  return NULL;
650 
651  for (;;)
652  {
653  /* Wait for, and collect, the next PGresult. */
654  PGresult *result;
655 
656  result = libpqrcv_PQgetResult(streamConn);
657  if (result == NULL)
658  break; /* query is complete, or failure */
659 
660  /*
661  * Emulate PQexec()'s behavior of returning the last result when there
662  * are many. We are fine with returning just last error message.
663  */
664  PQclear(lastResult);
665  lastResult = result;
666 
667  if (PQresultStatus(lastResult) == PGRES_COPY_IN ||
668  PQresultStatus(lastResult) == PGRES_COPY_OUT ||
669  PQresultStatus(lastResult) == PGRES_COPY_BOTH ||
670  PQstatus(streamConn) == CONNECTION_BAD)
671  break;
672  }
673 
674  return lastResult;
675 }
676 
677 /*
678  * Perform the equivalent of PQgetResult(), but watch for interrupts.
679  */
680 static PGresult *
682 {
683  /*
684  * Collect data until PQgetResult is ready to get the result without
685  * blocking.
686  */
687  while (PQisBusy(streamConn))
688  {
689  int rc;
690 
691  /*
692  * We don't need to break down the sleep into smaller increments,
693  * since we'll get interrupted by signals and can handle any
694  * interrupts here.
695  */
698  WL_LATCH_SET,
699  PQsocket(streamConn),
700  0,
702 
703  /* Interrupted? */
704  if (rc & WL_LATCH_SET)
705  {
708  }
709 
710  /* Consume whatever data is available from the socket */
711  if (PQconsumeInput(streamConn) == 0)
712  {
713  /* trouble; return NULL */
714  return NULL;
715  }
716  }
717 
718  /* Now we can collect and return the next PGresult */
719  return PQgetResult(streamConn);
720 }
721 
722 /*
723  * Disconnect connection to primary, if any.
724  */
725 static void
727 {
728  PQfinish(conn->streamConn);
729  if (conn->recvBuf != NULL)
730  PQfreemem(conn->recvBuf);
731  pfree(conn);
732 }
733 
734 /*
735  * Receive a message available from XLOG stream.
736  *
737  * Returns:
738  *
739  * If data was received, returns the length of the data. *buffer is set to
740  * point to a buffer holding the received message. The buffer is only valid
741  * until the next libpqrcv_* call.
742  *
743  * If no data was available immediately, returns 0, and *wait_fd is set to a
744  * socket descriptor which can be waited on before trying again.
745  *
746  * -1 if the server ended the COPY.
747  *
748  * ereports on error.
749  */
750 static int
751 libpqrcv_receive(WalReceiverConn *conn, char **buffer,
752  pgsocket *wait_fd)
753 {
754  int rawlen;
755 
756  if (conn->recvBuf != NULL)
757  PQfreemem(conn->recvBuf);
758  conn->recvBuf = NULL;
759 
760  /* Try to receive a CopyData message */
761  rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
762  if (rawlen == 0)
763  {
764  /* Try consuming some data. */
765  if (PQconsumeInput(conn->streamConn) == 0)
766  ereport(ERROR,
767  (errcode(ERRCODE_CONNECTION_FAILURE),
768  errmsg("could not receive data from WAL stream: %s",
769  pchomp(PQerrorMessage(conn->streamConn)))));
770 
771  /* Now that we've consumed some input, try again */
772  rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
773  if (rawlen == 0)
774  {
775  /* Tell caller to try again when our socket is ready. */
776  *wait_fd = PQsocket(conn->streamConn);
777  return 0;
778  }
779  }
780  if (rawlen == -1) /* end-of-streaming or error */
781  {
782  PGresult *res;
783 
784  res = libpqrcv_PQgetResult(conn->streamConn);
785  if (PQresultStatus(res) == PGRES_COMMAND_OK)
786  {
787  PQclear(res);
788 
789  /* Verify that there are no more results. */
790  res = libpqrcv_PQgetResult(conn->streamConn);
791  if (res != NULL)
792  {
793  PQclear(res);
794 
795  /*
796  * If the other side closed the connection orderly (otherwise
797  * we'd seen an error, or PGRES_COPY_IN) don't report an error
798  * here, but let callers deal with it.
799  */
800  if (PQstatus(conn->streamConn) == CONNECTION_BAD)
801  return -1;
802 
803  ereport(ERROR,
804  (errcode(ERRCODE_PROTOCOL_VIOLATION),
805  errmsg("unexpected result after CommandComplete: %s",
806  PQerrorMessage(conn->streamConn))));
807  }
808 
809  return -1;
810  }
811  else if (PQresultStatus(res) == PGRES_COPY_IN)
812  {
813  PQclear(res);
814  return -1;
815  }
816  else
817  {
818  PQclear(res);
819  ereport(ERROR,
820  (errcode(ERRCODE_PROTOCOL_VIOLATION),
821  errmsg("could not receive data from WAL stream: %s",
822  pchomp(PQerrorMessage(conn->streamConn)))));
823  }
824  }
825  if (rawlen < -1)
826  ereport(ERROR,
827  (errcode(ERRCODE_PROTOCOL_VIOLATION),
828  errmsg("could not receive data from WAL stream: %s",
829  pchomp(PQerrorMessage(conn->streamConn)))));
830 
831  /* Return received messages to caller */
832  *buffer = conn->recvBuf;
833  return rawlen;
834 }
835 
836 /*
837  * Send a message to XLOG stream.
838  *
839  * ereports on error.
840  */
841 static void
842 libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes)
843 {
844  if (PQputCopyData(conn->streamConn, buffer, nbytes) <= 0 ||
845  PQflush(conn->streamConn))
846  ereport(ERROR,
847  (errcode(ERRCODE_CONNECTION_FAILURE),
848  errmsg("could not send data to WAL stream: %s",
849  pchomp(PQerrorMessage(conn->streamConn)))));
850 }
851 
852 /*
853  * Create new replication slot.
854  * Returns the name of the exported snapshot for logical slot or NULL for
855  * physical slot.
856  */
857 static char *
858 libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
859  bool temporary, bool two_phase, CRSSnapshotAction snapshot_action,
860  XLogRecPtr *lsn)
861 {
862  PGresult *res;
863  StringInfoData cmd;
864  char *snapshot;
865  int use_new_options_syntax;
866 
867  use_new_options_syntax = (PQserverVersion(conn->streamConn) >= 150000);
868 
869  initStringInfo(&cmd);
870 
871  appendStringInfo(&cmd, "CREATE_REPLICATION_SLOT \"%s\"", slotname);
872 
873  if (temporary)
874  appendStringInfoString(&cmd, " TEMPORARY");
875 
876  if (conn->logical)
877  {
878  appendStringInfoString(&cmd, " LOGICAL pgoutput ");
879  if (use_new_options_syntax)
880  appendStringInfoChar(&cmd, '(');
881  if (two_phase)
882  {
883  appendStringInfoString(&cmd, "TWO_PHASE");
884  if (use_new_options_syntax)
885  appendStringInfoString(&cmd, ", ");
886  else
887  appendStringInfoChar(&cmd, ' ');
888  }
889 
890  if (use_new_options_syntax)
891  {
892  switch (snapshot_action)
893  {
894  case CRS_EXPORT_SNAPSHOT:
895  appendStringInfoString(&cmd, "SNAPSHOT 'export'");
896  break;
898  appendStringInfoString(&cmd, "SNAPSHOT 'nothing'");
899  break;
900  case CRS_USE_SNAPSHOT:
901  appendStringInfoString(&cmd, "SNAPSHOT 'use'");
902  break;
903  }
904  }
905  else
906  {
907  switch (snapshot_action)
908  {
909  case CRS_EXPORT_SNAPSHOT:
910  appendStringInfoString(&cmd, "EXPORT_SNAPSHOT");
911  break;
913  appendStringInfoString(&cmd, "NOEXPORT_SNAPSHOT");
914  break;
915  case CRS_USE_SNAPSHOT:
916  appendStringInfoString(&cmd, "USE_SNAPSHOT");
917  break;
918  }
919  }
920 
921  if (use_new_options_syntax)
922  appendStringInfoChar(&cmd, ')');
923  }
924  else
925  {
926  if (use_new_options_syntax)
927  appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
928  else
929  appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
930  }
931 
932  res = libpqrcv_PQexec(conn->streamConn, cmd.data);
933  pfree(cmd.data);
934 
935  if (PQresultStatus(res) != PGRES_TUPLES_OK)
936  {
937  PQclear(res);
938  ereport(ERROR,
939  (errcode(ERRCODE_PROTOCOL_VIOLATION),
940  errmsg("could not create replication slot \"%s\": %s",
941  slotname, pchomp(PQerrorMessage(conn->streamConn)))));
942  }
943 
944  if (lsn)
946  CStringGetDatum(PQgetvalue(res, 0, 1))));
947 
948  if (!PQgetisnull(res, 0, 2))
949  snapshot = pstrdup(PQgetvalue(res, 0, 2));
950  else
951  snapshot = NULL;
952 
953  PQclear(res);
954 
955  return snapshot;
956 }
957 
958 /*
959  * Return PID of remote backend process.
960  */
961 static pid_t
963 {
964  return PQbackendPID(conn->streamConn);
965 }
966 
967 /*
968  * Convert tuple query result to tuplestore.
969  */
970 static void
972  const int nRetTypes, const Oid *retTypes)
973 {
974  int tupn;
975  int coln;
976  int nfields = PQnfields(pgres);
977  HeapTuple tuple;
978  AttInMetadata *attinmeta;
979  MemoryContext rowcontext;
980  MemoryContext oldcontext;
981 
982  /* Make sure we got expected number of fields. */
983  if (nfields != nRetTypes)
984  ereport(ERROR,
985  (errcode(ERRCODE_PROTOCOL_VIOLATION),
986  errmsg("invalid query response"),
987  errdetail("Expected %d fields, got %d fields.",
988  nRetTypes, nfields)));
989 
990  walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
991 
992  /* Create tuple descriptor corresponding to expected result. */
993  walres->tupledesc = CreateTemplateTupleDesc(nRetTypes);
994  for (coln = 0; coln < nRetTypes; coln++)
995  TupleDescInitEntry(walres->tupledesc, (AttrNumber) coln + 1,
996  PQfname(pgres, coln), retTypes[coln], -1, 0);
997  attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
998 
999  /* No point in doing more here if there were no tuples returned. */
1000  if (PQntuples(pgres) == 0)
1001  return;
1002 
1003  /* Create temporary context for local allocations. */
1005  "libpqrcv query result context",
1007 
1008  /* Process returned rows. */
1009  for (tupn = 0; tupn < PQntuples(pgres); tupn++)
1010  {
1011  char *cstrs[MaxTupleAttributeNumber];
1012 
1014 
1015  /* Do the allocations in temporary context. */
1016  oldcontext = MemoryContextSwitchTo(rowcontext);
1017 
1018  /*
1019  * Fill cstrs with null-terminated strings of column values.
1020  */
1021  for (coln = 0; coln < nfields; coln++)
1022  {
1023  if (PQgetisnull(pgres, tupn, coln))
1024  cstrs[coln] = NULL;
1025  else
1026  cstrs[coln] = PQgetvalue(pgres, tupn, coln);
1027  }
1028 
1029  /* Convert row to a tuple, and add it to the tuplestore */
1030  tuple = BuildTupleFromCStrings(attinmeta, cstrs);
1031  tuplestore_puttuple(walres->tuplestore, tuple);
1032 
1033  /* Clean up */
1034  MemoryContextSwitchTo(oldcontext);
1035  MemoryContextReset(rowcontext);
1036  }
1037 
1038  MemoryContextDelete(rowcontext);
1039 }
1040 
1041 /*
1042  * Public interface for sending generic queries (and commands).
1043  *
1044  * This can only be called from process connected to database.
1045  */
1046 static WalRcvExecResult *
1047 libpqrcv_exec(WalReceiverConn *conn, const char *query,
1048  const int nRetTypes, const Oid *retTypes)
1049 {
1050  PGresult *pgres = NULL;
1051  WalRcvExecResult *walres = palloc0(sizeof(WalRcvExecResult));
1052  char *diag_sqlstate;
1053 
1054  if (MyDatabaseId == InvalidOid)
1055  ereport(ERROR,
1056  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1057  errmsg("the query interface requires a database connection")));
1058 
1059  pgres = libpqrcv_PQexec(conn->streamConn, query);
1060 
1061  switch (PQresultStatus(pgres))
1062  {
1063  case PGRES_SINGLE_TUPLE:
1064  case PGRES_TUPLES_OK:
1065  walres->status = WALRCV_OK_TUPLES;
1066  libpqrcv_processTuples(pgres, walres, nRetTypes, retTypes);
1067  break;
1068 
1069  case PGRES_COPY_IN:
1070  walres->status = WALRCV_OK_COPY_IN;
1071  break;
1072 
1073  case PGRES_COPY_OUT:
1074  walres->status = WALRCV_OK_COPY_OUT;
1075  break;
1076 
1077  case PGRES_COPY_BOTH:
1078  walres->status = WALRCV_OK_COPY_BOTH;
1079  break;
1080 
1081  case PGRES_COMMAND_OK:
1082  walres->status = WALRCV_OK_COMMAND;
1083  break;
1084 
1085  /* Empty query is considered error. */
1086  case PGRES_EMPTY_QUERY:
1087  walres->status = WALRCV_ERROR;
1088  walres->err = _("empty query");
1089  break;
1090 
1091  case PGRES_PIPELINE_SYNC:
1093  walres->status = WALRCV_ERROR;
1094  walres->err = _("unexpected pipeline mode");
1095  break;
1096 
1097  case PGRES_NONFATAL_ERROR:
1098  case PGRES_FATAL_ERROR:
1099  case PGRES_BAD_RESPONSE:
1100  walres->status = WALRCV_ERROR;
1101  walres->err = pchomp(PQerrorMessage(conn->streamConn));
1102  diag_sqlstate = PQresultErrorField(pgres, PG_DIAG_SQLSTATE);
1103  if (diag_sqlstate)
1104  walres->sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
1105  diag_sqlstate[1],
1106  diag_sqlstate[2],
1107  diag_sqlstate[3],
1108  diag_sqlstate[4]);
1109  break;
1110  }
1111 
1112  PQclear(pgres);
1113 
1114  return walres;
1115 }
1116 
1117 /*
1118  * Given a List of strings, return it as single comma separated
1119  * string, quoting identifiers as needed.
1120  *
1121  * This is essentially the reverse of SplitIdentifierString.
1122  *
1123  * The caller should free the result.
1124  */
1125 static char *
1127 {
1128  ListCell *lc;
1129  StringInfoData res;
1130  bool first = true;
1131 
1132  initStringInfo(&res);
1133 
1134  foreach(lc, strings)
1135  {
1136  char *val = strVal(lfirst(lc));
1137  char *val_escaped;
1138 
1139  if (first)
1140  first = false;
1141  else
1142  appendStringInfoChar(&res, ',');
1143 
1144  val_escaped = PQescapeIdentifier(conn, val, strlen(val));
1145  if (!val_escaped)
1146  {
1147  free(res.data);
1148  return NULL;
1149  }
1150  appendStringInfoString(&res, val_escaped);
1151  PQfreemem(val_escaped);
1152  }
1153 
1154  return res.data;
1155 }
int PQputCopyData(PGconn *conn, const char *buffer, int nbytes)
Definition: fe-exec.c:2544
int PQgetlength(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3653
static bool libpqrcv_startstreaming(WalReceiverConn *conn, const WalRcvStreamOptions *options)
int PQnfields(const PGresult *res)
Definition: fe-exec.c:3256
#define WL_SOCKET_WRITEABLE
Definition: latch.h:127
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6744
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:218
#define AllocSetContextCreate
Definition: memutils.h:173
static void libpqrcv_check_conninfo(const char *conninfo)
static int libpqrcv_receive(WalReceiverConn *conn, char **buffer, pgsocket *wait_fd)
uint32 TimeLineID
Definition: xlogdefs.h:59
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3642
#define MaxTupleAttributeNumber
Definition: htup_details.h:33
TupleDesc CreateTemplateTupleDesc(int natts)
Definition: tupdesc.c:45
char * PQfname(const PGresult *res, int field_num)
Definition: fe-exec.c:3334
static char * libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname, bool temporary, bool two_phase, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
void termPQExpBuffer(PQExpBuffer str)
Definition: pqexpbuffer.c:131
#define MAKE_SQLSTATE(ch1, ch2, ch3, ch4, ch5)
Definition: elog.h:64
void _PG_init(void)
char * pstrdup(const char *in)
Definition: mcxt.c:1299
void ProcessWalRcvInterrupts(void)
Definition: walreceiver.c:147
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
#define strVal(v)
Definition: value.h:65
union WalRcvStreamOptions::@104 proto
int errcode(int sqlerrcode)
Definition: elog.c:698
void PQfinish(PGconn *conn)
Definition: fe-connect.c:4231
char * PQport(const PGconn *conn)
Definition: fe-connect.c:6659
int PQputCopyEnd(PGconn *conn, const char *errormsg)
Definition: fe-exec.c:2600
#define WL_SOCKET_READABLE
Definition: latch.h:126
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:143
unsigned int Oid
Definition: postgres_ext.h:31
int PQserverVersion(const PGconn *conn)
Definition: fe-connect.c:6734
static void libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn, TimeLineID tli, char **filename, char **content, int *len)
int PQntuples(const PGresult *res)
Definition: fe-exec.c:3248
#define PG_DIAG_SQLSTATE
Definition: postgres_ext.h:57
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:3178
Datum pg_lsn_in(PG_FUNCTION_ARGS)
Definition: pg_lsn.c:64
void ResetLatch(Latch *latch)
Definition: latch.c:660
int PQgetCopyData(PGconn *conn, char **buffer, int async)
Definition: fe-exec.c:2668
HeapTuple BuildTupleFromCStrings(AttInMetadata *attinmeta, char **values)
Definition: execTuples.c:2146
static PGresult * libpqrcv_PQgetResult(PGconn *streamConn)
char * pchomp(const char *in)
Definition: mcxt.c:1327
int PQsendQuery(PGconn *conn, const char *query)
Definition: fe-exec.c:1326
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
char * PQescapeIdentifier(PGconn *conn, const char *str, size_t len)
Definition: fe-exec.c:4075
PQconninfoOption * PQconninfoParse(const char *conninfo, char **errmsg)
Definition: fe-connect.c:5337
void pfree(void *pointer)
Definition: mcxt.c:1169
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:91
TupleDesc tupledesc
Definition: walreceiver.h:220
static char * libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli)
#define ERROR
Definition: elog.h:46
static void libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres, const int nRetTypes, const Oid *retTypes)
static void libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
PG_MODULE_MAGIC
PGconn * conn
Definition: streamutil.c:54
void appendPQExpBuffer(PQExpBuffer str, const char *fmt,...)
Definition: pqexpbuffer.c:267
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:195
void tuplestore_puttuple(Tuplestorestate *state, HeapTuple tuple)
Definition: tuplestore.c:730
int PQflush(PGconn *conn)
Definition: fe-exec.c:3766
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:176
static char * buf
Definition: pg_test_fsync.c:68
static WalRcvExecResult * libpqrcv_exec(WalReceiverConn *conn, const char *query, const int nRetTypes, const Oid *retTypes)
int errdetail(const char *fmt,...)
Definition: elog.c:1042
void PQconninfoFree(PQconninfoOption *connOptions)
Definition: fe-connect.c:6572
#define CStringGetDatum(X)
Definition: postgres.h:622
Datum DirectFunctionCall1Coll(PGFunction func, Oid collation, Datum arg1)
Definition: fmgr.c:791
XLogRecPtr startpoint
Definition: walreceiver.h:170
int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:500
int pgsocket
Definition: port.h:31
PQconninfoOption * PQconninfo(PGconn *conn)
Definition: fe-connect.c:6528
MemoryContext CurrentMemoryContext
Definition: mcxt.c:42
void TupleDescInitEntry(TupleDesc desc, AttrNumber attributeNumber, const char *attributeName, Oid oidtypeid, int32 typmod, int attdim)
Definition: tupdesc.c:583
static WalReceiverFunctionsType PQWalReceiverFunctions
static AmcheckOptions opts
Definition: pg_amcheck.c:110
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn)
void appendStringInfoChar(StringInfo str, char ch)
Definition: stringinfo.c:188
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
Tuplestorestate * tuplestore_begin_heap(bool randomAccess, bool interXact, int maxKBytes)
Definition: tuplestore.c:318
void * palloc0(Size size)
Definition: mcxt.c:1093
int PQbackendPID(const PGconn *conn)
Definition: fe-connect.c:6778
WalReceiverFunctionsType * WalReceiverFunctions
Definition: walreceiver.c:95
Oid MyDatabaseId
Definition: globals.c:88
AttInMetadata * TupleDescGetAttInMetadata(TupleDesc tupdesc)
Definition: execTuples.c:2097
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1904
int work_mem
Definition: globals.c:124
static void libpqrcv_disconnect(WalReceiverConn *conn)
PostgresPollingStatusType PQconnectPoll(PGconn *conn)
Definition: fe-connect.c:2246
static bool two_phase
static int libpqrcv_server_version(WalReceiverConn *conn)
char * PQescapeLiteral(PGconn *conn, const char *str, size_t len)
Definition: fe-exec.c:4069
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
#define InvalidOid
Definition: postgres_ext.h:36
#define DatumGetLSN(X)
Definition: pg_lsn.h:21
#define ereport(elevel,...)
Definition: elog.h:157
char * PQhost(const PGconn *conn)
Definition: fe-connect.c:6623
void PQclear(PGresult *res)
Definition: fe-exec.c:694
#define free(a)
Definition: header.h:65
#define PQExpBufferDataBroken(buf)
Definition: pqexpbuffer.h:67
int PQendcopy(PGconn *conn)
Definition: fe-exec.c:2802
char * PQresultErrorField(const PGresult *res, int fieldcode)
Definition: fe-exec.c:3233
Tuplestorestate * tuplestore
Definition: walreceiver.h:219
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:804
#define lfirst(lc)
Definition: pg_list.h:169
WalRcvExecStatus status
Definition: walreceiver.h:216
const char * GetDatabaseEncodingName(void)
Definition: mbutils.c:1216
int PQisBusy(PGconn *conn)
Definition: fe-exec.c:1951
#define ALWAYS_SECURE_SEARCH_PATH_SQL
Definition: connect.h:25
static char * stringlist_to_identifierstr(PGconn *conn, List *strings)
int32 pg_strtoint32(const char *s)
Definition: numutils.c:263
PostgresPollingStatusType
Definition: libpq-fe.h:82
static WalReceiverConn * libpqrcv_connect(const char *conninfo, bool logical, const char *appname, char **err)
static char * filename
Definition: pg_dumpall.c:92
void * palloc(Size size)
Definition: mcxt.c:1062
int errmsg(const char *fmt,...)
Definition: elog.c:909
CRSSnapshotAction
Definition: walsender.h:20
static char * libpqrcv_get_conninfo(WalReceiverConn *conn)
#define elog(elevel,...)
Definition: elog.h:232
int i
PGconn * PQconnectStartParams(const char *const *keywords, const char *const *values, int expand_dbname)
Definition: fe-connect.c:760
struct Latch * MyLatch
Definition: globals.c:57
struct WalRcvStreamOptions::@104::@105 physical
static void libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes)
static void static void status(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:227
static void libpqrcv_get_senderinfo(WalReceiverConn *conn, char **sender_host, int *sender_port)
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3667
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6691
#define WL_SOCKET_CONNECTED
Definition: latch.h:135
Definition: pg_list.h:50
#define snprintf
Definition: port.h:217
#define WL_LATCH_SET
Definition: latch.h:125
int16 AttrNumber
Definition: attnum.h:21
#define _(x)
Definition: elog.c:89
void PQfreemem(void *ptr)
Definition: fe-exec.c:3796
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6770
long val
Definition: informix.c:664
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1978
void initPQExpBuffer(PQExpBuffer str)
Definition: pqexpbuffer.c:92
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:130