LCOV - code coverage report
Current view: top level - src/backend/replication/libpqwalreceiver - libpqwalreceiver.c (source / functions) Hit Total Coverage
Test: PostgreSQL 19devel Lines: 342 413 82.8 %
Date: 2025-07-29 04:18:44 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             :  * Apart from walreceiver, the libpq-specific routines are now being used by
      10             :  * logical replication workers and slot synchronization.
      11             :  *
      12             :  * Portions Copyright (c) 2010-2025, PostgreSQL Global Development Group
      13             :  *
      14             :  *
      15             :  * IDENTIFICATION
      16             :  *    src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
      17             :  *
      18             :  *-------------------------------------------------------------------------
      19             :  */
      20             : #include "postgres.h"
      21             : 
      22             : #include <unistd.h>
      23             : #include <sys/time.h>
      24             : 
      25             : #include "common/connect.h"
      26             : #include "funcapi.h"
      27             : #include "libpq-fe.h"
      28             : #include "libpq/libpq-be-fe-helpers.h"
      29             : #include "mb/pg_wchar.h"
      30             : #include "miscadmin.h"
      31             : #include "pgstat.h"
      32             : #include "pqexpbuffer.h"
      33             : #include "replication/walreceiver.h"
      34             : #include "storage/latch.h"
      35             : #include "utils/builtins.h"
      36             : #include "utils/memutils.h"
      37             : #include "utils/pg_lsn.h"
      38             : #include "utils/tuplestore.h"
      39             : 
      40        1816 : PG_MODULE_MAGIC_EXT(
      41             :                     .name = "libpqwalreceiver",
      42             :                     .version = PG_VERSION
      43             : );
      44             : 
      45             : struct WalReceiverConn
      46             : {
      47             :     /* Current connection to the primary, if any */
      48             :     PGconn     *streamConn;
      49             :     /* Used to remember if the connection is logical or physical */
      50             :     bool        logical;
      51             :     /* Buffer for currently read records */
      52             :     char       *recvBuf;
      53             : };
      54             : 
      55             : /* Prototypes for interface functions */
      56             : static WalReceiverConn *libpqrcv_connect(const char *conninfo,
      57             :                                          bool replication, bool logical,
      58             :                                          bool must_use_password,
      59             :                                          const char *appname, char **err);
      60             : static void libpqrcv_check_conninfo(const char *conninfo,
      61             :                                     bool must_use_password);
      62             : static char *libpqrcv_get_conninfo(WalReceiverConn *conn);
      63             : static void libpqrcv_get_senderinfo(WalReceiverConn *conn,
      64             :                                     char **sender_host, int *sender_port);
      65             : static char *libpqrcv_identify_system(WalReceiverConn *conn,
      66             :                                       TimeLineID *primary_tli);
      67             : static char *libpqrcv_get_dbname_from_conninfo(const char *connInfo);
      68             : static int  libpqrcv_server_version(WalReceiverConn *conn);
      69             : static void libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
      70             :                                              TimeLineID tli, char **filename,
      71             :                                              char **content, int *len);
      72             : static bool libpqrcv_startstreaming(WalReceiverConn *conn,
      73             :                                     const WalRcvStreamOptions *options);
      74             : static void libpqrcv_endstreaming(WalReceiverConn *conn,
      75             :                                   TimeLineID *next_tli);
      76             : static int  libpqrcv_receive(WalReceiverConn *conn, char **buffer,
      77             :                              pgsocket *wait_fd);
      78             : static void libpqrcv_send(WalReceiverConn *conn, const char *buffer,
      79             :                           int nbytes);
      80             : static char *libpqrcv_create_slot(WalReceiverConn *conn,
      81             :                                   const char *slotname,
      82             :                                   bool temporary,
      83             :                                   bool two_phase,
      84             :                                   bool failover,
      85             :                                   CRSSnapshotAction snapshot_action,
      86             :                                   XLogRecPtr *lsn);
      87             : static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
      88             :                                 const bool *failover, const bool *two_phase);
      89             : static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
      90             : static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
      91             :                                        const char *query,
      92             :                                        const int nRetTypes,
      93             :                                        const Oid *retTypes);
      94             : static void libpqrcv_disconnect(WalReceiverConn *conn);
      95             : 
      96             : static WalReceiverFunctionsType PQWalReceiverFunctions = {
      97             :     .walrcv_connect = libpqrcv_connect,
      98             :     .walrcv_check_conninfo = libpqrcv_check_conninfo,
      99             :     .walrcv_get_conninfo = libpqrcv_get_conninfo,
     100             :     .walrcv_get_senderinfo = libpqrcv_get_senderinfo,
     101             :     .walrcv_identify_system = libpqrcv_identify_system,
     102             :     .walrcv_server_version = libpqrcv_server_version,
     103             :     .walrcv_readtimelinehistoryfile = libpqrcv_readtimelinehistoryfile,
     104             :     .walrcv_startstreaming = libpqrcv_startstreaming,
     105             :     .walrcv_endstreaming = libpqrcv_endstreaming,
     106             :     .walrcv_receive = libpqrcv_receive,
     107             :     .walrcv_send = libpqrcv_send,
     108             :     .walrcv_create_slot = libpqrcv_create_slot,
     109             :     .walrcv_alter_slot = libpqrcv_alter_slot,
     110             :     .walrcv_get_dbname_from_conninfo = libpqrcv_get_dbname_from_conninfo,
     111             :     .walrcv_get_backend_pid = libpqrcv_get_backend_pid,
     112             :     .walrcv_exec = libpqrcv_exec,
     113             :     .walrcv_disconnect = libpqrcv_disconnect
     114             : };
     115             : 
     116             : /* Prototypes for private functions */
     117             : static char *stringlist_to_identifierstr(PGconn *conn, List *strings);
     118             : 
     119             : /*
     120             :  * Module initialization function
     121             :  */
     122             : void
     123        1816 : _PG_init(void)
     124             : {
     125        1816 :     if (WalReceiverFunctions != NULL)
     126           0 :         elog(ERROR, "libpqwalreceiver already loaded");
     127        1816 :     WalReceiverFunctions = &PQWalReceiverFunctions;
     128        1816 : }
     129             : 
     130             : /*
     131             :  * Establish the connection to the primary server.
     132             :  *
     133             :  * This function can be used for both replication and regular connections.
     134             :  * If it is a replication connection, it could be either logical or physical
     135             :  * based on input argument 'logical'.
     136             :  *
     137             :  * If an error occurs, this function will normally return NULL and set *err
     138             :  * to a palloc'ed error message. However, if must_use_password is true and
     139             :  * the connection fails to use the password, this function will ereport(ERROR).
     140             :  * We do this because in that case the error includes a detail and a hint for
     141             :  * consistency with other parts of the system, and it's not worth adding the
     142             :  * machinery to pass all of those back to the caller just to cover this one
     143             :  * case.
     144             :  */
     145             : static WalReceiverConn *
     146        1810 : libpqrcv_connect(const char *conninfo, bool replication, bool logical,
     147             :                  bool must_use_password, const char *appname, char **err)
     148             : {
     149             :     WalReceiverConn *conn;
     150             :     const char *keys[6];
     151             :     const char *vals[6];
     152        1810 :     int         i = 0;
     153             : 
     154             :     /*
     155             :      * Re-validate connection string. The validation already happened at DDL
     156             :      * time, but the subscription owner may have changed. If we don't recheck
     157             :      * with the correct must_use_password, it's possible that the connection
     158             :      * will obtain the password from a different source, such as PGPASSFILE or
     159             :      * PGPASSWORD.
     160             :      */
     161        1810 :     libpqrcv_check_conninfo(conninfo, must_use_password);
     162             : 
     163             :     /*
     164             :      * We use the expand_dbname parameter to process the connection string (or
     165             :      * URI), and pass some extra options.
     166             :      */
     167        1784 :     keys[i] = "dbname";
     168        1784 :     vals[i] = conninfo;
     169             : 
     170             :     /* We can not have logical without replication */
     171             :     Assert(replication || !logical);
     172             : 
     173        1784 :     if (replication)
     174             :     {
     175        1760 :         keys[++i] = "replication";
     176        1760 :         vals[i] = logical ? "database" : "true";
     177             : 
     178        1760 :         if (logical)
     179             :         {
     180             :             /* Tell the publisher to translate to our encoding */
     181        1294 :             keys[++i] = "client_encoding";
     182        1294 :             vals[i] = GetDatabaseEncodingName();
     183             : 
     184             :             /*
     185             :              * Force assorted GUC parameters to settings that ensure that the
     186             :              * publisher will output data values in a form that is unambiguous
     187             :              * to the subscriber.  (We don't want to modify the subscriber's
     188             :              * GUC settings, since that might surprise user-defined code
     189             :              * running in the subscriber, such as triggers.)  This should
     190             :              * match what pg_dump does.
     191             :              */
     192        1294 :             keys[++i] = "options";
     193        1294 :             vals[i] = "-c datestyle=ISO -c intervalstyle=postgres -c extra_float_digits=3";
     194             :         }
     195             :         else
     196             :         {
     197             :             /*
     198             :              * The database name is ignored by the server in replication mode,
     199             :              * but specify "replication" for .pgpass lookup.
     200             :              */
     201         466 :             keys[++i] = "dbname";
     202         466 :             vals[i] = "replication";
     203             :         }
     204             :     }
     205             : 
     206        1784 :     keys[++i] = "fallback_application_name";
     207        1784 :     vals[i] = appname;
     208             : 
     209        1784 :     keys[++i] = NULL;
     210        1784 :     vals[i] = NULL;
     211             : 
     212             :     Assert(i < lengthof(keys));
     213             : 
     214        1784 :     conn = palloc0(sizeof(WalReceiverConn));
     215        1782 :     conn->streamConn =
     216        1784 :         libpqsrv_connect_params(keys, vals,
     217             :                                  /* expand_dbname = */ true,
     218             :                                 WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);
     219             : 
     220        1782 :     if (PQstatus(conn->streamConn) != CONNECTION_OK)
     221         238 :         goto bad_connection_errmsg;
     222             : 
     223        1544 :     if (must_use_password && !PQconnectionUsedPassword(conn->streamConn))
     224             :     {
     225           0 :         libpqsrv_disconnect(conn->streamConn);
     226           0 :         pfree(conn);
     227             : 
     228           0 :         ereport(ERROR,
     229             :                 (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
     230             :                  errmsg("password is required"),
     231             :                  errdetail("Non-superuser cannot connect if the server does not request a password."),
     232             :                  errhint("Target server's authentication method must be changed, or set password_required=false in the subscription parameters.")));
     233             :     }
     234             : 
     235        1544 :     PQsetNoticeReceiver(conn->streamConn, libpqsrv_notice_receiver,
     236             :                         "received message via replication");
     237             : 
     238             :     /*
     239             :      * Set always-secure search path for the cases where the connection is
     240             :      * used to run SQL queries, so malicious users can't get control.
     241             :      */
     242        1544 :     if (!replication || logical)
     243             :     {
     244             :         PGresult   *res;
     245             : 
     246        1268 :         res = libpqsrv_exec(conn->streamConn,
     247             :                             ALWAYS_SECURE_SEARCH_PATH_SQL,
     248             :                             WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
     249        1268 :         if (PQresultStatus(res) != PGRES_TUPLES_OK)
     250             :         {
     251           0 :             PQclear(res);
     252           0 :             *err = psprintf(_("could not clear search path: %s"),
     253           0 :                             pchomp(PQerrorMessage(conn->streamConn)));
     254           0 :             goto bad_connection;
     255             :         }
     256        1268 :         PQclear(res);
     257             :     }
     258             : 
     259        1544 :     conn->logical = logical;
     260             : 
     261        1544 :     return conn;
     262             : 
     263             :     /* error path, using libpq's error message */
     264         238 : bad_connection_errmsg:
     265         238 :     *err = pchomp(PQerrorMessage(conn->streamConn));
     266             : 
     267             :     /* error path, error already set */
     268         238 : bad_connection:
     269         238 :     libpqsrv_disconnect(conn->streamConn);
     270         238 :     pfree(conn);
     271         238 :     return NULL;
     272             : }
     273             : 
     274             : /*
     275             :  * Validate connection info string.
     276             :  *
     277             :  * If the connection string can't be parsed, this function will raise
     278             :  * an error. If must_use_password is true, the function raises an error
     279             :  * if no password is provided in the connection string. In any other case
     280             :  * it successfully completes.
     281             :  */
     282             : static void
     283        2182 : libpqrcv_check_conninfo(const char *conninfo, bool must_use_password)
     284             : {
     285        2182 :     PQconninfoOption *opts = NULL;
     286             :     PQconninfoOption *opt;
     287        2182 :     char       *err = NULL;
     288             : 
     289        2182 :     opts = PQconninfoParse(conninfo, &err);
     290        2182 :     if (opts == NULL)
     291             :     {
     292             :         /* The error string is malloc'd, so we must free it explicitly */
     293          18 :         char       *errcopy = err ? pstrdup(err) : "out of memory";
     294             : 
     295          18 :         PQfreemem(err);
     296          18 :         ereport(ERROR,
     297             :                 (errcode(ERRCODE_SYNTAX_ERROR),
     298             :                  errmsg("invalid connection string syntax: %s", errcopy)));
     299             :     }
     300             : 
     301        2164 :     if (must_use_password)
     302             :     {
     303          44 :         bool        uses_password = false;
     304             : 
     305        1712 :         for (opt = opts; opt->keyword != NULL; ++opt)
     306             :         {
     307             :             /* Ignore connection options that are not present. */
     308        1680 :             if (opt->val == NULL)
     309        1574 :                 continue;
     310             : 
     311         106 :             if (strcmp(opt->keyword, "password") == 0 && opt->val[0] != '\0')
     312             :             {
     313          12 :                 uses_password = true;
     314          12 :                 break;
     315             :             }
     316             :         }
     317             : 
     318          44 :         if (!uses_password)
     319             :         {
     320             :             /* malloc'd, so we must free it explicitly */
     321          32 :             PQconninfoFree(opts);
     322             : 
     323          32 :             ereport(ERROR,
     324             :                     (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
     325             :                      errmsg("password is required"),
     326             :                      errdetail("Non-superusers must provide a password in the connection string.")));
     327             :         }
     328             :     }
     329             : 
     330        2132 :     PQconninfoFree(opts);
     331        2132 : }
     332             : 
     333             : /*
     334             :  * Return a user-displayable conninfo string.  Any security-sensitive fields
     335             :  * are obfuscated.
     336             :  */
     337             : static char *
     338         276 : libpqrcv_get_conninfo(WalReceiverConn *conn)
     339             : {
     340             :     PQconninfoOption *conn_opts;
     341             :     PQconninfoOption *conn_opt;
     342             :     PQExpBufferData buf;
     343             :     char       *retval;
     344             : 
     345             :     Assert(conn->streamConn != NULL);
     346             : 
     347         276 :     initPQExpBuffer(&buf);
     348         276 :     conn_opts = PQconninfo(conn->streamConn);
     349             : 
     350         276 :     if (conn_opts == NULL)
     351           0 :         ereport(ERROR,
     352             :                 (errcode(ERRCODE_OUT_OF_MEMORY),
     353             :                  errmsg("could not parse connection string: %s",
     354             :                         _("out of memory"))));
     355             : 
     356             :     /* build a clean connection string from pieces */
     357       14352 :     for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
     358             :     {
     359             :         bool        obfuscate;
     360             : 
     361             :         /* Skip debug and empty options */
     362       14076 :         if (strchr(conn_opt->dispchar, 'D') ||
     363       12972 :             conn_opt->val == NULL ||
     364        5266 :             conn_opt->val[0] == '\0')
     365        9086 :             continue;
     366             : 
     367             :         /* Obfuscate security-sensitive options */
     368        4990 :         obfuscate = strchr(conn_opt->dispchar, '*') != NULL;
     369             : 
     370        9980 :         appendPQExpBuffer(&buf, "%s%s=%s",
     371        4990 :                           buf.len == 0 ? "" : " ",
     372             :                           conn_opt->keyword,
     373             :                           obfuscate ? "********" : conn_opt->val);
     374             :     }
     375             : 
     376         276 :     PQconninfoFree(conn_opts);
     377             : 
     378         276 :     retval = PQExpBufferDataBroken(buf) ? NULL : pstrdup(buf.data);
     379         276 :     termPQExpBuffer(&buf);
     380         276 :     return retval;
     381             : }
     382             : 
     383             : /*
     384             :  * Provides information of sender this WAL receiver is connected to.
     385             :  */
     386             : static void
     387         276 : libpqrcv_get_senderinfo(WalReceiverConn *conn, char **sender_host,
     388             :                         int *sender_port)
     389             : {
     390         276 :     char       *ret = NULL;
     391             : 
     392         276 :     *sender_host = NULL;
     393         276 :     *sender_port = 0;
     394             : 
     395             :     Assert(conn->streamConn != NULL);
     396             : 
     397         276 :     ret = PQhost(conn->streamConn);
     398         276 :     if (ret && strlen(ret) != 0)
     399         276 :         *sender_host = pstrdup(ret);
     400             : 
     401         276 :     ret = PQport(conn->streamConn);
     402         276 :     if (ret && strlen(ret) != 0)
     403         276 :         *sender_port = atoi(ret);
     404         276 : }
     405             : 
     406             : /*
     407             :  * Check that primary's system identifier matches ours, and fetch the current
     408             :  * timeline ID of the primary.
     409             :  */
     410             : static char *
     411         672 : libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli)
     412             : {
     413             :     PGresult   *res;
     414             :     char       *primary_sysid;
     415             : 
     416             :     /*
     417             :      * Get the system identifier and timeline ID as a DataRow message from the
     418             :      * primary server.
     419             :      */
     420         672 :     res = libpqsrv_exec(conn->streamConn,
     421             :                         "IDENTIFY_SYSTEM",
     422             :                         WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
     423         672 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
     424           0 :         ereport(ERROR,
     425             :                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
     426             :                  errmsg("could not receive database system identifier and timeline ID from "
     427             :                         "the primary server: %s",
     428             :                         pchomp(PQerrorMessage(conn->streamConn)))));
     429             : 
     430             :     /*
     431             :      * IDENTIFY_SYSTEM returns 3 columns in 9.3 and earlier, and 4 columns in
     432             :      * 9.4 and onwards.
     433             :      */
     434         672 :     if (PQnfields(res) < 3 || PQntuples(res) != 1)
     435           0 :         ereport(ERROR,
     436             :                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
     437             :                  errmsg("invalid response from primary server"),
     438             :                  errdetail("Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
     439             :                            PQntuples(res), PQnfields(res), 1, 3)));
     440         672 :     primary_sysid = pstrdup(PQgetvalue(res, 0, 0));
     441         672 :     *primary_tli = pg_strtoint32(PQgetvalue(res, 0, 1));
     442         672 :     PQclear(res);
     443             : 
     444         672 :     return primary_sysid;
     445             : }
     446             : 
     447             : /*
     448             :  * Thin wrapper around libpq to obtain server version.
     449             :  */
     450             : static int
     451        1848 : libpqrcv_server_version(WalReceiverConn *conn)
     452             : {
     453        1848 :     return PQserverVersion(conn->streamConn);
     454             : }
     455             : 
     456             : /*
     457             :  * Get database name from the primary server's conninfo.
     458             :  *
     459             :  * If dbname is not found in connInfo, return NULL value.
     460             :  */
     461             : static char *
     462          26 : libpqrcv_get_dbname_from_conninfo(const char *connInfo)
     463             : {
     464             :     PQconninfoOption *opts;
     465          26 :     char       *dbname = NULL;
     466          26 :     char       *err = NULL;
     467             : 
     468          26 :     opts = PQconninfoParse(connInfo, &err);
     469          26 :     if (opts == NULL)
     470             :     {
     471             :         /* The error string is malloc'd, so we must free it explicitly */
     472           0 :         char       *errcopy = err ? pstrdup(err) : "out of memory";
     473             : 
     474           0 :         PQfreemem(err);
     475           0 :         ereport(ERROR,
     476             :                 (errcode(ERRCODE_SYNTAX_ERROR),
     477             :                  errmsg("invalid connection string syntax: %s", errcopy)));
     478             :     }
     479             : 
     480        1352 :     for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
     481             :     {
     482             :         /*
     483             :          * If multiple dbnames are specified, then the last one will be
     484             :          * returned
     485             :          */
     486        1326 :         if (strcmp(opt->keyword, "dbname") == 0 && opt->val &&
     487          24 :             *opt->val)
     488             :         {
     489          24 :             if (dbname)
     490           0 :                 pfree(dbname);
     491             : 
     492          24 :             dbname = pstrdup(opt->val);
     493             :         }
     494             :     }
     495             : 
     496          26 :     PQconninfoFree(opts);
     497          26 :     return dbname;
     498             : }
     499             : 
     500             : /*
     501             :  * Start streaming WAL data from given streaming options.
     502             :  *
     503             :  * Returns true if we switched successfully to copy-both mode. False
     504             :  * means the server received the command and executed it successfully, but
     505             :  * didn't switch to copy-mode.  That means that there was no WAL on the
     506             :  * requested timeline and starting point, because the server switched to
     507             :  * another timeline at or before the requested starting point. On failure,
     508             :  * throws an ERROR.
     509             :  */
     510             : static bool
     511        1038 : libpqrcv_startstreaming(WalReceiverConn *conn,
     512             :                         const WalRcvStreamOptions *options)
     513             : {
     514             :     StringInfoData cmd;
     515             :     PGresult   *res;
     516             : 
     517             :     Assert(options->logical == conn->logical);
     518             :     Assert(options->slotname || !options->logical);
     519             : 
     520        1038 :     initStringInfo(&cmd);
     521             : 
     522             :     /* Build the command. */
     523        1038 :     appendStringInfoString(&cmd, "START_REPLICATION");
     524        1038 :     if (options->slotname != NULL)
     525         850 :         appendStringInfo(&cmd, " SLOT \"%s\"",
     526         850 :                          options->slotname);
     527             : 
     528        1038 :     if (options->logical)
     529         762 :         appendStringInfoString(&cmd, " LOGICAL");
     530             : 
     531        1038 :     appendStringInfo(&cmd, " %X/%08X", LSN_FORMAT_ARGS(options->startpoint));
     532             : 
     533             :     /*
     534             :      * Additional options are different depending on if we are doing logical
     535             :      * or physical replication.
     536             :      */
     537        1038 :     if (options->logical)
     538             :     {
     539             :         char       *pubnames_str;
     540             :         List       *pubnames;
     541             :         char       *pubnames_literal;
     542             : 
     543         762 :         appendStringInfoString(&cmd, " (");
     544             : 
     545         762 :         appendStringInfo(&cmd, "proto_version '%u'",
     546         762 :                          options->proto.logical.proto_version);
     547             : 
     548         762 :         if (options->proto.logical.streaming_str)
     549         746 :             appendStringInfo(&cmd, ", streaming '%s'",
     550         746 :                              options->proto.logical.streaming_str);
     551             : 
     552         778 :         if (options->proto.logical.twophase &&
     553          16 :             PQserverVersion(conn->streamConn) >= 150000)
     554          16 :             appendStringInfoString(&cmd, ", two_phase 'on'");
     555             : 
     556        1524 :         if (options->proto.logical.origin &&
     557         762 :             PQserverVersion(conn->streamConn) >= 160000)
     558         762 :             appendStringInfo(&cmd, ", origin '%s'",
     559         762 :                              options->proto.logical.origin);
     560             : 
     561         762 :         pubnames = options->proto.logical.publication_names;
     562         762 :         pubnames_str = stringlist_to_identifierstr(conn->streamConn, pubnames);
     563         762 :         if (!pubnames_str)
     564           0 :             ereport(ERROR,
     565             :                     (errcode(ERRCODE_OUT_OF_MEMORY),    /* likely guess */
     566             :                      errmsg("could not start WAL streaming: %s",
     567             :                             pchomp(PQerrorMessage(conn->streamConn)))));
     568         762 :         pubnames_literal = PQescapeLiteral(conn->streamConn, pubnames_str,
     569             :                                            strlen(pubnames_str));
     570         762 :         if (!pubnames_literal)
     571           0 :             ereport(ERROR,
     572             :                     (errcode(ERRCODE_OUT_OF_MEMORY),    /* likely guess */
     573             :                      errmsg("could not start WAL streaming: %s",
     574             :                             pchomp(PQerrorMessage(conn->streamConn)))));
     575         762 :         appendStringInfo(&cmd, ", publication_names %s", pubnames_literal);
     576         762 :         PQfreemem(pubnames_literal);
     577         762 :         pfree(pubnames_str);
     578             : 
     579         782 :         if (options->proto.logical.binary &&
     580          20 :             PQserverVersion(conn->streamConn) >= 140000)
     581          20 :             appendStringInfoString(&cmd, ", binary 'true'");
     582             : 
     583         762 :         appendStringInfoChar(&cmd, ')');
     584             :     }
     585             :     else
     586         276 :         appendStringInfo(&cmd, " TIMELINE %u",
     587         276 :                          options->proto.physical.startpointTLI);
     588             : 
     589             :     /* Start streaming. */
     590        1038 :     res = libpqsrv_exec(conn->streamConn,
     591        1038 :                         cmd.data,
     592             :                         WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
     593        1038 :     pfree(cmd.data);
     594             : 
     595        1038 :     if (PQresultStatus(res) == PGRES_COMMAND_OK)
     596             :     {
     597           0 :         PQclear(res);
     598           0 :         return false;
     599             :     }
     600        1038 :     else if (PQresultStatus(res) != PGRES_COPY_BOTH)
     601           2 :         ereport(ERROR,
     602             :                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
     603             :                  errmsg("could not start WAL streaming: %s",
     604             :                         pchomp(PQerrorMessage(conn->streamConn)))));
     605        1036 :     PQclear(res);
     606        1036 :     return true;
     607             : }
     608             : 
     609             : /*
     610             :  * Stop streaming WAL data. Returns the next timeline's ID in *next_tli, as
     611             :  * reported by the server, or 0 if it did not report it.
     612             :  */
     613             : static void
     614         458 : libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
     615             : {
     616             :     PGresult   *res;
     617             : 
     618             :     /*
     619             :      * Send copy-end message.  As in libpqsrv_exec, this could theoretically
     620             :      * block, but the risk seems small.
     621             :      */
     622         846 :     if (PQputCopyEnd(conn->streamConn, NULL) <= 0 ||
     623         388 :         PQflush(conn->streamConn))
     624          70 :         ereport(ERROR,
     625             :                 (errcode(ERRCODE_CONNECTION_FAILURE),
     626             :                  errmsg("could not send end-of-streaming message to primary: %s",
     627             :                         pchomp(PQerrorMessage(conn->streamConn)))));
     628             : 
     629         388 :     *next_tli = 0;
     630             : 
     631             :     /*
     632             :      * After COPY is finished, we should receive a result set indicating the
     633             :      * next timeline's ID, or just CommandComplete if the server was shut
     634             :      * down.
     635             :      *
     636             :      * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
     637             :      * also possible in case we aborted the copy in mid-stream.
     638             :      */
     639         388 :     res = libpqsrv_get_result(conn->streamConn,
     640             :                               WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
     641         388 :     if (PQresultStatus(res) == PGRES_TUPLES_OK)
     642             :     {
     643             :         /*
     644             :          * Read the next timeline's ID. The server also sends the timeline's
     645             :          * starting point, but it is ignored.
     646             :          */
     647          22 :         if (PQnfields(res) < 2 || PQntuples(res) != 1)
     648           0 :             ereport(ERROR,
     649             :                     (errcode(ERRCODE_PROTOCOL_VIOLATION),
     650             :                      errmsg("unexpected result set after end-of-streaming")));
     651          22 :         *next_tli = pg_strtoint32(PQgetvalue(res, 0, 0));
     652          22 :         PQclear(res);
     653             : 
     654             :         /* the result set should be followed by CommandComplete */
     655          22 :         res = libpqsrv_get_result(conn->streamConn,
     656             :                                   WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
     657             :     }
     658         366 :     else if (PQresultStatus(res) == PGRES_COPY_OUT)
     659             :     {
     660         366 :         PQclear(res);
     661             : 
     662             :         /* End the copy */
     663         366 :         if (PQendcopy(conn->streamConn))
     664           0 :             ereport(ERROR,
     665             :                     (errcode(ERRCODE_CONNECTION_FAILURE),
     666             :                      errmsg("error while shutting down streaming COPY: %s",
     667             :                             pchomp(PQerrorMessage(conn->streamConn)))));
     668             : 
     669             :         /* CommandComplete should follow */
     670         366 :         res = libpqsrv_get_result(conn->streamConn,
     671             :                                   WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
     672             :     }
     673             : 
     674         388 :     if (PQresultStatus(res) != PGRES_COMMAND_OK)
     675           0 :         ereport(ERROR,
     676             :                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
     677             :                  errmsg("error reading result of streaming command: %s",
     678             :                         pchomp(PQerrorMessage(conn->streamConn)))));
     679         388 :     PQclear(res);
     680             : 
     681             :     /* Verify that there are no more results */
     682         388 :     res = libpqsrv_get_result(conn->streamConn,
     683             :                               WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
     684         388 :     if (res != NULL)
     685           0 :         ereport(ERROR,
     686             :                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
     687             :                  errmsg("unexpected result after CommandComplete: %s",
     688             :                         pchomp(PQerrorMessage(conn->streamConn)))));
     689         388 : }
     690             : 
     691             : /*
     692             :  * Fetch the timeline history file for 'tli' from primary.
     693             :  */
     694             : static void
     695          20 : libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
     696             :                                  TimeLineID tli, char **filename,
     697             :                                  char **content, int *len)
     698             : {
     699             :     PGresult   *res;
     700             :     char        cmd[64];
     701             : 
     702             :     Assert(!conn->logical);
     703             : 
     704             :     /*
     705             :      * Request the primary to send over the history file for given timeline.
     706             :      */
     707          20 :     snprintf(cmd, sizeof(cmd), "TIMELINE_HISTORY %u", tli);
     708          20 :     res = libpqsrv_exec(conn->streamConn,
     709             :                         cmd,
     710             :                         WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
     711          20 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
     712           0 :         ereport(ERROR,
     713             :                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
     714             :                  errmsg("could not receive timeline history file from "
     715             :                         "the primary server: %s",
     716             :                         pchomp(PQerrorMessage(conn->streamConn)))));
     717          20 :     if (PQnfields(res) != 2 || PQntuples(res) != 1)
     718           0 :         ereport(ERROR,
     719             :                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
     720             :                  errmsg("invalid response from primary server"),
     721             :                  errdetail("Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
     722             :                            PQntuples(res), PQnfields(res))));
     723          20 :     *filename = pstrdup(PQgetvalue(res, 0, 0));
     724             : 
     725          20 :     *len = PQgetlength(res, 0, 1);
     726          20 :     *content = palloc(*len);
     727          20 :     memcpy(*content, PQgetvalue(res, 0, 1), *len);
     728          20 :     PQclear(res);
     729          20 : }
     730             : 
     731             : /*
     732             :  * Disconnect connection to primary, if any.
     733             :  */
     734             : static void
     735        1544 : libpqrcv_disconnect(WalReceiverConn *conn)
     736             : {
     737        1544 :     libpqsrv_disconnect(conn->streamConn);
     738        1544 :     PQfreemem(conn->recvBuf);
     739        1544 :     pfree(conn);
     740        1544 : }
     741             : 
     742             : /*
     743             :  * Receive a message available from XLOG stream.
     744             :  *
     745             :  * Returns:
     746             :  *
     747             :  *   If data was received, returns the length of the data. *buffer is set to
     748             :  *   point to a buffer holding the received message. The buffer is only valid
     749             :  *   until the next libpqrcv_* call.
     750             :  *
     751             :  *   If no data was available immediately, returns 0, and *wait_fd is set to a
     752             :  *   socket descriptor which can be waited on before trying again.
     753             :  *
     754             :  *   -1 if the server ended the COPY.
     755             :  *
     756             :  * ereports on error.
     757             :  */
     758             : static int
     759      795774 : libpqrcv_receive(WalReceiverConn *conn, char **buffer,
     760             :                  pgsocket *wait_fd)
     761             : {
     762             :     int         rawlen;
     763             : 
     764      795774 :     PQfreemem(conn->recvBuf);
     765      795774 :     conn->recvBuf = NULL;
     766             : 
     767             :     /* Try to receive a CopyData message */
     768      795774 :     rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
     769      795774 :     if (rawlen == 0)
     770             :     {
     771             :         /* Try consuming some data. */
     772      540236 :         if (PQconsumeInput(conn->streamConn) == 0)
     773          76 :             ereport(ERROR,
     774             :                     (errcode(ERRCODE_CONNECTION_FAILURE),
     775             :                      errmsg("could not receive data from WAL stream: %s",
     776             :                             pchomp(PQerrorMessage(conn->streamConn)))));
     777             : 
     778             :         /* Now that we've consumed some input, try again */
     779      540160 :         rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
     780      540160 :         if (rawlen == 0)
     781             :         {
     782             :             /* Tell caller to try again when our socket is ready. */
     783      208234 :             *wait_fd = PQsocket(conn->streamConn);
     784      208234 :             return 0;
     785             :         }
     786             :     }
     787      587464 :     if (rawlen == -1)           /* end-of-streaming or error */
     788             :     {
     789             :         PGresult   *res;
     790             : 
     791         476 :         res = libpqsrv_get_result(conn->streamConn,
     792             :                                   WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
     793         476 :         if (PQresultStatus(res) == PGRES_COMMAND_OK)
     794             :         {
     795         448 :             PQclear(res);
     796             : 
     797             :             /* Verify that there are no more results. */
     798         448 :             res = libpqsrv_get_result(conn->streamConn,
     799             :                                       WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
     800         448 :             if (res != NULL)
     801             :             {
     802          70 :                 PQclear(res);
     803             : 
     804             :                 /*
     805             :                  * If the other side closed the connection orderly (otherwise
     806             :                  * we'd seen an error, or PGRES_COPY_IN) don't report an error
     807             :                  * here, but let callers deal with it.
     808             :                  */
     809          70 :                 if (PQstatus(conn->streamConn) == CONNECTION_BAD)
     810          70 :                     return -1;
     811             : 
     812           0 :                 ereport(ERROR,
     813             :                         (errcode(ERRCODE_PROTOCOL_VIOLATION),
     814             :                          errmsg("unexpected result after CommandComplete: %s",
     815             :                                 PQerrorMessage(conn->streamConn))));
     816             :             }
     817             : 
     818         378 :             return -1;
     819             :         }
     820          28 :         else if (PQresultStatus(res) == PGRES_COPY_IN)
     821             :         {
     822          22 :             PQclear(res);
     823          22 :             return -1;
     824             :         }
     825             :         else
     826           6 :             ereport(ERROR,
     827             :                     (errcode(ERRCODE_PROTOCOL_VIOLATION),
     828             :                      errmsg("could not receive data from WAL stream: %s",
     829             :                             pchomp(PQerrorMessage(conn->streamConn)))));
     830             :     }
     831      586988 :     if (rawlen < -1)
     832           0 :         ereport(ERROR,
     833             :                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
     834             :                  errmsg("could not receive data from WAL stream: %s",
     835             :                         pchomp(PQerrorMessage(conn->streamConn)))));
     836             : 
     837             :     /* Return received messages to caller */
     838      586988 :     *buffer = conn->recvBuf;
     839      586988 :     return rawlen;
     840             : }
     841             : 
     842             : /*
     843             :  * Send a message to XLOG stream.
     844             :  *
     845             :  * ereports on error.
     846             :  */
     847             : static void
     848      217958 : libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes)
     849             : {
     850      435916 :     if (PQputCopyData(conn->streamConn, buffer, nbytes) <= 0 ||
     851      217958 :         PQflush(conn->streamConn))
     852           0 :         ereport(ERROR,
     853             :                 (errcode(ERRCODE_CONNECTION_FAILURE),
     854             :                  errmsg("could not send data to WAL stream: %s",
     855             :                         pchomp(PQerrorMessage(conn->streamConn)))));
     856      217958 : }
     857             : 
     858             : /*
     859             :  * Create new replication slot.
     860             :  * Returns the name of the exported snapshot for logical slot or NULL for
     861             :  * physical slot.
     862             :  */
     863             : static char *
     864         612 : libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
     865             :                      bool temporary, bool two_phase, bool failover,
     866             :                      CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
     867             : {
     868             :     PGresult   *res;
     869             :     StringInfoData cmd;
     870             :     char       *snapshot;
     871             :     int         use_new_options_syntax;
     872             : 
     873         612 :     use_new_options_syntax = (PQserverVersion(conn->streamConn) >= 150000);
     874             : 
     875         612 :     initStringInfo(&cmd);
     876             : 
     877         612 :     appendStringInfo(&cmd, "CREATE_REPLICATION_SLOT \"%s\"", slotname);
     878             : 
     879         612 :     if (temporary)
     880           0 :         appendStringInfoString(&cmd, " TEMPORARY");
     881             : 
     882         612 :     if (conn->logical)
     883             :     {
     884         612 :         appendStringInfoString(&cmd, " LOGICAL pgoutput ");
     885         612 :         if (use_new_options_syntax)
     886         612 :             appendStringInfoChar(&cmd, '(');
     887         612 :         if (two_phase)
     888             :         {
     889           2 :             appendStringInfoString(&cmd, "TWO_PHASE");
     890           2 :             if (use_new_options_syntax)
     891           2 :                 appendStringInfoString(&cmd, ", ");
     892             :             else
     893           0 :                 appendStringInfoChar(&cmd, ' ');
     894             :         }
     895             : 
     896         612 :         if (failover)
     897             :         {
     898          12 :             appendStringInfoString(&cmd, "FAILOVER");
     899          12 :             if (use_new_options_syntax)
     900          12 :                 appendStringInfoString(&cmd, ", ");
     901             :             else
     902           0 :                 appendStringInfoChar(&cmd, ' ');
     903             :         }
     904             : 
     905         612 :         if (use_new_options_syntax)
     906             :         {
     907         612 :             switch (snapshot_action)
     908             :             {
     909           0 :                 case CRS_EXPORT_SNAPSHOT:
     910           0 :                     appendStringInfoString(&cmd, "SNAPSHOT 'export'");
     911           0 :                     break;
     912         224 :                 case CRS_NOEXPORT_SNAPSHOT:
     913         224 :                     appendStringInfoString(&cmd, "SNAPSHOT 'nothing'");
     914         224 :                     break;
     915         388 :                 case CRS_USE_SNAPSHOT:
     916         388 :                     appendStringInfoString(&cmd, "SNAPSHOT 'use'");
     917         388 :                     break;
     918             :             }
     919             :         }
     920             :         else
     921             :         {
     922           0 :             switch (snapshot_action)
     923             :             {
     924           0 :                 case CRS_EXPORT_SNAPSHOT:
     925           0 :                     appendStringInfoString(&cmd, "EXPORT_SNAPSHOT");
     926           0 :                     break;
     927           0 :                 case CRS_NOEXPORT_SNAPSHOT:
     928           0 :                     appendStringInfoString(&cmd, "NOEXPORT_SNAPSHOT");
     929           0 :                     break;
     930           0 :                 case CRS_USE_SNAPSHOT:
     931           0 :                     appendStringInfoString(&cmd, "USE_SNAPSHOT");
     932           0 :                     break;
     933             :             }
     934             :         }
     935             : 
     936         612 :         if (use_new_options_syntax)
     937         612 :             appendStringInfoChar(&cmd, ')');
     938             :     }
     939             :     else
     940             :     {
     941           0 :         if (use_new_options_syntax)
     942           0 :             appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
     943             :         else
     944           0 :             appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
     945             :     }
     946             : 
     947         612 :     res = libpqsrv_exec(conn->streamConn,
     948         612 :                         cmd.data,
     949             :                         WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
     950         612 :     pfree(cmd.data);
     951             : 
     952         612 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
     953           0 :         ereport(ERROR,
     954             :                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
     955             :                  errmsg("could not create replication slot \"%s\": %s",
     956             :                         slotname, pchomp(PQerrorMessage(conn->streamConn)))));
     957             : 
     958         612 :     if (lsn)
     959         388 :         *lsn = DatumGetLSN(DirectFunctionCall1Coll(pg_lsn_in, InvalidOid,
     960         388 :                                                    CStringGetDatum(PQgetvalue(res, 0, 1))));
     961             : 
     962         612 :     if (!PQgetisnull(res, 0, 2))
     963           0 :         snapshot = pstrdup(PQgetvalue(res, 0, 2));
     964             :     else
     965         612 :         snapshot = NULL;
     966             : 
     967         612 :     PQclear(res);
     968             : 
     969         612 :     return snapshot;
     970             : }
     971             : 
     972             : /*
     973             :  * Change the definition of the replication slot.
     974             :  */
     975             : static void
     976           8 : libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
     977             :                     const bool *failover, const bool *two_phase)
     978             : {
     979             :     StringInfoData cmd;
     980             :     PGresult   *res;
     981             : 
     982           8 :     initStringInfo(&cmd);
     983           8 :     appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( ",
     984             :                      quote_identifier(slotname));
     985             : 
     986           8 :     if (failover)
     987           6 :         appendStringInfo(&cmd, "FAILOVER %s",
     988           6 :                          *failover ? "true" : "false");
     989             : 
     990           8 :     if (failover && two_phase)
     991           0 :         appendStringInfoString(&cmd, ", ");
     992             : 
     993           8 :     if (two_phase)
     994           2 :         appendStringInfo(&cmd, "TWO_PHASE %s",
     995           2 :                          *two_phase ? "true" : "false");
     996             : 
     997           8 :     appendStringInfoString(&cmd, " );");
     998             : 
     999           8 :     res = libpqsrv_exec(conn->streamConn, cmd.data,
    1000             :                         WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
    1001           8 :     pfree(cmd.data);
    1002             : 
    1003           8 :     if (PQresultStatus(res) != PGRES_COMMAND_OK)
    1004           0 :         ereport(ERROR,
    1005             :                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
    1006             :                  errmsg("could not alter replication slot \"%s\": %s",
    1007             :                         slotname, pchomp(PQerrorMessage(conn->streamConn)))));
    1008             : 
    1009           8 :     PQclear(res);
    1010           8 : }
    1011             : 
    1012             : /*
    1013             :  * Return PID of remote backend process.
    1014             :  */
    1015             : static pid_t
    1016           0 : libpqrcv_get_backend_pid(WalReceiverConn *conn)
    1017             : {
    1018           0 :     return PQbackendPID(conn->streamConn);
    1019             : }
    1020             : 
    1021             : /*
    1022             :  * Convert tuple query result to tuplestore.
    1023             :  */
    1024             : static void
    1025        2212 : libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
    1026             :                        const int nRetTypes, const Oid *retTypes)
    1027             : {
    1028             :     int         tupn;
    1029             :     int         coln;
    1030        2212 :     int         nfields = PQnfields(pgres);
    1031             :     HeapTuple   tuple;
    1032             :     AttInMetadata *attinmeta;
    1033             :     MemoryContext rowcontext;
    1034             :     MemoryContext oldcontext;
    1035             : 
    1036             :     /* Make sure we got expected number of fields. */
    1037        2212 :     if (nfields != nRetTypes)
    1038           0 :         ereport(ERROR,
    1039             :                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
    1040             :                  errmsg("invalid query response"),
    1041             :                  errdetail("Expected %d fields, got %d fields.",
    1042             :                            nRetTypes, nfields)));
    1043             : 
    1044        2212 :     walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
    1045             : 
    1046             :     /* Create tuple descriptor corresponding to expected result. */
    1047        2212 :     walres->tupledesc = CreateTemplateTupleDesc(nRetTypes);
    1048        7782 :     for (coln = 0; coln < nRetTypes; coln++)
    1049        5570 :         TupleDescInitEntry(walres->tupledesc, (AttrNumber) coln + 1,
    1050        5570 :                            PQfname(pgres, coln), retTypes[coln], -1, 0);
    1051        2212 :     attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
    1052             : 
    1053             :     /* No point in doing more here if there were no tuples returned. */
    1054        2212 :     if (PQntuples(pgres) == 0)
    1055          46 :         return;
    1056             : 
    1057             :     /* Create temporary context for local allocations. */
    1058        2166 :     rowcontext = AllocSetContextCreate(CurrentMemoryContext,
    1059             :                                        "libpqrcv query result context",
    1060             :                                        ALLOCSET_DEFAULT_SIZES);
    1061             : 
    1062             :     /* Process returned rows. */
    1063        5034 :     for (tupn = 0; tupn < PQntuples(pgres); tupn++)
    1064             :     {
    1065             :         char       *cstrs[MaxTupleAttributeNumber];
    1066             : 
    1067        2868 :         CHECK_FOR_INTERRUPTS();
    1068             : 
    1069             :         /* Do the allocations in temporary context. */
    1070        2868 :         oldcontext = MemoryContextSwitchTo(rowcontext);
    1071             : 
    1072             :         /*
    1073             :          * Fill cstrs with null-terminated strings of column values.
    1074             :          */
    1075       11258 :         for (coln = 0; coln < nfields; coln++)
    1076             :         {
    1077        8390 :             if (PQgetisnull(pgres, tupn, coln))
    1078        1174 :                 cstrs[coln] = NULL;
    1079             :             else
    1080        7216 :                 cstrs[coln] = PQgetvalue(pgres, tupn, coln);
    1081             :         }
    1082             : 
    1083             :         /* Convert row to a tuple, and add it to the tuplestore */
    1084        2868 :         tuple = BuildTupleFromCStrings(attinmeta, cstrs);
    1085        2868 :         tuplestore_puttuple(walres->tuplestore, tuple);
    1086             : 
    1087             :         /* Clean up */
    1088        2868 :         MemoryContextSwitchTo(oldcontext);
    1089        2868 :         MemoryContextReset(rowcontext);
    1090             :     }
    1091             : 
    1092        2166 :     MemoryContextDelete(rowcontext);
    1093             : }
    1094             : 
    1095             : /*
    1096             :  * Public interface for sending generic queries (and commands).
    1097             :  *
    1098             :  * This can only be called from process connected to database.
    1099             :  */
    1100             : static WalRcvExecResult *
    1101        3878 : libpqrcv_exec(WalReceiverConn *conn, const char *query,
    1102             :               const int nRetTypes, const Oid *retTypes)
    1103             : {
    1104        3878 :     PGresult   *pgres = NULL;
    1105        3878 :     WalRcvExecResult *walres = palloc0(sizeof(WalRcvExecResult));
    1106             :     char       *diag_sqlstate;
    1107             : 
    1108        3878 :     if (MyDatabaseId == InvalidOid)
    1109           0 :         ereport(ERROR,
    1110             :                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    1111             :                  errmsg("the query interface requires a database connection")));
    1112             : 
    1113        3878 :     pgres = libpqsrv_exec(conn->streamConn,
    1114             :                           query,
    1115             :                           WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
    1116             : 
    1117        3878 :     switch (PQresultStatus(pgres))
    1118             :     {
    1119        2212 :         case PGRES_TUPLES_OK:
    1120             :         case PGRES_SINGLE_TUPLE:
    1121             :         case PGRES_TUPLES_CHUNK:
    1122        2212 :             walres->status = WALRCV_OK_TUPLES;
    1123        2212 :             libpqrcv_processTuples(pgres, walres, nRetTypes, retTypes);
    1124        2212 :             break;
    1125             : 
    1126           0 :         case PGRES_COPY_IN:
    1127           0 :             walres->status = WALRCV_OK_COPY_IN;
    1128           0 :             break;
    1129             : 
    1130         382 :         case PGRES_COPY_OUT:
    1131         382 :             walres->status = WALRCV_OK_COPY_OUT;
    1132         382 :             break;
    1133             : 
    1134           0 :         case PGRES_COPY_BOTH:
    1135           0 :             walres->status = WALRCV_OK_COPY_BOTH;
    1136           0 :             break;
    1137             : 
    1138        1280 :         case PGRES_COMMAND_OK:
    1139        1280 :             walres->status = WALRCV_OK_COMMAND;
    1140        1280 :             break;
    1141             : 
    1142             :             /* Empty query is considered error. */
    1143           0 :         case PGRES_EMPTY_QUERY:
    1144           0 :             walres->status = WALRCV_ERROR;
    1145           0 :             walres->err = _("empty query");
    1146           0 :             break;
    1147             : 
    1148           0 :         case PGRES_PIPELINE_SYNC:
    1149             :         case PGRES_PIPELINE_ABORTED:
    1150           0 :             walres->status = WALRCV_ERROR;
    1151           0 :             walres->err = _("unexpected pipeline mode");
    1152           0 :             break;
    1153             : 
    1154           4 :         case PGRES_NONFATAL_ERROR:
    1155             :         case PGRES_FATAL_ERROR:
    1156             :         case PGRES_BAD_RESPONSE:
    1157           4 :             walres->status = WALRCV_ERROR;
    1158           4 :             walres->err = pchomp(PQerrorMessage(conn->streamConn));
    1159           4 :             diag_sqlstate = PQresultErrorField(pgres, PG_DIAG_SQLSTATE);
    1160           4 :             if (diag_sqlstate)
    1161           4 :                 walres->sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
    1162             :                                                  diag_sqlstate[1],
    1163             :                                                  diag_sqlstate[2],
    1164             :                                                  diag_sqlstate[3],
    1165             :                                                  diag_sqlstate[4]);
    1166           4 :             break;
    1167             :     }
    1168             : 
    1169        3878 :     PQclear(pgres);
    1170             : 
    1171        3878 :     return walres;
    1172             : }
    1173             : 
    1174             : /*
    1175             :  * Given a List of strings, return it as single comma separated
    1176             :  * string, quoting identifiers as needed.
    1177             :  *
    1178             :  * This is essentially the reverse of SplitIdentifierString.
    1179             :  *
    1180             :  * The caller should free the result.
    1181             :  */
    1182             : static char *
    1183         762 : stringlist_to_identifierstr(PGconn *conn, List *strings)
    1184             : {
    1185             :     ListCell   *lc;
    1186             :     StringInfoData res;
    1187         762 :     bool        first = true;
    1188             : 
    1189         762 :     initStringInfo(&res);
    1190             : 
    1191        2016 :     foreach(lc, strings)
    1192             :     {
    1193        1254 :         char       *val = strVal(lfirst(lc));
    1194             :         char       *val_escaped;
    1195             : 
    1196        1254 :         if (first)
    1197         762 :             first = false;
    1198             :         else
    1199         492 :             appendStringInfoChar(&res, ',');
    1200             : 
    1201        1254 :         val_escaped = PQescapeIdentifier(conn, val, strlen(val));
    1202        1254 :         if (!val_escaped)
    1203             :         {
    1204           0 :             free(res.data);
    1205           0 :             return NULL;
    1206             :         }
    1207        1254 :         appendStringInfoString(&res, val_escaped);
    1208        1254 :         PQfreemem(val_escaped);
    1209             :     }
    1210             : 
    1211         762 :     return res.data;
    1212             : }

Generated by: LCOV version 1.16