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