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

Generated by: LCOV version 1.13