LCOV - code coverage report
Current view: top level - contrib/dblink - dblink.c (source / functions) Hit Total Coverage
Test: PostgreSQL 13beta1 Lines: 944 1093 86.4 %
Date: 2020-06-01 10:07:15 Functions: 74 76 97.4 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*
       2             :  * dblink.c
       3             :  *
       4             :  * Functions returning results from a remote database
       5             :  *
       6             :  * Joe Conway <mail@joeconway.com>
       7             :  * And contributors:
       8             :  * Darko Prenosil <Darko.Prenosil@finteh.hr>
       9             :  * Shridhar Daithankar <shridhar_daithankar@persistent.co.in>
      10             :  *
      11             :  * contrib/dblink/dblink.c
      12             :  * Copyright (c) 2001-2020, PostgreSQL Global Development Group
      13             :  * ALL RIGHTS RESERVED;
      14             :  *
      15             :  * Permission to use, copy, modify, and distribute this software and its
      16             :  * documentation for any purpose, without fee, and without a written agreement
      17             :  * is hereby granted, provided that the above copyright notice and this
      18             :  * paragraph and the following two paragraphs appear in all copies.
      19             :  *
      20             :  * IN NO EVENT SHALL THE AUTHOR OR DISTRIBUTORS BE LIABLE TO ANY PARTY FOR
      21             :  * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING
      22             :  * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS
      23             :  * DOCUMENTATION, EVEN IF THE AUTHOR OR DISTRIBUTORS HAVE BEEN ADVISED OF THE
      24             :  * POSSIBILITY OF SUCH DAMAGE.
      25             :  *
      26             :  * THE AUTHOR AND DISTRIBUTORS SPECIFICALLY DISCLAIMS ANY WARRANTIES,
      27             :  * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
      28             :  * AND FITNESS FOR A PARTICULAR PURPOSE.  THE SOFTWARE PROVIDED HEREUNDER IS
      29             :  * ON AN "AS IS" BASIS, AND THE AUTHOR AND DISTRIBUTORS HAS NO OBLIGATIONS TO
      30             :  * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
      31             :  *
      32             :  */
      33             : #include "postgres.h"
      34             : 
      35             : #include <limits.h>
      36             : 
      37             : #include "access/htup_details.h"
      38             : #include "access/relation.h"
      39             : #include "access/reloptions.h"
      40             : #include "access/table.h"
      41             : #include "catalog/indexing.h"
      42             : #include "catalog/namespace.h"
      43             : #include "catalog/pg_foreign_data_wrapper.h"
      44             : #include "catalog/pg_foreign_server.h"
      45             : #include "catalog/pg_type.h"
      46             : #include "catalog/pg_user_mapping.h"
      47             : #include "executor/spi.h"
      48             : #include "foreign/foreign.h"
      49             : #include "funcapi.h"
      50             : #include "lib/stringinfo.h"
      51             : #include "libpq-fe.h"
      52             : #include "mb/pg_wchar.h"
      53             : #include "miscadmin.h"
      54             : #include "parser/scansup.h"
      55             : #include "utils/acl.h"
      56             : #include "utils/builtins.h"
      57             : #include "utils/fmgroids.h"
      58             : #include "utils/guc.h"
      59             : #include "utils/lsyscache.h"
      60             : #include "utils/memutils.h"
      61             : #include "utils/rel.h"
      62             : #include "utils/varlena.h"
      63             : 
      64           6 : PG_MODULE_MAGIC;
      65             : 
      66             : typedef struct remoteConn
      67             : {
      68             :     PGconn     *conn;           /* Hold the remote connection */
      69             :     int         openCursorCount;    /* The number of open cursors */
      70             :     bool        newXactForCursor;   /* Opened a transaction for a cursor */
      71             : } remoteConn;
      72             : 
      73             : typedef struct storeInfo
      74             : {
      75             :     FunctionCallInfo fcinfo;
      76             :     Tuplestorestate *tuplestore;
      77             :     AttInMetadata *attinmeta;
      78             :     MemoryContext tmpcontext;
      79             :     char      **cstrs;
      80             :     /* temp storage for results to avoid leaks on exception */
      81             :     PGresult   *last_res;
      82             :     PGresult   *cur_res;
      83             : } storeInfo;
      84             : 
      85             : /*
      86             :  * Internal declarations
      87             :  */
      88             : static Datum dblink_record_internal(FunctionCallInfo fcinfo, bool is_async);
      89             : static void prepTuplestoreResult(FunctionCallInfo fcinfo);
      90             : static void materializeResult(FunctionCallInfo fcinfo, PGconn *conn,
      91             :                               PGresult *res);
      92             : static void materializeQueryResult(FunctionCallInfo fcinfo,
      93             :                                    PGconn *conn,
      94             :                                    const char *conname,
      95             :                                    const char *sql,
      96             :                                    bool fail);
      97             : static PGresult *storeQueryResult(volatile storeInfo *sinfo, PGconn *conn, const char *sql);
      98             : static void storeRow(volatile storeInfo *sinfo, PGresult *res, bool first);
      99             : static remoteConn *getConnectionByName(const char *name);
     100             : static HTAB *createConnHash(void);
     101             : static void createNewConnection(const char *name, remoteConn *rconn);
     102             : static void deleteConnection(const char *name);
     103             : static char **get_pkey_attnames(Relation rel, int16 *indnkeyatts);
     104             : static char **get_text_array_contents(ArrayType *array, int *numitems);
     105             : static char *get_sql_insert(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals, char **tgt_pkattvals);
     106             : static char *get_sql_delete(Relation rel, int *pkattnums, int pknumatts, char **tgt_pkattvals);
     107             : static char *get_sql_update(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals, char **tgt_pkattvals);
     108             : static char *quote_ident_cstr(char *rawstr);
     109             : static int  get_attnum_pk_pos(int *pkattnums, int pknumatts, int key);
     110             : static HeapTuple get_tuple_of_interest(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals);
     111             : static Relation get_rel_from_relname(text *relname_text, LOCKMODE lockmode, AclMode aclmode);
     112             : static char *generate_relation_name(Relation rel);
     113             : static void dblink_connstr_check(const char *connstr);
     114             : static void dblink_security_check(PGconn *conn, remoteConn *rconn);
     115             : static void dblink_res_error(PGconn *conn, const char *conname, PGresult *res,
     116             :                              bool fail, const char *fmt,...) pg_attribute_printf(5, 6);
     117             : static char *get_connect_string(const char *servername);
     118             : static char *escape_param_str(const char *from);
     119             : static void validate_pkattnums(Relation rel,
     120             :                                int2vector *pkattnums_arg, int32 pknumatts_arg,
     121             :                                int **pkattnums, int *pknumatts);
     122             : static bool is_valid_dblink_option(const PQconninfoOption *options,
     123             :                                    const char *option, Oid context);
     124             : static int  applyRemoteGucs(PGconn *conn);
     125             : static void restoreLocalGucs(int nestlevel);
     126             : 
     127             : /* Global */
     128             : static remoteConn *pconn = NULL;
     129             : static HTAB *remoteConnHash = NULL;
     130             : 
     131             : /*
     132             :  *  Following is list that holds multiple remote connections.
     133             :  *  Calling convention of each dblink function changes to accept
     134             :  *  connection name as the first parameter. The connection list is
     135             :  *  much like ecpg e.g. a mapping between a name and a PGconn object.
     136             :  */
     137             : 
     138             : typedef struct remoteConnHashEnt
     139             : {
     140             :     char        name[NAMEDATALEN];
     141             :     remoteConn *rconn;
     142             : } remoteConnHashEnt;
     143             : 
     144             : /* initial number of connection hashes */
     145             : #define NUMCONN 16
     146             : 
     147             : static char *
     148          96 : xpstrdup(const char *in)
     149             : {
     150          96 :     if (in == NULL)
     151          72 :         return NULL;
     152          24 :     return pstrdup(in);
     153             : }
     154             : 
     155             : static void
     156             : pg_attribute_noreturn()
     157           0 : dblink_res_internalerror(PGconn *conn, PGresult *res, const char *p2)
     158             : {
     159           0 :     char       *msg = pchomp(PQerrorMessage(conn));
     160             : 
     161           0 :     if (res)
     162           0 :         PQclear(res);
     163           0 :     elog(ERROR, "%s: %s", p2, msg);
     164             : }
     165             : 
     166             : static void
     167             : pg_attribute_noreturn()
     168           6 : dblink_conn_not_avail(const char *conname)
     169             : {
     170           6 :     if (conname)
     171           2 :         ereport(ERROR,
     172             :                 (errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST),
     173             :                  errmsg("connection \"%s\" not available", conname)));
     174             :     else
     175           4 :         ereport(ERROR,
     176             :                 (errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST),
     177             :                  errmsg("connection not available")));
     178             : }
     179             : 
     180             : static void
     181          66 : dblink_get_conn(char *conname_or_str,
     182             :                 PGconn *volatile *conn_p, char **conname_p, volatile bool *freeconn_p)
     183             : {
     184          66 :     remoteConn *rconn = getConnectionByName(conname_or_str);
     185             :     PGconn     *conn;
     186             :     char       *conname;
     187             :     bool        freeconn;
     188             : 
     189          66 :     if (rconn)
     190             :     {
     191          54 :         conn = rconn->conn;
     192          54 :         conname = conname_or_str;
     193          54 :         freeconn = false;
     194             :     }
     195             :     else
     196             :     {
     197             :         const char *connstr;
     198             : 
     199          12 :         connstr = get_connect_string(conname_or_str);
     200          12 :         if (connstr == NULL)
     201          12 :             connstr = conname_or_str;
     202          12 :         dblink_connstr_check(connstr);
     203             : 
     204             :         /*
     205             :          * We must obey fd.c's limit on non-virtual file descriptors.  Assume
     206             :          * that a PGconn represents one long-lived FD.  (Doing this here also
     207             :          * ensures that VFDs are closed if needed to make room.)
     208             :          */
     209          12 :         if (!AcquireExternalFD())
     210             :         {
     211             : #ifndef WIN32                   /* can't write #if within ereport() macro */
     212           0 :             ereport(ERROR,
     213             :                     (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
     214             :                      errmsg("could not establish connection"),
     215             :                      errdetail("There are too many open files on the local server."),
     216             :                      errhint("Raise the server's max_files_per_process and/or \"ulimit -n\" limits.")));
     217             : #else
     218             :             ereport(ERROR,
     219             :                     (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
     220             :                      errmsg("could not establish connection"),
     221             :                      errdetail("There are too many open files on the local server."),
     222             :                      errhint("Raise the server's max_files_per_process setting.")));
     223             : #endif
     224             :         }
     225             : 
     226             :         /* OK to make connection */
     227          12 :         conn = PQconnectdb(connstr);
     228             : 
     229          12 :         if (PQstatus(conn) == CONNECTION_BAD)
     230             :         {
     231           4 :             char       *msg = pchomp(PQerrorMessage(conn));
     232             : 
     233           4 :             PQfinish(conn);
     234           4 :             ReleaseExternalFD();
     235           4 :             ereport(ERROR,
     236             :                     (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
     237             :                      errmsg("could not establish connection"),
     238             :                      errdetail_internal("%s", msg)));
     239             :         }
     240           8 :         dblink_security_check(conn, rconn);
     241           8 :         if (PQclientEncoding(conn) != GetDatabaseEncoding())
     242           0 :             PQsetClientEncoding(conn, GetDatabaseEncodingName());
     243           8 :         freeconn = true;
     244           8 :         conname = NULL;
     245             :     }
     246             : 
     247          62 :     *conn_p = conn;
     248          62 :     *conname_p = conname;
     249          62 :     *freeconn_p = freeconn;
     250          62 : }
     251             : 
     252             : static PGconn *
     253          34 : dblink_get_named_conn(const char *conname)
     254             : {
     255          34 :     remoteConn *rconn = getConnectionByName(conname);
     256             : 
     257          34 :     if (rconn)
     258          34 :         return rconn->conn;
     259             : 
     260           0 :     dblink_conn_not_avail(conname);
     261             :     return NULL;                /* keep compiler quiet */
     262             : }
     263             : 
     264             : static void
     265         234 : dblink_init(void)
     266             : {
     267         234 :     if (!pconn)
     268             :     {
     269           6 :         pconn = (remoteConn *) MemoryContextAlloc(TopMemoryContext, sizeof(remoteConn));
     270           6 :         pconn->conn = NULL;
     271           6 :         pconn->openCursorCount = 0;
     272           6 :         pconn->newXactForCursor = false;
     273             :     }
     274         234 : }
     275             : 
     276             : /*
     277             :  * Create a persistent connection to another database
     278             :  */
     279          18 : PG_FUNCTION_INFO_V1(dblink_connect);
     280             : Datum
     281          28 : dblink_connect(PG_FUNCTION_ARGS)
     282             : {
     283          28 :     char       *conname_or_str = NULL;
     284          28 :     char       *connstr = NULL;
     285          28 :     char       *connname = NULL;
     286             :     char       *msg;
     287          28 :     PGconn     *conn = NULL;
     288          28 :     remoteConn *rconn = NULL;
     289             : 
     290          28 :     dblink_init();
     291             : 
     292          28 :     if (PG_NARGS() == 2)
     293             :     {
     294          22 :         conname_or_str = text_to_cstring(PG_GETARG_TEXT_PP(1));
     295          22 :         connname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     296             :     }
     297           6 :     else if (PG_NARGS() == 1)
     298           6 :         conname_or_str = text_to_cstring(PG_GETARG_TEXT_PP(0));
     299             : 
     300          28 :     if (connname)
     301             :     {
     302          22 :         rconn = (remoteConn *) MemoryContextAlloc(TopMemoryContext,
     303             :                                                   sizeof(remoteConn));
     304          22 :         rconn->conn = NULL;
     305          22 :         rconn->openCursorCount = 0;
     306          22 :         rconn->newXactForCursor = false;
     307             :     }
     308             : 
     309             :     /* first check for valid foreign data server */
     310          28 :     connstr = get_connect_string(conname_or_str);
     311          28 :     if (connstr == NULL)
     312          24 :         connstr = conname_or_str;
     313             : 
     314             :     /* check password in connection string if not superuser */
     315          28 :     dblink_connstr_check(connstr);
     316             : 
     317             :     /*
     318             :      * We must obey fd.c's limit on non-virtual file descriptors.  Assume that
     319             :      * a PGconn represents one long-lived FD.  (Doing this here also ensures
     320             :      * that VFDs are closed if needed to make room.)
     321             :      */
     322          26 :     if (!AcquireExternalFD())
     323             :     {
     324             : #ifndef WIN32                   /* can't write #if within ereport() macro */
     325           0 :         ereport(ERROR,
     326             :                 (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
     327             :                  errmsg("could not establish connection"),
     328             :                  errdetail("There are too many open files on the local server."),
     329             :                  errhint("Raise the server's max_files_per_process and/or \"ulimit -n\" limits.")));
     330             : #else
     331             :         ereport(ERROR,
     332             :                 (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
     333             :                  errmsg("could not establish connection"),
     334             :                  errdetail("There are too many open files on the local server."),
     335             :                  errhint("Raise the server's max_files_per_process setting.")));
     336             : #endif
     337             :     }
     338             : 
     339             :     /* OK to make connection */
     340          26 :     conn = PQconnectdb(connstr);
     341             : 
     342          26 :     if (PQstatus(conn) == CONNECTION_BAD)
     343             :     {
     344           0 :         msg = pchomp(PQerrorMessage(conn));
     345           0 :         PQfinish(conn);
     346           0 :         ReleaseExternalFD();
     347           0 :         if (rconn)
     348           0 :             pfree(rconn);
     349             : 
     350           0 :         ereport(ERROR,
     351             :                 (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
     352             :                  errmsg("could not establish connection"),
     353             :                  errdetail_internal("%s", msg)));
     354             :     }
     355             : 
     356             :     /* check password actually used if not superuser */
     357          26 :     dblink_security_check(conn, rconn);
     358             : 
     359             :     /* attempt to set client encoding to match server encoding, if needed */
     360          26 :     if (PQclientEncoding(conn) != GetDatabaseEncoding())
     361           0 :         PQsetClientEncoding(conn, GetDatabaseEncodingName());
     362             : 
     363          26 :     if (connname)
     364             :     {
     365          20 :         rconn->conn = conn;
     366          20 :         createNewConnection(connname, rconn);
     367             :     }
     368             :     else
     369             :     {
     370           6 :         if (pconn->conn)
     371             :         {
     372           0 :             PQfinish(pconn->conn);
     373           0 :             ReleaseExternalFD();
     374             :         }
     375           6 :         pconn->conn = conn;
     376             :     }
     377             : 
     378          24 :     PG_RETURN_TEXT_P(cstring_to_text("OK"));
     379             : }
     380             : 
     381             : /*
     382             :  * Clear a persistent connection to another database
     383             :  */
     384          12 : PG_FUNCTION_INFO_V1(dblink_disconnect);
     385             : Datum
     386          24 : dblink_disconnect(PG_FUNCTION_ARGS)
     387             : {
     388          24 :     char       *conname = NULL;
     389          24 :     remoteConn *rconn = NULL;
     390          24 :     PGconn     *conn = NULL;
     391             : 
     392          24 :     dblink_init();
     393             : 
     394          24 :     if (PG_NARGS() == 1)
     395             :     {
     396          18 :         conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     397          18 :         rconn = getConnectionByName(conname);
     398          18 :         if (rconn)
     399          16 :             conn = rconn->conn;
     400             :     }
     401             :     else
     402           6 :         conn = pconn->conn;
     403             : 
     404          24 :     if (!conn)
     405           2 :         dblink_conn_not_avail(conname);
     406             : 
     407          22 :     PQfinish(conn);
     408          22 :     ReleaseExternalFD();
     409          22 :     if (rconn)
     410             :     {
     411          16 :         deleteConnection(conname);
     412          16 :         pfree(rconn);
     413             :     }
     414             :     else
     415           6 :         pconn->conn = NULL;
     416             : 
     417          22 :     PG_RETURN_TEXT_P(cstring_to_text("OK"));
     418             : }
     419             : 
     420             : /*
     421             :  * opens a cursor using a persistent connection
     422             :  */
     423          18 : PG_FUNCTION_INFO_V1(dblink_open);
     424             : Datum
     425          18 : dblink_open(PG_FUNCTION_ARGS)
     426             : {
     427          18 :     PGresult   *res = NULL;
     428             :     PGconn     *conn;
     429          18 :     char       *curname = NULL;
     430          18 :     char       *sql = NULL;
     431          18 :     char       *conname = NULL;
     432             :     StringInfoData buf;
     433          18 :     remoteConn *rconn = NULL;
     434          18 :     bool        fail = true;    /* default to backward compatible behavior */
     435             : 
     436          18 :     dblink_init();
     437          18 :     initStringInfo(&buf);
     438             : 
     439          18 :     if (PG_NARGS() == 2)
     440             :     {
     441             :         /* text,text */
     442           4 :         curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     443           4 :         sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
     444           4 :         rconn = pconn;
     445             :     }
     446          14 :     else if (PG_NARGS() == 3)
     447             :     {
     448             :         /* might be text,text,text or text,text,bool */
     449          12 :         if (get_fn_expr_argtype(fcinfo->flinfo, 2) == BOOLOID)
     450             :         {
     451           2 :             curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     452           2 :             sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
     453           2 :             fail = PG_GETARG_BOOL(2);
     454           2 :             rconn = pconn;
     455             :         }
     456             :         else
     457             :         {
     458          10 :             conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     459          10 :             curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
     460          10 :             sql = text_to_cstring(PG_GETARG_TEXT_PP(2));
     461          10 :             rconn = getConnectionByName(conname);
     462             :         }
     463             :     }
     464           2 :     else if (PG_NARGS() == 4)
     465             :     {
     466             :         /* text,text,text,bool */
     467           2 :         conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     468           2 :         curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
     469           2 :         sql = text_to_cstring(PG_GETARG_TEXT_PP(2));
     470           2 :         fail = PG_GETARG_BOOL(3);
     471           2 :         rconn = getConnectionByName(conname);
     472             :     }
     473             : 
     474          18 :     if (!rconn || !rconn->conn)
     475           0 :         dblink_conn_not_avail(conname);
     476             : 
     477          18 :     conn = rconn->conn;
     478             : 
     479             :     /* If we are not in a transaction, start one */
     480          18 :     if (PQtransactionStatus(conn) == PQTRANS_IDLE)
     481             :     {
     482          14 :         res = PQexec(conn, "BEGIN");
     483          14 :         if (PQresultStatus(res) != PGRES_COMMAND_OK)
     484           0 :             dblink_res_internalerror(conn, res, "begin error");
     485          14 :         PQclear(res);
     486          14 :         rconn->newXactForCursor = true;
     487             : 
     488             :         /*
     489             :          * Since transaction state was IDLE, we force cursor count to
     490             :          * initially be 0. This is needed as a previous ABORT might have wiped
     491             :          * out our transaction without maintaining the cursor count for us.
     492             :          */
     493          14 :         rconn->openCursorCount = 0;
     494             :     }
     495             : 
     496             :     /* if we started a transaction, increment cursor count */
     497          18 :     if (rconn->newXactForCursor)
     498          18 :         (rconn->openCursorCount)++;
     499             : 
     500          18 :     appendStringInfo(&buf, "DECLARE %s CURSOR FOR %s", curname, sql);
     501          18 :     res = PQexec(conn, buf.data);
     502          18 :     if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
     503             :     {
     504           4 :         dblink_res_error(conn, conname, res, fail,
     505             :                          "while opening cursor \"%s\"", curname);
     506           4 :         PG_RETURN_TEXT_P(cstring_to_text("ERROR"));
     507             :     }
     508             : 
     509          14 :     PQclear(res);
     510          14 :     PG_RETURN_TEXT_P(cstring_to_text("OK"));
     511             : }
     512             : 
     513             : /*
     514             :  * closes a cursor
     515             :  */
     516          12 : PG_FUNCTION_INFO_V1(dblink_close);
     517             : Datum
     518          10 : dblink_close(PG_FUNCTION_ARGS)
     519             : {
     520             :     PGconn     *conn;
     521          10 :     PGresult   *res = NULL;
     522          10 :     char       *curname = NULL;
     523          10 :     char       *conname = NULL;
     524             :     StringInfoData buf;
     525          10 :     remoteConn *rconn = NULL;
     526          10 :     bool        fail = true;    /* default to backward compatible behavior */
     527             : 
     528          10 :     dblink_init();
     529          10 :     initStringInfo(&buf);
     530             : 
     531          10 :     if (PG_NARGS() == 1)
     532             :     {
     533             :         /* text */
     534           0 :         curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     535           0 :         rconn = pconn;
     536             :     }
     537          10 :     else if (PG_NARGS() == 2)
     538             :     {
     539             :         /* might be text,text or text,bool */
     540          10 :         if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
     541             :         {
     542           4 :             curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     543           4 :             fail = PG_GETARG_BOOL(1);
     544           4 :             rconn = pconn;
     545             :         }
     546             :         else
     547             :         {
     548           6 :             conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     549           6 :             curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
     550           6 :             rconn = getConnectionByName(conname);
     551             :         }
     552             :     }
     553          10 :     if (PG_NARGS() == 3)
     554             :     {
     555             :         /* text,text,bool */
     556           0 :         conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     557           0 :         curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
     558           0 :         fail = PG_GETARG_BOOL(2);
     559           0 :         rconn = getConnectionByName(conname);
     560             :     }
     561             : 
     562          10 :     if (!rconn || !rconn->conn)
     563           0 :         dblink_conn_not_avail(conname);
     564             : 
     565          10 :     conn = rconn->conn;
     566             : 
     567          10 :     appendStringInfo(&buf, "CLOSE %s", curname);
     568             : 
     569             :     /* close the cursor */
     570          10 :     res = PQexec(conn, buf.data);
     571          10 :     if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
     572             :     {
     573           2 :         dblink_res_error(conn, conname, res, fail,
     574             :                          "while closing cursor \"%s\"", curname);
     575           2 :         PG_RETURN_TEXT_P(cstring_to_text("ERROR"));
     576             :     }
     577             : 
     578           8 :     PQclear(res);
     579             : 
     580             :     /* if we started a transaction, decrement cursor count */
     581           8 :     if (rconn->newXactForCursor)
     582             :     {
     583           8 :         (rconn->openCursorCount)--;
     584             : 
     585             :         /* if count is zero, commit the transaction */
     586           8 :         if (rconn->openCursorCount == 0)
     587             :         {
     588           4 :             rconn->newXactForCursor = false;
     589             : 
     590           4 :             res = PQexec(conn, "COMMIT");
     591           4 :             if (PQresultStatus(res) != PGRES_COMMAND_OK)
     592           0 :                 dblink_res_internalerror(conn, res, "commit error");
     593           4 :             PQclear(res);
     594             :         }
     595             :     }
     596             : 
     597           8 :     PG_RETURN_TEXT_P(cstring_to_text("OK"));
     598             : }
     599             : 
     600             : /*
     601             :  * Fetch results from an open cursor
     602             :  */
     603          18 : PG_FUNCTION_INFO_V1(dblink_fetch);
     604             : Datum
     605          26 : dblink_fetch(PG_FUNCTION_ARGS)
     606             : {
     607          26 :     PGresult   *res = NULL;
     608          26 :     char       *conname = NULL;
     609          26 :     remoteConn *rconn = NULL;
     610          26 :     PGconn     *conn = NULL;
     611             :     StringInfoData buf;
     612          26 :     char       *curname = NULL;
     613          26 :     int         howmany = 0;
     614          26 :     bool        fail = true;    /* default to backward compatible */
     615             : 
     616          26 :     prepTuplestoreResult(fcinfo);
     617             : 
     618          26 :     dblink_init();
     619             : 
     620          26 :     if (PG_NARGS() == 4)
     621             :     {
     622             :         /* text,text,int,bool */
     623           2 :         conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     624           2 :         curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
     625           2 :         howmany = PG_GETARG_INT32(2);
     626           2 :         fail = PG_GETARG_BOOL(3);
     627             : 
     628           2 :         rconn = getConnectionByName(conname);
     629           2 :         if (rconn)
     630           2 :             conn = rconn->conn;
     631             :     }
     632          24 :     else if (PG_NARGS() == 3)
     633             :     {
     634             :         /* text,text,int or text,int,bool */
     635          16 :         if (get_fn_expr_argtype(fcinfo->flinfo, 2) == BOOLOID)
     636             :         {
     637           4 :             curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     638           4 :             howmany = PG_GETARG_INT32(1);
     639           4 :             fail = PG_GETARG_BOOL(2);
     640           4 :             conn = pconn->conn;
     641             :         }
     642             :         else
     643             :         {
     644          12 :             conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     645          12 :             curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
     646          12 :             howmany = PG_GETARG_INT32(2);
     647             : 
     648          12 :             rconn = getConnectionByName(conname);
     649          12 :             if (rconn)
     650          12 :                 conn = rconn->conn;
     651             :         }
     652             :     }
     653           8 :     else if (PG_NARGS() == 2)
     654             :     {
     655             :         /* text,int */
     656           8 :         curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     657           8 :         howmany = PG_GETARG_INT32(1);
     658           8 :         conn = pconn->conn;
     659             :     }
     660             : 
     661          26 :     if (!conn)
     662           0 :         dblink_conn_not_avail(conname);
     663             : 
     664          26 :     initStringInfo(&buf);
     665          26 :     appendStringInfo(&buf, "FETCH %d FROM %s", howmany, curname);
     666             : 
     667             :     /*
     668             :      * Try to execute the query.  Note that since libpq uses malloc, the
     669             :      * PGresult will be long-lived even though we are still in a short-lived
     670             :      * memory context.
     671             :      */
     672          26 :     res = PQexec(conn, buf.data);
     673          52 :     if (!res ||
     674          52 :         (PQresultStatus(res) != PGRES_COMMAND_OK &&
     675          26 :          PQresultStatus(res) != PGRES_TUPLES_OK))
     676             :     {
     677          10 :         dblink_res_error(conn, conname, res, fail,
     678             :                          "while fetching from cursor \"%s\"", curname);
     679           6 :         return (Datum) 0;
     680             :     }
     681          16 :     else if (PQresultStatus(res) == PGRES_COMMAND_OK)
     682             :     {
     683             :         /* cursor does not exist - closed already or bad name */
     684           0 :         PQclear(res);
     685           0 :         ereport(ERROR,
     686             :                 (errcode(ERRCODE_INVALID_CURSOR_NAME),
     687             :                  errmsg("cursor \"%s\" does not exist", curname)));
     688             :     }
     689             : 
     690          16 :     materializeResult(fcinfo, conn, res);
     691          14 :     return (Datum) 0;
     692             : }
     693             : 
     694             : /*
     695             :  * Note: this is the new preferred version of dblink
     696             :  */
     697          22 : PG_FUNCTION_INFO_V1(dblink_record);
     698             : Datum
     699          50 : dblink_record(PG_FUNCTION_ARGS)
     700             : {
     701          50 :     return dblink_record_internal(fcinfo, false);
     702             : }
     703             : 
     704           6 : PG_FUNCTION_INFO_V1(dblink_send_query);
     705             : Datum
     706          12 : dblink_send_query(PG_FUNCTION_ARGS)
     707             : {
     708             :     PGconn     *conn;
     709             :     char       *sql;
     710             :     int         retval;
     711             : 
     712          12 :     if (PG_NARGS() == 2)
     713             :     {
     714          12 :         conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
     715          12 :         sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
     716             :     }
     717             :     else
     718             :         /* shouldn't happen */
     719           0 :         elog(ERROR, "wrong number of arguments");
     720             : 
     721             :     /* async query send */
     722          12 :     retval = PQsendQuery(conn, sql);
     723          12 :     if (retval != 1)
     724           0 :         elog(NOTICE, "could not send query: %s", pchomp(PQerrorMessage(conn)));
     725             : 
     726          12 :     PG_RETURN_INT32(retval);
     727             : }
     728             : 
     729           8 : PG_FUNCTION_INFO_V1(dblink_get_result);
     730             : Datum
     731          16 : dblink_get_result(PG_FUNCTION_ARGS)
     732             : {
     733          16 :     return dblink_record_internal(fcinfo, true);
     734             : }
     735             : 
     736             : static Datum
     737          66 : dblink_record_internal(FunctionCallInfo fcinfo, bool is_async)
     738             : {
     739          66 :     PGconn     *volatile conn = NULL;
     740          66 :     volatile bool freeconn = false;
     741             : 
     742          66 :     prepTuplestoreResult(fcinfo);
     743             : 
     744          66 :     dblink_init();
     745             : 
     746          66 :     PG_TRY();
     747             :     {
     748          66 :         char       *sql = NULL;
     749          66 :         char       *conname = NULL;
     750          66 :         bool        fail = true;    /* default to backward compatible */
     751             : 
     752          66 :         if (!is_async)
     753             :         {
     754          50 :             if (PG_NARGS() == 3)
     755             :             {
     756             :                 /* text,text,bool */
     757           2 :                 conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     758           2 :                 sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
     759           2 :                 fail = PG_GETARG_BOOL(2);
     760           2 :                 dblink_get_conn(conname, &conn, &conname, &freeconn);
     761             :             }
     762          48 :             else if (PG_NARGS() == 2)
     763             :             {
     764             :                 /* text,text or text,bool */
     765          34 :                 if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
     766             :                 {
     767           2 :                     sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
     768           2 :                     fail = PG_GETARG_BOOL(1);
     769           2 :                     conn = pconn->conn;
     770             :                 }
     771             :                 else
     772             :                 {
     773          32 :                     conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     774          32 :                     sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
     775          32 :                     dblink_get_conn(conname, &conn, &conname, &freeconn);
     776             :                 }
     777             :             }
     778          14 :             else if (PG_NARGS() == 1)
     779             :             {
     780             :                 /* text */
     781          14 :                 conn = pconn->conn;
     782          14 :                 sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
     783             :             }
     784             :             else
     785             :                 /* shouldn't happen */
     786           0 :                 elog(ERROR, "wrong number of arguments");
     787             :         }
     788             :         else                    /* is_async */
     789             :         {
     790             :             /* get async result */
     791          16 :             conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
     792             : 
     793          16 :             if (PG_NARGS() == 2)
     794             :             {
     795             :                 /* text,bool */
     796           0 :                 fail = PG_GETARG_BOOL(1);
     797           0 :                 conn = dblink_get_named_conn(conname);
     798             :             }
     799          16 :             else if (PG_NARGS() == 1)
     800             :             {
     801             :                 /* text */
     802          16 :                 conn = dblink_get_named_conn(conname);
     803             :             }
     804             :             else
     805             :                 /* shouldn't happen */
     806           0 :                 elog(ERROR, "wrong number of arguments");
     807             :         }
     808             : 
     809          62 :         if (!conn)
     810           4 :             dblink_conn_not_avail(conname);
     811             : 
     812          58 :         if (!is_async)
     813             :         {
     814             :             /* synchronous query, use efficient tuple collection method */
     815          42 :             materializeQueryResult(fcinfo, conn, conname, sql, fail);
     816             :         }
     817             :         else
     818             :         {
     819             :             /* async result retrieval, do it the old way */
     820          16 :             PGresult   *res = PQgetResult(conn);
     821             : 
     822             :             /* NULL means we're all done with the async results */
     823          16 :             if (res)
     824             :             {
     825          10 :                 if (PQresultStatus(res) != PGRES_COMMAND_OK &&
     826          10 :                     PQresultStatus(res) != PGRES_TUPLES_OK)
     827             :                 {
     828           0 :                     dblink_res_error(conn, conname, res, fail,
     829             :                                      "while executing query");
     830             :                     /* if fail isn't set, we'll return an empty query result */
     831             :                 }
     832             :                 else
     833             :                 {
     834          10 :                     materializeResult(fcinfo, conn, res);
     835             :                 }
     836             :             }
     837             :         }
     838             :     }
     839           8 :     PG_FINALLY();
     840             :     {
     841             :         /* if needed, close the connection to the database */
     842          66 :         if (freeconn)
     843             :         {
     844           6 :             PQfinish(conn);
     845           6 :             ReleaseExternalFD();
     846             :         }
     847             :     }
     848          66 :     PG_END_TRY();
     849             : 
     850          58 :     return (Datum) 0;
     851             : }
     852             : 
     853             : /*
     854             :  * Verify function caller can handle a tuplestore result, and set up for that.
     855             :  *
     856             :  * Note: if the caller returns without actually creating a tuplestore, the
     857             :  * executor will treat the function result as an empty set.
     858             :  */
     859             : static void
     860          96 : prepTuplestoreResult(FunctionCallInfo fcinfo)
     861             : {
     862          96 :     ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
     863             : 
     864             :     /* check to see if query supports us returning a tuplestore */
     865          96 :     if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
     866           0 :         ereport(ERROR,
     867             :                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
     868             :                  errmsg("set-valued function called in context that cannot accept a set")));
     869          96 :     if (!(rsinfo->allowedModes & SFRM_Materialize))
     870           0 :         ereport(ERROR,
     871             :                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
     872             :                  errmsg("materialize mode required, but it is not allowed in this context")));
     873             : 
     874             :     /* let the executor know we're sending back a tuplestore */
     875          96 :     rsinfo->returnMode = SFRM_Materialize;
     876             : 
     877             :     /* caller must fill these to return a non-empty result */
     878          96 :     rsinfo->setResult = NULL;
     879          96 :     rsinfo->setDesc = NULL;
     880          96 : }
     881             : 
     882             : /*
     883             :  * Copy the contents of the PGresult into a tuplestore to be returned
     884             :  * as the result of the current function.
     885             :  * The PGresult will be released in this function.
     886             :  */
     887             : static void
     888          26 : materializeResult(FunctionCallInfo fcinfo, PGconn *conn, PGresult *res)
     889             : {
     890          26 :     ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
     891             : 
     892             :     /* prepTuplestoreResult must have been called previously */
     893             :     Assert(rsinfo->returnMode == SFRM_Materialize);
     894             : 
     895          26 :     PG_TRY();
     896             :     {
     897             :         TupleDesc   tupdesc;
     898             :         bool        is_sql_cmd;
     899             :         int         ntuples;
     900             :         int         nfields;
     901             : 
     902          26 :         if (PQresultStatus(res) == PGRES_COMMAND_OK)
     903             :         {
     904           0 :             is_sql_cmd = true;
     905             : 
     906             :             /*
     907             :              * need a tuple descriptor representing one TEXT column to return
     908             :              * the command status string as our result tuple
     909             :              */
     910           0 :             tupdesc = CreateTemplateTupleDesc(1);
     911           0 :             TupleDescInitEntry(tupdesc, (AttrNumber) 1, "status",
     912             :                                TEXTOID, -1, 0);
     913           0 :             ntuples = 1;
     914           0 :             nfields = 1;
     915             :         }
     916             :         else
     917             :         {
     918             :             Assert(PQresultStatus(res) == PGRES_TUPLES_OK);
     919             : 
     920          26 :             is_sql_cmd = false;
     921             : 
     922             :             /* get a tuple descriptor for our result type */
     923          26 :             switch (get_call_result_type(fcinfo, NULL, &tupdesc))
     924             :             {
     925          26 :                 case TYPEFUNC_COMPOSITE:
     926             :                     /* success */
     927          26 :                     break;
     928           0 :                 case TYPEFUNC_RECORD:
     929             :                     /* failed to determine actual type of RECORD */
     930           0 :                     ereport(ERROR,
     931             :                             (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
     932             :                              errmsg("function returning record called in context "
     933             :                                     "that cannot accept type record")));
     934             :                     break;
     935           0 :                 default:
     936             :                     /* result type isn't composite */
     937           0 :                     elog(ERROR, "return type must be a row type");
     938             :                     break;
     939             :             }
     940             : 
     941             :             /* make sure we have a persistent copy of the tupdesc */
     942          26 :             tupdesc = CreateTupleDescCopy(tupdesc);
     943          26 :             ntuples = PQntuples(res);
     944          26 :             nfields = PQnfields(res);
     945             :         }
     946             : 
     947             :         /*
     948             :          * check result and tuple descriptor have the same number of columns
     949             :          */
     950          26 :         if (nfields != tupdesc->natts)
     951           0 :             ereport(ERROR,
     952             :                     (errcode(ERRCODE_DATATYPE_MISMATCH),
     953             :                      errmsg("remote query result rowtype does not match "
     954             :                             "the specified FROM clause rowtype")));
     955             : 
     956          26 :         if (ntuples > 0)
     957             :         {
     958             :             AttInMetadata *attinmeta;
     959          26 :             int         nestlevel = -1;
     960             :             Tuplestorestate *tupstore;
     961             :             MemoryContext oldcontext;
     962             :             int         row;
     963             :             char      **values;
     964             : 
     965          26 :             attinmeta = TupleDescGetAttInMetadata(tupdesc);
     966             : 
     967             :             /* Set GUCs to ensure we read GUC-sensitive data types correctly */
     968          26 :             if (!is_sql_cmd)
     969          26 :                 nestlevel = applyRemoteGucs(conn);
     970             : 
     971          26 :             oldcontext = MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory);
     972          26 :             tupstore = tuplestore_begin_heap(true, false, work_mem);
     973          26 :             rsinfo->setResult = tupstore;
     974          26 :             rsinfo->setDesc = tupdesc;
     975          26 :             MemoryContextSwitchTo(oldcontext);
     976             : 
     977          26 :             values = (char **) palloc(nfields * sizeof(char *));
     978             : 
     979             :             /* put all tuples into the tuplestore */
     980          98 :             for (row = 0; row < ntuples; row++)
     981             :             {
     982             :                 HeapTuple   tuple;
     983             : 
     984          74 :                 if (!is_sql_cmd)
     985             :                 {
     986             :                     int         i;
     987             : 
     988         276 :                     for (i = 0; i < nfields; i++)
     989             :                     {
     990         202 :                         if (PQgetisnull(res, row, i))
     991           0 :                             values[i] = NULL;
     992             :                         else
     993         202 :                             values[i] = PQgetvalue(res, row, i);
     994             :                     }
     995             :                 }
     996             :                 else
     997             :                 {
     998           0 :                     values[0] = PQcmdStatus(res);
     999             :                 }
    1000             : 
    1001             :                 /* build the tuple and put it into the tuplestore. */
    1002          74 :                 tuple = BuildTupleFromCStrings(attinmeta, values);
    1003          72 :                 tuplestore_puttuple(tupstore, tuple);
    1004             :             }
    1005             : 
    1006             :             /* clean up GUC settings, if we changed any */
    1007          24 :             restoreLocalGucs(nestlevel);
    1008             : 
    1009             :             /* clean up and return the tuplestore */
    1010             :             tuplestore_donestoring(tupstore);
    1011             :         }
    1012             :     }
    1013           2 :     PG_FINALLY();
    1014             :     {
    1015             :         /* be sure to release the libpq result */
    1016          26 :         PQclear(res);
    1017             :     }
    1018          26 :     PG_END_TRY();
    1019          24 : }
    1020             : 
    1021             : /*
    1022             :  * Execute the given SQL command and store its results into a tuplestore
    1023             :  * to be returned as the result of the current function.
    1024             :  *
    1025             :  * This is equivalent to PQexec followed by materializeResult, but we make
    1026             :  * use of libpq's single-row mode to avoid accumulating the whole result
    1027             :  * inside libpq before it gets transferred to the tuplestore.
    1028             :  */
    1029             : static void
    1030          42 : materializeQueryResult(FunctionCallInfo fcinfo,
    1031             :                        PGconn *conn,
    1032             :                        const char *conname,
    1033             :                        const char *sql,
    1034             :                        bool fail)
    1035             : {
    1036          42 :     ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
    1037          42 :     PGresult   *volatile res = NULL;
    1038          42 :     volatile storeInfo sinfo = {0};
    1039             : 
    1040             :     /* prepTuplestoreResult must have been called previously */
    1041             :     Assert(rsinfo->returnMode == SFRM_Materialize);
    1042             : 
    1043          42 :     sinfo.fcinfo = fcinfo;
    1044             : 
    1045          42 :     PG_TRY();
    1046             :     {
    1047             :         /* Create short-lived memory context for data conversions */
    1048          42 :         sinfo.tmpcontext = AllocSetContextCreate(CurrentMemoryContext,
    1049             :                                                  "dblink temporary context",
    1050             :                                                  ALLOCSET_DEFAULT_SIZES);
    1051             : 
    1052             :         /* execute query, collecting any tuples into the tuplestore */
    1053          42 :         res = storeQueryResult(&sinfo, conn, sql);
    1054             : 
    1055          42 :         if (!res ||
    1056          42 :             (PQresultStatus(res) != PGRES_COMMAND_OK &&
    1057          42 :              PQresultStatus(res) != PGRES_TUPLES_OK))
    1058           4 :         {
    1059             :             /*
    1060             :              * dblink_res_error will clear the passed PGresult, so we need
    1061             :              * this ugly dance to avoid doing so twice during error exit
    1062             :              */
    1063           4 :             PGresult   *res1 = res;
    1064             : 
    1065           4 :             res = NULL;
    1066           4 :             dblink_res_error(conn, conname, res1, fail,
    1067             :                              "while executing query");
    1068             :             /* if fail isn't set, we'll return an empty query result */
    1069             :         }
    1070          38 :         else if (PQresultStatus(res) == PGRES_COMMAND_OK)
    1071             :         {
    1072             :             /*
    1073             :              * storeRow didn't get called, so we need to convert the command
    1074             :              * status string to a tuple manually
    1075             :              */
    1076             :             TupleDesc   tupdesc;
    1077             :             AttInMetadata *attinmeta;
    1078             :             Tuplestorestate *tupstore;
    1079             :             HeapTuple   tuple;
    1080             :             char       *values[1];
    1081             :             MemoryContext oldcontext;
    1082             : 
    1083             :             /*
    1084             :              * need a tuple descriptor representing one TEXT column to return
    1085             :              * the command status string as our result tuple
    1086             :              */
    1087           0 :             tupdesc = CreateTemplateTupleDesc(1);
    1088           0 :             TupleDescInitEntry(tupdesc, (AttrNumber) 1, "status",
    1089             :                                TEXTOID, -1, 0);
    1090           0 :             attinmeta = TupleDescGetAttInMetadata(tupdesc);
    1091             : 
    1092           0 :             oldcontext = MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory);
    1093           0 :             tupstore = tuplestore_begin_heap(true, false, work_mem);
    1094           0 :             rsinfo->setResult = tupstore;
    1095           0 :             rsinfo->setDesc = tupdesc;
    1096           0 :             MemoryContextSwitchTo(oldcontext);
    1097             : 
    1098           0 :             values[0] = PQcmdStatus(res);
    1099             : 
    1100             :             /* build the tuple and put it into the tuplestore. */
    1101           0 :             tuple = BuildTupleFromCStrings(attinmeta, values);
    1102           0 :             tuplestore_puttuple(tupstore, tuple);
    1103             : 
    1104           0 :             PQclear(res);
    1105           0 :             res = NULL;
    1106             :         }
    1107             :         else
    1108             :         {
    1109             :             Assert(PQresultStatus(res) == PGRES_TUPLES_OK);
    1110             :             /* storeRow should have created a tuplestore */
    1111             :             Assert(rsinfo->setResult != NULL);
    1112             : 
    1113          38 :             PQclear(res);
    1114          38 :             res = NULL;
    1115             :         }
    1116             : 
    1117             :         /* clean up data conversion short-lived memory context */
    1118          42 :         if (sinfo.tmpcontext != NULL)
    1119          42 :             MemoryContextDelete(sinfo.tmpcontext);
    1120          42 :         sinfo.tmpcontext = NULL;
    1121             : 
    1122          42 :         PQclear(sinfo.last_res);
    1123          42 :         sinfo.last_res = NULL;
    1124          42 :         PQclear(sinfo.cur_res);
    1125          42 :         sinfo.cur_res = NULL;
    1126             :     }
    1127           0 :     PG_CATCH();
    1128             :     {
    1129             :         /* be sure to release any libpq result we collected */
    1130           0 :         PQclear(res);
    1131           0 :         PQclear(sinfo.last_res);
    1132           0 :         PQclear(sinfo.cur_res);
    1133             :         /* and clear out any pending data in libpq */
    1134           0 :         while ((res = PQgetResult(conn)) != NULL)
    1135           0 :             PQclear(res);
    1136           0 :         PG_RE_THROW();
    1137             :     }
    1138          42 :     PG_END_TRY();
    1139          42 : }
    1140             : 
    1141             : /*
    1142             :  * Execute query, and send any result rows to sinfo->tuplestore.
    1143             :  */
    1144             : static PGresult *
    1145          42 : storeQueryResult(volatile storeInfo *sinfo, PGconn *conn, const char *sql)
    1146             : {
    1147          42 :     bool        first = true;
    1148          42 :     int         nestlevel = -1;
    1149             :     PGresult   *res;
    1150             : 
    1151          42 :     if (!PQsendQuery(conn, sql))
    1152           0 :         elog(ERROR, "could not send query: %s", pchomp(PQerrorMessage(conn)));
    1153             : 
    1154          42 :     if (!PQsetSingleRowMode(conn))  /* shouldn't fail */
    1155           0 :         elog(ERROR, "failed to set single-row mode for dblink query");
    1156             : 
    1157             :     for (;;)
    1158             :     {
    1159         348 :         CHECK_FOR_INTERRUPTS();
    1160             : 
    1161         348 :         sinfo->cur_res = PQgetResult(conn);
    1162         348 :         if (!sinfo->cur_res)
    1163          42 :             break;
    1164             : 
    1165         306 :         if (PQresultStatus(sinfo->cur_res) == PGRES_SINGLE_TUPLE)
    1166             :         {
    1167             :             /* got one row from possibly-bigger resultset */
    1168             : 
    1169             :             /*
    1170             :              * Set GUCs to ensure we read GUC-sensitive data types correctly.
    1171             :              * We shouldn't do this until we have a row in hand, to ensure
    1172             :              * libpq has seen any earlier ParameterStatus protocol messages.
    1173             :              */
    1174         264 :             if (first && nestlevel < 0)
    1175          38 :                 nestlevel = applyRemoteGucs(conn);
    1176             : 
    1177         264 :             storeRow(sinfo, sinfo->cur_res, first);
    1178             : 
    1179         264 :             PQclear(sinfo->cur_res);
    1180         264 :             sinfo->cur_res = NULL;
    1181         264 :             first = false;
    1182             :         }
    1183             :         else
    1184             :         {
    1185             :             /* if empty resultset, fill tuplestore header */
    1186          42 :             if (first && PQresultStatus(sinfo->cur_res) == PGRES_TUPLES_OK)
    1187           0 :                 storeRow(sinfo, sinfo->cur_res, first);
    1188             : 
    1189             :             /* store completed result at last_res */
    1190          42 :             PQclear(sinfo->last_res);
    1191          42 :             sinfo->last_res = sinfo->cur_res;
    1192          42 :             sinfo->cur_res = NULL;
    1193          42 :             first = true;
    1194             :         }
    1195             :     }
    1196             : 
    1197             :     /* clean up GUC settings, if we changed any */
    1198          42 :     restoreLocalGucs(nestlevel);
    1199             : 
    1200             :     /* return last_res */
    1201          42 :     res = sinfo->last_res;
    1202          42 :     sinfo->last_res = NULL;
    1203          42 :     return res;
    1204             : }
    1205             : 
    1206             : /*
    1207             :  * Send single row to sinfo->tuplestore.
    1208             :  *
    1209             :  * If "first" is true, create the tuplestore using PGresult's metadata
    1210             :  * (in this case the PGresult might contain either zero or one row).
    1211             :  */
    1212             : static void
    1213         264 : storeRow(volatile storeInfo *sinfo, PGresult *res, bool first)
    1214             : {
    1215         264 :     int         nfields = PQnfields(res);
    1216             :     HeapTuple   tuple;
    1217             :     int         i;
    1218             :     MemoryContext oldcontext;
    1219             : 
    1220         264 :     if (first)
    1221             :     {
    1222             :         /* Prepare for new result set */
    1223          38 :         ReturnSetInfo *rsinfo = (ReturnSetInfo *) sinfo->fcinfo->resultinfo;
    1224             :         TupleDesc   tupdesc;
    1225             : 
    1226             :         /*
    1227             :          * It's possible to get more than one result set if the query string
    1228             :          * contained multiple SQL commands.  In that case, we follow PQexec's
    1229             :          * traditional behavior of throwing away all but the last result.
    1230             :          */
    1231          38 :         if (sinfo->tuplestore)
    1232           0 :             tuplestore_end(sinfo->tuplestore);
    1233          38 :         sinfo->tuplestore = NULL;
    1234             : 
    1235             :         /* get a tuple descriptor for our result type */
    1236          38 :         switch (get_call_result_type(sinfo->fcinfo, NULL, &tupdesc))
    1237             :         {
    1238          38 :             case TYPEFUNC_COMPOSITE:
    1239             :                 /* success */
    1240          38 :                 break;
    1241           0 :             case TYPEFUNC_RECORD:
    1242             :                 /* failed to determine actual type of RECORD */
    1243           0 :                 ereport(ERROR,
    1244             :                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    1245             :                          errmsg("function returning record called in context "
    1246             :                                 "that cannot accept type record")));
    1247             :                 break;
    1248           0 :             default:
    1249             :                 /* result type isn't composite */
    1250           0 :                 elog(ERROR, "return type must be a row type");
    1251             :                 break;
    1252             :         }
    1253             : 
    1254             :         /* make sure we have a persistent copy of the tupdesc */
    1255          38 :         tupdesc = CreateTupleDescCopy(tupdesc);
    1256             : 
    1257             :         /* check result and tuple descriptor have the same number of columns */
    1258          38 :         if (nfields != tupdesc->natts)
    1259           0 :             ereport(ERROR,
    1260             :                     (errcode(ERRCODE_DATATYPE_MISMATCH),
    1261             :                      errmsg("remote query result rowtype does not match "
    1262             :                             "the specified FROM clause rowtype")));
    1263             : 
    1264             :         /* Prepare attinmeta for later data conversions */
    1265          38 :         sinfo->attinmeta = TupleDescGetAttInMetadata(tupdesc);
    1266             : 
    1267             :         /* Create a new, empty tuplestore */
    1268          38 :         oldcontext = MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory);
    1269          38 :         sinfo->tuplestore = tuplestore_begin_heap(true, false, work_mem);
    1270          38 :         rsinfo->setResult = sinfo->tuplestore;
    1271          38 :         rsinfo->setDesc = tupdesc;
    1272          38 :         MemoryContextSwitchTo(oldcontext);
    1273             : 
    1274             :         /* Done if empty resultset */
    1275          38 :         if (PQntuples(res) == 0)
    1276           0 :             return;
    1277             : 
    1278             :         /*
    1279             :          * Set up sufficiently-wide string pointers array; this won't change
    1280             :          * in size so it's easy to preallocate.
    1281             :          */
    1282          38 :         if (sinfo->cstrs)
    1283           0 :             pfree(sinfo->cstrs);
    1284          38 :         sinfo->cstrs = (char **) palloc(nfields * sizeof(char *));
    1285             :     }
    1286             : 
    1287             :     /* Should have a single-row result if we get here */
    1288             :     Assert(PQntuples(res) == 1);
    1289             : 
    1290             :     /*
    1291             :      * Do the following work in a temp context that we reset after each tuple.
    1292             :      * This cleans up not only the data we have direct access to, but any
    1293             :      * cruft the I/O functions might leak.
    1294             :      */
    1295         264 :     oldcontext = MemoryContextSwitchTo(sinfo->tmpcontext);
    1296             : 
    1297             :     /*
    1298             :      * Fill cstrs with null-terminated strings of column values.
    1299             :      */
    1300        1020 :     for (i = 0; i < nfields; i++)
    1301             :     {
    1302         756 :         if (PQgetisnull(res, 0, i))
    1303           0 :             sinfo->cstrs[i] = NULL;
    1304             :         else
    1305         756 :             sinfo->cstrs[i] = PQgetvalue(res, 0, i);
    1306             :     }
    1307             : 
    1308             :     /* Convert row to a tuple, and add it to the tuplestore */
    1309         264 :     tuple = BuildTupleFromCStrings(sinfo->attinmeta, sinfo->cstrs);
    1310             : 
    1311         264 :     tuplestore_puttuple(sinfo->tuplestore, tuple);
    1312             : 
    1313             :     /* Clean up */
    1314         264 :     MemoryContextSwitchTo(oldcontext);
    1315         264 :     MemoryContextReset(sinfo->tmpcontext);
    1316             : }
    1317             : 
    1318             : /*
    1319             :  * List all open dblink connections by name.
    1320             :  * Returns an array of all connection names.
    1321             :  * Takes no params
    1322             :  */
    1323           4 : PG_FUNCTION_INFO_V1(dblink_get_connections);
    1324             : Datum
    1325           2 : dblink_get_connections(PG_FUNCTION_ARGS)
    1326             : {
    1327             :     HASH_SEQ_STATUS status;
    1328             :     remoteConnHashEnt *hentry;
    1329           2 :     ArrayBuildState *astate = NULL;
    1330             : 
    1331           2 :     if (remoteConnHash)
    1332             :     {
    1333           2 :         hash_seq_init(&status, remoteConnHash);
    1334           8 :         while ((hentry = (remoteConnHashEnt *) hash_seq_search(&status)) != NULL)
    1335             :         {
    1336             :             /* stash away current value */
    1337           6 :             astate = accumArrayResult(astate,
    1338           6 :                                       CStringGetTextDatum(hentry->name),
    1339             :                                       false, TEXTOID, CurrentMemoryContext);
    1340             :         }
    1341             :     }
    1342             : 
    1343           2 :     if (astate)
    1344           2 :         PG_RETURN_ARRAYTYPE_P(makeArrayResult(astate,
    1345             :                                               CurrentMemoryContext));
    1346             :     else
    1347           0 :         PG_RETURN_NULL();
    1348             : }
    1349             : 
    1350             : /*
    1351             :  * Checks if a given remote connection is busy
    1352             :  *
    1353             :  * Returns 1 if the connection is busy, 0 otherwise
    1354             :  * Params:
    1355             :  *  text connection_name - name of the connection to check
    1356             :  *
    1357             :  */
    1358           4 : PG_FUNCTION_INFO_V1(dblink_is_busy);
    1359             : Datum
    1360           2 : dblink_is_busy(PG_FUNCTION_ARGS)
    1361             : {
    1362             :     PGconn     *conn;
    1363             : 
    1364           2 :     dblink_init();
    1365           2 :     conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
    1366             : 
    1367           2 :     PQconsumeInput(conn);
    1368           2 :     PG_RETURN_INT32(PQisBusy(conn));
    1369             : }
    1370             : 
    1371             : /*
    1372             :  * Cancels a running request on a connection
    1373             :  *
    1374             :  * Returns text:
    1375             :  *  "OK" if the cancel request has been sent correctly,
    1376             :  *      an error message otherwise
    1377             :  *
    1378             :  * Params:
    1379             :  *  text connection_name - name of the connection to check
    1380             :  *
    1381             :  */
    1382           4 : PG_FUNCTION_INFO_V1(dblink_cancel_query);
    1383             : Datum
    1384           2 : dblink_cancel_query(PG_FUNCTION_ARGS)
    1385             : {
    1386             :     int         res;
    1387             :     PGconn     *conn;
    1388             :     PGcancel   *cancel;
    1389             :     char        errbuf[256];
    1390             : 
    1391           2 :     dblink_init();
    1392           2 :     conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
    1393           2 :     cancel = PQgetCancel(conn);
    1394             : 
    1395           2 :     res = PQcancel(cancel, errbuf, 256);
    1396           2 :     PQfreeCancel(cancel);
    1397             : 
    1398           2 :     if (res == 1)
    1399           2 :         PG_RETURN_TEXT_P(cstring_to_text("OK"));
    1400             :     else
    1401           0 :         PG_RETURN_TEXT_P(cstring_to_text(errbuf));
    1402             : }
    1403             : 
    1404             : 
    1405             : /*
    1406             :  * Get error message from a connection
    1407             :  *
    1408             :  * Returns text:
    1409             :  *  "OK" if no error, an error message otherwise
    1410             :  *
    1411             :  * Params:
    1412             :  *  text connection_name - name of the connection to check
    1413             :  *
    1414             :  */
    1415           4 : PG_FUNCTION_INFO_V1(dblink_error_message);
    1416             : Datum
    1417           2 : dblink_error_message(PG_FUNCTION_ARGS)
    1418             : {
    1419             :     char       *msg;
    1420             :     PGconn     *conn;
    1421             : 
    1422           2 :     dblink_init();
    1423           2 :     conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
    1424             : 
    1425           2 :     msg = PQerrorMessage(conn);
    1426           2 :     if (msg == NULL || msg[0] == '\0')
    1427           2 :         PG_RETURN_TEXT_P(cstring_to_text("OK"));
    1428             :     else
    1429           0 :         PG_RETURN_TEXT_P(cstring_to_text(pchomp(msg)));
    1430             : }
    1431             : 
    1432             : /*
    1433             :  * Execute an SQL non-SELECT command
    1434             :  */
    1435          18 : PG_FUNCTION_INFO_V1(dblink_exec);
    1436             : Datum
    1437          52 : dblink_exec(PG_FUNCTION_ARGS)
    1438             : {
    1439          52 :     text       *volatile sql_cmd_status = NULL;
    1440          52 :     PGconn     *volatile conn = NULL;
    1441          52 :     volatile bool freeconn = false;
    1442             : 
    1443          52 :     dblink_init();
    1444             : 
    1445          52 :     PG_TRY();
    1446             :     {
    1447          52 :         PGresult   *res = NULL;
    1448          52 :         char       *sql = NULL;
    1449          52 :         char       *conname = NULL;
    1450          52 :         bool        fail = true;    /* default to backward compatible behavior */
    1451             : 
    1452          52 :         if (PG_NARGS() == 3)
    1453             :         {
    1454             :             /* must be text,text,bool */
    1455           0 :             conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
    1456           0 :             sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
    1457           0 :             fail = PG_GETARG_BOOL(2);
    1458           0 :             dblink_get_conn(conname, &conn, &conname, &freeconn);
    1459             :         }
    1460          52 :         else if (PG_NARGS() == 2)
    1461             :         {
    1462             :             /* might be text,text or text,bool */
    1463          34 :             if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
    1464             :             {
    1465           2 :                 sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
    1466           2 :                 fail = PG_GETARG_BOOL(1);
    1467           2 :                 conn = pconn->conn;
    1468             :             }
    1469             :             else
    1470             :             {
    1471          32 :                 conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
    1472          32 :                 sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
    1473          32 :                 dblink_get_conn(conname, &conn, &conname, &freeconn);
    1474             :             }
    1475             :         }
    1476          18 :         else if (PG_NARGS() == 1)
    1477             :         {
    1478             :             /* must be single text argument */
    1479          18 :             conn = pconn->conn;
    1480          18 :             sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
    1481             :         }
    1482             :         else
    1483             :             /* shouldn't happen */
    1484           0 :             elog(ERROR, "wrong number of arguments");
    1485             : 
    1486          52 :         if (!conn)
    1487           0 :             dblink_conn_not_avail(conname);
    1488             : 
    1489          52 :         res = PQexec(conn, sql);
    1490          52 :         if (!res ||
    1491          52 :             (PQresultStatus(res) != PGRES_COMMAND_OK &&
    1492           4 :              PQresultStatus(res) != PGRES_TUPLES_OK))
    1493             :         {
    1494           4 :             dblink_res_error(conn, conname, res, fail,
    1495             :                              "while executing command");
    1496             : 
    1497             :             /*
    1498             :              * and save a copy of the command status string to return as our
    1499             :              * result tuple
    1500             :              */
    1501           2 :             sql_cmd_status = cstring_to_text("ERROR");
    1502             :         }
    1503          48 :         else if (PQresultStatus(res) == PGRES_COMMAND_OK)
    1504             :         {
    1505             :             /*
    1506             :              * and save a copy of the command status string to return as our
    1507             :              * result tuple
    1508             :              */
    1509          48 :             sql_cmd_status = cstring_to_text(PQcmdStatus(res));
    1510          48 :             PQclear(res);
    1511             :         }
    1512             :         else
    1513             :         {
    1514           0 :             PQclear(res);
    1515           0 :             ereport(ERROR,
    1516             :                     (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
    1517             :                      errmsg("statement returning results not allowed")));
    1518             :         }
    1519             :     }
    1520           2 :     PG_FINALLY();
    1521             :     {
    1522             :         /* if needed, close the connection to the database */
    1523          52 :         if (freeconn)
    1524             :         {
    1525           2 :             PQfinish(conn);
    1526           2 :             ReleaseExternalFD();
    1527             :         }
    1528             :     }
    1529          52 :     PG_END_TRY();
    1530             : 
    1531          50 :     PG_RETURN_TEXT_P(sql_cmd_status);
    1532             : }
    1533             : 
    1534             : 
    1535             : /*
    1536             :  * dblink_get_pkey
    1537             :  *
    1538             :  * Return list of primary key fields for the supplied relation,
    1539             :  * or NULL if none exists.
    1540             :  */
    1541           4 : PG_FUNCTION_INFO_V1(dblink_get_pkey);
    1542             : Datum
    1543          18 : dblink_get_pkey(PG_FUNCTION_ARGS)
    1544             : {
    1545             :     int16       indnkeyatts;
    1546             :     char      **results;
    1547             :     FuncCallContext *funcctx;
    1548             :     int32       call_cntr;
    1549             :     int32       max_calls;
    1550             :     AttInMetadata *attinmeta;
    1551             :     MemoryContext oldcontext;
    1552             : 
    1553             :     /* stuff done only on the first call of the function */
    1554          18 :     if (SRF_IS_FIRSTCALL())
    1555             :     {
    1556             :         Relation    rel;
    1557             :         TupleDesc   tupdesc;
    1558             : 
    1559             :         /* create a function context for cross-call persistence */
    1560           6 :         funcctx = SRF_FIRSTCALL_INIT();
    1561             : 
    1562             :         /*
    1563             :          * switch to memory context appropriate for multiple function calls
    1564             :          */
    1565           6 :         oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
    1566             : 
    1567             :         /* open target relation */
    1568           6 :         rel = get_rel_from_relname(PG_GETARG_TEXT_PP(0), AccessShareLock, ACL_SELECT);
    1569             : 
    1570             :         /* get the array of attnums */
    1571           6 :         results = get_pkey_attnames(rel, &indnkeyatts);
    1572             : 
    1573           6 :         relation_close(rel, AccessShareLock);
    1574             : 
    1575             :         /*
    1576             :          * need a tuple descriptor representing one INT and one TEXT column
    1577             :          */
    1578           6 :         tupdesc = CreateTemplateTupleDesc(2);
    1579           6 :         TupleDescInitEntry(tupdesc, (AttrNumber) 1, "position",
    1580             :                            INT4OID, -1, 0);
    1581           6 :         TupleDescInitEntry(tupdesc, (AttrNumber) 2, "colname",
    1582             :                            TEXTOID, -1, 0);
    1583             : 
    1584             :         /*
    1585             :          * Generate attribute metadata needed later to produce tuples from raw
    1586             :          * C strings
    1587             :          */
    1588           6 :         attinmeta = TupleDescGetAttInMetadata(tupdesc);
    1589           6 :         funcctx->attinmeta = attinmeta;
    1590             : 
    1591           6 :         if ((results != NULL) && (indnkeyatts > 0))
    1592             :         {
    1593           6 :             funcctx->max_calls = indnkeyatts;
    1594             : 
    1595             :             /* got results, keep track of them */
    1596           6 :             funcctx->user_fctx = results;
    1597             :         }
    1598             :         else
    1599             :         {
    1600             :             /* fast track when no results */
    1601           0 :             MemoryContextSwitchTo(oldcontext);
    1602           0 :             SRF_RETURN_DONE(funcctx);
    1603             :         }
    1604             : 
    1605           6 :         MemoryContextSwitchTo(oldcontext);
    1606             :     }
    1607             : 
    1608             :     /* stuff done on every call of the function */
    1609          18 :     funcctx = SRF_PERCALL_SETUP();
    1610             : 
    1611             :     /*
    1612             :      * initialize per-call variables
    1613             :      */
    1614          18 :     call_cntr = funcctx->call_cntr;
    1615          18 :     max_calls = funcctx->max_calls;
    1616             : 
    1617          18 :     results = (char **) funcctx->user_fctx;
    1618          18 :     attinmeta = funcctx->attinmeta;
    1619             : 
    1620          18 :     if (call_cntr < max_calls)   /* do when there is more left to send */
    1621             :     {
    1622             :         char      **values;
    1623             :         HeapTuple   tuple;
    1624             :         Datum       result;
    1625             : 
    1626          12 :         values = (char **) palloc(2 * sizeof(char *));
    1627          12 :         values[0] = psprintf("%d", call_cntr + 1);
    1628          12 :         values[1] = results[call_cntr];
    1629             : 
    1630             :         /* build the tuple */
    1631          12 :         tuple = BuildTupleFromCStrings(attinmeta, values);
    1632             : 
    1633             :         /* make the tuple into a datum */
    1634          12 :         result = HeapTupleGetDatum(tuple);
    1635             : 
    1636          12 :         SRF_RETURN_NEXT(funcctx, result);
    1637             :     }
    1638             :     else
    1639             :     {
    1640             :         /* do when there is no more left */
    1641           6 :         SRF_RETURN_DONE(funcctx);
    1642             :     }
    1643             : }
    1644             : 
    1645             : 
    1646             : /*
    1647             :  * dblink_build_sql_insert
    1648             :  *
    1649             :  * Used to generate an SQL insert statement
    1650             :  * based on an existing tuple in a local relation.
    1651             :  * This is useful for selectively replicating data
    1652             :  * to another server via dblink.
    1653             :  *
    1654             :  * API:
    1655             :  * <relname> - name of local table of interest
    1656             :  * <pkattnums> - an int2vector of attnums which will be used
    1657             :  * to identify the local tuple of interest
    1658             :  * <pknumatts> - number of attnums in pkattnums
    1659             :  * <src_pkattvals_arry> - text array of key values which will be used
    1660             :  * to identify the local tuple of interest
    1661             :  * <tgt_pkattvals_arry> - text array of key values which will be used
    1662             :  * to build the string for execution remotely. These are substituted
    1663             :  * for their counterparts in src_pkattvals_arry
    1664             :  */
    1665           6 : PG_FUNCTION_INFO_V1(dblink_build_sql_insert);
    1666             : Datum
    1667          12 : dblink_build_sql_insert(PG_FUNCTION_ARGS)
    1668             : {
    1669          12 :     text       *relname_text = PG_GETARG_TEXT_PP(0);
    1670          12 :     int2vector *pkattnums_arg = (int2vector *) PG_GETARG_POINTER(1);
    1671          12 :     int32       pknumatts_arg = PG_GETARG_INT32(2);
    1672          12 :     ArrayType  *src_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
    1673          12 :     ArrayType  *tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(4);
    1674             :     Relation    rel;
    1675             :     int        *pkattnums;
    1676             :     int         pknumatts;
    1677             :     char      **src_pkattvals;
    1678             :     char      **tgt_pkattvals;
    1679             :     int         src_nitems;
    1680             :     int         tgt_nitems;
    1681             :     char       *sql;
    1682             : 
    1683             :     /*
    1684             :      * Open target relation.
    1685             :      */
    1686          12 :     rel = get_rel_from_relname(relname_text, AccessShareLock, ACL_SELECT);
    1687             : 
    1688             :     /*
    1689             :      * Process pkattnums argument.
    1690             :      */
    1691          12 :     validate_pkattnums(rel, pkattnums_arg, pknumatts_arg,
    1692             :                        &pkattnums, &pknumatts);
    1693             : 
    1694             :     /*
    1695             :      * Source array is made up of key values that will be used to locate the
    1696             :      * tuple of interest from the local system.
    1697             :      */
    1698           8 :     src_pkattvals = get_text_array_contents(src_pkattvals_arry, &src_nitems);
    1699             : 
    1700             :     /*
    1701             :      * There should be one source array key value for each key attnum
    1702             :      */
    1703           8 :     if (src_nitems != pknumatts)
    1704           0 :         ereport(ERROR,
    1705             :                 (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
    1706             :                  errmsg("source key array length must match number of key attributes")));
    1707             : 
    1708             :     /*
    1709             :      * Target array is made up of key values that will be used to build the
    1710             :      * SQL string for use on the remote system.
    1711             :      */
    1712           8 :     tgt_pkattvals = get_text_array_contents(tgt_pkattvals_arry, &tgt_nitems);
    1713             : 
    1714             :     /*
    1715             :      * There should be one target array key value for each key attnum
    1716             :      */
    1717           8 :     if (tgt_nitems != pknumatts)
    1718           0 :         ereport(ERROR,
    1719             :                 (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
    1720             :                  errmsg("target key array length must match number of key attributes")));
    1721             : 
    1722             :     /*
    1723             :      * Prep work is finally done. Go get the SQL string.
    1724             :      */
    1725           8 :     sql = get_sql_insert(rel, pkattnums, pknumatts, src_pkattvals, tgt_pkattvals);
    1726             : 
    1727             :     /*
    1728             :      * Now we can close the relation.
    1729             :      */
    1730           8 :     relation_close(rel, AccessShareLock);
    1731             : 
    1732             :     /*
    1733             :      * And send it
    1734             :      */
    1735           8 :     PG_RETURN_TEXT_P(cstring_to_text(sql));
    1736             : }
    1737             : 
    1738             : 
    1739             : /*
    1740             :  * dblink_build_sql_delete
    1741             :  *
    1742             :  * Used to generate an SQL delete statement.
    1743             :  * This is useful for selectively replicating a
    1744             :  * delete to another server via dblink.
    1745             :  *
    1746             :  * API:
    1747             :  * <relname> - name of remote table of interest
    1748             :  * <pkattnums> - an int2vector of attnums which will be used
    1749             :  * to identify the remote tuple of interest
    1750             :  * <pknumatts> - number of attnums in pkattnums
    1751             :  * <tgt_pkattvals_arry> - text array of key values which will be used
    1752             :  * to build the string for execution remotely.
    1753             :  */
    1754           6 : PG_FUNCTION_INFO_V1(dblink_build_sql_delete);
    1755             : Datum
    1756          12 : dblink_build_sql_delete(PG_FUNCTION_ARGS)
    1757             : {
    1758          12 :     text       *relname_text = PG_GETARG_TEXT_PP(0);
    1759          12 :     int2vector *pkattnums_arg = (int2vector *) PG_GETARG_POINTER(1);
    1760          12 :     int32       pknumatts_arg = PG_GETARG_INT32(2);
    1761          12 :     ArrayType  *tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
    1762             :     Relation    rel;
    1763             :     int        *pkattnums;
    1764             :     int         pknumatts;
    1765             :     char      **tgt_pkattvals;
    1766             :     int         tgt_nitems;
    1767             :     char       *sql;
    1768             : 
    1769             :     /*
    1770             :      * Open target relation.
    1771             :      */
    1772          12 :     rel = get_rel_from_relname(relname_text, AccessShareLock, ACL_SELECT);
    1773             : 
    1774             :     /*
    1775             :      * Process pkattnums argument.
    1776             :      */
    1777          12 :     validate_pkattnums(rel, pkattnums_arg, pknumatts_arg,
    1778             :                        &pkattnums, &pknumatts);
    1779             : 
    1780             :     /*
    1781             :      * Target array is made up of key values that will be used to build the
    1782             :      * SQL string for use on the remote system.
    1783             :      */
    1784           8 :     tgt_pkattvals = get_text_array_contents(tgt_pkattvals_arry, &tgt_nitems);
    1785             : 
    1786             :     /*
    1787             :      * There should be one target array key value for each key attnum
    1788             :      */
    1789           8 :     if (tgt_nitems != pknumatts)
    1790           0 :         ereport(ERROR,
    1791             :                 (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
    1792             :                  errmsg("target key array length must match number of key attributes")));
    1793             : 
    1794             :     /*
    1795             :      * Prep work is finally done. Go get the SQL string.
    1796             :      */
    1797           8 :     sql = get_sql_delete(rel, pkattnums, pknumatts, tgt_pkattvals);
    1798             : 
    1799             :     /*
    1800             :      * Now we can close the relation.
    1801             :      */
    1802           8 :     relation_close(rel, AccessShareLock);
    1803             : 
    1804             :     /*
    1805             :      * And send it
    1806             :      */
    1807           8 :     PG_RETURN_TEXT_P(cstring_to_text(sql));
    1808             : }
    1809             : 
    1810             : 
    1811             : /*
    1812             :  * dblink_build_sql_update
    1813             :  *
    1814             :  * Used to generate an SQL update statement
    1815             :  * based on an existing tuple in a local relation.
    1816             :  * This is useful for selectively replicating data
    1817             :  * to another server via dblink.
    1818             :  *
    1819             :  * API:
    1820             :  * <relname> - name of local table of interest
    1821             :  * <pkattnums> - an int2vector of attnums which will be used
    1822             :  * to identify the local tuple of interest
    1823             :  * <pknumatts> - number of attnums in pkattnums
    1824             :  * <src_pkattvals_arry> - text array of key values which will be used
    1825             :  * to identify the local tuple of interest
    1826             :  * <tgt_pkattvals_arry> - text array of key values which will be used
    1827             :  * to build the string for execution remotely. These are substituted
    1828             :  * for their counterparts in src_pkattvals_arry
    1829             :  */
    1830           6 : PG_FUNCTION_INFO_V1(dblink_build_sql_update);
    1831             : Datum
    1832          12 : dblink_build_sql_update(PG_FUNCTION_ARGS)
    1833             : {
    1834          12 :     text       *relname_text = PG_GETARG_TEXT_PP(0);
    1835          12 :     int2vector *pkattnums_arg = (int2vector *) PG_GETARG_POINTER(1);
    1836          12 :     int32       pknumatts_arg = PG_GETARG_INT32(2);
    1837          12 :     ArrayType  *src_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
    1838          12 :     ArrayType  *tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(4);
    1839             :     Relation    rel;
    1840             :     int        *pkattnums;
    1841             :     int         pknumatts;
    1842             :     char      **src_pkattvals;
    1843             :     char      **tgt_pkattvals;
    1844             :     int         src_nitems;
    1845             :     int         tgt_nitems;
    1846             :     char       *sql;
    1847             : 
    1848             :     /*
    1849             :      * Open target relation.
    1850             :      */
    1851          12 :     rel = get_rel_from_relname(relname_text, AccessShareLock, ACL_SELECT);
    1852             : 
    1853             :     /*
    1854             :      * Process pkattnums argument.
    1855             :      */
    1856          12 :     validate_pkattnums(rel, pkattnums_arg, pknumatts_arg,
    1857             :                        &pkattnums, &pknumatts);
    1858             : 
    1859             :     /*
    1860             :      * Source array is made up of key values that will be used to locate the
    1861             :      * tuple of interest from the local system.
    1862             :      */
    1863           8 :     src_pkattvals = get_text_array_contents(src_pkattvals_arry, &src_nitems);
    1864             : 
    1865             :     /*
    1866             :      * There should be one source array key value for each key attnum
    1867             :      */
    1868           8 :     if (src_nitems != pknumatts)
    1869           0 :         ereport(ERROR,
    1870             :                 (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
    1871             :                  errmsg("source key array length must match number of key attributes")));
    1872             : 
    1873             :     /*
    1874             :      * Target array is made up of key values that will be used to build the
    1875             :      * SQL string for use on the remote system.
    1876             :      */
    1877           8 :     tgt_pkattvals = get_text_array_contents(tgt_pkattvals_arry, &tgt_nitems);
    1878             : 
    1879             :     /*
    1880             :      * There should be one target array key value for each key attnum
    1881             :      */
    1882           8 :     if (tgt_nitems != pknumatts)
    1883           0 :         ereport(ERROR,
    1884             :                 (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
    1885             :                  errmsg("target key array length must match number of key attributes")));
    1886             : 
    1887             :     /*
    1888             :      * Prep work is finally done. Go get the SQL string.
    1889             :      */
    1890           8 :     sql = get_sql_update(rel, pkattnums, pknumatts, src_pkattvals, tgt_pkattvals);
    1891             : 
    1892             :     /*
    1893             :      * Now we can close the relation.
    1894             :      */
    1895           8 :     relation_close(rel, AccessShareLock);
    1896             : 
    1897             :     /*
    1898             :      * And send it
    1899             :      */
    1900           8 :     PG_RETURN_TEXT_P(cstring_to_text(sql));
    1901             : }
    1902             : 
    1903             : /*
    1904             :  * dblink_current_query
    1905             :  * return the current query string
    1906             :  * to allow its use in (among other things)
    1907             :  * rewrite rules
    1908             :  */
    1909           2 : PG_FUNCTION_INFO_V1(dblink_current_query);
    1910             : Datum
    1911           0 : dblink_current_query(PG_FUNCTION_ARGS)
    1912             : {
    1913             :     /* This is now just an alias for the built-in function current_query() */
    1914           0 :     PG_RETURN_DATUM(current_query(fcinfo));
    1915             : }
    1916             : 
    1917             : /*
    1918             :  * Retrieve async notifications for a connection.
    1919             :  *
    1920             :  * Returns a setof record of notifications, or an empty set if none received.
    1921             :  * Can optionally take a named connection as parameter, but uses the unnamed
    1922             :  * connection per default.
    1923             :  *
    1924             :  */
    1925             : #define DBLINK_NOTIFY_COLS      3
    1926             : 
    1927           6 : PG_FUNCTION_INFO_V1(dblink_get_notify);
    1928             : Datum
    1929           4 : dblink_get_notify(PG_FUNCTION_ARGS)
    1930             : {
    1931             :     PGconn     *conn;
    1932             :     PGnotify   *notify;
    1933           4 :     ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
    1934             :     TupleDesc   tupdesc;
    1935             :     Tuplestorestate *tupstore;
    1936             :     MemoryContext per_query_ctx;
    1937             :     MemoryContext oldcontext;
    1938             : 
    1939           4 :     prepTuplestoreResult(fcinfo);
    1940             : 
    1941           4 :     dblink_init();
    1942           4 :     if (PG_NARGS() == 1)
    1943           0 :         conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
    1944             :     else
    1945           4 :         conn = pconn->conn;
    1946             : 
    1947             :     /* create the tuplestore in per-query memory */
    1948           4 :     per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
    1949           4 :     oldcontext = MemoryContextSwitchTo(per_query_ctx);
    1950             : 
    1951           4 :     tupdesc = CreateTemplateTupleDesc(DBLINK_NOTIFY_COLS);
    1952           4 :     TupleDescInitEntry(tupdesc, (AttrNumber) 1, "notify_name",
    1953             :                        TEXTOID, -1, 0);
    1954           4 :     TupleDescInitEntry(tupdesc, (AttrNumber) 2, "be_pid",
    1955             :                        INT4OID, -1, 0);
    1956           4 :     TupleDescInitEntry(tupdesc, (AttrNumber) 3, "extra",
    1957             :                        TEXTOID, -1, 0);
    1958             : 
    1959           4 :     tupstore = tuplestore_begin_heap(true, false, work_mem);
    1960           4 :     rsinfo->setResult = tupstore;
    1961           4 :     rsinfo->setDesc = tupdesc;
    1962             : 
    1963           4 :     MemoryContextSwitchTo(oldcontext);
    1964             : 
    1965           4 :     PQconsumeInput(conn);
    1966           8 :     while ((notify = PQnotifies(conn)) != NULL)
    1967             :     {
    1968             :         Datum       values[DBLINK_NOTIFY_COLS];
    1969             :         bool        nulls[DBLINK_NOTIFY_COLS];
    1970             : 
    1971           4 :         memset(values, 0, sizeof(values));
    1972           4 :         memset(nulls, 0, sizeof(nulls));
    1973             : 
    1974           4 :         if (notify->relname != NULL)
    1975           4 :             values[0] = CStringGetTextDatum(notify->relname);
    1976             :         else
    1977           0 :             nulls[0] = true;
    1978             : 
    1979           4 :         values[1] = Int32GetDatum(notify->be_pid);
    1980             : 
    1981           4 :         if (notify->extra != NULL)
    1982           4 :             values[2] = CStringGetTextDatum(notify->extra);
    1983             :         else
    1984           0 :             nulls[2] = true;
    1985             : 
    1986           4 :         tuplestore_putvalues(tupstore, tupdesc, values, nulls);
    1987             : 
    1988           4 :         PQfreemem(notify);
    1989           4 :         PQconsumeInput(conn);
    1990             :     }
    1991             : 
    1992             :     /* clean up and return the tuplestore */
    1993             :     tuplestore_donestoring(tupstore);
    1994             : 
    1995           4 :     return (Datum) 0;
    1996             : }
    1997             : 
    1998             : /*
    1999             :  * Validate the options given to a dblink foreign server or user mapping.
    2000             :  * Raise an error if any option is invalid.
    2001             :  *
    2002             :  * We just check the names of options here, so semantic errors in options,
    2003             :  * such as invalid numeric format, will be detected at the attempt to connect.
    2004             :  */
    2005           4 : PG_FUNCTION_INFO_V1(dblink_fdw_validator);
    2006             : Datum
    2007           8 : dblink_fdw_validator(PG_FUNCTION_ARGS)
    2008             : {
    2009           8 :     List       *options_list = untransformRelOptions(PG_GETARG_DATUM(0));
    2010           8 :     Oid         context = PG_GETARG_OID(1);
    2011             :     ListCell   *cell;
    2012             : 
    2013             :     static const PQconninfoOption *options = NULL;
    2014             : 
    2015             :     /*
    2016             :      * Get list of valid libpq options.
    2017             :      *
    2018             :      * To avoid unnecessary work, we get the list once and use it throughout
    2019             :      * the lifetime of this backend process.  We don't need to care about
    2020             :      * memory context issues, because PQconndefaults allocates with malloc.
    2021             :      */
    2022           8 :     if (!options)
    2023             :     {
    2024           2 :         options = PQconndefaults();
    2025           2 :         if (!options)           /* assume reason for failure is OOM */
    2026           0 :             ereport(ERROR,
    2027             :                     (errcode(ERRCODE_FDW_OUT_OF_MEMORY),
    2028             :                      errmsg("out of memory"),
    2029             :                      errdetail("Could not get libpq's default connection options.")));
    2030             :     }
    2031             : 
    2032             :     /* Validate each supplied option. */
    2033          14 :     foreach(cell, options_list)
    2034             :     {
    2035           8 :         DefElem    *def = (DefElem *) lfirst(cell);
    2036             : 
    2037           8 :         if (!is_valid_dblink_option(options, def->defname, context))
    2038             :         {
    2039             :             /*
    2040             :              * Unknown option, or invalid option for the context specified, so
    2041             :              * complain about it.  Provide a hint with list of valid options
    2042             :              * for the context.
    2043             :              */
    2044             :             StringInfoData buf;
    2045             :             const PQconninfoOption *opt;
    2046             : 
    2047           2 :             initStringInfo(&buf);
    2048          74 :             for (opt = options; opt->keyword; opt++)
    2049             :             {
    2050          72 :                 if (is_valid_dblink_option(options, opt->keyword, context))
    2051           6 :                     appendStringInfo(&buf, "%s%s",
    2052           6 :                                      (buf.len > 0) ? ", " : "",
    2053             :                                      opt->keyword);
    2054             :             }
    2055           2 :             ereport(ERROR,
    2056             :                     (errcode(ERRCODE_FDW_OPTION_NAME_NOT_FOUND),
    2057             :                      errmsg("invalid option \"%s\"", def->defname),
    2058             :                      errhint("Valid options in this context are: %s",
    2059             :                              buf.data)));
    2060             :         }
    2061             :     }
    2062             : 
    2063           6 :     PG_RETURN_VOID();
    2064             : }
    2065             : 
    2066             : 
    2067             : /*************************************************************
    2068             :  * internal functions
    2069             :  */
    2070             : 
    2071             : 
    2072             : /*
    2073             :  * get_pkey_attnames
    2074             :  *
    2075             :  * Get the primary key attnames for the given relation.
    2076             :  * Return NULL, and set indnkeyatts = 0, if no primary key exists.
    2077             :  */
    2078             : static char **
    2079           6 : get_pkey_attnames(Relation rel, int16 *indnkeyatts)
    2080             : {
    2081             :     Relation    indexRelation;
    2082             :     ScanKeyData skey;
    2083             :     SysScanDesc scan;
    2084             :     HeapTuple   indexTuple;
    2085             :     int         i;
    2086           6 :     char      **result = NULL;
    2087             :     TupleDesc   tupdesc;
    2088             : 
    2089             :     /* initialize indnkeyatts to 0 in case no primary key exists */
    2090           6 :     *indnkeyatts = 0;
    2091             : 
    2092           6 :     tupdesc = rel->rd_att;
    2093             : 
    2094             :     /* Prepare to scan pg_index for entries having indrelid = this rel. */
    2095           6 :     indexRelation = table_open(IndexRelationId, AccessShareLock);
    2096           6 :     ScanKeyInit(&skey,
    2097             :                 Anum_pg_index_indrelid,
    2098             :                 BTEqualStrategyNumber, F_OIDEQ,
    2099           6 :                 ObjectIdGetDatum(RelationGetRelid(rel)));
    2100             : 
    2101           6 :     scan = systable_beginscan(indexRelation, IndexIndrelidIndexId, true,
    2102             :                               NULL, 1, &skey);
    2103             : 
    2104           6 :     while (HeapTupleIsValid(indexTuple = systable_getnext(scan)))
    2105             :     {
    2106           6 :         Form_pg_index index = (Form_pg_index) GETSTRUCT(indexTuple);
    2107             : 
    2108             :         /* we're only interested if it is the primary key */
    2109           6 :         if (index->indisprimary)
    2110             :         {
    2111           6 :             *indnkeyatts = index->indnkeyatts;
    2112           6 :             if (*indnkeyatts > 0)
    2113             :             {
    2114           6 :                 result = (char **) palloc(*indnkeyatts * sizeof(char *));
    2115             : 
    2116          18 :                 for (i = 0; i < *indnkeyatts; i++)
    2117          12 :                     result[i] = SPI_fname(tupdesc, index->indkey.values[i]);
    2118             :             }
    2119           6 :             break;
    2120             :         }
    2121             :     }
    2122             : 
    2123           6 :     systable_endscan(scan);
    2124           6 :     table_close(indexRelation, AccessShareLock);
    2125             : 
    2126           6 :     return result;
    2127             : }
    2128             : 
    2129             : /*
    2130             :  * Deconstruct a text[] into C-strings (note any NULL elements will be
    2131             :  * returned as NULL pointers)
    2132             :  */
    2133             : static char **
    2134          40 : get_text_array_contents(ArrayType *array, int *numitems)
    2135             : {
    2136          40 :     int         ndim = ARR_NDIM(array);
    2137          40 :     int        *dims = ARR_DIMS(array);
    2138             :     int         nitems;
    2139             :     int16       typlen;
    2140             :     bool        typbyval;
    2141             :     char        typalign;
    2142             :     char      **values;
    2143             :     char       *ptr;
    2144             :     bits8      *bitmap;
    2145             :     int         bitmask;
    2146             :     int         i;
    2147             : 
    2148             :     Assert(ARR_ELEMTYPE(array) == TEXTOID);
    2149             : 
    2150          40 :     *numitems = nitems = ArrayGetNItems(ndim, dims);
    2151             : 
    2152          40 :     get_typlenbyvalalign(ARR_ELEMTYPE(array),
    2153             :                          &typlen, &typbyval, &typalign);
    2154             : 
    2155          40 :     values = (char **) palloc(nitems * sizeof(char *));
    2156             : 
    2157          40 :     ptr = ARR_DATA_PTR(array);
    2158          40 :     bitmap = ARR_NULLBITMAP(array);
    2159          40 :     bitmask = 1;
    2160             : 
    2161         110 :     for (i = 0; i < nitems; i++)
    2162             :     {
    2163          70 :         if (bitmap && (*bitmap & bitmask) == 0)
    2164             :         {
    2165           0 :             values[i] = NULL;
    2166             :         }
    2167             :         else
    2168             :         {
    2169          70 :             values[i] = TextDatumGetCString(PointerGetDatum(ptr));
    2170          70 :             ptr = att_addlength_pointer(ptr, typlen, ptr);
    2171          70 :             ptr = (char *) att_align_nominal(ptr, typalign);
    2172             :         }
    2173             : 
    2174             :         /* advance bitmap pointer if any */
    2175          70 :         if (bitmap)
    2176             :         {
    2177           0 :             bitmask <<= 1;
    2178           0 :             if (bitmask == 0x100)
    2179             :             {
    2180           0 :                 bitmap++;
    2181           0 :                 bitmask = 1;
    2182             :             }
    2183             :         }
    2184             :     }
    2185             : 
    2186          40 :     return values;
    2187             : }
    2188             : 
    2189             : static char *
    2190           8 : get_sql_insert(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals, char **tgt_pkattvals)
    2191             : {
    2192             :     char       *relname;
    2193             :     HeapTuple   tuple;
    2194             :     TupleDesc   tupdesc;
    2195             :     int         natts;
    2196             :     StringInfoData buf;
    2197             :     char       *val;
    2198             :     int         key;
    2199             :     int         i;
    2200             :     bool        needComma;
    2201             : 
    2202           8 :     initStringInfo(&buf);
    2203             : 
    2204             :     /* get relation name including any needed schema prefix and quoting */
    2205           8 :     relname = generate_relation_name(rel);
    2206             : 
    2207           8 :     tupdesc = rel->rd_att;
    2208           8 :     natts = tupdesc->natts;
    2209             : 
    2210           8 :     tuple = get_tuple_of_interest(rel, pkattnums, pknumatts, src_pkattvals);
    2211           8 :     if (!tuple)
    2212           0 :         ereport(ERROR,
    2213             :                 (errcode(ERRCODE_CARDINALITY_VIOLATION),
    2214             :                  errmsg("source row not found")));
    2215             : 
    2216           8 :     appendStringInfo(&buf, "INSERT INTO %s(", relname);
    2217             : 
    2218           8 :     needComma = false;
    2219          38 :     for (i = 0; i < natts; i++)
    2220             :     {
    2221          30 :         Form_pg_attribute att = TupleDescAttr(tupdesc, i);
    2222             : 
    2223          30 :         if (att->attisdropped)
    2224           4 :             continue;
    2225             : 
    2226          26 :         if (needComma)
    2227          18 :             appendStringInfoChar(&buf, ',');
    2228             : 
    2229          26 :         appendStringInfoString(&buf,
    2230          26 :                                quote_ident_cstr(NameStr(att->attname)));
    2231          26 :         needComma = true;
    2232             :     }
    2233             : 
    2234           8 :     appendStringInfoString(&buf, ") VALUES(");
    2235             : 
    2236             :     /*
    2237             :      * Note: i is physical column number (counting from 0).
    2238             :      */
    2239           8 :     needComma = false;
    2240          38 :     for (i = 0; i < natts; i++)
    2241             :     {
    2242          30 :         if (TupleDescAttr(tupdesc, i)->attisdropped)
    2243           4 :             continue;
    2244             : 
    2245          26 :         if (needComma)
    2246          18 :             appendStringInfoChar(&buf, ',');
    2247             : 
    2248          26 :         key = get_attnum_pk_pos(pkattnums, pknumatts, i);
    2249             : 
    2250          26 :         if (key >= 0)
    2251          14 :             val = tgt_pkattvals[key] ? pstrdup(tgt_pkattvals[key]) : NULL;
    2252             :         else
    2253          12 :             val = SPI_getvalue(tuple, tupdesc, i + 1);
    2254             : 
    2255          26 :         if (val != NULL)
    2256             :         {
    2257          26 :             appendStringInfoString(&buf, quote_literal_cstr(val));
    2258          26 :             pfree(val);
    2259             :         }
    2260             :         else
    2261           0 :             appendStringInfoString(&buf, "NULL");
    2262          26 :         needComma = true;
    2263             :     }
    2264           8 :     appendStringInfoChar(&buf, ')');
    2265             : 
    2266           8 :     return buf.data;
    2267             : }
    2268             : 
    2269             : static char *
    2270           8 : get_sql_delete(Relation rel, int *pkattnums, int pknumatts, char **tgt_pkattvals)
    2271             : {
    2272             :     char       *relname;
    2273             :     TupleDesc   tupdesc;
    2274             :     StringInfoData buf;
    2275             :     int         i;
    2276             : 
    2277           8 :     initStringInfo(&buf);
    2278             : 
    2279             :     /* get relation name including any needed schema prefix and quoting */
    2280           8 :     relname = generate_relation_name(rel);
    2281             : 
    2282           8 :     tupdesc = rel->rd_att;
    2283             : 
    2284           8 :     appendStringInfo(&buf, "DELETE FROM %s WHERE ", relname);
    2285          22 :     for (i = 0; i < pknumatts; i++)
    2286             :     {
    2287          14 :         int         pkattnum = pkattnums[i];
    2288          14 :         Form_pg_attribute attr = TupleDescAttr(tupdesc, pkattnum);
    2289             : 
    2290          14 :         if (i > 0)
    2291           6 :             appendStringInfoString(&buf, " AND ");
    2292             : 
    2293          14 :         appendStringInfoString(&buf,
    2294          14 :                                quote_ident_cstr(NameStr(attr->attname)));
    2295             : 
    2296          14 :         if (tgt_pkattvals[i] != NULL)
    2297          14 :             appendStringInfo(&buf, " = %s",
    2298          14 :                              quote_literal_cstr(tgt_pkattvals[i]));
    2299             :         else
    2300           0 :             appendStringInfoString(&buf, " IS NULL");
    2301             :     }
    2302             : 
    2303           8 :     return buf.data;
    2304             : }
    2305             : 
    2306             : static char *
    2307           8 : get_sql_update(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals, char **tgt_pkattvals)
    2308             : {
    2309             :     char       *relname;
    2310             :     HeapTuple   tuple;
    2311             :     TupleDesc   tupdesc;
    2312             :     int         natts;
    2313             :     StringInfoData buf;
    2314             :     char       *val;
    2315             :     int         key;
    2316             :     int         i;
    2317             :     bool        needComma;
    2318             : 
    2319           8 :     initStringInfo(&buf);
    2320             : 
    2321             :     /* get relation name including any needed schema prefix and quoting */
    2322           8 :     relname = generate_relation_name(rel);
    2323             : 
    2324           8 :     tupdesc = rel->rd_att;
    2325           8 :     natts = tupdesc->natts;
    2326             : 
    2327           8 :     tuple = get_tuple_of_interest(rel, pkattnums, pknumatts, src_pkattvals);
    2328           8 :     if (!tuple)
    2329           0 :         ereport(ERROR,
    2330             :                 (errcode(ERRCODE_CARDINALITY_VIOLATION),
    2331             :                  errmsg("source row not found")));
    2332             : 
    2333           8 :     appendStringInfo(&buf, "UPDATE %s SET ", relname);
    2334             : 
    2335             :     /*
    2336             :      * Note: i is physical column number (counting from 0).
    2337             :      */
    2338           8 :     needComma = false;
    2339          38 :     for (i = 0; i < natts; i++)
    2340             :     {
    2341          30 :         Form_pg_attribute attr = TupleDescAttr(tupdesc, i);
    2342             : 
    2343          30 :         if (attr->attisdropped)
    2344           4 :             continue;
    2345             : 
    2346          26 :         if (needComma)
    2347          18 :             appendStringInfoString(&buf, ", ");
    2348             : 
    2349          26 :         appendStringInfo(&buf, "%s = ",
    2350          26 :                          quote_ident_cstr(NameStr(attr->attname)));
    2351             : 
    2352          26 :         key = get_attnum_pk_pos(pkattnums, pknumatts, i);
    2353             : 
    2354          26 :         if (key >= 0)
    2355          14 :             val = tgt_pkattvals[key] ? pstrdup(tgt_pkattvals[key]) : NULL;
    2356             :         else
    2357          12 :             val = SPI_getvalue(tuple, tupdesc, i + 1);
    2358             : 
    2359          26 :         if (val != NULL)
    2360             :         {
    2361          26 :             appendStringInfoString(&buf, quote_literal_cstr(val));
    2362          26 :             pfree(val);
    2363             :         }
    2364             :         else
    2365           0 :             appendStringInfoString(&buf, "NULL");
    2366          26 :         needComma = true;
    2367             :     }
    2368             : 
    2369           8 :     appendStringInfoString(&buf, " WHERE ");
    2370             : 
    2371          22 :     for (i = 0; i < pknumatts; i++)
    2372             :     {
    2373          14 :         int         pkattnum = pkattnums[i];
    2374          14 :         Form_pg_attribute attr = TupleDescAttr(tupdesc, pkattnum);
    2375             : 
    2376          14 :         if (i > 0)
    2377           6 :             appendStringInfoString(&buf, " AND ");
    2378             : 
    2379          14 :         appendStringInfoString(&buf,
    2380          14 :                                quote_ident_cstr(NameStr(attr->attname)));
    2381             : 
    2382          14 :         val = tgt_pkattvals[i];
    2383             : 
    2384          14 :         if (val != NULL)
    2385          14 :             appendStringInfo(&buf, " = %s", quote_literal_cstr(val));
    2386             :         else
    2387           0 :             appendStringInfoString(&buf, " IS NULL");
    2388             :     }
    2389             : 
    2390           8 :     return buf.data;
    2391             : }
    2392             : 
    2393             : /*
    2394             :  * Return a properly quoted identifier.
    2395             :  * Uses quote_ident in quote.c
    2396             :  */
    2397             : static char *
    2398         160 : quote_ident_cstr(char *rawstr)
    2399             : {
    2400             :     text       *rawstr_text;
    2401             :     text       *result_text;
    2402             :     char       *result;
    2403             : 
    2404         160 :     rawstr_text = cstring_to_text(rawstr);
    2405         160 :     result_text = DatumGetTextPP(DirectFunctionCall1(quote_ident,
    2406             :                                                      PointerGetDatum(rawstr_text)));
    2407         160 :     result = text_to_cstring(result_text);
    2408             : 
    2409         160 :     return result;
    2410             : }
    2411             : 
    2412             : static int
    2413          52 : get_attnum_pk_pos(int *pkattnums, int pknumatts, int key)
    2414             : {
    2415             :     int         i;
    2416             : 
    2417             :     /*
    2418             :      * Not likely a long list anyway, so just scan for the value
    2419             :      */
    2420         100 :     for (i = 0; i < pknumatts; i++)
    2421          76 :         if (key == pkattnums[i])
    2422          28 :             return i;
    2423             : 
    2424          24 :     return -1;
    2425             : }
    2426             : 
    2427             : static HeapTuple
    2428          16 : get_tuple_of_interest(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals)
    2429             : {
    2430             :     char       *relname;
    2431             :     TupleDesc   tupdesc;
    2432             :     int         natts;
    2433             :     StringInfoData buf;
    2434             :     int         ret;
    2435             :     HeapTuple   tuple;
    2436             :     int         i;
    2437             : 
    2438             :     /*
    2439             :      * Connect to SPI manager
    2440             :      */
    2441          16 :     if ((ret = SPI_connect()) < 0)
    2442             :         /* internal error */
    2443           0 :         elog(ERROR, "SPI connect failure - returned %d", ret);
    2444             : 
    2445          16 :     initStringInfo(&buf);
    2446             : 
    2447             :     /* get relation name including any needed schema prefix and quoting */
    2448          16 :     relname = generate_relation_name(rel);
    2449             : 
    2450          16 :     tupdesc = rel->rd_att;
    2451          16 :     natts = tupdesc->natts;
    2452             : 
    2453             :     /*
    2454             :      * Build sql statement to look up tuple of interest, ie, the one matching
    2455             :      * src_pkattvals.  We used to use "SELECT *" here, but it's simpler to
    2456             :      * generate a result tuple that matches the table's physical structure,
    2457             :      * with NULLs for any dropped columns.  Otherwise we have to deal with two
    2458             :      * different tupdescs and everything's very confusing.
    2459             :      */
    2460          16 :     appendStringInfoString(&buf, "SELECT ");
    2461             : 
    2462          76 :     for (i = 0; i < natts; i++)
    2463             :     {
    2464          60 :         Form_pg_attribute attr = TupleDescAttr(tupdesc, i);
    2465             : 
    2466          60 :         if (i > 0)
    2467          44 :             appendStringInfoString(&buf, ", ");
    2468             : 
    2469          60 :         if (attr->attisdropped)
    2470           8 :             appendStringInfoString(&buf, "NULL");
    2471             :         else
    2472          52 :             appendStringInfoString(&buf,
    2473          52 :                                    quote_ident_cstr(NameStr(attr->attname)));
    2474             :     }
    2475             : 
    2476          16 :     appendStringInfo(&buf, " FROM %s WHERE ", relname);
    2477             : 
    2478          44 :     for (i = 0; i < pknumatts; i++)
    2479             :     {
    2480          28 :         int         pkattnum = pkattnums[i];
    2481          28 :         Form_pg_attribute attr = TupleDescAttr(tupdesc, pkattnum);
    2482             : 
    2483          28 :         if (i > 0)
    2484          12 :             appendStringInfoString(&buf, " AND ");
    2485             : 
    2486          28 :         appendStringInfoString(&buf,
    2487          28 :                                quote_ident_cstr(NameStr(attr->attname)));
    2488             : 
    2489          28 :         if (src_pkattvals[i] != NULL)
    2490          28 :             appendStringInfo(&buf, " = %s",
    2491          28 :                              quote_literal_cstr(src_pkattvals[i]));
    2492             :         else
    2493           0 :             appendStringInfoString(&buf, " IS NULL");
    2494             :     }
    2495             : 
    2496             :     /*
    2497             :      * Retrieve the desired tuple
    2498             :      */
    2499          16 :     ret = SPI_exec(buf.data, 0);
    2500          16 :     pfree(buf.data);
    2501             : 
    2502             :     /*
    2503             :      * Only allow one qualifying tuple
    2504             :      */
    2505          16 :     if ((ret == SPI_OK_SELECT) && (SPI_processed > 1))
    2506           0 :         ereport(ERROR,
    2507             :                 (errcode(ERRCODE_CARDINALITY_VIOLATION),
    2508             :                  errmsg("source criteria matched more than one record")));
    2509             : 
    2510          16 :     else if (ret == SPI_OK_SELECT && SPI_processed == 1)
    2511             :     {
    2512          16 :         SPITupleTable *tuptable = SPI_tuptable;
    2513             : 
    2514          16 :         tuple = SPI_copytuple(tuptable->vals[0]);
    2515          16 :         SPI_finish();
    2516             : 
    2517          16 :         return tuple;
    2518             :     }
    2519             :     else
    2520             :     {
    2521             :         /*
    2522             :          * no qualifying tuples
    2523             :          */
    2524           0 :         SPI_finish();
    2525             : 
    2526           0 :         return NULL;
    2527             :     }
    2528             : 
    2529             :     /*
    2530             :      * never reached, but keep compiler quiet
    2531             :      */
    2532             :     return NULL;
    2533             : }
    2534             : 
    2535             : /*
    2536             :  * Open the relation named by relname_text, acquire specified type of lock,
    2537             :  * verify we have specified permissions.
    2538             :  * Caller must close rel when done with it.
    2539             :  */
    2540             : static Relation
    2541          42 : get_rel_from_relname(text *relname_text, LOCKMODE lockmode, AclMode aclmode)
    2542             : {
    2543             :     RangeVar   *relvar;
    2544             :     Relation    rel;
    2545             :     AclResult   aclresult;
    2546             : 
    2547          42 :     relvar = makeRangeVarFromNameList(textToQualifiedNameList(relname_text));
    2548          42 :     rel = table_openrv(relvar, lockmode);
    2549             : 
    2550          42 :     aclresult = pg_class_aclcheck(RelationGetRelid(rel), GetUserId(),
    2551             :                                   aclmode);
    2552          42 :     if (aclresult != ACLCHECK_OK)
    2553           0 :         aclcheck_error(aclresult, get_relkind_objtype(rel->rd_rel->relkind),
    2554           0 :                        RelationGetRelationName(rel));
    2555             : 
    2556          42 :     return rel;
    2557             : }
    2558             : 
    2559             : /*
    2560             :  * generate_relation_name - copied from ruleutils.c
    2561             :  *      Compute the name to display for a relation
    2562             :  *
    2563             :  * The result includes all necessary quoting and schema-prefixing.
    2564             :  */
    2565             : static char *
    2566          40 : generate_relation_name(Relation rel)
    2567             : {
    2568             :     char       *nspname;
    2569             :     char       *result;
    2570             : 
    2571             :     /* Qualify the name if not visible in search path */
    2572          40 :     if (RelationIsVisible(RelationGetRelid(rel)))
    2573          30 :         nspname = NULL;
    2574             :     else
    2575          10 :         nspname = get_namespace_name(rel->rd_rel->relnamespace);
    2576             : 
    2577          40 :     result = quote_qualified_identifier(nspname, RelationGetRelationName(rel));
    2578             : 
    2579          40 :     return result;
    2580             : }
    2581             : 
    2582             : 
    2583             : static remoteConn *
    2584         150 : getConnectionByName(const char *name)
    2585             : {
    2586             :     remoteConnHashEnt *hentry;
    2587             :     char       *key;
    2588             : 
    2589         150 :     if (!remoteConnHash)
    2590           4 :         remoteConnHash = createConnHash();
    2591             : 
    2592         150 :     key = pstrdup(name);
    2593         150 :     truncate_identifier(key, strlen(key), false);
    2594         150 :     hentry = (remoteConnHashEnt *) hash_search(remoteConnHash,
    2595             :                                                key, HASH_FIND, NULL);
    2596             : 
    2597         150 :     if (hentry)
    2598         136 :         return hentry->rconn;
    2599             : 
    2600          14 :     return NULL;
    2601             : }
    2602             : 
    2603             : static HTAB *
    2604           6 : createConnHash(void)
    2605             : {
    2606             :     HASHCTL     ctl;
    2607             : 
    2608           6 :     ctl.keysize = NAMEDATALEN;
    2609           6 :     ctl.entrysize = sizeof(remoteConnHashEnt);
    2610             : 
    2611           6 :     return hash_create("Remote Con hash", NUMCONN, &ctl, HASH_ELEM);
    2612             : }
    2613             : 
    2614             : static void
    2615          20 : createNewConnection(const char *name, remoteConn *rconn)
    2616             : {
    2617             :     remoteConnHashEnt *hentry;
    2618             :     bool        found;
    2619             :     char       *key;
    2620             : 
    2621          20 :     if (!remoteConnHash)
    2622           2 :         remoteConnHash = createConnHash();
    2623             : 
    2624          20 :     key = pstrdup(name);
    2625          20 :     truncate_identifier(key, strlen(key), true);
    2626          20 :     hentry = (remoteConnHashEnt *) hash_search(remoteConnHash, key,
    2627             :                                                HASH_ENTER, &found);
    2628             : 
    2629          20 :     if (found)
    2630             :     {
    2631           2 :         PQfinish(rconn->conn);
    2632           2 :         ReleaseExternalFD();
    2633           2 :         pfree(rconn);
    2634             : 
    2635           2 :         ereport(ERROR,
    2636             :                 (errcode(ERRCODE_DUPLICATE_OBJECT),
    2637             :                  errmsg("duplicate connection name")));
    2638             :     }
    2639             : 
    2640          18 :     hentry->rconn = rconn;
    2641          18 :     strlcpy(hentry->name, name, sizeof(hentry->name));
    2642          18 : }
    2643             : 
    2644             : static void
    2645          16 : deleteConnection(const char *name)
    2646             : {
    2647             :     remoteConnHashEnt *hentry;
    2648             :     bool        found;
    2649             :     char       *key;
    2650             : 
    2651          16 :     if (!remoteConnHash)
    2652           0 :         remoteConnHash = createConnHash();
    2653             : 
    2654          16 :     key = pstrdup(name);
    2655          16 :     truncate_identifier(key, strlen(key), false);
    2656          16 :     hentry = (remoteConnHashEnt *) hash_search(remoteConnHash,
    2657             :                                                key, HASH_REMOVE, &found);
    2658             : 
    2659          16 :     if (!hentry)
    2660           0 :         ereport(ERROR,
    2661             :                 (errcode(ERRCODE_UNDEFINED_OBJECT),
    2662             :                  errmsg("undefined connection name")));
    2663             : 
    2664          16 : }
    2665             : 
    2666             : static void
    2667          34 : dblink_security_check(PGconn *conn, remoteConn *rconn)
    2668             : {
    2669          34 :     if (!superuser())
    2670             :     {
    2671           0 :         if (!PQconnectionUsedPassword(conn))
    2672             :         {
    2673           0 :             PQfinish(conn);
    2674           0 :             ReleaseExternalFD();
    2675           0 :             if (rconn)
    2676           0 :                 pfree(rconn);
    2677             : 
    2678           0 :             ereport(ERROR,
    2679             :                     (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
    2680             :                      errmsg("password is required"),
    2681             :                      errdetail("Non-superuser cannot connect if the server does not request a password."),
    2682             :                      errhint("Target server's authentication method must be changed.")));
    2683             :         }
    2684             :     }
    2685          34 : }
    2686             : 
    2687             : /*
    2688             :  * For non-superusers, insist that the connstr specify a password.  This
    2689             :  * prevents a password from being picked up from .pgpass, a service file,
    2690             :  * the environment, etc.  We don't want the postgres user's passwords
    2691             :  * to be accessible to non-superusers.
    2692             :  */
    2693             : static void
    2694          40 : dblink_connstr_check(const char *connstr)
    2695             : {
    2696          40 :     if (!superuser())
    2697             :     {
    2698             :         PQconninfoOption *options;
    2699             :         PQconninfoOption *option;
    2700           2 :         bool        connstr_gives_password = false;
    2701             : 
    2702           2 :         options = PQconninfoParse(connstr, NULL);
    2703           2 :         if (options)
    2704             :         {
    2705          74 :             for (option = options; option->keyword != NULL; option++)
    2706             :             {
    2707          72 :                 if (strcmp(option->keyword, "password") == 0)
    2708             :                 {
    2709           2 :                     if (option->val != NULL && option->val[0] != '\0')
    2710             :                     {
    2711           0 :                         connstr_gives_password = true;
    2712           0 :                         break;
    2713             :                     }
    2714             :                 }
    2715             :             }
    2716           2 :             PQconninfoFree(options);
    2717             :         }
    2718             : 
    2719           2 :         if (!connstr_gives_password)
    2720           2 :             ereport(ERROR,
    2721             :                     (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
    2722             :                      errmsg("password is required"),
    2723             :                      errdetail("Non-superusers must provide a password in the connection string.")));
    2724             :     }
    2725          38 : }
    2726             : 
    2727             : /*
    2728             :  * Report an error received from the remote server
    2729             :  *
    2730             :  * res: the received error result (will be freed)
    2731             :  * fail: true for ERROR ereport, false for NOTICE
    2732             :  * fmt and following args: sprintf-style format and values for errcontext;
    2733             :  * the resulting string should be worded like "while <some action>"
    2734             :  */
    2735             : static void
    2736          24 : dblink_res_error(PGconn *conn, const char *conname, PGresult *res,
    2737             :                  bool fail, const char *fmt,...)
    2738             : {
    2739             :     int         level;
    2740          24 :     char       *pg_diag_sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
    2741          24 :     char       *pg_diag_message_primary = PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY);
    2742          24 :     char       *pg_diag_message_detail = PQresultErrorField(res, PG_DIAG_MESSAGE_DETAIL);
    2743          24 :     char       *pg_diag_message_hint = PQresultErrorField(res, PG_DIAG_MESSAGE_HINT);
    2744          24 :     char       *pg_diag_context = PQresultErrorField(res, PG_DIAG_CONTEXT);
    2745             :     int         sqlstate;
    2746             :     char       *message_primary;
    2747             :     char       *message_detail;
    2748             :     char       *message_hint;
    2749             :     char       *message_context;
    2750             :     va_list     ap;
    2751             :     char        dblink_context_msg[512];
    2752             : 
    2753          24 :     if (fail)
    2754           6 :         level = ERROR;
    2755             :     else
    2756          18 :         level = NOTICE;
    2757             : 
    2758          24 :     if (pg_diag_sqlstate)
    2759          24 :         sqlstate = MAKE_SQLSTATE(pg_diag_sqlstate[0],
    2760             :                                  pg_diag_sqlstate[1],
    2761             :                                  pg_diag_sqlstate[2],
    2762             :                                  pg_diag_sqlstate[3],
    2763             :                                  pg_diag_sqlstate[4]);
    2764             :     else
    2765           0 :         sqlstate = ERRCODE_CONNECTION_FAILURE;
    2766             : 
    2767          24 :     message_primary = xpstrdup(pg_diag_message_primary);
    2768          24 :     message_detail = xpstrdup(pg_diag_message_detail);
    2769          24 :     message_hint = xpstrdup(pg_diag_message_hint);
    2770          24 :     message_context = xpstrdup(pg_diag_context);
    2771             : 
    2772             :     /*
    2773             :      * If we don't get a message from the PGresult, try the PGconn.  This is
    2774             :      * needed because for connection-level failures, PQexec may just return
    2775             :      * NULL, not a PGresult at all.
    2776             :      */
    2777          24 :     if (message_primary == NULL)
    2778           0 :         message_primary = pchomp(PQerrorMessage(conn));
    2779             : 
    2780             :     /*
    2781             :      * Now that we've copied all the data we need out of the PGresult, it's
    2782             :      * safe to free it.  We must do this to avoid PGresult leakage.  We're
    2783             :      * leaking all the strings too, but those are in palloc'd memory that will
    2784             :      * get cleaned up eventually.
    2785             :      */
    2786          24 :     if (res)
    2787          24 :         PQclear(res);
    2788             : 
    2789             :     /*
    2790             :      * Format the basic errcontext string.  Below, we'll add on something
    2791             :      * about the connection name.  That's a violation of the translatability
    2792             :      * guidelines about constructing error messages out of parts, but since
    2793             :      * there's no translation support for dblink, there's no need to worry
    2794             :      * about that (yet).
    2795             :      */
    2796          24 :     va_start(ap, fmt);
    2797          24 :     vsnprintf(dblink_context_msg, sizeof(dblink_context_msg), fmt, ap);
    2798          24 :     va_end(ap);
    2799             : 
    2800          24 :     ereport(level,
    2801             :             (errcode(sqlstate),
    2802             :              message_primary ? errmsg_internal("%s", message_primary) :
    2803             :              errmsg("could not obtain message string for remote error"),
    2804             :              message_detail ? errdetail_internal("%s", message_detail) : 0,
    2805             :              message_hint ? errhint("%s", message_hint) : 0,
    2806             :              message_context ? (errcontext("%s", message_context)) : 0,
    2807             :              conname ?
    2808             :              (errcontext("%s on dblink connection named \"%s\"",
    2809             :                          dblink_context_msg, conname)) :
    2810             :              (errcontext("%s on unnamed dblink connection",
    2811             :                          dblink_context_msg))));
    2812          18 : }
    2813             : 
    2814             : /*
    2815             :  * Obtain connection string for a foreign server
    2816             :  */
    2817             : static char *
    2818          40 : get_connect_string(const char *servername)
    2819             : {
    2820          40 :     ForeignServer *foreign_server = NULL;
    2821             :     UserMapping *user_mapping;
    2822             :     ListCell   *cell;
    2823             :     StringInfoData buf;
    2824             :     ForeignDataWrapper *fdw;
    2825             :     AclResult   aclresult;
    2826             :     char       *srvname;
    2827             : 
    2828             :     static const PQconninfoOption *options = NULL;
    2829             : 
    2830          40 :     initStringInfo(&buf);
    2831             : 
    2832             :     /*
    2833             :      * Get list of valid libpq options.
    2834             :      *
    2835             :      * To avoid unnecessary work, we get the list once and use it throughout
    2836             :      * the lifetime of this backend process.  We don't need to care about
    2837             :      * memory context issues, because PQconndefaults allocates with malloc.
    2838             :      */
    2839          40 :     if (!options)
    2840             :     {
    2841           6 :         options = PQconndefaults();
    2842           6 :         if (!options)           /* assume reason for failure is OOM */
    2843           0 :             ereport(ERROR,
    2844             :                     (errcode(ERRCODE_FDW_OUT_OF_MEMORY),
    2845             :                      errmsg("out of memory"),
    2846             :                      errdetail("Could not get libpq's default connection options.")));
    2847             :     }
    2848             : 
    2849             :     /* first gather the server connstr options */
    2850          40 :     srvname = pstrdup(servername);
    2851          40 :     truncate_identifier(srvname, strlen(srvname), false);
    2852          40 :     foreign_server = GetForeignServerByName(srvname, true);
    2853             : 
    2854          40 :     if (foreign_server)
    2855             :     {
    2856           4 :         Oid         serverid = foreign_server->serverid;
    2857           4 :         Oid         fdwid = foreign_server->fdwid;
    2858           4 :         Oid         userid = GetUserId();
    2859             : 
    2860           4 :         user_mapping = GetUserMapping(userid, serverid);
    2861           4 :         fdw = GetForeignDataWrapper(fdwid);
    2862             : 
    2863             :         /* Check permissions, user must have usage on the server. */
    2864           4 :         aclresult = pg_foreign_server_aclcheck(serverid, userid, ACL_USAGE);
    2865           4 :         if (aclresult != ACLCHECK_OK)
    2866           0 :             aclcheck_error(aclresult, OBJECT_FOREIGN_SERVER, foreign_server->servername);
    2867             : 
    2868           4 :         foreach(cell, fdw->options)
    2869             :         {
    2870           0 :             DefElem    *def = lfirst(cell);
    2871             : 
    2872           0 :             if (is_valid_dblink_option(options, def->defname, ForeignDataWrapperRelationId))
    2873           0 :                 appendStringInfo(&buf, "%s='%s' ", def->defname,
    2874           0 :                                  escape_param_str(strVal(def->arg)));
    2875             :         }
    2876             : 
    2877          12 :         foreach(cell, foreign_server->options)
    2878             :         {
    2879           8 :             DefElem    *def = lfirst(cell);
    2880             : 
    2881           8 :             if (is_valid_dblink_option(options, def->defname, ForeignServerRelationId))
    2882           8 :                 appendStringInfo(&buf, "%s='%s' ", def->defname,
    2883           8 :                                  escape_param_str(strVal(def->arg)));
    2884             :         }
    2885             : 
    2886           8 :         foreach(cell, user_mapping->options)
    2887             :         {
    2888             : 
    2889           4 :             DefElem    *def = lfirst(cell);
    2890             : 
    2891           4 :             if (is_valid_dblink_option(options, def->defname, UserMappingRelationId))
    2892           4 :                 appendStringInfo(&buf, "%s='%s' ", def->defname,
    2893           4 :                                  escape_param_str(strVal(def->arg)));
    2894             :         }
    2895             : 
    2896           4 :         return buf.data;
    2897             :     }
    2898             :     else
    2899          36 :         return NULL;
    2900             : }
    2901             : 
    2902             : /*
    2903             :  * Escaping libpq connect parameter strings.
    2904             :  *
    2905             :  * Replaces "'" with "\'" and "\" with "\\".
    2906             :  */
    2907             : static char *
    2908          12 : escape_param_str(const char *str)
    2909             : {
    2910             :     const char *cp;
    2911             :     StringInfoData buf;
    2912             : 
    2913          12 :     initStringInfo(&buf);
    2914             : 
    2915         136 :     for (cp = str; *cp; cp++)
    2916             :     {
    2917         124 :         if (*cp == '\\' || *cp == '\'')
    2918           0 :             appendStringInfoChar(&buf, '\\');
    2919         124 :         appendStringInfoChar(&buf, *cp);
    2920             :     }
    2921             : 
    2922          12 :     return buf.data;
    2923             : }
    2924             : 
    2925             : /*
    2926             :  * Validate the PK-attnums argument for dblink_build_sql_insert() and related
    2927             :  * functions, and translate to the internal representation.
    2928             :  *
    2929             :  * The user supplies an int2vector of 1-based logical attnums, plus a count
    2930             :  * argument (the need for the separate count argument is historical, but we
    2931             :  * still check it).  We check that each attnum corresponds to a valid,
    2932             :  * non-dropped attribute of the rel.  We do *not* prevent attnums from being
    2933             :  * listed twice, though the actual use-case for such things is dubious.
    2934             :  * Note that before Postgres 9.0, the user's attnums were interpreted as
    2935             :  * physical not logical column numbers; this was changed for future-proofing.
    2936             :  *
    2937             :  * The internal representation is a palloc'd int array of 0-based physical
    2938             :  * attnums.
    2939             :  */
    2940             : static void
    2941          36 : validate_pkattnums(Relation rel,
    2942             :                    int2vector *pkattnums_arg, int32 pknumatts_arg,
    2943             :                    int **pkattnums, int *pknumatts)
    2944             : {
    2945          36 :     TupleDesc   tupdesc = rel->rd_att;
    2946          36 :     int         natts = tupdesc->natts;
    2947             :     int         i;
    2948             : 
    2949             :     /* Don't take more array elements than there are */
    2950          36 :     pknumatts_arg = Min(pknumatts_arg, pkattnums_arg->dim1);
    2951             : 
    2952             :     /* Must have at least one pk attnum selected */
    2953          36 :     if (pknumatts_arg <= 0)
    2954           0 :         ereport(ERROR,
    2955             :                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
    2956             :                  errmsg("number of key attributes must be > 0")));
    2957             : 
    2958             :     /* Allocate output array */
    2959          36 :     *pkattnums = (int *) palloc(pknumatts_arg * sizeof(int));
    2960          36 :     *pknumatts = pknumatts_arg;
    2961             : 
    2962             :     /* Validate attnums and convert to internal form */
    2963         114 :     for (i = 0; i < pknumatts_arg; i++)
    2964             :     {
    2965          90 :         int         pkattnum = pkattnums_arg->values[i];
    2966             :         int         lnum;
    2967             :         int         j;
    2968             : 
    2969             :         /* Can throw error immediately if out of range */
    2970          90 :         if (pkattnum <= 0 || pkattnum > natts)
    2971          12 :             ereport(ERROR,
    2972             :                     (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
    2973             :                      errmsg("invalid attribute number %d", pkattnum)));
    2974             : 
    2975             :         /* Identify which physical column has this logical number */
    2976          78 :         lnum = 0;
    2977         138 :         for (j = 0; j < natts; j++)
    2978             :         {
    2979             :             /* dropped columns don't count */
    2980         138 :             if (TupleDescAttr(tupdesc, j)->attisdropped)
    2981           6 :                 continue;
    2982             : 
    2983         132 :             if (++lnum == pkattnum)
    2984          78 :                 break;
    2985             :         }
    2986             : 
    2987          78 :         if (j < natts)
    2988          78 :             (*pkattnums)[i] = j;
    2989             :         else
    2990           0 :             ereport(ERROR,
    2991             :                     (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
    2992             :                      errmsg("invalid attribute number %d", pkattnum)));
    2993             :     }
    2994          24 : }
    2995             : 
    2996             : /*
    2997             :  * Check if the specified connection option is valid.
    2998             :  *
    2999             :  * We basically allow whatever libpq thinks is an option, with these
    3000             :  * restrictions:
    3001             :  *      debug options: disallowed
    3002             :  *      "client_encoding": disallowed
    3003             :  *      "user": valid only in USER MAPPING options
    3004             :  *      secure options (eg password): valid only in USER MAPPING options
    3005             :  *      others: valid only in FOREIGN SERVER options
    3006             :  *
    3007             :  * We disallow client_encoding because it would be overridden anyway via
    3008             :  * PQclientEncoding; allowing it to be specified would merely promote
    3009             :  * confusion.
    3010             :  */
    3011             : static bool
    3012          92 : is_valid_dblink_option(const PQconninfoOption *options, const char *option,
    3013             :                        Oid context)
    3014             : {
    3015             :     const PQconninfoOption *opt;
    3016             : 
    3017             :     /* Look up the option in libpq result */
    3018        1538 :     for (opt = options; opt->keyword; opt++)
    3019             :     {
    3020        1536 :         if (strcmp(opt->keyword, option) == 0)
    3021          90 :             break;
    3022             :     }
    3023          92 :     if (opt->keyword == NULL)
    3024           2 :         return false;
    3025             : 
    3026             :     /* Disallow debug options (particularly "replication") */
    3027          90 :     if (strchr(opt->dispchar, 'D'))
    3028           6 :         return false;
    3029             : 
    3030             :     /* Disallow "client_encoding" */
    3031          84 :     if (strcmp(opt->keyword, "client_encoding") == 0)
    3032           2 :         return false;
    3033             : 
    3034             :     /*
    3035             :      * If the option is "user" or marked secure, it should be specified only
    3036             :      * in USER MAPPING.  Others should be specified only in SERVER.
    3037             :      */
    3038          82 :     if (strcmp(opt->keyword, "user") == 0 || strchr(opt->dispchar, '*'))
    3039             :     {
    3040          12 :         if (context != UserMappingRelationId)
    3041           0 :             return false;
    3042             :     }
    3043             :     else
    3044             :     {
    3045          70 :         if (context != ForeignServerRelationId)
    3046          58 :             return false;
    3047             :     }
    3048             : 
    3049          24 :     return true;
    3050             : }
    3051             : 
    3052             : /*
    3053             :  * Copy the remote session's values of GUCs that affect datatype I/O
    3054             :  * and apply them locally in a new GUC nesting level.  Returns the new
    3055             :  * nestlevel (which is needed by restoreLocalGucs to undo the settings),
    3056             :  * or -1 if no new nestlevel was needed.
    3057             :  *
    3058             :  * We use the equivalent of a function SET option to allow the settings to
    3059             :  * persist only until the caller calls restoreLocalGucs.  If an error is
    3060             :  * thrown in between, guc.c will take care of undoing the settings.
    3061             :  */
    3062             : static int
    3063          64 : applyRemoteGucs(PGconn *conn)
    3064             : {
    3065             :     static const char *const GUCsAffectingIO[] = {
    3066             :         "DateStyle",
    3067             :         "IntervalStyle"
    3068             :     };
    3069             : 
    3070          64 :     int         nestlevel = -1;
    3071             :     int         i;
    3072             : 
    3073         192 :     for (i = 0; i < lengthof(GUCsAffectingIO); i++)
    3074             :     {
    3075         128 :         const char *gucName = GUCsAffectingIO[i];
    3076         128 :         const char *remoteVal = PQparameterStatus(conn, gucName);
    3077             :         const char *localVal;
    3078             : 
    3079             :         /*
    3080             :          * If the remote server is pre-8.4, it won't have IntervalStyle, but
    3081             :          * that's okay because its output format won't be ambiguous.  So just
    3082             :          * skip the GUC if we don't get a value for it.  (We might eventually
    3083             :          * need more complicated logic with remote-version checks here.)
    3084             :          */
    3085         128 :         if (remoteVal == NULL)
    3086           0 :             continue;
    3087             : 
    3088             :         /*
    3089             :          * Avoid GUC-setting overhead if the remote and local GUCs already
    3090             :          * have the same value.
    3091             :          */
    3092         128 :         localVal = GetConfigOption(gucName, false, false);
    3093             :         Assert(localVal != NULL);
    3094             : 
    3095         128 :         if (strcmp(remoteVal, localVal) == 0)
    3096          94 :             continue;
    3097             : 
    3098             :         /* Create new GUC nest level if we didn't already */
    3099          34 :         if (nestlevel < 0)
    3100          18 :             nestlevel = NewGUCNestLevel();
    3101             : 
    3102             :         /* Apply the option (this will throw error on failure) */
    3103          34 :         (void) set_config_option(gucName, remoteVal,
    3104             :                                  PGC_USERSET, PGC_S_SESSION,
    3105             :                                  GUC_ACTION_SAVE, true, 0, false);
    3106             :     }
    3107             : 
    3108          64 :     return nestlevel;
    3109             : }
    3110             : 
    3111             : /*
    3112             :  * Restore local GUCs after they have been overlaid with remote settings.
    3113             :  */
    3114             : static void
    3115          66 : restoreLocalGucs(int nestlevel)
    3116             : {
    3117             :     /* Do nothing if no new nestlevel was created */
    3118          66 :     if (nestlevel > 0)
    3119          16 :         AtEOXact_GUC(true, nestlevel);
    3120          66 : }

Generated by: LCOV version 1.13