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-2023, 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"
28 #include "mb/pg_wchar.h"
29 #include "miscadmin.h"
30 #include "pgstat.h"
31 #include "pqexpbuffer.h"
33 #include "utils/builtins.h"
34 #include "utils/memutils.h"
35 #include "utils/pg_lsn.h"
36 #include "utils/tuplestore.h"
37 
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  bool two_phase,
76  CRSSnapshotAction snapshot_action,
77  XLogRecPtr *lsn);
80  const char *query,
81  const int nRetTypes,
82  const Oid *retTypes);
84 
87  .walrcv_check_conninfo = libpqrcv_check_conninfo,
88  .walrcv_get_conninfo = libpqrcv_get_conninfo,
89  .walrcv_get_senderinfo = libpqrcv_get_senderinfo,
90  .walrcv_identify_system = libpqrcv_identify_system,
91  .walrcv_server_version = libpqrcv_server_version,
92  .walrcv_readtimelinehistoryfile = libpqrcv_readtimelinehistoryfile,
93  .walrcv_startstreaming = libpqrcv_startstreaming,
94  .walrcv_endstreaming = libpqrcv_endstreaming,
95  .walrcv_receive = libpqrcv_receive,
96  .walrcv_send = libpqrcv_send,
97  .walrcv_create_slot = libpqrcv_create_slot,
98  .walrcv_get_backend_pid = libpqrcv_get_backend_pid,
99  .walrcv_exec = libpqrcv_exec,
100  .walrcv_disconnect = libpqrcv_disconnect
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 {
129  const char *keys[6];
130  const char *vals[6];
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  /* Tell the publisher to translate to our encoding */
155  keys[++i] = "client_encoding";
156  vals[i] = GetDatabaseEncodingName();
157 
158  /*
159  * Force assorted GUC parameters to settings that ensure that the
160  * publisher will output data values in a form that is unambiguous to
161  * the subscriber. (We don't want to modify the subscriber's GUC
162  * settings, since that might surprise user-defined code running in
163  * the subscriber, such as triggers.) This should match what pg_dump
164  * does.
165  */
166  keys[++i] = "options";
167  vals[i] = "-c datestyle=ISO -c intervalstyle=postgres -c extra_float_digits=3";
168  }
169  keys[++i] = NULL;
170  vals[i] = NULL;
171 
172  Assert(i < sizeof(keys));
173 
174  conn = palloc0(sizeof(WalReceiverConn));
175  conn->streamConn =
176  libpqsrv_connect_params(keys, vals,
177  /* expand_dbname = */ true,
179 
180  if (PQstatus(conn->streamConn) != CONNECTION_OK)
181  goto bad_connection_errmsg;
182 
183  if (logical)
184  {
185  PGresult *res;
186 
187  res = libpqrcv_PQexec(conn->streamConn,
190  {
191  PQclear(res);
192  *err = psprintf(_("could not clear search path: %s"),
193  pchomp(PQerrorMessage(conn->streamConn)));
194  goto bad_connection;
195  }
196  PQclear(res);
197  }
198 
199  conn->logical = logical;
200 
201  return conn;
202 
203  /* error path, using libpq's error message */
204 bad_connection_errmsg:
205  *err = pchomp(PQerrorMessage(conn->streamConn));
206 
207  /* error path, error already set */
208 bad_connection:
209  libpqsrv_disconnect(conn->streamConn);
210  pfree(conn);
211  return NULL;
212 }
213 
214 /*
215  * Validate connection info string (just try to parse it)
216  */
217 static void
218 libpqrcv_check_conninfo(const char *conninfo)
219 {
220  PQconninfoOption *opts = NULL;
221  char *err = NULL;
222 
223  opts = PQconninfoParse(conninfo, &err);
224  if (opts == NULL)
225  {
226  /* The error string is malloc'd, so we must free it explicitly */
227  char *errcopy = err ? pstrdup(err) : "out of memory";
228 
229  PQfreemem(err);
230  ereport(ERROR,
231  (errcode(ERRCODE_SYNTAX_ERROR),
232  errmsg("invalid connection string syntax: %s", errcopy)));
233  }
234 
236 }
237 
238 /*
239  * Return a user-displayable conninfo string. Any security-sensitive fields
240  * are obfuscated.
241  */
242 static char *
244 {
245  PQconninfoOption *conn_opts;
246  PQconninfoOption *conn_opt;
248  char *retval;
249 
250  Assert(conn->streamConn != NULL);
251 
253  conn_opts = PQconninfo(conn->streamConn);
254 
255  if (conn_opts == NULL)
256  ereport(ERROR,
257  (errcode(ERRCODE_OUT_OF_MEMORY),
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);
285  return retval;
286 }
287 
288 /*
289  * Provides information of sender this WAL receiver is connected to.
290  */
291 static void
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");
327  {
328  PQclear(res);
329  ereport(ERROR,
330  (errcode(ERRCODE_PROTOCOL_VIOLATION),
331  errmsg("could not receive database system identifier and timeline ID from "
332  "the primary server: %s",
333  pchomp(PQerrorMessage(conn->streamConn)))));
334  }
335  if (PQnfields(res) < 3 || PQntuples(res) != 1)
336  {
337  int ntuples = PQntuples(res);
338  int nfields = PQnfields(res);
339 
340  PQclear(res);
341  ereport(ERROR,
342  (errcode(ERRCODE_PROTOCOL_VIOLATION),
343  errmsg("invalid response from primary server"),
344  errdetail("Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
345  ntuples, nfields, 3, 1)));
346  }
347  primary_sysid = pstrdup(PQgetvalue(res, 0, 0));
348  *primary_tli = pg_strtoint32(PQgetvalue(res, 0, 1));
349  PQclear(res);
350 
351  return primary_sysid;
352 }
353 
354 /*
355  * Thin wrapper around libpq to obtain server version.
356  */
357 static int
359 {
360  return PQserverVersion(conn->streamConn);
361 }
362 
363 /*
364  * Start streaming WAL data from given streaming options.
365  *
366  * Returns true if we switched successfully to copy-both mode. False
367  * means the server received the command and executed it successfully, but
368  * didn't switch to copy-mode. That means that there was no WAL on the
369  * requested timeline and starting point, because the server switched to
370  * another timeline at or before the requested starting point. On failure,
371  * throws an ERROR.
372  */
373 static bool
376 {
377  StringInfoData cmd;
378  PGresult *res;
379 
380  Assert(options->logical == conn->logical);
381  Assert(options->slotname || !options->logical);
382 
383  initStringInfo(&cmd);
384 
385  /* Build the command. */
386  appendStringInfoString(&cmd, "START_REPLICATION");
387  if (options->slotname != NULL)
388  appendStringInfo(&cmd, " SLOT \"%s\"",
389  options->slotname);
390 
391  if (options->logical)
392  appendStringInfoString(&cmd, " LOGICAL");
393 
394  appendStringInfo(&cmd, " %X/%X", LSN_FORMAT_ARGS(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  if (options->proto.logical.streaming_str)
412  appendStringInfo(&cmd, ", streaming '%s'",
413  options->proto.logical.streaming_str);
414 
415  if (options->proto.logical.twophase &&
416  PQserverVersion(conn->streamConn) >= 150000)
417  appendStringInfoString(&cmd, ", two_phase 'on'");
418 
419  if (options->proto.logical.origin &&
420  PQserverVersion(conn->streamConn) >= 160000)
421  appendStringInfo(&cmd, ", origin '%s'",
422  options->proto.logical.origin);
423 
424  pubnames = options->proto.logical.publication_names;
425  pubnames_str = stringlist_to_identifierstr(conn->streamConn, pubnames);
426  if (!pubnames_str)
427  ereport(ERROR,
428  (errcode(ERRCODE_OUT_OF_MEMORY), /* likely guess */
429  errmsg("could not start WAL streaming: %s",
430  pchomp(PQerrorMessage(conn->streamConn)))));
431  pubnames_literal = PQescapeLiteral(conn->streamConn, pubnames_str,
432  strlen(pubnames_str));
433  if (!pubnames_literal)
434  ereport(ERROR,
435  (errcode(ERRCODE_OUT_OF_MEMORY), /* likely guess */
436  errmsg("could not start WAL streaming: %s",
437  pchomp(PQerrorMessage(conn->streamConn)))));
438  appendStringInfo(&cmd, ", publication_names %s", pubnames_literal);
439  PQfreemem(pubnames_literal);
440  pfree(pubnames_str);
441 
442  if (options->proto.logical.binary &&
443  PQserverVersion(conn->streamConn) >= 140000)
444  appendStringInfoString(&cmd, ", binary 'true'");
445 
446  appendStringInfoChar(&cmd, ')');
447  }
448  else
449  appendStringInfo(&cmd, " TIMELINE %u",
450  options->proto.physical.startpointTLI);
451 
452  /* Start streaming. */
453  res = libpqrcv_PQexec(conn->streamConn, cmd.data);
454  pfree(cmd.data);
455 
457  {
458  PQclear(res);
459  return false;
460  }
461  else if (PQresultStatus(res) != PGRES_COPY_BOTH)
462  {
463  PQclear(res);
464  ereport(ERROR,
465  (errcode(ERRCODE_PROTOCOL_VIOLATION),
466  errmsg("could not start WAL streaming: %s",
467  pchomp(PQerrorMessage(conn->streamConn)))));
468  }
469  PQclear(res);
470  return true;
471 }
472 
473 /*
474  * Stop streaming WAL data. Returns the next timeline's ID in *next_tli, as
475  * reported by the server, or 0 if it did not report it.
476  */
477 static void
479 {
480  PGresult *res;
481 
482  /*
483  * Send copy-end message. As in libpqrcv_PQexec, this could theoretically
484  * block, but the risk seems small.
485  */
486  if (PQputCopyEnd(conn->streamConn, NULL) <= 0 ||
487  PQflush(conn->streamConn))
488  ereport(ERROR,
489  (errcode(ERRCODE_CONNECTION_FAILURE),
490  errmsg("could not send end-of-streaming message to primary: %s",
491  pchomp(PQerrorMessage(conn->streamConn)))));
492 
493  *next_tli = 0;
494 
495  /*
496  * After COPY is finished, we should receive a result set indicating the
497  * next timeline's ID, or just CommandComplete if the server was shut
498  * down.
499  *
500  * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
501  * also possible in case we aborted the copy in mid-stream.
502  */
503  res = libpqrcv_PQgetResult(conn->streamConn);
505  {
506  /*
507  * Read the next timeline's ID. The server also sends the timeline's
508  * starting point, but it is ignored.
509  */
510  if (PQnfields(res) < 2 || PQntuples(res) != 1)
511  ereport(ERROR,
512  (errcode(ERRCODE_PROTOCOL_VIOLATION),
513  errmsg("unexpected result set after end-of-streaming")));
514  *next_tli = pg_strtoint32(PQgetvalue(res, 0, 0));
515  PQclear(res);
516 
517  /* the result set should be followed by CommandComplete */
518  res = libpqrcv_PQgetResult(conn->streamConn);
519  }
520  else if (PQresultStatus(res) == PGRES_COPY_OUT)
521  {
522  PQclear(res);
523 
524  /* End the copy */
525  if (PQendcopy(conn->streamConn))
526  ereport(ERROR,
527  (errcode(ERRCODE_CONNECTION_FAILURE),
528  errmsg("error while shutting down streaming COPY: %s",
529  pchomp(PQerrorMessage(conn->streamConn)))));
530 
531  /* CommandComplete should follow */
532  res = libpqrcv_PQgetResult(conn->streamConn);
533  }
534 
536  ereport(ERROR,
537  (errcode(ERRCODE_PROTOCOL_VIOLATION),
538  errmsg("error reading result of streaming command: %s",
539  pchomp(PQerrorMessage(conn->streamConn)))));
540  PQclear(res);
541 
542  /* Verify that there are no more results */
543  res = libpqrcv_PQgetResult(conn->streamConn);
544  if (res != NULL)
545  ereport(ERROR,
546  (errcode(ERRCODE_PROTOCOL_VIOLATION),
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);
570  {
571  PQclear(res);
572  ereport(ERROR,
573  (errcode(ERRCODE_PROTOCOL_VIOLATION),
574  errmsg("could not receive timeline history file from "
575  "the primary server: %s",
576  pchomp(PQerrorMessage(conn->streamConn)))));
577  }
578  if (PQnfields(res) != 2 || PQntuples(res) != 1)
579  {
580  int ntuples = PQntuples(res);
581  int nfields = PQnfields(res);
582 
583  PQclear(res);
584  ereport(ERROR,
585  (errcode(ERRCODE_PROTOCOL_VIOLATION),
586  errmsg("invalid response from primary server"),
587  errdetail("Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
588  ntuples, nfields)));
589  }
590  *filename = pstrdup(PQgetvalue(res, 0, 0));
591 
592  *len = PQgetlength(res, 0, 1);
593  *content = palloc(*len);
594  memcpy(*content, PQgetvalue(res, 0, 1), *len);
595  PQclear(res);
596 }
597 
598 /*
599  * Send a query and wait for the results by using the asynchronous libpq
600  * functions and socket readiness events.
601  *
602  * We must not use the regular blocking libpq functions like PQexec()
603  * since they are uninterruptible by signals on some platforms, such as
604  * Windows.
605  *
606  * The function is modeled on PQexec() in libpq, but only implements
607  * those parts that are in use in the walreceiver api.
608  *
609  * May return NULL, rather than an error result, on failure.
610  */
611 static PGresult *
612 libpqrcv_PQexec(PGconn *streamConn, const char *query)
613 {
614  PGresult *lastResult = NULL;
615 
616  /*
617  * PQexec() silently discards any prior query results on the connection.
618  * This is not required for this function as it's expected that the caller
619  * (which is this library in all cases) will behave correctly and we don't
620  * have to be backwards compatible with old libpq.
621  */
622 
623  /*
624  * Submit the query. Since we don't use non-blocking mode, this could
625  * theoretically block. In practice, since we don't send very long query
626  * strings, the risk seems negligible.
627  */
628  if (!PQsendQuery(streamConn, query))
629  return NULL;
630 
631  for (;;)
632  {
633  /* Wait for, and collect, the next PGresult. */
634  PGresult *result;
635 
636  result = libpqrcv_PQgetResult(streamConn);
637  if (result == NULL)
638  break; /* query is complete, or failure */
639 
640  /*
641  * Emulate PQexec()'s behavior of returning the last result when there
642  * are many. We are fine with returning just last error message.
643  */
644  PQclear(lastResult);
645  lastResult = result;
646 
647  if (PQresultStatus(lastResult) == PGRES_COPY_IN ||
648  PQresultStatus(lastResult) == PGRES_COPY_OUT ||
649  PQresultStatus(lastResult) == PGRES_COPY_BOTH ||
650  PQstatus(streamConn) == CONNECTION_BAD)
651  break;
652  }
653 
654  return lastResult;
655 }
656 
657 /*
658  * Perform the equivalent of PQgetResult(), but watch for interrupts.
659  */
660 static PGresult *
662 {
663  /*
664  * Collect data until PQgetResult is ready to get the result without
665  * blocking.
666  */
667  while (PQisBusy(streamConn))
668  {
669  int rc;
670 
671  /*
672  * We don't need to break down the sleep into smaller increments,
673  * since we'll get interrupted by signals and can handle any
674  * interrupts here.
675  */
678  WL_LATCH_SET,
679  PQsocket(streamConn),
680  0,
682 
683  /* Interrupted? */
684  if (rc & WL_LATCH_SET)
685  {
688  }
689 
690  /* Consume whatever data is available from the socket */
691  if (PQconsumeInput(streamConn) == 0)
692  {
693  /* trouble; return NULL */
694  return NULL;
695  }
696  }
697 
698  /* Now we can collect and return the next PGresult */
699  return PQgetResult(streamConn);
700 }
701 
702 /*
703  * Disconnect connection to primary, if any.
704  */
705 static void
707 {
708  libpqsrv_disconnect(conn->streamConn);
709  PQfreemem(conn->recvBuf);
710  pfree(conn);
711 }
712 
713 /*
714  * Receive a message available from XLOG stream.
715  *
716  * Returns:
717  *
718  * If data was received, returns the length of the data. *buffer is set to
719  * point to a buffer holding the received message. The buffer is only valid
720  * until the next libpqrcv_* call.
721  *
722  * If no data was available immediately, returns 0, and *wait_fd is set to a
723  * socket descriptor which can be waited on before trying again.
724  *
725  * -1 if the server ended the COPY.
726  *
727  * ereports on error.
728  */
729 static int
731  pgsocket *wait_fd)
732 {
733  int rawlen;
734 
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  (errcode(ERRCODE_CONNECTION_FAILURE),
746  errmsg("could not receive data from WAL stream: %s",
747  pchomp(PQerrorMessage(conn->streamConn)))));
748 
749  /* Now that we've consumed some input, try again */
750  rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
751  if (rawlen == 0)
752  {
753  /* Tell caller to try again when our socket is ready. */
754  *wait_fd = PQsocket(conn->streamConn);
755  return 0;
756  }
757  }
758  if (rawlen == -1) /* end-of-streaming or error */
759  {
760  PGresult *res;
761 
762  res = libpqrcv_PQgetResult(conn->streamConn);
764  {
765  PQclear(res);
766 
767  /* Verify that there are no more results. */
768  res = libpqrcv_PQgetResult(conn->streamConn);
769  if (res != NULL)
770  {
771  PQclear(res);
772 
773  /*
774  * If the other side closed the connection orderly (otherwise
775  * we'd seen an error, or PGRES_COPY_IN) don't report an error
776  * here, but let callers deal with it.
777  */
778  if (PQstatus(conn->streamConn) == CONNECTION_BAD)
779  return -1;
780 
781  ereport(ERROR,
782  (errcode(ERRCODE_PROTOCOL_VIOLATION),
783  errmsg("unexpected result after CommandComplete: %s",
784  PQerrorMessage(conn->streamConn))));
785  }
786 
787  return -1;
788  }
789  else if (PQresultStatus(res) == PGRES_COPY_IN)
790  {
791  PQclear(res);
792  return -1;
793  }
794  else
795  {
796  PQclear(res);
797  ereport(ERROR,
798  (errcode(ERRCODE_PROTOCOL_VIOLATION),
799  errmsg("could not receive data from WAL stream: %s",
800  pchomp(PQerrorMessage(conn->streamConn)))));
801  }
802  }
803  if (rawlen < -1)
804  ereport(ERROR,
805  (errcode(ERRCODE_PROTOCOL_VIOLATION),
806  errmsg("could not receive data from WAL stream: %s",
807  pchomp(PQerrorMessage(conn->streamConn)))));
808 
809  /* Return received messages to caller */
810  *buffer = conn->recvBuf;
811  return rawlen;
812 }
813 
814 /*
815  * Send a message to XLOG stream.
816  *
817  * ereports on error.
818  */
819 static void
820 libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes)
821 {
822  if (PQputCopyData(conn->streamConn, buffer, nbytes) <= 0 ||
823  PQflush(conn->streamConn))
824  ereport(ERROR,
825  (errcode(ERRCODE_CONNECTION_FAILURE),
826  errmsg("could not send data to WAL stream: %s",
827  pchomp(PQerrorMessage(conn->streamConn)))));
828 }
829 
830 /*
831  * Create new replication slot.
832  * Returns the name of the exported snapshot for logical slot or NULL for
833  * physical slot.
834  */
835 static char *
837  bool temporary, bool two_phase, CRSSnapshotAction snapshot_action,
838  XLogRecPtr *lsn)
839 {
840  PGresult *res;
841  StringInfoData cmd;
842  char *snapshot;
843  int use_new_options_syntax;
844 
845  use_new_options_syntax = (PQserverVersion(conn->streamConn) >= 150000);
846 
847  initStringInfo(&cmd);
848 
849  appendStringInfo(&cmd, "CREATE_REPLICATION_SLOT \"%s\"", slotname);
850 
851  if (temporary)
852  appendStringInfoString(&cmd, " TEMPORARY");
853 
854  if (conn->logical)
855  {
856  appendStringInfoString(&cmd, " LOGICAL pgoutput ");
857  if (use_new_options_syntax)
858  appendStringInfoChar(&cmd, '(');
859  if (two_phase)
860  {
861  appendStringInfoString(&cmd, "TWO_PHASE");
862  if (use_new_options_syntax)
863  appendStringInfoString(&cmd, ", ");
864  else
865  appendStringInfoChar(&cmd, ' ');
866  }
867 
868  if (use_new_options_syntax)
869  {
870  switch (snapshot_action)
871  {
872  case CRS_EXPORT_SNAPSHOT:
873  appendStringInfoString(&cmd, "SNAPSHOT 'export'");
874  break;
876  appendStringInfoString(&cmd, "SNAPSHOT 'nothing'");
877  break;
878  case CRS_USE_SNAPSHOT:
879  appendStringInfoString(&cmd, "SNAPSHOT 'use'");
880  break;
881  }
882  }
883  else
884  {
885  switch (snapshot_action)
886  {
887  case CRS_EXPORT_SNAPSHOT:
888  appendStringInfoString(&cmd, "EXPORT_SNAPSHOT");
889  break;
891  appendStringInfoString(&cmd, "NOEXPORT_SNAPSHOT");
892  break;
893  case CRS_USE_SNAPSHOT:
894  appendStringInfoString(&cmd, "USE_SNAPSHOT");
895  break;
896  }
897  }
898 
899  if (use_new_options_syntax)
900  appendStringInfoChar(&cmd, ')');
901  }
902  else
903  {
904  if (use_new_options_syntax)
905  appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
906  else
907  appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
908  }
909 
910  res = libpqrcv_PQexec(conn->streamConn, cmd.data);
911  pfree(cmd.data);
912 
914  {
915  PQclear(res);
916  ereport(ERROR,
917  (errcode(ERRCODE_PROTOCOL_VIOLATION),
918  errmsg("could not create replication slot \"%s\": %s",
919  slotname, pchomp(PQerrorMessage(conn->streamConn)))));
920  }
921 
922  if (lsn)
924  CStringGetDatum(PQgetvalue(res, 0, 1))));
925 
926  if (!PQgetisnull(res, 0, 2))
927  snapshot = pstrdup(PQgetvalue(res, 0, 2));
928  else
929  snapshot = NULL;
930 
931  PQclear(res);
932 
933  return snapshot;
934 }
935 
936 /*
937  * Return PID of remote backend process.
938  */
939 static pid_t
941 {
942  return PQbackendPID(conn->streamConn);
943 }
944 
945 /*
946  * Convert tuple query result to tuplestore.
947  */
948 static void
950  const int nRetTypes, const Oid *retTypes)
951 {
952  int tupn;
953  int coln;
954  int nfields = PQnfields(pgres);
955  HeapTuple tuple;
956  AttInMetadata *attinmeta;
957  MemoryContext rowcontext;
958  MemoryContext oldcontext;
959 
960  /* Make sure we got expected number of fields. */
961  if (nfields != nRetTypes)
962  ereport(ERROR,
963  (errcode(ERRCODE_PROTOCOL_VIOLATION),
964  errmsg("invalid query response"),
965  errdetail("Expected %d fields, got %d fields.",
966  nRetTypes, nfields)));
967 
968  walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
969 
970  /* Create tuple descriptor corresponding to expected result. */
971  walres->tupledesc = CreateTemplateTupleDesc(nRetTypes);
972  for (coln = 0; coln < nRetTypes; coln++)
973  TupleDescInitEntry(walres->tupledesc, (AttrNumber) coln + 1,
974  PQfname(pgres, coln), retTypes[coln], -1, 0);
975  attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
976 
977  /* No point in doing more here if there were no tuples returned. */
978  if (PQntuples(pgres) == 0)
979  return;
980 
981  /* Create temporary context for local allocations. */
983  "libpqrcv query result context",
985 
986  /* Process returned rows. */
987  for (tupn = 0; tupn < PQntuples(pgres); tupn++)
988  {
989  char *cstrs[MaxTupleAttributeNumber];
990 
992 
993  /* Do the allocations in temporary context. */
994  oldcontext = MemoryContextSwitchTo(rowcontext);
995 
996  /*
997  * Fill cstrs with null-terminated strings of column values.
998  */
999  for (coln = 0; coln < nfields; coln++)
1000  {
1001  if (PQgetisnull(pgres, tupn, coln))
1002  cstrs[coln] = NULL;
1003  else
1004  cstrs[coln] = PQgetvalue(pgres, tupn, coln);
1005  }
1006 
1007  /* Convert row to a tuple, and add it to the tuplestore */
1008  tuple = BuildTupleFromCStrings(attinmeta, cstrs);
1009  tuplestore_puttuple(walres->tuplestore, tuple);
1010 
1011  /* Clean up */
1012  MemoryContextSwitchTo(oldcontext);
1013  MemoryContextReset(rowcontext);
1014  }
1015 
1016  MemoryContextDelete(rowcontext);
1017 }
1018 
1019 /*
1020  * Public interface for sending generic queries (and commands).
1021  *
1022  * This can only be called from process connected to database.
1023  */
1024 static WalRcvExecResult *
1025 libpqrcv_exec(WalReceiverConn *conn, const char *query,
1026  const int nRetTypes, const Oid *retTypes)
1027 {
1028  PGresult *pgres = NULL;
1029  WalRcvExecResult *walres = palloc0(sizeof(WalRcvExecResult));
1030  char *diag_sqlstate;
1031 
1032  if (MyDatabaseId == InvalidOid)
1033  ereport(ERROR,
1034  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1035  errmsg("the query interface requires a database connection")));
1036 
1037  pgres = libpqrcv_PQexec(conn->streamConn, query);
1038 
1039  switch (PQresultStatus(pgres))
1040  {
1041  case PGRES_SINGLE_TUPLE:
1042  case PGRES_TUPLES_OK:
1043  walres->status = WALRCV_OK_TUPLES;
1044  libpqrcv_processTuples(pgres, walres, nRetTypes, retTypes);
1045  break;
1046 
1047  case PGRES_COPY_IN:
1048  walres->status = WALRCV_OK_COPY_IN;
1049  break;
1050 
1051  case PGRES_COPY_OUT:
1052  walres->status = WALRCV_OK_COPY_OUT;
1053  break;
1054 
1055  case PGRES_COPY_BOTH:
1056  walres->status = WALRCV_OK_COPY_BOTH;
1057  break;
1058 
1059  case PGRES_COMMAND_OK:
1060  walres->status = WALRCV_OK_COMMAND;
1061  break;
1062 
1063  /* Empty query is considered error. */
1064  case PGRES_EMPTY_QUERY:
1065  walres->status = WALRCV_ERROR;
1066  walres->err = _("empty query");
1067  break;
1068 
1069  case PGRES_PIPELINE_SYNC:
1071  walres->status = WALRCV_ERROR;
1072  walres->err = _("unexpected pipeline mode");
1073  break;
1074 
1075  case PGRES_NONFATAL_ERROR:
1076  case PGRES_FATAL_ERROR:
1077  case PGRES_BAD_RESPONSE:
1078  walres->status = WALRCV_ERROR;
1079  walres->err = pchomp(PQerrorMessage(conn->streamConn));
1080  diag_sqlstate = PQresultErrorField(pgres, PG_DIAG_SQLSTATE);
1081  if (diag_sqlstate)
1082  walres->sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
1083  diag_sqlstate[1],
1084  diag_sqlstate[2],
1085  diag_sqlstate[3],
1086  diag_sqlstate[4]);
1087  break;
1088  }
1089 
1090  PQclear(pgres);
1091 
1092  return walres;
1093 }
1094 
1095 /*
1096  * Given a List of strings, return it as single comma separated
1097  * string, quoting identifiers as needed.
1098  *
1099  * This is essentially the reverse of SplitIdentifierString.
1100  *
1101  * The caller should free the result.
1102  */
1103 static char *
1105 {
1106  ListCell *lc;
1108  bool first = true;
1109 
1110  initStringInfo(&res);
1111 
1112  foreach(lc, strings)
1113  {
1114  char *val = strVal(lfirst(lc));
1115  char *val_escaped;
1116 
1117  if (first)
1118  first = false;
1119  else
1120  appendStringInfoChar(&res, ',');
1121 
1122  val_escaped = PQescapeIdentifier(conn, val, strlen(val));
1123  if (!val_escaped)
1124  {
1125  free(res.data);
1126  return NULL;
1127  }
1128  appendStringInfoString(&res, val_escaped);
1129  PQfreemem(val_escaped);
1130  }
1131 
1132  return res.data;
1133 }
int16 AttrNumber
Definition: attnum.h:21
#define ALWAYS_SECURE_SEARCH_PATH_SQL
Definition: connect.h:25
int errdetail(const char *fmt,...)
Definition: elog.c:1202
int errcode(int sqlerrcode)
Definition: elog.c:858
int errmsg(const char *fmt,...)
Definition: elog.c:1069
#define _(x)
Definition: elog.c:91
#define ERROR
Definition: elog.h:39
#define MAKE_SQLSTATE(ch1, ch2, ch3, ch4, ch5)
Definition: elog.h:56
#define ereport(elevel,...)
Definition: elog.h:149
void err(int eval, const char *fmt,...)
Definition: err.c:43
HeapTuple BuildTupleFromCStrings(AttInMetadata *attinmeta, char **values)
Definition: execTuples.c:2135
AttInMetadata * TupleDescGetAttInMetadata(TupleDesc tupdesc)
Definition: execTuples.c:2086
int PQserverVersion(const PGconn *conn)
Definition: fe-connect.c:6955
PQconninfoOption * PQconninfoParse(const char *conninfo, char **errmsg)
Definition: fe-connect.c:5579
char * PQhost(const PGconn *conn)
Definition: fe-connect.c:6844
void PQconninfoFree(PQconninfoOption *connOptions)
Definition: fe-connect.c:6798
PQconninfoOption * PQconninfo(PGconn *conn)
Definition: fe-connect.c:6754
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6965
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6912
int PQbackendPID(const PGconn *conn)
Definition: fe-connect.c:6999
char * PQport(const PGconn *conn)
Definition: fe-connect.c:6880
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6991
int PQgetlength(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3716
int PQflush(PGconn *conn)
Definition: fe-exec.c:3833
void PQfreemem(void *ptr)
Definition: fe-exec.c:3865
char * PQescapeIdentifier(PGconn *conn, const char *str, size_t len)
Definition: fe-exec.c:4143
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:3240
int PQendcopy(PGconn *conn)
Definition: fe-exec.c:2831
int PQputCopyEnd(PGconn *conn, const char *errormsg)
Definition: fe-exec.c:2631
int PQntuples(const PGresult *res)
Definition: fe-exec.c:3310
int PQputCopyData(PGconn *conn, const char *buffer, int nbytes)
Definition: fe-exec.c:2576
char * PQfname(const PGresult *res, int field_num)
Definition: fe-exec.c:3396
char * PQescapeLiteral(PGconn *conn, const char *str, size_t len)
Definition: fe-exec.c:4137
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1953
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3705
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3730
int PQsendQuery(PGconn *conn, const char *query)
Definition: fe-exec.c:1418
int PQisBusy(PGconn *conn)
Definition: fe-exec.c:2000
char * PQresultErrorField(const PGresult *res, int fieldcode)
Definition: fe-exec.c:3295
int PQnfields(const PGresult *res)
Definition: fe-exec.c:3318
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:2031
int PQgetCopyData(PGconn *conn, char **buffer, int async)
Definition: fe-exec.c:2698
Datum DirectFunctionCall1Coll(PGFunction func, Oid collation, Datum arg1)
Definition: fmgr.c:779
int work_mem
Definition: globals.c:125
struct Latch * MyLatch
Definition: globals.c:58
Oid MyDatabaseId
Definition: globals.c:89
#define free(a)
Definition: header.h:65
#define MaxTupleAttributeNumber
Definition: htup_details.h:34
long val
Definition: informix.c:664
int i
Definition: isn.c:73
int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:540
void ResetLatch(Latch *latch)
Definition: latch.c:699
#define WL_SOCKET_READABLE
Definition: latch.h:126
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:130
#define WL_LATCH_SET
Definition: latch.h:125
static void libpqsrv_disconnect(PGconn *conn)
static PGconn * libpqsrv_connect_params(const char *const *keywords, const char *const *values, int expand_dbname, uint32 wait_event_info)
@ CONNECTION_BAD
Definition: libpq-fe.h:61
@ CONNECTION_OK
Definition: libpq-fe.h:60
@ PGRES_COPY_IN
Definition: libpq-fe.h:104
@ PGRES_COPY_BOTH
Definition: libpq-fe.h:109
@ PGRES_COMMAND_OK
Definition: libpq-fe.h:97
@ PGRES_FATAL_ERROR
Definition: libpq-fe.h:108
@ PGRES_SINGLE_TUPLE
Definition: libpq-fe.h:110
@ PGRES_COPY_OUT
Definition: libpq-fe.h:103
@ PGRES_EMPTY_QUERY
Definition: libpq-fe.h:96
@ PGRES_PIPELINE_SYNC
Definition: libpq-fe.h:111
@ PGRES_BAD_RESPONSE
Definition: libpq-fe.h:105
@ PGRES_PIPELINE_ABORTED
Definition: libpq-fe.h:112
@ PGRES_NONFATAL_ERROR
Definition: libpq-fe.h:107
@ PGRES_TUPLES_OK
Definition: libpq-fe.h:100
Assert(fmt[strlen(fmt) - 1] !='\n')
static void libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn, TimeLineID tli, char **filename, char **content, int *len)
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn)
static void libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes)
void _PG_init(void)
static void libpqrcv_disconnect(WalReceiverConn *conn)
static char * libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli)
static PGresult * libpqrcv_PQgetResult(PGconn *streamConn)
static WalRcvExecResult * libpqrcv_exec(WalReceiverConn *conn, const char *query, const int nRetTypes, const Oid *retTypes)
PG_MODULE_MAGIC
static void libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
static void libpqrcv_get_senderinfo(WalReceiverConn *conn, char **sender_host, int *sender_port)
static int libpqrcv_server_version(WalReceiverConn *conn)
static char * stringlist_to_identifierstr(PGconn *conn, List *strings)
static WalReceiverConn * libpqrcv_connect(const char *conninfo, bool logical, const char *appname, char **err)
static bool libpqrcv_startstreaming(WalReceiverConn *conn, const WalRcvStreamOptions *options)
static int libpqrcv_receive(WalReceiverConn *conn, char **buffer, pgsocket *wait_fd)
static char * libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname, bool temporary, bool two_phase, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
static void libpqrcv_check_conninfo(const char *conninfo)
static WalReceiverFunctionsType PQWalReceiverFunctions
static char * libpqrcv_get_conninfo(WalReceiverConn *conn)
static void libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres, const int nRetTypes, const Oid *retTypes)
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
const char * GetDatabaseEncodingName(void)
Definition: mbutils.c:1274
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:314
char * pchomp(const char *in)
Definition: mcxt.c:1652
char * pstrdup(const char *in)
Definition: mcxt.c:1624
void pfree(void *pointer)
Definition: mcxt.c:1436
void * palloc0(Size size)
Definition: mcxt.c:1241
MemoryContext CurrentMemoryContext
Definition: mcxt.c:135
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:387
void * palloc(Size size)
Definition: mcxt.c:1210
#define AllocSetContextCreate
Definition: memutils.h:129
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:153
int32 pg_strtoint32(const char *s)
Definition: numutils.c:291
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:138
static AmcheckOptions opts
Definition: pg_amcheck.c:110
const void size_t len
static char * filename
Definition: pg_dumpall.c:119
#define lfirst(lc)
Definition: pg_list.h:172
Datum pg_lsn_in(PG_FUNCTION_ARGS)
Definition: pg_lsn.c:64
static XLogRecPtr DatumGetLSN(Datum X)
Definition: pg_lsn.h:22
static bool two_phase
static char * buf
Definition: pg_test_fsync.c:67
int pgsocket
Definition: port.h:29
#define snprintf
Definition: port.h:238
static Datum CStringGetDatum(const char *X)
Definition: postgres.h:350
#define InvalidOid
Definition: postgres_ext.h:36
unsigned int Oid
Definition: postgres_ext.h:31
#define PG_DIAG_SQLSTATE
Definition: postgres_ext.h:56
void initPQExpBuffer(PQExpBuffer str)
Definition: pqexpbuffer.c:90
void appendPQExpBuffer(PQExpBuffer str, const char *fmt,...)
Definition: pqexpbuffer.c:265
void termPQExpBuffer(PQExpBuffer str)
Definition: pqexpbuffer.c:129
#define PQExpBufferDataBroken(buf)
Definition: pqexpbuffer.h:67
char * psprintf(const char *fmt,...)
Definition: psprintf.c:46
PGconn * conn
Definition: streamutil.c:54
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:91
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:176
void appendStringInfoChar(StringInfo str, char ch)
Definition: stringinfo.c:188
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
Definition: pg_list.h:54
Tuplestorestate * tuplestore
Definition: walreceiver.h:223
TupleDesc tupledesc
Definition: walreceiver.h:224
WalRcvExecStatus status
Definition: walreceiver.h:220
walrcv_connect_fn walrcv_connect
Definition: walreceiver.h:389
TupleDesc CreateTemplateTupleDesc(int natts)
Definition: tupdesc.c:45
void TupleDescInitEntry(TupleDesc desc, AttrNumber attributeNumber, const char *attributeName, Oid oidtypeid, int32 typmod, int attdim)
Definition: tupdesc.c:583
Tuplestorestate * tuplestore_begin_heap(bool randomAccess, bool interXact, int maxKBytes)
Definition: tuplestore.c:318
void tuplestore_puttuple(Tuplestorestate *state, HeapTuple tuple)
Definition: tuplestore.c:730
#define strVal(v)
Definition: value.h:82
@ WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE
Definition: wait_event.h:67
@ WAIT_EVENT_LIBPQWALRECEIVER_CONNECT
Definition: wait_event.h:66
void ProcessWalRcvInterrupts(void)
Definition: walreceiver.c:166
WalReceiverFunctionsType * WalReceiverFunctions
Definition: walreceiver.c:96
@ WALRCV_OK_COPY_IN
Definition: walreceiver.h:208
@ WALRCV_OK_COMMAND
Definition: walreceiver.h:205
@ WALRCV_ERROR
Definition: walreceiver.h:204
@ WALRCV_OK_TUPLES
Definition: walreceiver.h:207
@ WALRCV_OK_COPY_OUT
Definition: walreceiver.h:209
@ WALRCV_OK_COPY_BOTH
Definition: walreceiver.h:210
CRSSnapshotAction
Definition: walsender.h:21
@ CRS_USE_SNAPSHOT
Definition: walsender.h:24
@ CRS_NOEXPORT_SNAPSHOT
Definition: walsender.h:23
@ CRS_EXPORT_SNAPSHOT
Definition: walsender.h:22
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
uint64 XLogRecPtr
Definition: xlogdefs.h:21
uint32 TimeLineID
Definition: xlogdefs.h:59