dblink.c File Reference

#include "postgres.h"
#include <limits.h>
#include "access/htup_details.h"
#include "access/relation.h"
#include "access/reloptions.h"
#include "access/table.h"
#include "catalog/namespace.h"
#include "catalog/pg_foreign_data_wrapper.h"
#include "catalog/pg_foreign_server.h"
#include "catalog/pg_type.h"
#include "catalog/pg_user_mapping.h"
#include "commands/defrem.h"
#include "common/base64.h"
#include "executor/spi.h"
#include "foreign/foreign.h"
#include "funcapi.h"
#include "lib/stringinfo.h"
#include "libpq-fe.h"
#include "libpq/libpq-be.h"
#include "libpq/libpq-be-fe-helpers.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "parser/scansup.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/guc.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/tuplestore.h"
#include "utils/varlena.h"
#include "utils/wait_event.h"

Include dependency graph for dblink.c:

Go to the source code of this file.

Data Structures
struct	remoteConn
struct	storeInfo
struct	remoteConnHashEnt

Macros
#define	NUMCONN   16
#define	DBLINK_NOTIFY_COLS   3

Typedefs
typedef struct remoteConn	remoteConn
typedef struct storeInfo	storeInfo
typedef struct remoteConnHashEnt	remoteConnHashEnt

Functions
	PG_MODULE_MAGIC_EXT (.name="dblink",.version=PG_VERSION)
static Datum	dblink_record_internal (FunctionCallInfo fcinfo, bool is_async)
static void	prepTuplestoreResult (FunctionCallInfo fcinfo)
static void	materializeResult (FunctionCallInfo fcinfo, PGconn *conn, PGresult *res)
static void	materializeQueryResult (FunctionCallInfo fcinfo, PGconn *conn, const char *conname, const char *sql, bool fail)
static PGresult *	storeQueryResult (storeInfo *sinfo, PGconn *conn, const char *sql)
static void	storeRow (storeInfo *sinfo, PGresult *res, bool first)
static remoteConn *	getConnectionByName (const char *name)
static HTAB *	createConnHash (void)
static remoteConn *	createNewConnection (const char *name)
static void	deleteConnection (const char *name)
static char **	get_pkey_attnames (Relation rel, int16 *indnkeyatts)
static char **	get_text_array_contents (ArrayType *array, int *numitems)
static char *	get_sql_insert (Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals, char **tgt_pkattvals)
static char *	get_sql_delete (Relation rel, int *pkattnums, int pknumatts, char **tgt_pkattvals)
static char *	get_sql_update (Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals, char **tgt_pkattvals)
static char *	quote_ident_cstr (char *rawstr)
static int	get_attnum_pk_pos (int *pkattnums, int pknumatts, int key)
static HeapTuple	get_tuple_of_interest (Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals)
static Relation	get_rel_from_relname (text *relname_text, LOCKMODE lockmode, AclMode aclmode)
static char *	generate_relation_name (Relation rel)
static void	dblink_connstr_check (const char *connstr)
static bool	dblink_connstr_has_pw (const char *connstr)
static void	dblink_security_check (PGconn *conn, const char *connname, const char *connstr)
static void	dblink_res_error (PGconn *conn, const char *conname, PGresult *res, bool fail, const char *fmt,...) pg_attribute_printf(5
static void static char *	get_connect_string (const char *servername)
static char *	escape_param_str (const char *str)
static void	validate_pkattnums (Relation rel, int2vector *pkattnums_arg, int32 pknumatts_arg, int **pkattnums, int *pknumatts)
static bool	is_valid_dblink_option (const PQconninfoOption *options, const char *option, Oid context)
static int	applyRemoteGucs (PGconn *conn)
static void	restoreLocalGucs (int nestlevel)
static bool	UseScramPassthrough (ForeignServer *foreign_server, UserMapping *user)
static void	appendSCRAMKeysInfo (StringInfo buf)
static bool	is_valid_dblink_fdw_option (const PQconninfoOption *options, const char *option, Oid context)
static bool	dblink_connstr_has_required_scram_options (const char *connstr)
static pg_noreturn void	dblink_res_internalerror (PGconn *conn, PGresult *res, const char *p2)
static pg_noreturn void	dblink_conn_not_avail (const char *conname)
static void	dblink_get_conn (char *conname_or_str, PGconn *volatile *conn_p, char **conname_p, volatile bool *freeconn_p)
static PGconn *	dblink_get_named_conn (const char *conname)
static void	dblink_init (void)
	PG_FUNCTION_INFO_V1 (dblink_connect)
Datum	dblink_connect (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_disconnect)
Datum	dblink_disconnect (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_open)
Datum	dblink_open (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_close)
Datum	dblink_close (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_fetch)
Datum	dblink_fetch (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_record)
Datum	dblink_record (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_send_query)
Datum	dblink_send_query (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_get_result)
Datum	dblink_get_result (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_get_connections)
Datum	dblink_get_connections (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_is_busy)
Datum	dblink_is_busy (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_cancel_query)
Datum	dblink_cancel_query (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_error_message)
Datum	dblink_error_message (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_exec)
Datum	dblink_exec (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_get_pkey)
Datum	dblink_get_pkey (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_build_sql_insert)
Datum	dblink_build_sql_insert (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_build_sql_delete)
Datum	dblink_build_sql_delete (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_build_sql_update)
Datum	dblink_build_sql_update (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_current_query)
Datum	dblink_current_query (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_get_notify)
Datum	dblink_get_notify (PG_FUNCTION_ARGS)
	PG_FUNCTION_INFO_V1 (dblink_fdw_validator)
Datum	dblink_fdw_validator (PG_FUNCTION_ARGS)
static void	RangeVarCallbackForDblink (const RangeVar *relation, Oid relId, Oid oldRelId, void *arg)

Variables
static remoteConn *	pconn = NULL
static HTAB *	remoteConnHash = NULL
static uint32	dblink_we_connect = 0
static uint32	dblink_we_get_conn = 0
static uint32	dblink_we_get_result = 0

Macro Definition Documentation

DBLINK_NOTIFY_COLS

#define DBLINK_NOTIFY_COLS   3

Definition at line 1878 of file dblink.c.

NUMCONN

#define NUMCONN   16

Definition at line 171 of file dblink.c.

Typedef Documentation

remoteConn

typedef struct remoteConn remoteConn

remoteConnHashEnt

typedef struct remoteConnHashEnt remoteConnHashEnt

storeInfo

typedef struct storeInfo storeInfo

Function Documentation

appendSCRAMKeysInfo()

static void appendSCRAMKeysInfo ( StringInfo  buf )

static

Definition at line 3198 of file dblink.c.

{
    int         len;
    int         encoded_len;
    char       *client_key;
    char       *server_key;
 
    len = pg_b64_enc_len(sizeof(MyProcPort->scram_ClientKey));
    /* don't forget the zero-terminator */
    client_key = palloc0(len + 1);
    encoded_len = pg_b64_encode(MyProcPort->scram_ClientKey,
                                sizeof(MyProcPort->scram_ClientKey),
                                client_key, len);
    if (encoded_len < 0)
        elog(ERROR, "could not encode SCRAM client key");
 
    len = pg_b64_enc_len(sizeof(MyProcPort->scram_ServerKey));
    /* don't forget the zero-terminator */
    server_key = palloc0(len + 1);
    encoded_len = pg_b64_encode(MyProcPort->scram_ServerKey,
                                sizeof(MyProcPort->scram_ServerKey),
                                server_key, len);
    if (encoded_len < 0)
        elog(ERROR, "could not encode SCRAM server key");
 
    appendStringInfo(buf, "scram_client_key='%s' ", client_key);
    appendStringInfo(buf, "scram_server_key='%s' ", server_key);
    appendStringInfoString(buf, "require_auth='scram-sha-256' ");
 
    pfree(client_key);
    pfree(server_key);
}

References appendStringInfo(), appendStringInfoString(), buf, elog, ERROR, fb(), len, MyProcPort, palloc0(), pfree(), pg_b64_enc_len(), pg_b64_encode(), Port::scram_ClientKey, and Port::scram_ServerKey.

Referenced by get_connect_string().

applyRemoteGucs()

static int applyRemoteGucs ( PGconn *  conn )

static

Definition at line 3134 of file dblink.c.

{
    static const char *const GUCsAffectingIO[] = {
        "DateStyle",
        "IntervalStyle"
    };
 
    int         nestlevel = -1;
    int         i;
 
    for (i = 0; i < lengthof(GUCsAffectingIO); i++)
    {
        const char *gucName = GUCsAffectingIO[i];
        const char *remoteVal = PQparameterStatus(conn, gucName);
        const char *localVal;
 
        /*
         * If the remote server is pre-8.4, it won't have IntervalStyle, but
         * that's okay because its output format won't be ambiguous.  So just
         * skip the GUC if we don't get a value for it.  (We might eventually
         * need more complicated logic with remote-version checks here.)
         */
        if (remoteVal == NULL)
            continue;
 
        /*
         * Avoid GUC-setting overhead if the remote and local GUCs already
         * have the same value.
         */
        localVal = GetConfigOption(gucName, false, false);
        Assert(localVal != NULL);
 
        if (strcmp(remoteVal, localVal) == 0)
            continue;
 
        /* Create new GUC nest level if we didn't already */
        if (nestlevel < 0)
            nestlevel = NewGUCNestLevel();
 
        /* Apply the option (this will throw error on failure) */
        (void) set_config_option(gucName, remoteVal,
                                 PGC_USERSET, PGC_S_SESSION,
                                 GUC_ACTION_SAVE, true, 0, false);
    }
 
    return nestlevel;
}

References Assert, conn, fb(), GetConfigOption(), GUC_ACTION_SAVE, i, lengthof, NewGUCNestLevel(), PGC_S_SESSION, PGC_USERSET, PQparameterStatus(), and set_config_option().

Referenced by materializeResult(), and storeQueryResult().

createConnHash()

static HTAB * createConnHash ( void  )

static

Definition at line 2548 of file dblink.c.

{
    HASHCTL     ctl;
 
    ctl.keysize = NAMEDATALEN;
    ctl.entrysize = sizeof(remoteConnHashEnt);
 
    return hash_create("Remote Con hash", NUMCONN, &ctl,
                       HASH_ELEM | HASH_STRINGS);
}

References ctl, hash_create(), HASH_ELEM, HASH_STRINGS, HASHCTL::keysize, NAMEDATALEN, and NUMCONN.

Referenced by createNewConnection(), deleteConnection(), and getConnectionByName().

createNewConnection()

static remoteConn * createNewConnection ( const char *  name )

static

Definition at line 2560 of file dblink.c.

{
    remoteConnHashEnt *hentry;
    bool        found;
    char       *key;
 
    if (!remoteConnHash)
        remoteConnHash = createConnHash();
 
    key = pstrdup(name);
    truncate_identifier(key, strlen(key), true);
    hentry = (remoteConnHashEnt *) hash_search(remoteConnHash, key,
                                               HASH_ENTER, &found);
 
    if (found && hentry->rconn.conn != NULL)
        ereport(ERROR,
                (errcode(ERRCODE_DUPLICATE_OBJECT),
                 errmsg("duplicate connection name")));
 
    /* New, or reusable, so initialize the rconn struct to zeroes */
    memset(&hentry->rconn, 0, sizeof(remoteConn));
 
    return &hentry->rconn;
}

References createConnHash(), ereport, errcode(), ERRCODE_DUPLICATE_OBJECT, errmsg, ERROR, fb(), HASH_ENTER, hash_search(), name, pstrdup(), remoteConnHash, and truncate_identifier().

Referenced by dblink_connect().

dblink_build_sql_delete()

Datum dblink_build_sql_delete

(

PG_FUNCTION_ARGS

)

Definition at line 1709 of file dblink.c.

{
    text       *relname_text = PG_GETARG_TEXT_PP(0);
    int2vector *pkattnums_arg = (int2vector *) PG_GETARG_POINTER(1);
    int32       pknumatts_arg = PG_GETARG_INT32(2);
    ArrayType  *tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
    Relation    rel;
    int        *pkattnums;
    int         pknumatts;
    char      **tgt_pkattvals;
    int         tgt_nitems;
    char       *sql;
 
    /*
     * Open target relation.
     */
    rel = get_rel_from_relname(relname_text, AccessShareLock, ACL_SELECT);
 
    /*
     * Process pkattnums argument.
     */
    validate_pkattnums(rel, pkattnums_arg, pknumatts_arg,
                       &pkattnums, &pknumatts);
 
    /*
     * Target array is made up of key values that will be used to build the
     * SQL string for use on the remote system.
     */
    tgt_pkattvals = get_text_array_contents(tgt_pkattvals_arry, &tgt_nitems);
 
    /*
     * There should be one target array key value for each key attnum
     */
    if (tgt_nitems != pknumatts)
        ereport(ERROR,
                (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
                 errmsg("target key array length must match number of key attributes")));
 
    /*
     * Prep work is finally done. Go get the SQL string.
     */
    sql = get_sql_delete(rel, pkattnums, pknumatts, tgt_pkattvals);
 
    /*
     * Now we can close the relation.
     */
    relation_close(rel, AccessShareLock);
 
    /*
     * And send it
     */
    PG_RETURN_TEXT_P(cstring_to_text(sql));
}

References AccessShareLock, ACL_SELECT, cstring_to_text(), ereport, errcode(), errmsg, ERROR, fb(), get_rel_from_relname(), get_sql_delete(), get_text_array_contents(), PG_GETARG_ARRAYTYPE_P, PG_GETARG_INT32, PG_GETARG_POINTER, PG_GETARG_TEXT_PP, PG_RETURN_TEXT_P, relation_close(), and validate_pkattnums().

dblink_build_sql_insert()

Datum dblink_build_sql_insert

(

PG_FUNCTION_ARGS

)

Definition at line 1620 of file dblink.c.

{
    text       *relname_text = PG_GETARG_TEXT_PP(0);
    int2vector *pkattnums_arg = (int2vector *) PG_GETARG_POINTER(1);
    int32       pknumatts_arg = PG_GETARG_INT32(2);
    ArrayType  *src_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
    ArrayType  *tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(4);
    Relation    rel;
    int        *pkattnums;
    int         pknumatts;
    char      **src_pkattvals;
    char      **tgt_pkattvals;
    int         src_nitems;
    int         tgt_nitems;
    char       *sql;
 
    /*
     * Open target relation.
     */
    rel = get_rel_from_relname(relname_text, AccessShareLock, ACL_SELECT);
 
    /*
     * Process pkattnums argument.
     */
    validate_pkattnums(rel, pkattnums_arg, pknumatts_arg,
                       &pkattnums, &pknumatts);
 
    /*
     * Source array is made up of key values that will be used to locate the
     * tuple of interest from the local system.
     */
    src_pkattvals = get_text_array_contents(src_pkattvals_arry, &src_nitems);
 
    /*
     * There should be one source array key value for each key attnum
     */
    if (src_nitems != pknumatts)
        ereport(ERROR,
                (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
                 errmsg("source key array length must match number of key attributes")));
 
    /*
     * Target array is made up of key values that will be used to build the
     * SQL string for use on the remote system.
     */
    tgt_pkattvals = get_text_array_contents(tgt_pkattvals_arry, &tgt_nitems);
 
    /*
     * There should be one target array key value for each key attnum
     */
    if (tgt_nitems != pknumatts)
        ereport(ERROR,
                (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
                 errmsg("target key array length must match number of key attributes")));
 
    /*
     * Prep work is finally done. Go get the SQL string.
     */
    sql = get_sql_insert(rel, pkattnums, pknumatts, src_pkattvals, tgt_pkattvals);
 
    /*
     * Now we can close the relation.
     */
    relation_close(rel, AccessShareLock);
 
    /*
     * And send it
     */
    PG_RETURN_TEXT_P(cstring_to_text(sql));
}

References AccessShareLock, ACL_SELECT, cstring_to_text(), ereport, errcode(), errmsg, ERROR, fb(), get_rel_from_relname(), get_sql_insert(), get_text_array_contents(), PG_GETARG_ARRAYTYPE_P, PG_GETARG_INT32, PG_GETARG_POINTER, PG_GETARG_TEXT_PP, PG_RETURN_TEXT_P, relation_close(), and validate_pkattnums().

dblink_build_sql_update()

Datum dblink_build_sql_update

(

PG_FUNCTION_ARGS

)

Definition at line 1785 of file dblink.c.

{
    text       *relname_text = PG_GETARG_TEXT_PP(0);
    int2vector *pkattnums_arg = (int2vector *) PG_GETARG_POINTER(1);
    int32       pknumatts_arg = PG_GETARG_INT32(2);
    ArrayType  *src_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
    ArrayType  *tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(4);
    Relation    rel;
    int        *pkattnums;
    int         pknumatts;
    char      **src_pkattvals;
    char      **tgt_pkattvals;
    int         src_nitems;
    int         tgt_nitems;
    char       *sql;
 
    /*
     * Open target relation.
     */
    rel = get_rel_from_relname(relname_text, AccessShareLock, ACL_SELECT);
 
    /*
     * Process pkattnums argument.
     */
    validate_pkattnums(rel, pkattnums_arg, pknumatts_arg,
                       &pkattnums, &pknumatts);
 
    /*
     * Source array is made up of key values that will be used to locate the
     * tuple of interest from the local system.
     */
    src_pkattvals = get_text_array_contents(src_pkattvals_arry, &src_nitems);
 
    /*
     * There should be one source array key value for each key attnum
     */
    if (src_nitems != pknumatts)
        ereport(ERROR,
                (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
                 errmsg("source key array length must match number of key attributes")));
 
    /*
     * Target array is made up of key values that will be used to build the
     * SQL string for use on the remote system.
     */
    tgt_pkattvals = get_text_array_contents(tgt_pkattvals_arry, &tgt_nitems);
 
    /*
     * There should be one target array key value for each key attnum
     */
    if (tgt_nitems != pknumatts)
        ereport(ERROR,
                (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
                 errmsg("target key array length must match number of key attributes")));
 
    /*
     * Prep work is finally done. Go get the SQL string.
     */
    sql = get_sql_update(rel, pkattnums, pknumatts, src_pkattvals, tgt_pkattvals);
 
    /*
     * Now we can close the relation.
     */
    relation_close(rel, AccessShareLock);
 
    /*
     * And send it
     */
    PG_RETURN_TEXT_P(cstring_to_text(sql));
}

References AccessShareLock, ACL_SELECT, cstring_to_text(), ereport, errcode(), errmsg, ERROR, fb(), get_rel_from_relname(), get_sql_update(), get_text_array_contents(), PG_GETARG_ARRAYTYPE_P, PG_GETARG_INT32, PG_GETARG_POINTER, PG_GETARG_TEXT_PP, PG_RETURN_TEXT_P, relation_close(), and validate_pkattnums().

dblink_cancel_query()

Datum dblink_cancel_query

(

PG_FUNCTION_ARGS

)

Definition at line 1341 of file dblink.c.

{
    PGconn     *conn;
    const char *msg;
    TimestampTz endtime;
 
    dblink_init();
    conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
    endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(),
                                          30000);
    msg = libpqsrv_cancel(conn, endtime);
    if (msg == NULL)
        msg = "OK";
 
    PG_RETURN_TEXT_P(cstring_to_text(msg));
}

References conn, cstring_to_text(), dblink_get_named_conn(), dblink_init(), fb(), GetCurrentTimestamp(), libpqsrv_cancel(), PG_GETARG_TEXT_PP, PG_RETURN_TEXT_P, text_to_cstring(), and TimestampTzPlusMilliseconds.

dblink_close()

Datum dblink_close

(

PG_FUNCTION_ARGS

)

Definition at line 496 of file dblink.c.

{
    PGconn     *conn;
    PGresult   *res = NULL;
    char       *curname = NULL;
    char       *conname = NULL;
    StringInfoData buf;
    remoteConn *rconn = NULL;
    bool        fail = true;    /* default to backward compatible behavior */
 
    dblink_init();
    initStringInfo(&buf);
 
    if (PG_NARGS() == 1)
    {
        /* text */
        curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
        rconn = pconn;
    }
    else if (PG_NARGS() == 2)
    {
        /* might be text,text or text,bool */
        if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
        {
            curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
            fail = PG_GETARG_BOOL(1);
            rconn = pconn;
        }
        else
        {
            conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
            curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
            rconn = getConnectionByName(conname);
        }
    }
    if (PG_NARGS() == 3)
    {
        /* text,text,bool */
        conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
        curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
        fail = PG_GETARG_BOOL(2);
        rconn = getConnectionByName(conname);
    }
 
    if (!rconn || !rconn->conn)
        dblink_conn_not_avail(conname);
 
    conn = rconn->conn;
 
    appendStringInfo(&buf, "CLOSE %s", curname);
 
    /* close the cursor */
    res = libpqsrv_exec(conn, buf.data, dblink_we_get_result);
    if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
    {
        dblink_res_error(conn, conname, res, fail,
                         "while closing cursor \"%s\"", curname);
        PG_RETURN_TEXT_P(cstring_to_text("ERROR"));
    }
 
    PQclear(res);
 
    /* if we started a transaction, decrement cursor count */
    if (rconn->newXactForCursor)
    {
        (rconn->openCursorCount)--;
 
        /* if count is zero, commit the transaction */
        if (rconn->openCursorCount == 0)
        {
            rconn->newXactForCursor = false;
 
            res = libpqsrv_exec(conn, "COMMIT", dblink_we_get_result);
            if (PQresultStatus(res) != PGRES_COMMAND_OK)
                dblink_res_internalerror(conn, res, "commit error");
            PQclear(res);
        }
    }
 
    PG_RETURN_TEXT_P(cstring_to_text("OK"));
}

References appendStringInfo(), buf, remoteConn::conn, conn, cstring_to_text(), dblink_conn_not_avail(), dblink_init(), dblink_res_error(), dblink_res_internalerror(), dblink_we_get_result, fb(), get_fn_expr_argtype(), getConnectionByName(), initStringInfo(), libpqsrv_exec(), remoteConn::newXactForCursor, remoteConn::openCursorCount, pconn, PG_GETARG_BOOL, PG_GETARG_TEXT_PP, PG_NARGS, PG_RETURN_TEXT_P, PGRES_COMMAND_OK, PQclear, PQresultStatus, and text_to_cstring().

dblink_conn_not_avail()

static pg_noreturn void dblink_conn_not_avail ( const char *  conname )

static

Definition at line 183 of file dblink.c.

{
    if (conname)
        ereport(ERROR,
                (errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST),
                 errmsg("connection \"%s\" not available", conname)));
    else
        ereport(ERROR,
                (errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST),
                 errmsg("connection not available")));
}

References ereport, errcode(), errmsg, ERROR, and fb().

Referenced by dblink_close(), dblink_disconnect(), dblink_exec(), dblink_fetch(), dblink_get_named_conn(), dblink_open(), and dblink_record_internal().

dblink_connect()

Datum dblink_connect

(

PG_FUNCTION_ARGS

)

Definition at line 284 of file dblink.c.

{
    char       *conname_or_str = NULL;
    char       *connstr = NULL;
    char       *connname = NULL;
    char       *msg;
    PGconn     *conn = NULL;
    remoteConn *rconn = NULL;
 
    dblink_init();
 
    if (PG_NARGS() == 2)
    {
        conname_or_str = text_to_cstring(PG_GETARG_TEXT_PP(1));
        connname = text_to_cstring(PG_GETARG_TEXT_PP(0));
    }
    else if (PG_NARGS() == 1)
        conname_or_str = text_to_cstring(PG_GETARG_TEXT_PP(0));
 
    /* first check for valid foreign data server */
    connstr = get_connect_string(conname_or_str);
    if (connstr == NULL)
        connstr = conname_or_str;
 
    /* check password in connection string if not superuser */
    dblink_connstr_check(connstr);
 
    /* first time, allocate or get the custom wait event */
    if (dblink_we_connect == 0)
        dblink_we_connect = WaitEventExtensionNew("DblinkConnect");
 
    /* if we need a hashtable entry, make that first, since it might fail */
    if (connname)
    {
        rconn = createNewConnection(connname);
        Assert(rconn->conn == NULL);
    }
 
    /* OK to make connection */
    conn = libpqsrv_connect(connstr, dblink_we_connect);
 
    if (PQstatus(conn) == CONNECTION_BAD)
    {
        msg = pchomp(PQerrorMessage(conn));
        libpqsrv_disconnect(conn);
        if (connname)
            deleteConnection(connname);
 
        ereport(ERROR,
                (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
                 errmsg("could not establish connection"),
                 errdetail_internal("%s", msg)));
    }
 
    PQsetNoticeReceiver(conn, libpqsrv_notice_receiver,
                        "received message via remote connection");
 
    /* check password actually used if not superuser */
    dblink_security_check(conn, connname, connstr);
 
    /* attempt to set client encoding to match server encoding, if needed */
    if (PQclientEncoding(conn) != GetDatabaseEncoding())
        PQsetClientEncoding(conn, GetDatabaseEncodingName());
 
    /* all OK, save away the conn */
    if (connname)
    {
        rconn->conn = conn;
    }
    else
    {
        if (pconn->conn)
            libpqsrv_disconnect(pconn->conn);
        pconn->conn = conn;
    }
 
    PG_RETURN_TEXT_P(cstring_to_text("OK"));
}

References Assert, remoteConn::conn, conn, CONNECTION_BAD, connstr, createNewConnection(), cstring_to_text(), dblink_connstr_check(), dblink_init(), dblink_security_check(), dblink_we_connect, deleteConnection(), ereport, errcode(), errdetail_internal(), errmsg, ERROR, fb(), get_connect_string(), GetDatabaseEncoding(), GetDatabaseEncodingName(), libpqsrv_connect(), libpqsrv_disconnect(), libpqsrv_notice_receiver(), pchomp(), pconn, PG_GETARG_TEXT_PP, PG_NARGS, PG_RETURN_TEXT_P, PQclientEncoding(), PQerrorMessage(), PQsetClientEncoding(), PQsetNoticeReceiver(), PQstatus(), text_to_cstring(), and WaitEventExtensionNew().

dblink_connstr_check()

static void dblink_connstr_check ( const char *  connstr )

static

Definition at line 2760 of file dblink.c.

{
    if (superuser())
        return;
 
    if (dblink_connstr_has_pw(connstr))
        return;
 
    if (MyProcPort != NULL && MyProcPort->has_scram_keys && dblink_connstr_has_required_scram_options(connstr))
        return;
 
#ifdef ENABLE_GSS
    if (be_gssapi_get_delegation(MyProcPort))
        return;
#endif
 
    ereport(ERROR,
            (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
             errmsg("password or GSSAPI delegated credentials required"),
             errdetail("Non-superusers must provide a password in the connection string or send delegated GSSAPI credentials.")));
}

References be_gssapi_get_delegation(), connstr, dblink_connstr_has_pw(), dblink_connstr_has_required_scram_options(), ereport, errcode(), errdetail(), errmsg, ERROR, fb(), Port::has_scram_keys, MyProcPort, and superuser().

Referenced by dblink_connect(), and dblink_get_conn().

dblink_connstr_has_pw()

static bool dblink_connstr_has_pw ( const char *  connstr )

static

Definition at line 2723 of file dblink.c.

{
    PQconninfoOption *options;
    PQconninfoOption *option;
    bool        connstr_gives_password = false;
 
    options = PQconninfoParse(connstr, NULL);
    if (options)
    {
        for (option = options; option->keyword != NULL; option++)
        {
            if (strcmp(option->keyword, "password") == 0)
            {
                if (option->val != NULL && option->val[0] != '\0')
                {
                    connstr_gives_password = true;
                    break;
                }
            }
        }
        PQconninfoFree(options);
    }
 
    return connstr_gives_password;
}

References connstr, fb(), PQconninfoFree(), PQconninfoParse(), and option::val.

Referenced by dblink_connstr_check(), and dblink_security_check().

dblink_connstr_has_required_scram_options()

bool dblink_connstr_has_required_scram_options ( const char *  connstr )

static

Definition at line 2617 of file dblink.c.

{
    PQconninfoOption *options;
    bool        has_scram_server_key = false;
    bool        has_scram_client_key = false;
    bool        has_require_auth = false;
    bool        has_scram_keys = false;
 
    options = PQconninfoParse(connstr, NULL);
    if (options)
    {
        /*
         * Continue iterating even if we found the keys that we need to
         * validate to make sure that there is no other declaration of these
         * keys that can overwrite the first.
         */
        for (PQconninfoOption *option = options; option->keyword != NULL; option++)
        {
            if (strcmp(option->keyword, "require_auth") == 0)
            {
                if (option->val != NULL && strcmp(option->val, "scram-sha-256") == 0)
                    has_require_auth = true;
                else
                    has_require_auth = false;
            }
 
            if (strcmp(option->keyword, "scram_client_key") == 0)
            {
                if (option->val != NULL && option->val[0] != '\0')
                    has_scram_client_key = true;
                else
                    has_scram_client_key = false;
            }
 
            if (strcmp(option->keyword, "scram_server_key") == 0)
            {
                if (option->val != NULL && option->val[0] != '\0')
                    has_scram_server_key = true;
                else
                    has_scram_server_key = false;
            }
        }
        PQconninfoFree(options);
    }
 
    has_scram_keys = has_scram_client_key && has_scram_server_key && MyProcPort != NULL && MyProcPort->has_scram_keys;
 
    return (has_scram_keys && has_require_auth);
}

References connstr, fb(), Port::has_scram_keys, MyProcPort, PQconninfoFree(), PQconninfoParse(), and option::val.

Referenced by dblink_connstr_check(), and dblink_security_check().

dblink_current_query()

Datum dblink_current_query

(

PG_FUNCTION_ARGS

)

Definition at line 1864 of file dblink.c.

{
    /* This is now just an alias for the built-in function current_query() */
    PG_RETURN_DATUM(current_query(fcinfo));
}

References current_query(), and PG_RETURN_DATUM.

dblink_disconnect()

Datum dblink_disconnect

(

PG_FUNCTION_ARGS

)

Definition at line 368 of file dblink.c.

{
    char       *conname = NULL;
    remoteConn *rconn = NULL;
    PGconn     *conn = NULL;
 
    dblink_init();
 
    if (PG_NARGS() == 1)
    {
        conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
        rconn = getConnectionByName(conname);
        if (rconn)
            conn = rconn->conn;
    }
    else
        conn = pconn->conn;
 
    if (!conn)
        dblink_conn_not_avail(conname);
 
    libpqsrv_disconnect(conn);
    if (rconn)
        deleteConnection(conname);
    else
        pconn->conn = NULL;
 
    PG_RETURN_TEXT_P(cstring_to_text("OK"));
}

References remoteConn::conn, conn, cstring_to_text(), dblink_conn_not_avail(), dblink_init(), deleteConnection(), fb(), getConnectionByName(), libpqsrv_disconnect(), pconn, PG_GETARG_TEXT_PP, PG_NARGS, PG_RETURN_TEXT_P, and text_to_cstring().

dblink_error_message()

Datum dblink_error_message

(

PG_FUNCTION_ARGS

)

Definition at line 1371 of file dblink.c.

{
    char       *msg;
    PGconn     *conn;
 
    dblink_init();
    conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
 
    msg = PQerrorMessage(conn);
    if (msg == NULL || msg[0] == '\0')
        PG_RETURN_TEXT_P(cstring_to_text("OK"));
    else
        PG_RETURN_TEXT_P(cstring_to_text(pchomp(msg)));
}

References conn, cstring_to_text(), dblink_get_named_conn(), dblink_init(), fb(), pchomp(), PG_GETARG_TEXT_PP, PG_RETURN_TEXT_P, PQerrorMessage(), and text_to_cstring().

dblink_exec()

Datum dblink_exec

(

PG_FUNCTION_ARGS

)

Definition at line 1391 of file dblink.c.

{
    text       *volatile sql_cmd_status = NULL;
    PGconn     *volatile conn = NULL;
    volatile bool freeconn = false;
 
    dblink_init();
 
    PG_TRY();
    {
        PGresult   *res = NULL;
        char       *sql = NULL;
        char       *conname = NULL;
        bool        fail = true;    /* default to backward compatible behavior */
 
        if (PG_NARGS() == 3)
        {
            /* must be text,text,bool */
            conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
            sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
            fail = PG_GETARG_BOOL(2);
            dblink_get_conn(conname, &conn, &conname, &freeconn);
        }
        else if (PG_NARGS() == 2)
        {
            /* might be text,text or text,bool */
            if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
            {
                sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
                fail = PG_GETARG_BOOL(1);
                conn = pconn->conn;
            }
            else
            {
                conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
                sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
                dblink_get_conn(conname, &conn, &conname, &freeconn);
            }
        }
        else if (PG_NARGS() == 1)
        {
            /* must be single text argument */
            conn = pconn->conn;
            sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
        }
        else
            /* shouldn't happen */
            elog(ERROR, "wrong number of arguments");
 
        if (!conn)
            dblink_conn_not_avail(conname);
 
        res = libpqsrv_exec(conn, sql, dblink_we_get_result);
        if (!res ||
            (PQresultStatus(res) != PGRES_COMMAND_OK &&
             PQresultStatus(res) != PGRES_TUPLES_OK))
        {
            dblink_res_error(conn, conname, res, fail,
                             "while executing command");
 
            /*
             * and save a copy of the command status string to return as our
             * result tuple
             */
            sql_cmd_status = cstring_to_text("ERROR");
        }
        else if (PQresultStatus(res) == PGRES_COMMAND_OK)
        {
            /*
             * and save a copy of the command status string to return as our
             * result tuple
             */
            sql_cmd_status = cstring_to_text(PQcmdStatus(res));
            PQclear(res);
        }
        else
        {
            PQclear(res);
            ereport(ERROR,
                    (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
                     errmsg("statement returning results not allowed")));
        }
    }
    PG_FINALLY();
    {
        /* if needed, close the connection to the database */
        if (freeconn)
            libpqsrv_disconnect(conn);
    }
    PG_END_TRY();
 
    PG_RETURN_TEXT_P(sql_cmd_status);
}

References remoteConn::conn, conn, cstring_to_text(), dblink_conn_not_avail(), dblink_get_conn(), dblink_init(), dblink_res_error(), dblink_we_get_result, elog, ereport, errcode(), errmsg, ERROR, fb(), get_fn_expr_argtype(), libpqsrv_disconnect(), libpqsrv_exec(), pconn, PG_END_TRY, PG_FINALLY, PG_GETARG_BOOL, PG_GETARG_TEXT_PP, PG_NARGS, PG_RETURN_TEXT_P, PG_TRY, PGRES_COMMAND_OK, PGRES_TUPLES_OK, PQclear, PQcmdStatus, PQresultStatus, and text_to_cstring().

dblink_fdw_validator()

Datum dblink_fdw_validator

(

PG_FUNCTION_ARGS

)

Definition at line 1935 of file dblink.c.

{
    List       *options_list = untransformRelOptions(PG_GETARG_DATUM(0));
    Oid         context = PG_GETARG_OID(1);
    ListCell   *cell;
 
    static const PQconninfoOption *options = NULL;
 
    /*
     * Get list of valid libpq options.
     *
     * To avoid unnecessary work, we get the list once and use it throughout
     * the lifetime of this backend process.  We don't need to care about
     * memory context issues, because PQconndefaults allocates with malloc.
     */
    if (!options)
    {
        options = PQconndefaults();
        if (!options)           /* assume reason for failure is OOM */
            ereport(ERROR,
                    (errcode(ERRCODE_FDW_OUT_OF_MEMORY),
                     errmsg("out of memory"),
                     errdetail("Could not get libpq's default connection options.")));
    }
 
    /* Validate each supplied option. */
    foreach(cell, options_list)
    {
        DefElem    *def = (DefElem *) lfirst(cell);
 
        if (!is_valid_dblink_fdw_option(options, def->defname, context))
        {
            /*
             * Unknown option, or invalid option for the context specified, so
             * complain about it.  Provide a hint with a valid option that
             * looks similar, if there is one.
             */
            const PQconninfoOption *opt;
            const char *closest_match;
            ClosestMatchState match_state;
            bool        has_valid_options = false;
 
            initClosestMatch(&match_state, def->defname, 4);
            for (opt = options; opt->keyword; opt++)
            {
                if (is_valid_dblink_option(options, opt->keyword, context))
                {
                    has_valid_options = true;
                    updateClosestMatch(&match_state, opt->keyword);
                }
            }
 
            closest_match = getClosestMatch(&match_state);
            ereport(ERROR,
                    (errcode(ERRCODE_FDW_OPTION_NAME_NOT_FOUND),
                     errmsg("invalid option \"%s\"", def->defname),
                     has_valid_options ? closest_match ?
                     errhint("Perhaps you meant the option \"%s\".",
                             closest_match) : 0 :
                     errhint("There are no valid options in this context.")));
        }
    }
 
    PG_RETURN_VOID();
}

References DefElem::defname, ereport, errcode(), errdetail(), errhint(), errmsg, ERROR, fb(), getClosestMatch(), initClosestMatch(), is_valid_dblink_fdw_option(), is_valid_dblink_option(), _PQconninfoOption::keyword, lfirst, match_state, PG_GETARG_DATUM, PG_GETARG_OID, PG_RETURN_VOID, PQconndefaults(), untransformRelOptions(), and updateClosestMatch().

dblink_fetch()

Datum dblink_fetch

(

PG_FUNCTION_ARGS

)

Definition at line 583 of file dblink.c.

{
    PGresult   *res = NULL;
    char       *conname = NULL;
    remoteConn *rconn = NULL;
    PGconn     *conn = NULL;
    StringInfoData buf;
    char       *curname = NULL;
    int         howmany = 0;
    bool        fail = true;    /* default to backward compatible */
 
    prepTuplestoreResult(fcinfo);
 
    dblink_init();
 
    if (PG_NARGS() == 4)
    {
        /* text,text,int,bool */
        conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
        curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
        howmany = PG_GETARG_INT32(2);
        fail = PG_GETARG_BOOL(3);
 
        rconn = getConnectionByName(conname);
        if (rconn)
            conn = rconn->conn;
    }
    else if (PG_NARGS() == 3)
    {
        /* text,text,int or text,int,bool */
        if (get_fn_expr_argtype(fcinfo->flinfo, 2) == BOOLOID)
        {
            curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
            howmany = PG_GETARG_INT32(1);
            fail = PG_GETARG_BOOL(2);
            conn = pconn->conn;
        }
        else
        {
            conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
            curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
            howmany = PG_GETARG_INT32(2);
 
            rconn = getConnectionByName(conname);
            if (rconn)
                conn = rconn->conn;
        }
    }
    else if (PG_NARGS() == 2)
    {
        /* text,int */
        curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
        howmany = PG_GETARG_INT32(1);
        conn = pconn->conn;
    }
 
    if (!conn)
        dblink_conn_not_avail(conname);
 
    initStringInfo(&buf);
    appendStringInfo(&buf, "FETCH %d FROM %s", howmany, curname);
 
    /*
     * Try to execute the query.  Note that since libpq uses malloc, the
     * PGresult will be long-lived even though we are still in a short-lived
     * memory context.
     */
    res = libpqsrv_exec(conn, buf.data, dblink_we_get_result);
    if (!res ||
        (PQresultStatus(res) != PGRES_COMMAND_OK &&
         PQresultStatus(res) != PGRES_TUPLES_OK))
    {
        dblink_res_error(conn, conname, res, fail,
                         "while fetching from cursor \"%s\"", curname);
        return (Datum) 0;
    }
    else if (PQresultStatus(res) == PGRES_COMMAND_OK)
    {
        /* cursor does not exist - closed already or bad name */
        PQclear(res);
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_CURSOR_NAME),
                 errmsg("cursor \"%s\" does not exist", curname)));
    }
 
    materializeResult(fcinfo, conn, res);
    return (Datum) 0;
}

References appendStringInfo(), buf, remoteConn::conn, conn, dblink_conn_not_avail(), dblink_init(), dblink_res_error(), dblink_we_get_result, ereport, errcode(), errmsg, ERROR, fb(), get_fn_expr_argtype(), getConnectionByName(), initStringInfo(), libpqsrv_exec(), materializeResult(), pconn, PG_GETARG_BOOL, PG_GETARG_INT32, PG_GETARG_TEXT_PP, PG_NARGS, PGRES_COMMAND_OK, PGRES_TUPLES_OK, PQclear, PQresultStatus, prepTuplestoreResult(), and text_to_cstring().

dblink_get_conn()

static void dblink_get_conn ( char *  conname_or_str, PGconn *volatile *  conn_p, char **  conname_p, volatile bool *  freeconn_p  )

static

Definition at line 196 of file dblink.c.

{
    remoteConn *rconn = getConnectionByName(conname_or_str);
    PGconn     *conn;
    char       *conname;
    bool        freeconn;
 
    if (rconn)
    {
        conn = rconn->conn;
        conname = conname_or_str;
        freeconn = false;
    }
    else
    {
        const char *connstr;
 
        connstr = get_connect_string(conname_or_str);
        if (connstr == NULL)
            connstr = conname_or_str;
        dblink_connstr_check(connstr);
 
        /* first time, allocate or get the custom wait event */
        if (dblink_we_get_conn == 0)
            dblink_we_get_conn = WaitEventExtensionNew("DblinkGetConnect");
 
        /* OK to make connection */
        conn = libpqsrv_connect(connstr, dblink_we_get_conn);
 
        if (PQstatus(conn) == CONNECTION_BAD)
        {
            char       *msg = pchomp(PQerrorMessage(conn));
 
            libpqsrv_disconnect(conn);
            ereport(ERROR,
                    (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
                     errmsg("could not establish connection"),
                     errdetail_internal("%s", msg)));
        }
 
        PQsetNoticeReceiver(conn, libpqsrv_notice_receiver,
                            "received message via remote connection");
 
        dblink_security_check(conn, NULL, connstr);
        if (PQclientEncoding(conn) != GetDatabaseEncoding())
            PQsetClientEncoding(conn, GetDatabaseEncodingName());
        freeconn = true;
        conname = NULL;
    }
 
    *conn_p = conn;
    *conname_p = conname;
    *freeconn_p = freeconn;
}

References remoteConn::conn, conn, CONNECTION_BAD, connstr, dblink_connstr_check(), dblink_security_check(), dblink_we_get_conn, ereport, errcode(), errdetail_internal(), errmsg, ERROR, fb(), get_connect_string(), getConnectionByName(), GetDatabaseEncoding(), GetDatabaseEncodingName(), libpqsrv_connect(), libpqsrv_disconnect(), libpqsrv_notice_receiver(), pchomp(), PQclientEncoding(), PQerrorMessage(), PQsetClientEncoding(), PQsetNoticeReceiver(), PQstatus(), and WaitEventExtensionNew().

Referenced by dblink_exec(), and dblink_record_internal().

dblink_get_connections()

Datum dblink_get_connections

(

PG_FUNCTION_ARGS

)

Definition at line 1279 of file dblink.c.

{
    HASH_SEQ_STATUS status;
    remoteConnHashEnt *hentry;
    ArrayBuildState *astate = NULL;
 
    if (remoteConnHash)
    {
        hash_seq_init(&status, remoteConnHash);
        while ((hentry = (remoteConnHashEnt *) hash_seq_search(&status)) != NULL)
        {
            /* ignore it if it's not an open connection */
            if (hentry->rconn.conn == NULL)
                continue;
            /* stash away current value */
            astate = accumArrayResult(astate,
                                      CStringGetTextDatum(hentry->name),
                                      false, TEXTOID, CurrentMemoryContext);
        }
    }
 
    if (astate)
        PG_RETURN_DATUM(makeArrayResult(astate,
                                        CurrentMemoryContext));
    else
        PG_RETURN_NULL();
}

References accumArrayResult(), CStringGetTextDatum, CurrentMemoryContext, fb(), hash_seq_init(), hash_seq_search(), makeArrayResult(), PG_RETURN_DATUM, PG_RETURN_NULL, and remoteConnHash.

dblink_get_named_conn()

static PGconn * dblink_get_named_conn ( const char *  conname )

static

Definition at line 253 of file dblink.c.

{
    remoteConn *rconn = getConnectionByName(conname);
 
    if (rconn)
        return rconn->conn;
 
    dblink_conn_not_avail(conname);
    return NULL;                /* keep compiler quiet */
}

References remoteConn::conn, dblink_conn_not_avail(), fb(), and getConnectionByName().

Referenced by dblink_cancel_query(), dblink_error_message(), dblink_get_notify(), dblink_is_busy(), dblink_record_internal(), and dblink_send_query().

dblink_get_notify()

Datum dblink_get_notify

(

PG_FUNCTION_ARGS

)

Definition at line 1882 of file dblink.c.

{
    PGconn     *conn;
    PGnotify   *notify;
    ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 
    dblink_init();
    if (PG_NARGS() == 1)
        conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
    else
        conn = pconn->conn;
 
    InitMaterializedSRF(fcinfo, 0);
 
    PQconsumeInput(conn);
    while ((notify = PQnotifies(conn)) != NULL)
    {
        Datum       values[DBLINK_NOTIFY_COLS];
        bool        nulls[DBLINK_NOTIFY_COLS];
 
        memset(values, 0, sizeof(values));
        memset(nulls, 0, sizeof(nulls));
 
        if (notify->relname != NULL)
            values[0] = CStringGetTextDatum(notify->relname);
        else
            nulls[0] = true;
 
        values[1] = Int32GetDatum(notify->be_pid);
 
        if (notify->extra != NULL)
            values[2] = CStringGetTextDatum(notify->extra);
        else
            nulls[2] = true;
 
        tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls);
 
        PQfreemem(notify);
        PQconsumeInput(conn);
    }
 
    return (Datum) 0;
}

References remoteConn::conn, conn, CStringGetTextDatum, dblink_get_named_conn(), dblink_init(), DBLINK_NOTIFY_COLS, fb(), InitMaterializedSRF(), Int32GetDatum(), pconn, PG_GETARG_TEXT_PP, PG_NARGS, PQconsumeInput(), PQfreemem(), PQnotifies(), text_to_cstring(), tuplestore_putvalues(), and values.

dblink_get_pkey()

Datum dblink_get_pkey

(

PG_FUNCTION_ARGS

)

Definition at line 1494 of file dblink.c.

{
    int16       indnkeyatts;
    char      **results;
    FuncCallContext *funcctx;
    int32       call_cntr;
    int32       max_calls;
    AttInMetadata *attinmeta;
    MemoryContext oldcontext;
 
    /* stuff done only on the first call of the function */
    if (SRF_IS_FIRSTCALL())
    {
        Relation    rel;
        TupleDesc   tupdesc;
 
        /* create a function context for cross-call persistence */
        funcctx = SRF_FIRSTCALL_INIT();
 
        /*
         * switch to memory context appropriate for multiple function calls
         */
        oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
 
        /* open target relation */
        rel = get_rel_from_relname(PG_GETARG_TEXT_PP(0), AccessShareLock, ACL_SELECT);
 
        /* get the array of attnums */
        results = get_pkey_attnames(rel, &indnkeyatts);
 
        relation_close(rel, AccessShareLock);
 
        /*
         * need a tuple descriptor representing one INT and one TEXT column
         */
        tupdesc = CreateTemplateTupleDesc(2);
        TupleDescInitEntry(tupdesc, (AttrNumber) 1, "position",
                           INT4OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber) 2, "colname",
                           TEXTOID, -1, 0);
 
        TupleDescFinalize(tupdesc);
 
        /*
         * Generate attribute metadata needed later to produce tuples from raw
         * C strings
         */
        attinmeta = TupleDescGetAttInMetadata(tupdesc);
        funcctx->attinmeta = attinmeta;
 
        if ((results != NULL) && (indnkeyatts > 0))
        {
            funcctx->max_calls = indnkeyatts;
 
            /* got results, keep track of them */
            funcctx->user_fctx = results;
        }
        else
        {
            /* fast track when no results */
            MemoryContextSwitchTo(oldcontext);
            SRF_RETURN_DONE(funcctx);
        }
 
        MemoryContextSwitchTo(oldcontext);
    }
 
    /* stuff done on every call of the function */
    funcctx = SRF_PERCALL_SETUP();
 
    /*
     * initialize per-call variables
     */
    call_cntr = funcctx->call_cntr;
    max_calls = funcctx->max_calls;
 
    results = (char **) funcctx->user_fctx;
    attinmeta = funcctx->attinmeta;
 
    if (call_cntr < max_calls)  /* do when there is more left to send */
    {
        char      **values;
        HeapTuple   tuple;
        Datum       result;
 
        values = palloc_array(char *, 2);
        values[0] = psprintf("%d", call_cntr + 1);
        values[1] = results[call_cntr];
 
        /* build the tuple */
        tuple = BuildTupleFromCStrings(attinmeta, values);
 
        /* make the tuple into a datum */
        result = HeapTupleGetDatum(tuple);
 
        SRF_RETURN_NEXT(funcctx, result);
    }
    else
    {
        /* do when there is no more left */
        SRF_RETURN_DONE(funcctx);
    }
}

References AccessShareLock, ACL_SELECT, BuildTupleFromCStrings(), CreateTemplateTupleDesc(), fb(), get_pkey_attnames(), get_rel_from_relname(), HeapTupleGetDatum(), MemoryContextSwitchTo(), palloc_array, PG_GETARG_TEXT_PP, psprintf(), relation_close(), SRF_FIRSTCALL_INIT, SRF_IS_FIRSTCALL, SRF_PERCALL_SETUP, SRF_RETURN_DONE, SRF_RETURN_NEXT, TupleDescFinalize(), TupleDescGetAttInMetadata(), TupleDescInitEntry(), and values.

dblink_get_result()

Datum dblink_get_result

(

PG_FUNCTION_ARGS

)

Definition at line 709 of file dblink.c.

{
    return dblink_record_internal(fcinfo, true);
}

References dblink_record_internal().

dblink_init()

static void dblink_init ( void  )

static

Definition at line 265 of file dblink.c.

{
    if (!pconn)
    {
        if (dblink_we_get_result == 0)
            dblink_we_get_result = WaitEventExtensionNew("DblinkGetResult");
 
        pconn = (remoteConn *) MemoryContextAlloc(TopMemoryContext, sizeof(remoteConn));
        pconn->conn = NULL;
        pconn->openCursorCount = 0;
        pconn->newXactForCursor = false;
    }
}

References remoteConn::conn, dblink_we_get_result, fb(), MemoryContextAlloc(), remoteConn::newXactForCursor, remoteConn::openCursorCount, pconn, TopMemoryContext, and WaitEventExtensionNew().

Referenced by dblink_cancel_query(), dblink_close(), dblink_connect(), dblink_disconnect(), dblink_error_message(), dblink_exec(), dblink_fetch(), dblink_get_notify(), dblink_is_busy(), dblink_open(), and dblink_record_internal().

dblink_is_busy()

Datum dblink_is_busy

(

PG_FUNCTION_ARGS

)

Definition at line 1317 of file dblink.c.

{
    PGconn     *conn;
 
    dblink_init();
    conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
 
    PQconsumeInput(conn);
    PG_RETURN_INT32(PQisBusy(conn));
}

References conn, dblink_get_named_conn(), dblink_init(), PG_GETARG_TEXT_PP, PG_RETURN_INT32, PQconsumeInput(), PQisBusy(), and text_to_cstring().

dblink_open()

Datum dblink_open

(

PG_FUNCTION_ARGS

)

Definition at line 403 of file dblink.c.

{
    PGresult   *res = NULL;
    PGconn     *conn;
    char       *curname = NULL;
    char       *sql = NULL;
    char       *conname = NULL;
    StringInfoData buf;
    remoteConn *rconn = NULL;
    bool        fail = true;    /* default to backward compatible behavior */
 
    dblink_init();
    initStringInfo(&buf);
 
    if (PG_NARGS() == 2)
    {
        /* text,text */
        curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
        sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
        rconn = pconn;
    }
    else if (PG_NARGS() == 3)
    {
        /* might be text,text,text or text,text,bool */
        if (get_fn_expr_argtype(fcinfo->flinfo, 2) == BOOLOID)
        {
            curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
            sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
            fail = PG_GETARG_BOOL(2);
            rconn = pconn;
        }
        else
        {
            conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
            curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
            sql = text_to_cstring(PG_GETARG_TEXT_PP(2));
            rconn = getConnectionByName(conname);
        }
    }
    else if (PG_NARGS() == 4)
    {
        /* text,text,text,bool */
        conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
        curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
        sql = text_to_cstring(PG_GETARG_TEXT_PP(2));
        fail = PG_GETARG_BOOL(3);
        rconn = getConnectionByName(conname);
    }
 
    if (!rconn || !rconn->conn)
        dblink_conn_not_avail(conname);
 
    conn = rconn->conn;
 
    /* If we are not in a transaction, start one */
    if (PQtransactionStatus(conn) == PQTRANS_IDLE)
    {
        res = libpqsrv_exec(conn, "BEGIN", dblink_we_get_result);
        if (PQresultStatus(res) != PGRES_COMMAND_OK)
            dblink_res_internalerror(conn, res, "begin error");
        PQclear(res);
        rconn->newXactForCursor = true;
 
        /*
         * Since transaction state was IDLE, we force cursor count to
         * initially be 0. This is needed as a previous ABORT might have wiped
         * out our transaction without maintaining the cursor count for us.
         */
        rconn->openCursorCount = 0;
    }
 
    /* if we started a transaction, increment cursor count */
    if (rconn->newXactForCursor)
        (rconn->openCursorCount)++;
 
    appendStringInfo(&buf, "DECLARE %s CURSOR FOR %s", curname, sql);
    res = libpqsrv_exec(conn, buf.data, dblink_we_get_result);
    if (!res || PQresultStatus(res) != PGRES_COMMAND_OK)
    {
        dblink_res_error(conn, conname, res, fail,
                         "while opening cursor \"%s\"", curname);
        PG_RETURN_TEXT_P(cstring_to_text("ERROR"));
    }
 
    PQclear(res);
    PG_RETURN_TEXT_P(cstring_to_text("OK"));
}

References appendStringInfo(), buf, remoteConn::conn, conn, cstring_to_text(), dblink_conn_not_avail(), dblink_init(), dblink_res_error(), dblink_res_internalerror(), dblink_we_get_result, fb(), get_fn_expr_argtype(), getConnectionByName(), initStringInfo(), libpqsrv_exec(), remoteConn::newXactForCursor, remoteConn::openCursorCount, pconn, PG_GETARG_BOOL, PG_GETARG_TEXT_PP, PG_NARGS, PG_RETURN_TEXT_P, PGRES_COMMAND_OK, PQclear, PQresultStatus, PQTRANS_IDLE, PQtransactionStatus(), and text_to_cstring().

dblink_record()

Datum dblink_record

(

PG_FUNCTION_ARGS

)

Definition at line 677 of file dblink.c.

{
    return dblink_record_internal(fcinfo, false);
}

References dblink_record_internal().

dblink_record_internal()

static Datum dblink_record_internal ( FunctionCallInfo  fcinfo, bool  is_async  )

static

Definition at line 715 of file dblink.c.

{
    PGconn     *volatile conn = NULL;
    volatile bool freeconn = false;
 
    prepTuplestoreResult(fcinfo);
 
    dblink_init();
 
    PG_TRY();
    {
        char       *sql = NULL;
        char       *conname = NULL;
        bool        fail = true;    /* default to backward compatible */
 
        if (!is_async)
        {
            if (PG_NARGS() == 3)
            {
                /* text,text,bool */
                conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
                sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
                fail = PG_GETARG_BOOL(2);
                dblink_get_conn(conname, &conn, &conname, &freeconn);
            }
            else if (PG_NARGS() == 2)
            {
                /* text,text or text,bool */
                if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
                {
                    sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
                    fail = PG_GETARG_BOOL(1);
                    conn = pconn->conn;
                }
                else
                {
                    conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
                    sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
                    dblink_get_conn(conname, &conn, &conname, &freeconn);
                }
            }
            else if (PG_NARGS() == 1)
            {
                /* text */
                conn = pconn->conn;
                sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
            }
            else
                /* shouldn't happen */
                elog(ERROR, "wrong number of arguments");
        }
        else                    /* is_async */
        {
            /* get async result */
            conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
 
            if (PG_NARGS() == 2)
            {
                /* text,bool */
                fail = PG_GETARG_BOOL(1);
                conn = dblink_get_named_conn(conname);
            }
            else if (PG_NARGS() == 1)
            {
                /* text */
                conn = dblink_get_named_conn(conname);
            }
            else
                /* shouldn't happen */
                elog(ERROR, "wrong number of arguments");
        }
 
        if (!conn)
            dblink_conn_not_avail(conname);
 
        if (!is_async)
        {
            /* synchronous query, use efficient tuple collection method */
            materializeQueryResult(fcinfo, conn, conname, sql, fail);
        }
        else
        {
            /* async result retrieval, do it the old way */
            PGresult   *res = libpqsrv_get_result(conn, dblink_we_get_result);
 
            /* NULL means we're all done with the async results */
            if (res)
            {
                if (PQresultStatus(res) != PGRES_COMMAND_OK &&
                    PQresultStatus(res) != PGRES_TUPLES_OK)
                {
                    dblink_res_error(conn, conname, res, fail,
                                     "while executing query");
                    /* if fail isn't set, we'll return an empty query result */
                }
                else
                {
                    materializeResult(fcinfo, conn, res);
                }
            }
        }
    }
    PG_FINALLY();
    {
        /* if needed, close the connection to the database */
        if (freeconn)
            libpqsrv_disconnect(conn);
    }
    PG_END_TRY();
 
    return (Datum) 0;
}

References remoteConn::conn, conn, dblink_conn_not_avail(), dblink_get_conn(), dblink_get_named_conn(), dblink_init(), dblink_res_error(), dblink_we_get_result, elog, ERROR, fb(), FunctionCallInfoBaseData::flinfo, get_fn_expr_argtype(), libpqsrv_disconnect(), libpqsrv_get_result(), materializeQueryResult(), materializeResult(), pconn, PG_END_TRY, PG_FINALLY, PG_GETARG_BOOL, PG_GETARG_TEXT_PP, PG_NARGS, PG_TRY, PGRES_COMMAND_OK, PGRES_TUPLES_OK, PQresultStatus, prepTuplestoreResult(), and text_to_cstring().

Referenced by dblink_get_result(), and dblink_record().

dblink_res_error()

static void dblink_res_error ( PGconn *  conn, const char *  conname, PGresult *  res, bool  fail, const char *  fmt,   ...  )

static

Definition at line 2794 of file dblink.c.

{
    int         level;
    char       *pg_diag_sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
    char       *message_primary = PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY);
    char       *message_detail = PQresultErrorField(res, PG_DIAG_MESSAGE_DETAIL);
    char       *message_hint = PQresultErrorField(res, PG_DIAG_MESSAGE_HINT);
    char       *message_context = PQresultErrorField(res, PG_DIAG_CONTEXT);
    int         sqlstate;
    va_list     ap;
    char        dblink_context_msg[512];
 
    if (fail)
        level = ERROR;
    else
        level = NOTICE;
 
    if (pg_diag_sqlstate)
        sqlstate = MAKE_SQLSTATE(pg_diag_sqlstate[0],
                                 pg_diag_sqlstate[1],
                                 pg_diag_sqlstate[2],
                                 pg_diag_sqlstate[3],
                                 pg_diag_sqlstate[4]);
    else
        sqlstate = ERRCODE_CONNECTION_FAILURE;
 
    /*
     * If we don't get a message from the PGresult, try the PGconn.  This is
     * needed because for connection-level failures, PQgetResult may just
     * return NULL, not a PGresult at all.
     */
    if (message_primary == NULL)
        message_primary = pchomp(PQerrorMessage(conn));
 
    /*
     * Format the basic errcontext string.  Below, we'll add on something
     * about the connection name.  That's a violation of the translatability
     * guidelines about constructing error messages out of parts, but since
     * there's no translation support for dblink, there's no need to worry
     * about that (yet).
     */
    va_start(ap, fmt);
    vsnprintf(dblink_context_msg, sizeof(dblink_context_msg), fmt, ap);
    va_end(ap);
 
    ereport(level,
            (errcode(sqlstate),
             (message_primary != NULL && message_primary[0] != '\0') ?
             errmsg_internal("%s", message_primary) :
             errmsg("could not obtain message string for remote error"),
             message_detail ? errdetail_internal("%s", message_detail) : 0,
             message_hint ? errhint("%s", message_hint) : 0,
             message_context ? (errcontext("%s", message_context)) : 0,
             conname ?
             (errcontext("%s on dblink connection named \"%s\"",
                         dblink_context_msg, conname)) :
             (errcontext("%s on unnamed dblink connection",
                         dblink_context_msg))));
    PQclear(res);
}

References conn, ereport, errcode(), errcontext, errdetail_internal(), errhint(), errmsg, errmsg_internal(), ERROR, fb(), MAKE_SQLSTATE, NOTICE, pchomp(), PG_DIAG_CONTEXT, PG_DIAG_MESSAGE_DETAIL, PG_DIAG_MESSAGE_HINT, PG_DIAG_MESSAGE_PRIMARY, PG_DIAG_SQLSTATE, PQclear, PQerrorMessage(), PQresultErrorField, and vsnprintf.

Referenced by dblink_close(), dblink_exec(), dblink_fetch(), dblink_open(), dblink_record_internal(), and materializeQueryResult().

dblink_res_internalerror()

static pg_noreturn void dblink_res_internalerror ( PGconn *  conn, PGresult *  res, const char *  p2  )

static

Definition at line 174 of file dblink.c.

{
    char       *msg = pchomp(PQerrorMessage(conn));
 
    PQclear(res);
    elog(ERROR, "%s: %s", p2, msg);
}

References conn, elog, ERROR, fb(), pchomp(), PQclear, and PQerrorMessage().

Referenced by dblink_close(), and dblink_open().

dblink_security_check()

static void dblink_security_check ( PGconn *  conn, const char *  connname, const char *  connstr  )

static

Definition at line 2676 of file dblink.c.

{
    /* Superuser bypasses security check */
    if (superuser())
        return;
 
    /* If password was used to connect, make sure it was one provided */
    if (PQconnectionUsedPassword(conn) && dblink_connstr_has_pw(connstr))
        return;
 
    /*
     * Password was not used to connect, check if SCRAM pass-through is in
     * use.
     *
     * If dblink_connstr_has_required_scram_options is true we assume that
     * UseScramPassthrough is also true because the required SCRAM keys are
     * only added if UseScramPassthrough is set, and the user is not allowed
     * to add the SCRAM keys on fdw and user mapping options.
     */
    if (MyProcPort != NULL && MyProcPort->has_scram_keys && dblink_connstr_has_required_scram_options(connstr))
        return;
 
#ifdef ENABLE_GSS
    /* If GSSAPI creds used to connect, make sure it was one delegated */
    if (PQconnectionUsedGSSAPI(conn) && be_gssapi_get_delegation(MyProcPort))
        return;
#endif
 
    /* Otherwise, fail out */
    libpqsrv_disconnect(conn);
    if (connname)
        deleteConnection(connname);
 
    ereport(ERROR,
            (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
             errmsg("password or GSSAPI delegated credentials required"),
             errdetail("Non-superusers may only connect using credentials they provide, eg: password in connection string or delegated GSSAPI credentials"),
             errhint("Ensure provided credentials match target server's authentication method.")));
}

References be_gssapi_get_delegation(), conn, connstr, dblink_connstr_has_pw(), dblink_connstr_has_required_scram_options(), deleteConnection(), ereport, errcode(), errdetail(), errhint(), errmsg, ERROR, fb(), Port::has_scram_keys, libpqsrv_disconnect(), MyProcPort, PQconnectionUsedGSSAPI(), PQconnectionUsedPassword(), and superuser().

Referenced by dblink_connect(), and dblink_get_conn().

dblink_send_query()

Datum dblink_send_query

(

PG_FUNCTION_ARGS

)

Definition at line 684 of file dblink.c.

{
    PGconn     *conn;
    char       *sql;
    int         retval;
 
    if (PG_NARGS() == 2)
    {
        conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
        sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
    }
    else
        /* shouldn't happen */
        elog(ERROR, "wrong number of arguments");
 
    /* async query send */
    retval = PQsendQuery(conn, sql);
    if (retval != 1)
        elog(NOTICE, "could not send query: %s", pchomp(PQerrorMessage(conn)));
 
    PG_RETURN_INT32(retval);
}

References conn, dblink_get_named_conn(), elog, ERROR, NOTICE, pchomp(), PG_GETARG_TEXT_PP, PG_NARGS, PG_RETURN_INT32, PQerrorMessage(), PQsendQuery(), and text_to_cstring().

deleteConnection()

static void deleteConnection ( const char *  name )

static

Definition at line 2586 of file dblink.c.

{
    remoteConnHashEnt *hentry;
    bool        found;
    char       *key;
 
    if (!remoteConnHash)
        remoteConnHash = createConnHash();
 
    key = pstrdup(name);
    truncate_identifier(key, strlen(key), false);
    hentry = (remoteConnHashEnt *) hash_search(remoteConnHash,
                                               key, HASH_REMOVE, &found);
 
    if (!hentry)
        ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_OBJECT),
                 errmsg("undefined connection name")));
}

References createConnHash(), ereport, errcode(), errmsg, ERROR, fb(), HASH_REMOVE, hash_search(), name, pstrdup(), remoteConnHash, and truncate_identifier().

Referenced by dblink_connect(), dblink_disconnect(), and dblink_security_check().

escape_param_str()

static char * escape_param_str ( const char *  str )

static

Definition at line 2958 of file dblink.c.

{
    const char *cp;
    StringInfoData buf;
 
    initStringInfo(&buf);
 
    for (cp = str; *cp; cp++)
    {
        if (*cp == '\\' || *cp == '\'')
            appendStringInfoChar(&buf, '\\');
        appendStringInfoChar(&buf, *cp);
    }
 
    return buf.data;
}

References appendStringInfoChar(), buf, fb(), initStringInfo(), and str.

Referenced by get_connect_string().

generate_relation_name()

static char * generate_relation_name ( Relation  rel )

static

Definition at line 2510 of file dblink.c.

{
    char       *nspname;
    char       *result;
 
    /* Qualify the name if not visible in search path */
    if (RelationIsVisible(RelationGetRelid(rel)))
        nspname = NULL;
    else
        nspname = get_namespace_name(rel->rd_rel->relnamespace);
 
    result = quote_qualified_identifier(nspname, RelationGetRelationName(rel));
 
    return result;
}

References fb(), get_namespace_name(), quote_qualified_identifier(), RelationData::rd_rel, RelationGetRelationName, RelationGetRelid, and RelationIsVisible().

Referenced by get_sql_delete(), get_sql_insert(), get_sql_update(), and get_tuple_of_interest().

get_attnum_pk_pos()

static int get_attnum_pk_pos ( int *  pkattnums, int  pknumatts, int  key  )

static

Definition at line 2350 of file dblink.c.

{
    int         i;
 
    /*
     * Not likely a long list anyway, so just scan for the value
     */
    for (i = 0; i < pknumatts; i++)
        if (key == pkattnums[i])
            return i;
 
    return -1;
}

References fb(), and i.

Referenced by get_sql_insert(), and get_sql_update().

get_connect_string()

static char * get_connect_string ( const char *  servername )

static

Definition at line 2860 of file dblink.c.

{
    ForeignServer *foreign_server = NULL;
    UserMapping *user_mapping;
    ListCell   *cell;
    StringInfoData buf;
    ForeignDataWrapper *fdw;
    AclResult   aclresult;
    char       *srvname;
 
    static const PQconninfoOption *options = NULL;
 
    initStringInfo(&buf);
 
    /*
     * Get list of valid libpq options.
     *
     * To avoid unnecessary work, we get the list once and use it throughout
     * the lifetime of this backend process.  We don't need to care about
     * memory context issues, because PQconndefaults allocates with malloc.
     */
    if (!options)
    {
        options = PQconndefaults();
        if (!options)           /* assume reason for failure is OOM */
            ereport(ERROR,
                    (errcode(ERRCODE_FDW_OUT_OF_MEMORY),
                     errmsg("out of memory"),
                     errdetail("Could not get libpq's default connection options.")));
    }
 
    /* first gather the server connstr options */
    srvname = pstrdup(servername);
    truncate_identifier(srvname, strlen(srvname), false);
    foreign_server = GetForeignServerByName(srvname, true);
 
    if (foreign_server)
    {
        Oid         serverid = foreign_server->serverid;
        Oid         fdwid = foreign_server->fdwid;
        Oid         userid = GetUserId();
 
        user_mapping = GetUserMapping(userid, serverid);
        fdw = GetForeignDataWrapper(fdwid);
 
        /* Check permissions, user must have usage on the server. */
        aclresult = object_aclcheck(ForeignServerRelationId, serverid, userid, ACL_USAGE);
        if (aclresult != ACLCHECK_OK)
            aclcheck_error(aclresult, OBJECT_FOREIGN_SERVER, foreign_server->servername);
 
        /*
         * First append hardcoded options needed for SCRAM pass-through, so if
         * the user overwrites these options we can ereport on
         * dblink_connstr_check and dblink_security_check.
         */
        if (MyProcPort != NULL && MyProcPort->has_scram_keys && UseScramPassthrough(foreign_server, user_mapping))
            appendSCRAMKeysInfo(&buf);
 
        foreach(cell, fdw->options)
        {
            DefElem    *def = lfirst(cell);
 
            if (is_valid_dblink_option(options, def->defname, ForeignDataWrapperRelationId))
                appendStringInfo(&buf, "%s='%s' ", def->defname,
                                 escape_param_str(strVal(def->arg)));
        }
 
        foreach(cell, foreign_server->options)
        {
            DefElem    *def = lfirst(cell);
 
            if (is_valid_dblink_option(options, def->defname, ForeignServerRelationId))
                appendStringInfo(&buf, "%s='%s' ", def->defname,
                                 escape_param_str(strVal(def->arg)));
        }
 
        foreach(cell, user_mapping->options)
        {
 
            DefElem    *def = lfirst(cell);
 
            if (is_valid_dblink_option(options, def->defname, UserMappingRelationId))
                appendStringInfo(&buf, "%s='%s' ", def->defname,
                                 escape_param_str(strVal(def->arg)));
        }
 
        return buf.data;
    }
    else
        return NULL;
}

References ACL_USAGE, aclcheck_error(), ACLCHECK_OK, appendSCRAMKeysInfo(), appendStringInfo(), DefElem::arg, buf, DefElem::defname, ereport, errcode(), errdetail(), errmsg, ERROR, escape_param_str(), fb(), ForeignServer::fdwid, GetForeignDataWrapper(), GetForeignServerByName(), GetUserId(), GetUserMapping(), Port::has_scram_keys, initStringInfo(), is_valid_dblink_option(), lfirst, MyProcPort, object_aclcheck(), OBJECT_FOREIGN_SERVER, ForeignServer::options, PQconndefaults(), pstrdup(), ForeignServer::serverid, ForeignServer::servername, strVal, truncate_identifier(), and UseScramPassthrough().

Referenced by dblink_connect(), and dblink_get_conn().

get_pkey_attnames()

static char ** get_pkey_attnames ( Relation  rel, int16 *  indnkeyatts  )

static

Definition at line 2014 of file dblink.c.

{
    Relation    indexRelation;
    ScanKeyData skey;
    SysScanDesc scan;
    HeapTuple   indexTuple;
    int         i;
    char      **result = NULL;
    TupleDesc   tupdesc;
 
    /* initialize indnkeyatts to 0 in case no primary key exists */
    *indnkeyatts = 0;
 
    tupdesc = rel->rd_att;
 
    /* Prepare to scan pg_index for entries having indrelid = this rel. */
    indexRelation = table_open(IndexRelationId, AccessShareLock);
    ScanKeyInit(&skey,
                Anum_pg_index_indrelid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(RelationGetRelid(rel)));
 
    scan = systable_beginscan(indexRelation, IndexIndrelidIndexId, true,
                              NULL, 1, &skey);
 
    while (HeapTupleIsValid(indexTuple = systable_getnext(scan)))
    {
        Form_pg_index index = (Form_pg_index) GETSTRUCT(indexTuple);
 
        /* we're only interested if it is the primary key */
        if (index->indisprimary)
        {
            *indnkeyatts = index->indnkeyatts;
            if (*indnkeyatts > 0)
            {
                result = palloc_array(char *, *indnkeyatts);
 
                for (i = 0; i < *indnkeyatts; i++)
                    result[i] = SPI_fname(tupdesc, index->indkey.values[i]);
            }
            break;
        }
    }
 
    systable_endscan(scan);
    table_close(indexRelation, AccessShareLock);
 
    return result;
}

References AccessShareLock, BTEqualStrategyNumber, fb(), Form_pg_index, GETSTRUCT(), HeapTupleIsValid, i, ObjectIdGetDatum(), palloc_array, RelationData::rd_att, RelationGetRelid, ScanKeyInit(), SPI_fname(), systable_beginscan(), systable_endscan(), systable_getnext(), table_close(), and table_open().

Referenced by dblink_get_pkey().

get_rel_from_relname()

static Relation get_rel_from_relname ( text *  relname_text, LOCKMODE  lockmode, AclMode  aclmode  )

static

Definition at line 2491 of file dblink.c.

{
    RangeVar   *relvar;
    Oid         relid;
 
    relvar = makeRangeVarFromNameList(textToQualifiedNameList(relname_text));
    relid = RangeVarGetRelidExtended(relvar, lockmode, 0,
                                     RangeVarCallbackForDblink, &aclmode);
 
    return table_open(relid, NoLock);
}

References fb(), makeRangeVarFromNameList(), NoLock, RangeVarCallbackForDblink(), RangeVarGetRelidExtended(), table_open(), and textToQualifiedNameList().

Referenced by dblink_build_sql_delete(), dblink_build_sql_insert(), dblink_build_sql_update(), and dblink_get_pkey().

get_sql_delete()

static char * get_sql_delete ( Relation  rel, int *  pkattnums, int  pknumatts, char **  tgt_pkattvals  )

static

Definition at line 2207 of file dblink.c.

{
    char       *relname;
    TupleDesc   tupdesc;
    StringInfoData buf;
    int         i;
 
    initStringInfo(&buf);
 
    /* get relation name including any needed schema prefix and quoting */
    relname = generate_relation_name(rel);
 
    tupdesc = rel->rd_att;
 
    appendStringInfo(&buf, "DELETE FROM %s WHERE ", relname);
    for (i = 0; i < pknumatts; i++)
    {
        int         pkattnum = pkattnums[i];
        Form_pg_attribute attr = TupleDescAttr(tupdesc, pkattnum);
 
        if (i > 0)
            appendStringInfoString(&buf, " AND ");
 
        appendStringInfoString(&buf,
                               quote_ident_cstr(NameStr(attr->attname)));
 
        if (tgt_pkattvals[i] != NULL)
            appendStringInfo(&buf, " = %s",
                             quote_literal_cstr(tgt_pkattvals[i]));
        else
            appendStringInfoString(&buf, " IS NULL");
    }
 
    return buf.data;
}

References appendStringInfo(), appendStringInfoString(), buf, fb(), generate_relation_name(), i, initStringInfo(), NameStr, quote_ident_cstr(), quote_literal_cstr(), RelationData::rd_att, relname, and TupleDescAttr().

Referenced by dblink_build_sql_delete().

get_sql_insert()

static char * get_sql_insert ( Relation  rel, int *  pkattnums, int  pknumatts, char **  src_pkattvals, char **  tgt_pkattvals  )

static

Definition at line 2127 of file dblink.c.

{
    char       *relname;
    HeapTuple   tuple;
    TupleDesc   tupdesc;
    int         natts;
    StringInfoData buf;
    char       *val;
    int         key;
    int         i;
    bool        needComma;
 
    initStringInfo(&buf);
 
    /* get relation name including any needed schema prefix and quoting */
    relname = generate_relation_name(rel);
 
    tupdesc = rel->rd_att;
    natts = tupdesc->natts;
 
    tuple = get_tuple_of_interest(rel, pkattnums, pknumatts, src_pkattvals);
    if (!tuple)
        ereport(ERROR,
                (errcode(ERRCODE_CARDINALITY_VIOLATION),
                 errmsg("source row not found")));
 
    appendStringInfo(&buf, "INSERT INTO %s(", relname);
 
    needComma = false;
    for (i = 0; i < natts; i++)
    {
        Form_pg_attribute att = TupleDescAttr(tupdesc, i);
 
        if (att->attisdropped)
            continue;
 
        if (needComma)
            appendStringInfoChar(&buf, ',');
 
        appendStringInfoString(&buf,
                               quote_ident_cstr(NameStr(att->attname)));
        needComma = true;
    }
 
    appendStringInfoString(&buf, ") VALUES(");
 
    /*
     * Note: i is physical column number (counting from 0).
     */
    needComma = false;
    for (i = 0; i < natts; i++)
    {
        if (TupleDescAttr(tupdesc, i)->attisdropped)
            continue;
 
        if (needComma)
            appendStringInfoChar(&buf, ',');
 
        key = get_attnum_pk_pos(pkattnums, pknumatts, i);
 
        if (key >= 0)
            val = tgt_pkattvals[key] ? pstrdup(tgt_pkattvals[key]) : NULL;
        else
            val = SPI_getvalue(tuple, tupdesc, i + 1);
 
        if (val != NULL)
        {
            appendStringInfoString(&buf, quote_literal_cstr(val));
            pfree(val);
        }
        else
            appendStringInfoString(&buf, "NULL");
        needComma = true;
    }
    appendStringInfoChar(&buf, ')');
 
    return buf.data;
}

References appendStringInfo(), appendStringInfoChar(), appendStringInfoString(), buf, ereport, errcode(), errmsg, ERROR, fb(), generate_relation_name(), get_attnum_pk_pos(), get_tuple_of_interest(), i, initStringInfo(), NameStr, TupleDescData::natts, pfree(), pstrdup(), quote_ident_cstr(), quote_literal_cstr(), RelationData::rd_att, relname, SPI_getvalue(), TupleDescAttr(), and val.

Referenced by dblink_build_sql_insert().

get_sql_update()

static char * get_sql_update ( Relation  rel, int *  pkattnums, int  pknumatts, char **  src_pkattvals, char **  tgt_pkattvals  )

static

Definition at line 2244 of file dblink.c.

{
    char       *relname;
    HeapTuple   tuple;
    TupleDesc   tupdesc;
    int         natts;
    StringInfoData buf;
    char       *val;
    int         key;
    int         i;
    bool        needComma;
 
    initStringInfo(&buf);
 
    /* get relation name including any needed schema prefix and quoting */
    relname = generate_relation_name(rel);
 
    tupdesc = rel->rd_att;
    natts = tupdesc->natts;
 
    tuple = get_tuple_of_interest(rel, pkattnums, pknumatts, src_pkattvals);
    if (!tuple)
        ereport(ERROR,
                (errcode(ERRCODE_CARDINALITY_VIOLATION),
                 errmsg("source row not found")));
 
    appendStringInfo(&buf, "UPDATE %s SET ", relname);
 
    /*
     * Note: i is physical column number (counting from 0).
     */
    needComma = false;
    for (i = 0; i < natts; i++)
    {
        Form_pg_attribute attr = TupleDescAttr(tupdesc, i);
 
        if (attr->attisdropped)
            continue;
 
        if (needComma)
            appendStringInfoString(&buf, ", ");
 
        appendStringInfo(&buf, "%s = ",
                         quote_ident_cstr(NameStr(attr->attname)));
 
        key = get_attnum_pk_pos(pkattnums, pknumatts, i);
 
        if (key >= 0)
            val = tgt_pkattvals[key] ? pstrdup(tgt_pkattvals[key]) : NULL;
        else
            val = SPI_getvalue(tuple, tupdesc, i + 1);
 
        if (val != NULL)
        {
            appendStringInfoString(&buf, quote_literal_cstr(val));
            pfree(val);
        }
        else
            appendStringInfoString(&buf, "NULL");
        needComma = true;
    }
 
    appendStringInfoString(&buf, " WHERE ");
 
    for (i = 0; i < pknumatts; i++)
    {
        int         pkattnum = pkattnums[i];
        Form_pg_attribute attr = TupleDescAttr(tupdesc, pkattnum);
 
        if (i > 0)
            appendStringInfoString(&buf, " AND ");
 
        appendStringInfoString(&buf,
                               quote_ident_cstr(NameStr(attr->attname)));
 
        val = tgt_pkattvals[i];
 
        if (val != NULL)
            appendStringInfo(&buf, " = %s", quote_literal_cstr(val));
        else
            appendStringInfoString(&buf, " IS NULL");
    }
 
    return buf.data;
}

References appendStringInfo(), appendStringInfoString(), buf, ereport, errcode(), errmsg, ERROR, fb(), generate_relation_name(), get_attnum_pk_pos(), get_tuple_of_interest(), i, initStringInfo(), NameStr, TupleDescData::natts, pfree(), pstrdup(), quote_ident_cstr(), quote_literal_cstr(), RelationData::rd_att, relname, SPI_getvalue(), TupleDescAttr(), and val.

Referenced by dblink_build_sql_update().

get_text_array_contents()

static char ** get_text_array_contents ( ArrayType *  array, int *  numitems  )

static

Definition at line 2069 of file dblink.c.

{
    int         ndim = ARR_NDIM(array);
    int        *dims = ARR_DIMS(array);
    int         nitems;
    int16       typlen;
    bool        typbyval;
    char        typalign;
    uint8       typalignby;
    char      **values;
    char       *ptr;
    bits8      *bitmap;
    int         bitmask;
    int         i;
 
    Assert(ARR_ELEMTYPE(array) == TEXTOID);
 
    *numitems = nitems = ArrayGetNItems(ndim, dims);
 
    get_typlenbyvalalign(ARR_ELEMTYPE(array),
                         &typlen, &typbyval, &typalign);
    typalignby = typalign_to_alignby(typalign);
 
    values = palloc_array(char *, nitems);
 
    ptr = ARR_DATA_PTR(array);
    bitmap = ARR_NULLBITMAP(array);
    bitmask = 1;
 
    for (i = 0; i < nitems; i++)
    {
        if (bitmap && (*bitmap & bitmask) == 0)
        {
            values[i] = NULL;
        }
        else
        {
            values[i] = TextDatumGetCString(PointerGetDatum(ptr));
            ptr = att_addlength_pointer(ptr, typlen, ptr);
            ptr = (char *) att_nominal_alignby(ptr, typalignby);
        }
 
        /* advance bitmap pointer if any */
        if (bitmap)
        {
            bitmask <<= 1;
            if (bitmask == 0x100)
            {
                bitmap++;
                bitmask = 1;
            }
        }
    }
 
    return values;
}

References ARR_DATA_PTR, ARR_DIMS, ARR_ELEMTYPE, ARR_NDIM, ARR_NULLBITMAP, ArrayGetNItems(), Assert, att_addlength_pointer, att_nominal_alignby, fb(), get_typlenbyvalalign(), i, nitems, palloc_array, PointerGetDatum(), TextDatumGetCString, typalign, typalign_to_alignby(), and values.

Referenced by dblink_build_sql_delete(), dblink_build_sql_insert(), and dblink_build_sql_update().

get_tuple_of_interest()

static HeapTuple get_tuple_of_interest ( Relation  rel, int *  pkattnums, int  pknumatts, char **  src_pkattvals  )

static

Definition at line 2365 of file dblink.c.

{
    char       *relname;
    TupleDesc   tupdesc;
    int         natts;
    StringInfoData buf;
    int         ret;
    HeapTuple   tuple;
    int         i;
 
    /*
     * Connect to SPI manager
     */
    SPI_connect();
 
    initStringInfo(&buf);
 
    /* get relation name including any needed schema prefix and quoting */
    relname = generate_relation_name(rel);
 
    tupdesc = rel->rd_att;
    natts = tupdesc->natts;
 
    /*
     * Build sql statement to look up tuple of interest, ie, the one matching
     * src_pkattvals.  We used to use "SELECT *" here, but it's simpler to
     * generate a result tuple that matches the table's physical structure,
     * with NULLs for any dropped columns.  Otherwise we have to deal with two
     * different tupdescs and everything's very confusing.
     */
    appendStringInfoString(&buf, "SELECT ");
 
    for (i = 0; i < natts; i++)
    {
        Form_pg_attribute attr = TupleDescAttr(tupdesc, i);
 
        if (i > 0)
            appendStringInfoString(&buf, ", ");
 
        if (attr->attisdropped)
            appendStringInfoString(&buf, "NULL");
        else
            appendStringInfoString(&buf,
                                   quote_ident_cstr(NameStr(attr->attname)));
    }
 
    appendStringInfo(&buf, " FROM %s WHERE ", relname);
 
    for (i = 0; i < pknumatts; i++)
    {
        int         pkattnum = pkattnums[i];
        Form_pg_attribute attr = TupleDescAttr(tupdesc, pkattnum);
 
        if (i > 0)
            appendStringInfoString(&buf, " AND ");
 
        appendStringInfoString(&buf,
                               quote_ident_cstr(NameStr(attr->attname)));
 
        if (src_pkattvals[i] != NULL)
            appendStringInfo(&buf, " = %s",
                             quote_literal_cstr(src_pkattvals[i]));
        else
            appendStringInfoString(&buf, " IS NULL");
    }
 
    /*
     * Retrieve the desired tuple
     */
    ret = SPI_exec(buf.data, 0);
    pfree(buf.data);
 
    /*
     * Only allow one qualifying tuple
     */
    if ((ret == SPI_OK_SELECT) && (SPI_processed > 1))
        ereport(ERROR,
                (errcode(ERRCODE_CARDINALITY_VIOLATION),
                 errmsg("source criteria matched more than one record")));
 
    else if (ret == SPI_OK_SELECT && SPI_processed == 1)
    {
        SPITupleTable *tuptable = SPI_tuptable;
 
        tuple = SPI_copytuple(tuptable->vals[0]);
        SPI_finish();
 
        return tuple;
    }
    else
    {
        /*
         * no qualifying tuples
         */
        SPI_finish();
 
        return NULL;
    }
 
    /*
     * never reached, but keep compiler quiet
     */
    return NULL;
}

References appendStringInfo(), appendStringInfoString(), buf, ereport, errcode(), errmsg, ERROR, fb(), generate_relation_name(), i, initStringInfo(), NameStr, TupleDescData::natts, pfree(), quote_ident_cstr(), quote_literal_cstr(), RelationData::rd_att, relname, SPI_connect(), SPI_copytuple(), SPI_exec(), SPI_finish(), SPI_OK_SELECT, SPI_processed, SPI_tuptable, TupleDescAttr(), and SPITupleTable::vals.

Referenced by get_sql_insert(), and get_sql_update().

getConnectionByName()

static remoteConn * getConnectionByName ( const char *  name )

static

Definition at line 2528 of file dblink.c.

{
    remoteConnHashEnt *hentry;
    char       *key;
 
    if (!remoteConnHash)
        remoteConnHash = createConnHash();
 
    key = pstrdup(name);
    truncate_identifier(key, strlen(key), false);
    hentry = (remoteConnHashEnt *) hash_search(remoteConnHash,
                                               key, HASH_FIND, NULL);
 
    if (hentry && hentry->rconn.conn != NULL)
        return &hentry->rconn;
 
    return NULL;
}

References createConnHash(), fb(), HASH_FIND, hash_search(), name, pstrdup(), remoteConnHash, and truncate_identifier().

Referenced by dblink_close(), dblink_disconnect(), dblink_fetch(), dblink_get_conn(), dblink_get_named_conn(), and dblink_open().

is_valid_dblink_fdw_option()

static bool is_valid_dblink_fdw_option ( const PQconninfoOption *  options, const char *  option, Oid  context  )

static

Definition at line 3114 of file dblink.c.

{
    if (strcmp(option, "use_scram_passthrough") == 0)
        return true;
 
    return is_valid_dblink_option(options, option, context);
}

References fb(), and is_valid_dblink_option().

Referenced by dblink_fdw_validator().

is_valid_dblink_option()

static bool is_valid_dblink_option ( const PQconninfoOption *  options, const char *  option, Oid  context  )

static

Definition at line 3062 of file dblink.c.

{
    const PQconninfoOption *opt;
 
    /* Look up the option in libpq result */
    for (opt = options; opt->keyword; opt++)
    {
        if (strcmp(opt->keyword, option) == 0)
            break;
    }
    if (opt->keyword == NULL)
        return false;
 
    /* Disallow debug options (particularly "replication") */
    if (strchr(opt->dispchar, 'D'))
        return false;
 
    /* Disallow "client_encoding" */
    if (strcmp(opt->keyword, "client_encoding") == 0)
        return false;
 
    /*
     * Disallow OAuth options for now, since the builtin flow communicates on
     * stderr by default and can't cache tokens yet.
     */
    if (strncmp(opt->keyword, "oauth_", strlen("oauth_")) == 0)
        return false;
 
    /*
     * If the option is "user" or marked secure, it should be specified only
     * in USER MAPPING.  Others should be specified only in SERVER.
     */
    if (strcmp(opt->keyword, "user") == 0 || strchr(opt->dispchar, '*'))
    {
        if (context != UserMappingRelationId)
            return false;
    }
    else
    {
        if (context != ForeignServerRelationId)
            return false;
    }
 
    return true;
}

References _PQconninfoOption::dispchar, fb(), and _PQconninfoOption::keyword.

Referenced by dblink_fdw_validator(), get_connect_string(), and is_valid_dblink_fdw_option().

materializeQueryResult()

static void materializeQueryResult ( FunctionCallInfo  fcinfo, PGconn *  conn, const char *  conname, const char *  sql, bool  fail  )

static

Definition at line 995 of file dblink.c.

{
    ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 
    /* prepTuplestoreResult must have been called previously */
    Assert(rsinfo->returnMode == SFRM_Materialize);
 
    /* Use a PG_TRY block to ensure we pump libpq dry of results */
    PG_TRY();
    {
        storeInfo   sinfo = {0};
        PGresult   *res;
 
        sinfo.fcinfo = fcinfo;
        /* Create short-lived memory context for data conversions */
        sinfo.tmpcontext = AllocSetContextCreate(CurrentMemoryContext,
                                                 "dblink temporary context",
                                                 ALLOCSET_DEFAULT_SIZES);
 
        /* execute query, collecting any tuples into the tuplestore */
        res = storeQueryResult(&sinfo, conn, sql);
 
        if (!res ||
            (PQresultStatus(res) != PGRES_COMMAND_OK &&
             PQresultStatus(res) != PGRES_TUPLES_OK))
        {
            dblink_res_error(conn, conname, res, fail,
                             "while executing query");
            /* if fail isn't set, we'll return an empty query result */
        }
        else if (PQresultStatus(res) == PGRES_COMMAND_OK)
        {
            /*
             * storeRow didn't get called, so we need to convert the command
             * status string to a tuple manually
             */
            TupleDesc   tupdesc;
            AttInMetadata *attinmeta;
            Tuplestorestate *tupstore;
            HeapTuple   tuple;
            char       *values[1];
            MemoryContext oldcontext;
 
            /*
             * need a tuple descriptor representing one TEXT column to return
             * the command status string as our result tuple
             */
            tupdesc = CreateTemplateTupleDesc(1);
            TupleDescInitEntry(tupdesc, (AttrNumber) 1, "status",
                               TEXTOID, -1, 0);
            TupleDescFinalize(tupdesc);
            attinmeta = TupleDescGetAttInMetadata(tupdesc);
 
            oldcontext = MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory);
            tupstore = tuplestore_begin_heap(true, false, work_mem);
            rsinfo->setResult = tupstore;
            rsinfo->setDesc = tupdesc;
            MemoryContextSwitchTo(oldcontext);
 
            values[0] = PQcmdStatus(res);
 
            /* build the tuple and put it into the tuplestore. */
            tuple = BuildTupleFromCStrings(attinmeta, values);
            tuplestore_puttuple(tupstore, tuple);
 
            PQclear(res);
        }
        else
        {
            Assert(PQresultStatus(res) == PGRES_TUPLES_OK);
            /* storeRow should have created a tuplestore */
            Assert(rsinfo->setResult != NULL);
 
            PQclear(res);
        }
 
        /* clean up data conversion short-lived memory context */
        if (sinfo.tmpcontext != NULL)
            MemoryContextDelete(sinfo.tmpcontext);
 
        PQclear(sinfo.last_res);
        PQclear(sinfo.cur_res);
    }
    PG_CATCH();
    {
        PGresult   *res;
 
        /* be sure to clear out any pending data in libpq */
        while ((res = libpqsrv_get_result(conn, dblink_we_get_result)) !=
               NULL)
            PQclear(res);
        PG_RE_THROW();
    }
    PG_END_TRY();
}

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, Assert, BuildTupleFromCStrings(), conn, CreateTemplateTupleDesc(), storeInfo::cur_res, CurrentMemoryContext, dblink_res_error(), dblink_we_get_result, fb(), storeInfo::fcinfo, storeInfo::last_res, libpqsrv_get_result(), MemoryContextDelete(), MemoryContextSwitchTo(), PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, PGRES_COMMAND_OK, PGRES_TUPLES_OK, PQclear, PQcmdStatus, PQresultStatus, FunctionCallInfoBaseData::resultinfo, SFRM_Materialize, storeQueryResult(), storeInfo::tmpcontext, TupleDescFinalize(), TupleDescGetAttInMetadata(), TupleDescInitEntry(), tuplestore_begin_heap(), tuplestore_puttuple(), values, and work_mem.

Referenced by dblink_record_internal().

materializeResult()

static void materializeResult ( FunctionCallInfo  fcinfo, PGconn *  conn, PGresult *  res  )

static

Definition at line 863 of file dblink.c.

{
    ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
    TupleDesc   tupdesc;
    bool        is_sql_cmd;
    int         ntuples;
    int         nfields;
 
    /* prepTuplestoreResult must have been called previously */
    Assert(rsinfo->returnMode == SFRM_Materialize);
 
    if (PQresultStatus(res) == PGRES_COMMAND_OK)
    {
        is_sql_cmd = true;
 
        /*
         * need a tuple descriptor representing one TEXT column to return the
         * command status string as our result tuple
         */
        tupdesc = CreateTemplateTupleDesc(1);
        TupleDescInitEntry(tupdesc, (AttrNumber) 1, "status",
                           TEXTOID, -1, 0);
        TupleDescFinalize(tupdesc);
        ntuples = 1;
        nfields = 1;
    }
    else
    {
        Assert(PQresultStatus(res) == PGRES_TUPLES_OK);
 
        is_sql_cmd = false;
 
        /* get a tuple descriptor for our result type */
        switch (get_call_result_type(fcinfo, NULL, &tupdesc))
        {
            case TYPEFUNC_COMPOSITE:
                /* success */
                break;
            case TYPEFUNC_RECORD:
                /* failed to determine actual type of RECORD */
                ereport(ERROR,
                        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                         errmsg("function returning record called in context "
                                "that cannot accept type record")));
                break;
            default:
                /* result type isn't composite */
                elog(ERROR, "return type must be a row type");
                break;
        }
 
        /* make sure we have a persistent copy of the tupdesc */
        tupdesc = CreateTupleDescCopy(tupdesc);
        ntuples = PQntuples(res);
        nfields = PQnfields(res);
    }
 
    /*
     * check result and tuple descriptor have the same number of columns
     */
    if (nfields != tupdesc->natts)
        ereport(ERROR,
                (errcode(ERRCODE_DATATYPE_MISMATCH),
                 errmsg("remote query result rowtype does not match "
                        "the specified FROM clause rowtype")));
 
    if (ntuples > 0)
    {
        AttInMetadata *attinmeta;
        int         nestlevel = -1;
        Tuplestorestate *tupstore;
        MemoryContext oldcontext;
        int         row;
        char      **values;
 
        attinmeta = TupleDescGetAttInMetadata(tupdesc);
 
        /* Set GUCs to ensure we read GUC-sensitive data types correctly */
        if (!is_sql_cmd)
            nestlevel = applyRemoteGucs(conn);
 
        oldcontext = MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory);
        tupstore = tuplestore_begin_heap(true, false, work_mem);
        rsinfo->setResult = tupstore;
        rsinfo->setDesc = tupdesc;
        MemoryContextSwitchTo(oldcontext);
 
        values = palloc_array(char *, nfields);
 
        /* put all tuples into the tuplestore */
        for (row = 0; row < ntuples; row++)
        {
            HeapTuple   tuple;
 
            if (!is_sql_cmd)
            {
                int         i;
 
                for (i = 0; i < nfields; i++)
                {
                    if (PQgetisnull(res, row, i))
                        values[i] = NULL;
                    else
                        values[i] = PQgetvalue(res, row, i);
                }
            }
            else
            {
                values[0] = PQcmdStatus(res);
            }
 
            /* build the tuple and put it into the tuplestore. */
            tuple = BuildTupleFromCStrings(attinmeta, values);
            tuplestore_puttuple(tupstore, tuple);
        }
 
        /* clean up GUC settings, if we changed any */
        restoreLocalGucs(nestlevel);
    }
 
    PQclear(res);
}

References applyRemoteGucs(), Assert, BuildTupleFromCStrings(), conn, CreateTemplateTupleDesc(), CreateTupleDescCopy(), elog, ereport, errcode(), errmsg, ERROR, fb(), get_call_result_type(), i, MemoryContextSwitchTo(), TupleDescData::natts, palloc_array, PGRES_COMMAND_OK, PGRES_TUPLES_OK, PQclear, PQcmdStatus, PQgetisnull, PQgetvalue, PQnfields, PQntuples, PQresultStatus, restoreLocalGucs(), FunctionCallInfoBaseData::resultinfo, SFRM_Materialize, TupleDescFinalize(), TupleDescGetAttInMetadata(), TupleDescInitEntry(), tuplestore_begin_heap(), tuplestore_puttuple(), TYPEFUNC_COMPOSITE, TYPEFUNC_RECORD, values, and work_mem.

Referenced by dblink_fetch(), and dblink_record_internal().

PG_FUNCTION_INFO_V1() [1/20]

PG_FUNCTION_INFO_V1

(

dblink_build_sql_delete

)

PG_FUNCTION_INFO_V1() [2/20]

PG_FUNCTION_INFO_V1

(

dblink_build_sql_insert

)

PG_FUNCTION_INFO_V1() [3/20]

PG_FUNCTION_INFO_V1

(

dblink_build_sql_update

)

PG_FUNCTION_INFO_V1() [4/20]

PG_FUNCTION_INFO_V1

(

dblink_cancel_query

)

PG_FUNCTION_INFO_V1() [5/20]

PG_FUNCTION_INFO_V1

(

dblink_close

)

PG_FUNCTION_INFO_V1() [6/20]

PG_FUNCTION_INFO_V1

(

dblink_connect

)

PG_FUNCTION_INFO_V1() [7/20]

PG_FUNCTION_INFO_V1

(

dblink_current_query

)

PG_FUNCTION_INFO_V1() [8/20]

PG_FUNCTION_INFO_V1

(

dblink_disconnect

)

PG_FUNCTION_INFO_V1() [9/20]

PG_FUNCTION_INFO_V1

(

dblink_error_message

)

PG_FUNCTION_INFO_V1() [10/20]

PG_FUNCTION_INFO_V1

(

dblink_exec

)

PG_FUNCTION_INFO_V1() [11/20]

PG_FUNCTION_INFO_V1

(

dblink_fdw_validator

)

PG_FUNCTION_INFO_V1() [12/20]

PG_FUNCTION_INFO_V1

(

dblink_fetch

)

PG_FUNCTION_INFO_V1() [13/20]

PG_FUNCTION_INFO_V1

(

dblink_get_connections

)

PG_FUNCTION_INFO_V1() [14/20]

PG_FUNCTION_INFO_V1

(

dblink_get_notify

)

PG_FUNCTION_INFO_V1() [15/20]

PG_FUNCTION_INFO_V1

(

dblink_get_pkey

)

PG_FUNCTION_INFO_V1() [16/20]

PG_FUNCTION_INFO_V1

(

dblink_get_result

)

PG_FUNCTION_INFO_V1() [17/20]

PG_FUNCTION_INFO_V1

(

dblink_is_busy

)

PG_FUNCTION_INFO_V1() [18/20]

PG_FUNCTION_INFO_V1

(

dblink_open

)

PG_FUNCTION_INFO_V1() [19/20]

PG_FUNCTION_INFO_V1

(

dblink_record

)

PG_FUNCTION_INFO_V1() [20/20]

PG_FUNCTION_INFO_V1

(

dblink_send_query

)

PG_MODULE_MAGIC_EXT()

PG_MODULE_MAGIC_EXT	(	.	name = "dblink",
		.	version = PG_VERSION
	)

prepTuplestoreResult()

static void prepTuplestoreResult ( FunctionCallInfo  fcinfo )

static

Definition at line 835 of file dblink.c.

{
    ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 
    /* check to see if query supports us returning a tuplestore */
    if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("set-valued function called in context that cannot accept a set")));
    if (!(rsinfo->allowedModes & SFRM_Materialize))
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("materialize mode required, but it is not allowed in this context")));
 
    /* let the executor know we're sending back a tuplestore */
    rsinfo->returnMode = SFRM_Materialize;
 
    /* caller must fill these to return a non-empty result */
    rsinfo->setResult = NULL;
    rsinfo->setDesc = NULL;
}

References ereport, errcode(), errmsg, ERROR, fb(), IsA, FunctionCallInfoBaseData::resultinfo, and SFRM_Materialize.

Referenced by dblink_fetch(), and dblink_record_internal().

quote_ident_cstr()

static char * quote_ident_cstr ( char *  rawstr )

static

Definition at line 2335 of file dblink.c.

{
    text       *rawstr_text;
    text       *result_text;
    char       *result;
 
    rawstr_text = cstring_to_text(rawstr);
    result_text = DatumGetTextPP(DirectFunctionCall1(quote_ident,
                                                     PointerGetDatum(rawstr_text)));
    result = text_to_cstring(result_text);
 
    return result;
}

References cstring_to_text(), DatumGetTextPP, DirectFunctionCall1, fb(), PointerGetDatum(), quote_ident(), and text_to_cstring().

Referenced by get_sql_delete(), get_sql_insert(), get_sql_update(), and get_tuple_of_interest().

RangeVarCallbackForDblink()

static void RangeVarCallbackForDblink ( const RangeVar *  relation, Oid  relId, Oid  oldRelId, void *  arg  )

static

Definition at line 2471 of file dblink.c.

{
    AclResult   aclresult;
 
    if (!OidIsValid(relId))
        return;
 
    aclresult = pg_class_aclcheck(relId, GetUserId(), *((AclMode *) arg));
    if (aclresult != ACLCHECK_OK)
        aclcheck_error(aclresult, get_relkind_objtype(get_rel_relkind(relId)),
                       relation->relname);
}

References aclcheck_error(), ACLCHECK_OK, arg, fb(), get_rel_relkind(), get_relkind_objtype(), GetUserId(), OidIsValid, pg_class_aclcheck(), and RangeVar::relname.

Referenced by get_rel_from_relname().

restoreLocalGucs()

static void restoreLocalGucs ( int  nestlevel )

static

Definition at line 3186 of file dblink.c.

{
    /* Do nothing if no new nestlevel was created */
    if (nestlevel > 0)
        AtEOXact_GUC(true, nestlevel);
}

References AtEOXact_GUC(), and fb().

Referenced by materializeResult(), and storeQueryResult().

storeQueryResult()

static PGresult * storeQueryResult ( storeInfo *  sinfo, PGconn *  conn, const char *  sql  )

static

Definition at line 1099 of file dblink.c.

{
    bool        first = true;
    int         nestlevel = -1;
    PGresult   *res;
 
    if (!PQsendQuery(conn, sql))
        elog(ERROR, "could not send query: %s", pchomp(PQerrorMessage(conn)));
 
    if (!PQsetSingleRowMode(conn))  /* shouldn't fail */
        elog(ERROR, "failed to set single-row mode for dblink query");
 
    for (;;)
    {
        CHECK_FOR_INTERRUPTS();
 
        sinfo->cur_res = libpqsrv_get_result(conn, dblink_we_get_result);
        if (!sinfo->cur_res)
            break;
 
        if (PQresultStatus(sinfo->cur_res) == PGRES_SINGLE_TUPLE)
        {
            /* got one row from possibly-bigger resultset */
 
            /*
             * Set GUCs to ensure we read GUC-sensitive data types correctly.
             * We shouldn't do this until we have a row in hand, to ensure
             * libpq has seen any earlier ParameterStatus protocol messages.
             */
            if (first && nestlevel < 0)
                nestlevel = applyRemoteGucs(conn);
 
            storeRow(sinfo, sinfo->cur_res, first);
 
            PQclear(sinfo->cur_res);
            sinfo->cur_res = NULL;
            first = false;
        }
        else
        {
            /* if empty resultset, fill tuplestore header */
            if (first && PQresultStatus(sinfo->cur_res) == PGRES_TUPLES_OK)
                storeRow(sinfo, sinfo->cur_res, first);
 
            /* store completed result at last_res */
            PQclear(sinfo->last_res);
            sinfo->last_res = sinfo->cur_res;
            sinfo->cur_res = NULL;
            first = true;
        }
    }
 
    /* clean up GUC settings, if we changed any */
    restoreLocalGucs(nestlevel);
 
    /* return last_res */
    res = sinfo->last_res;
    sinfo->last_res = NULL;
    return res;
}

References applyRemoteGucs(), CHECK_FOR_INTERRUPTS, conn, storeInfo::cur_res, dblink_we_get_result, elog, ERROR, fb(), storeInfo::last_res, libpqsrv_get_result(), pchomp(), PGRES_SINGLE_TUPLE, PGRES_TUPLES_OK, PQclear, PQerrorMessage(), PQresultStatus, PQsendQuery(), PQsetSingleRowMode(), restoreLocalGucs(), and storeRow().

Referenced by materializeQueryResult().

storeRow()

static void storeRow ( storeInfo *  sinfo, PGresult *  res, bool  first  )

static

Definition at line 1167 of file dblink.c.

{
    int         nfields = PQnfields(res);
    HeapTuple   tuple;
    int         i;
    MemoryContext oldcontext;
 
    if (first)
    {
        /* Prepare for new result set */
        ReturnSetInfo *rsinfo = (ReturnSetInfo *) sinfo->fcinfo->resultinfo;
        TupleDesc   tupdesc;
 
        /*
         * It's possible to get more than one result set if the query string
         * contained multiple SQL commands.  In that case, we follow PQexec's
         * traditional behavior of throwing away all but the last result.
         */
        if (sinfo->tuplestore)
            tuplestore_end(sinfo->tuplestore);
        sinfo->tuplestore = NULL;
 
        /* get a tuple descriptor for our result type */
        switch (get_call_result_type(sinfo->fcinfo, NULL, &tupdesc))
        {
            case TYPEFUNC_COMPOSITE:
                /* success */
                break;
            case TYPEFUNC_RECORD:
                /* failed to determine actual type of RECORD */
                ereport(ERROR,
                        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                         errmsg("function returning record called in context "
                                "that cannot accept type record")));
                break;
            default:
                /* result type isn't composite */
                elog(ERROR, "return type must be a row type");
                break;
        }
 
        /* make sure we have a persistent copy of the tupdesc */
        tupdesc = CreateTupleDescCopy(tupdesc);
 
        /* check result and tuple descriptor have the same number of columns */
        if (nfields != tupdesc->natts)
            ereport(ERROR,
                    (errcode(ERRCODE_DATATYPE_MISMATCH),
                     errmsg("remote query result rowtype does not match "
                            "the specified FROM clause rowtype")));
 
        /* Prepare attinmeta for later data conversions */
        sinfo->attinmeta = TupleDescGetAttInMetadata(tupdesc);
 
        /* Create a new, empty tuplestore */
        oldcontext = MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory);
        sinfo->tuplestore = tuplestore_begin_heap(true, false, work_mem);
        rsinfo->setResult = sinfo->tuplestore;
        rsinfo->setDesc = tupdesc;
        MemoryContextSwitchTo(oldcontext);
 
        /* Done if empty resultset */
        if (PQntuples(res) == 0)
            return;
 
        /*
         * Set up sufficiently-wide string pointers array; this won't change
         * in size so it's easy to preallocate.
         */
        if (sinfo->cstrs)
            pfree(sinfo->cstrs);
        sinfo->cstrs = palloc_array(char *, nfields);
    }
 
    /* Should have a single-row result if we get here */
    Assert(PQntuples(res) == 1);
 
    /*
     * Do the following work in a temp context that we reset after each tuple.
     * This cleans up not only the data we have direct access to, but any
     * cruft the I/O functions might leak.
     */
    oldcontext = MemoryContextSwitchTo(sinfo->tmpcontext);
 
    /*
     * Fill cstrs with null-terminated strings of column values.
     */
    for (i = 0; i < nfields; i++)
    {
        if (PQgetisnull(res, 0, i))
            sinfo->cstrs[i] = NULL;
        else
            sinfo->cstrs[i] = PQgetvalue(res, 0, i);
    }
 
    /* Convert row to a tuple, and add it to the tuplestore */
    tuple = BuildTupleFromCStrings(sinfo->attinmeta, sinfo->cstrs);
 
    tuplestore_puttuple(sinfo->tuplestore, tuple);
 
    /* Clean up */
    MemoryContextSwitchTo(oldcontext);
    MemoryContextReset(sinfo->tmpcontext);
}

References Assert, storeInfo::attinmeta, BuildTupleFromCStrings(), CreateTupleDescCopy(), storeInfo::cstrs, elog, ereport, errcode(), errmsg, ERROR, fb(), storeInfo::fcinfo, get_call_result_type(), i, MemoryContextReset(), MemoryContextSwitchTo(), TupleDescData::natts, palloc_array, pfree(), PQgetisnull, PQgetvalue, PQnfields, PQntuples, FunctionCallInfoBaseData::resultinfo, storeInfo::tmpcontext, TupleDescGetAttInMetadata(), storeInfo::tuplestore, tuplestore_begin_heap(), tuplestore_end(), tuplestore_puttuple(), TYPEFUNC_COMPOSITE, TYPEFUNC_RECORD, and work_mem.

Referenced by storeQueryResult().

UseScramPassthrough()

static bool UseScramPassthrough ( ForeignServer *  foreign_server, UserMapping *  user  )

static

Definition at line 3233 of file dblink.c.

{
    ListCell   *cell;
 
    foreach(cell, foreign_server->options)
    {
        DefElem    *def = lfirst(cell);
 
        if (strcmp(def->defname, "use_scram_passthrough") == 0)
            return defGetBoolean(def);
    }
 
    foreach(cell, user->options)
    {
        DefElem    *def = (DefElem *) lfirst(cell);
 
        if (strcmp(def->defname, "use_scram_passthrough") == 0)
            return defGetBoolean(def);
    }
 
    return false;
}

References defGetBoolean(), DefElem::defname, fb(), lfirst, ForeignServer::options, and user.

Referenced by get_connect_string().

validate_pkattnums()

static void validate_pkattnums ( Relation  rel, int2vector *  pkattnums_arg, int32  pknumatts_arg, int **  pkattnums, int *  pknumatts  )

static

Definition at line 2991 of file dblink.c.

{
    TupleDesc   tupdesc = rel->rd_att;
    int         natts = tupdesc->natts;
    int         i;
 
    /* Don't take more array elements than there are */
    pknumatts_arg = Min(pknumatts_arg, pkattnums_arg->dim1);
 
    /* Must have at least one pk attnum selected */
    if (pknumatts_arg <= 0)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("number of key attributes must be > 0")));
 
    /* Allocate output array */
    *pkattnums = palloc_array(int, pknumatts_arg);
    *pknumatts = pknumatts_arg;
 
    /* Validate attnums and convert to internal form */
    for (i = 0; i < pknumatts_arg; i++)
    {
        int         pkattnum = pkattnums_arg->values[i];
        int         lnum;
        int         j;
 
        /* Can throw error immediately if out of range */
        if (pkattnum <= 0 || pkattnum > natts)
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                     errmsg("invalid attribute number %d", pkattnum)));
 
        /* Identify which physical column has this logical number */
        lnum = 0;
        for (j = 0; j < natts; j++)
        {
            /* dropped columns don't count */
            if (TupleDescCompactAttr(tupdesc, j)->attisdropped)
                continue;
 
            if (++lnum == pkattnum)
                break;
        }
 
        if (j < natts)
            (*pkattnums)[i] = j;
        else
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                     errmsg("invalid attribute number %d", pkattnum)));
    }
}

References ereport, errcode(), errmsg, ERROR, fb(), i, j, Min, TupleDescData::natts, palloc_array, RelationData::rd_att, and TupleDescCompactAttr().

Referenced by dblink_build_sql_delete(), dblink_build_sql_insert(), and dblink_build_sql_update().

Variable Documentation

dblink_we_connect

uint32 dblink_we_connect = 0

static

Definition at line 147 of file dblink.c.

Referenced by dblink_connect().

dblink_we_get_conn

uint32 dblink_we_get_conn = 0

static

Definition at line 148 of file dblink.c.

Referenced by dblink_get_conn().

dblink_we_get_result

uint32 dblink_we_get_result = 0

static

Definition at line 149 of file dblink.c.

Referenced by dblink_close(), dblink_exec(), dblink_fetch(), dblink_init(), dblink_open(), dblink_record_internal(), materializeQueryResult(), and storeQueryResult().

pconn

remoteConn* pconn = NULL

static

Definition at line 143 of file dblink.c.

Referenced by dblink_close(), dblink_connect(), dblink_disconnect(), dblink_exec(), dblink_fetch(), dblink_get_notify(), dblink_init(), dblink_open(), and dblink_record_internal().

remoteConnHash

HTAB* remoteConnHash = NULL

static

Definition at line 144 of file dblink.c.

Referenced by createNewConnection(), dblink_get_connections(), deleteConnection(), and getConnectionByName().