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