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