LCOV - code coverage report
Current view: top level - src/backend/replication/libpqwalreceiver - libpqwalreceiver.c (source / functions) Hit Total Coverage
Test: PostgreSQL 13devel Lines: 272 330 82.4 %
Date: 2019-11-15 22:06:47 Functions: 20 20 100.0 %
Legend: Lines: hit not hit

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

Generated by: LCOV version 1.13