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