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 "executor/spi.h"
#include "foreign/foreign.h"
#include "funcapi.h"
#include "lib/stringinfo.h"
#include "libpq-fe.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/varlena.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
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 (volatile storeInfo *sinfo, PGconn *conn, const char *sql)
static void	storeRow (volatile storeInfo *sinfo, PGresult *res, bool first)
static remoteConn *	getConnectionByName (const char *name)
static HTAB *	createConnHash (void)
static void	createNewConnection (const char *name, remoteConn *rconn)
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 void	dblink_security_check (PGconn *conn, remoteConn *rconn)
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 *from)
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 char *	xpstrdup (const char *in)
static void	pg_attribute_noreturn ()
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)

Variables
	PG_MODULE_MAGIC
static remoteConn *	pconn = NULL
static HTAB *	remoteConnHash = NULL

Macro Definition Documentation

DBLINK_NOTIFY_COLS

#define DBLINK_NOTIFY_COLS   3

Definition at line 1924 of file dblink.c.

Referenced by dblink_get_notify().

NUMCONN

#define NUMCONN   16

Definition at line 144 of file dblink.c.

Referenced by createConnHash().

Typedef Documentation

remoteConn

typedef struct remoteConn remoteConn

remoteConnHashEnt

typedef struct remoteConnHashEnt remoteConnHashEnt

storeInfo

typedef struct storeInfo storeInfo

Function Documentation

applyRemoteGucs()

static int applyRemoteGucs ( PGconn *  conn )

static

Definition at line 3063 of file dblink.c.

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

Referenced by materializeResult(), and storeQueryResult().

{
    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;
}

createConnHash()

static HTAB * createConnHash ( void  )

static

Definition at line 2603 of file dblink.c.

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

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

{
    HASHCTL     ctl;

    ctl.keysize = NAMEDATALEN;
    ctl.entrysize = sizeof(remoteConnHashEnt);

    return hash_create("Remote Con hash", NUMCONN, &ctl,
                       HASH_ELEM | HASH_STRINGS);
}

createNewConnection()

static void createNewConnection ( const char *  name, remoteConn *  rconn  )

static

Definition at line 2615 of file dblink.c.

References remoteConn::conn, createConnHash(), ereport, errcode(), ERRCODE_DUPLICATE_OBJECT, errmsg(), ERROR, HASH_ENTER, hash_search(), sort-test::key, remoteConnHashEnt::name, pfree(), PQfinish(), pstrdup(), remoteConnHashEnt::rconn, ReleaseExternalFD(), remoteConnHash, strlcpy(), and truncate_identifier().

Referenced by dblink_connect().

{
    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)
    {
        PQfinish(rconn->conn);
        ReleaseExternalFD();
        pfree(rconn);

        ereport(ERROR,
                (errcode(ERRCODE_DUPLICATE_OBJECT),
                 errmsg("duplicate connection name")));
    }

    hentry->rconn = rconn;
    strlcpy(hentry->name, name, sizeof(hentry->name));
}

dblink_build_sql_delete()

Datum dblink_build_sql_delete

(

PG_FUNCTION_ARGS

)

Definition at line 1755 of file dblink.c.

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

Referenced by dblink_build_sql_insert().

{
    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));
}

dblink_build_sql_insert()

Datum dblink_build_sql_insert

(

PG_FUNCTION_ARGS

)

Definition at line 1666 of file dblink.c.

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

Referenced by dblink_get_pkey().

{
    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));
}

dblink_build_sql_update()

Datum dblink_build_sql_update

(

PG_FUNCTION_ARGS

)

Definition at line 1831 of file dblink.c.

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

Referenced by dblink_build_sql_delete().

{
    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));
}

dblink_cancel_query()

Datum dblink_cancel_query

(

PG_FUNCTION_ARGS

)

Definition at line 1383 of file dblink.c.

References remoteConn::conn, cstring_to_text(), dblink_error_message(), dblink_get_named_conn(), dblink_init(), PG_FUNCTION_INFO_V1(), PG_GETARG_TEXT_PP, PG_RETURN_TEXT_P, PQcancel(), PQfreeCancel(), PQgetCancel(), and text_to_cstring().

Referenced by dblink_is_busy().

{
    int         res;
    PGconn     *conn;
    PGcancel   *cancel;
    char        errbuf[256];

    dblink_init();
    conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
    cancel = PQgetCancel(conn);

    res = PQcancel(cancel, errbuf, 256);
    PQfreeCancel(cancel);

    if (res == 1)
        PG_RETURN_TEXT_P(cstring_to_text("OK"));
    else
        PG_RETURN_TEXT_P(cstring_to_text(errbuf));
}

dblink_close()

Datum dblink_close

(

PG_FUNCTION_ARGS

)

Definition at line 517 of file dblink.c.

References appendStringInfo(), buf, remoteConn::conn, cstring_to_text(), StringInfoData::data, dblink_fetch(), dblink_init(), dblink_res_error(), get_fn_expr_argtype(), getConnectionByName(), initStringInfo(), remoteConn::newXactForCursor, remoteConn::openCursorCount, pconn, PG_FUNCTION_INFO_V1(), PG_GETARG_BOOL, PG_GETARG_TEXT_PP, PG_NARGS, PG_RETURN_TEXT_P, PGRES_COMMAND_OK, PQclear(), PQexec(), PQresultStatus(), and text_to_cstring().

Referenced by dblink_open().

{
    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 = PQexec(conn, buf.data);
    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 = PQexec(conn, "COMMIT");
            if (PQresultStatus(res) != PGRES_COMMAND_OK)
                dblink_res_internalerror(conn, res, "commit error");
            PQclear(res);
        }
    }

    PG_RETURN_TEXT_P(cstring_to_text("OK"));
}

dblink_connect()

Datum dblink_connect

(

PG_FUNCTION_ARGS

)

Definition at line 280 of file dblink.c.

References AcquireExternalFD(), remoteConn::conn, CONNECTION_BAD, connstr, createNewConnection(), cstring_to_text(), dblink_connstr_check(), dblink_disconnect(), dblink_init(), dblink_security_check(), ereport, errcode(), errdetail(), errdetail_internal(), errhint(), errmsg(), ERROR, get_connect_string(), GetDatabaseEncoding(), GetDatabaseEncodingName(), MemoryContextAlloc(), remoteConn::newXactForCursor, remoteConn::openCursorCount, pchomp(), pconn, pfree(), PG_FUNCTION_INFO_V1(), PG_GETARG_TEXT_PP, PG_NARGS, PG_RETURN_TEXT_P, PQclientEncoding(), PQconnectdb(), PQerrorMessage(), PQfinish(), PQsetClientEncoding(), PQstatus(), ReleaseExternalFD(), text_to_cstring(), and TopMemoryContext.

Referenced by dblink_init().

{
    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));

    if (connname)
    {
        rconn = (remoteConn *) MemoryContextAlloc(TopMemoryContext,
                                                  sizeof(remoteConn));
        rconn->conn = NULL;
        rconn->openCursorCount = 0;
        rconn->newXactForCursor = false;
    }

    /* 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);

    /*
     * We must obey fd.c's limit on non-virtual file descriptors.  Assume that
     * a PGconn represents one long-lived FD.  (Doing this here also ensures
     * that VFDs are closed if needed to make room.)
     */
    if (!AcquireExternalFD())
    {
#ifndef WIN32                   /* can't write #if within ereport() macro */
        ereport(ERROR,
                (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
                 errmsg("could not establish connection"),
                 errdetail("There are too many open files on the local server."),
                 errhint("Raise the server's max_files_per_process and/or \"ulimit -n\" limits.")));
#else
        ereport(ERROR,
                (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
                 errmsg("could not establish connection"),
                 errdetail("There are too many open files on the local server."),
                 errhint("Raise the server's max_files_per_process setting.")));
#endif
    }

    /* OK to make connection */
    conn = PQconnectdb(connstr);

    if (PQstatus(conn) == CONNECTION_BAD)
    {
        msg = pchomp(PQerrorMessage(conn));
        PQfinish(conn);
        ReleaseExternalFD();
        if (rconn)
            pfree(rconn);

        ereport(ERROR,
                (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
                 errmsg("could not establish connection"),
                 errdetail_internal("%s", msg)));
    }

    /* check password actually used if not superuser */
    dblink_security_check(conn, rconn);

    /* attempt to set client encoding to match server encoding, if needed */
    if (PQclientEncoding(conn) != GetDatabaseEncoding())
        PQsetClientEncoding(conn, GetDatabaseEncodingName());

    if (connname)
    {
        rconn->conn = conn;
        createNewConnection(connname, rconn);
    }
    else
    {
        if (pconn->conn)
        {
            PQfinish(pconn->conn);
            ReleaseExternalFD();
        }
        pconn->conn = conn;
    }

    PG_RETURN_TEXT_P(cstring_to_text("OK"));
}

dblink_connstr_check()

static void dblink_connstr_check ( const char *  connstr )

static

Definition at line 2694 of file dblink.c.

References ereport, errcode(), errdetail(), errmsg(), ERROR, _PQconninfoOption::keyword, options, PQconninfoFree(), PQconninfoParse(), superuser(), and _PQconninfoOption::val.

Referenced by dblink_connect(), and dblink_get_conn().

{
    if (!superuser())
    {
        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);
        }

        if (!connstr_gives_password)
            ereport(ERROR,
                    (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
                     errmsg("password is required"),
                     errdetail("Non-superusers must provide a password in the connection string.")));
    }
}

dblink_current_query()

Datum dblink_current_query

(

PG_FUNCTION_ARGS

)

Definition at line 1910 of file dblink.c.

References current_query(), and PG_RETURN_DATUM.

Referenced by dblink_build_sql_update().

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

dblink_disconnect()

Datum dblink_disconnect

(

PG_FUNCTION_ARGS

)

Definition at line 385 of file dblink.c.

References remoteConn::conn, cstring_to_text(), dblink_init(), dblink_open(), deleteConnection(), getConnectionByName(), pconn, pfree(), PG_FUNCTION_INFO_V1(), PG_GETARG_TEXT_PP, PG_NARGS, PG_RETURN_TEXT_P, PQfinish(), ReleaseExternalFD(), and text_to_cstring().

Referenced by dblink_connect().

{
    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);

    PQfinish(conn);
    ReleaseExternalFD();
    if (rconn)
    {
        deleteConnection(conname);
        pfree(rconn);
    }
    else
        pconn->conn = NULL;

    PG_RETURN_TEXT_P(cstring_to_text("OK"));
}

dblink_error_message()

Datum dblink_error_message

(

PG_FUNCTION_ARGS

)

Definition at line 1416 of file dblink.c.

References remoteConn::conn, cstring_to_text(), dblink_exec(), dblink_get_named_conn(), dblink_init(), pchomp(), PG_FUNCTION_INFO_V1(), PG_GETARG_TEXT_PP, PG_RETURN_TEXT_P, PQerrorMessage(), and text_to_cstring().

Referenced by dblink_cancel_query().

{
    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)));
}

dblink_exec()

Datum dblink_exec

(

PG_FUNCTION_ARGS

)

Definition at line 1436 of file dblink.c.

References remoteConn::conn, cstring_to_text(), dblink_get_conn(), dblink_get_pkey(), dblink_init(), dblink_res_error(), elog, ereport, errcode(), errmsg(), ERROR, get_fn_expr_argtype(), pconn, PG_END_TRY, PG_FINALLY, PG_FUNCTION_INFO_V1(), PG_GETARG_BOOL, PG_GETARG_TEXT_PP, PG_NARGS, PG_RETURN_TEXT_P, PG_TRY, PGRES_COMMAND_OK, PGRES_TUPLES_OK, PQclear(), PQcmdStatus(), PQexec(), PQfinish(), PQresultStatus(), ReleaseExternalFD(), and text_to_cstring().

Referenced by dblink_error_message().

{
    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 = PQexec(conn, sql);
        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)
        {
            PQfinish(conn);
            ReleaseExternalFD();
        }
    }
    PG_END_TRY();

    PG_RETURN_TEXT_P(sql_cmd_status);
}

dblink_fdw_validator()

Datum dblink_fdw_validator

(

PG_FUNCTION_ARGS

)

Definition at line 2006 of file dblink.c.

References appendStringInfo(), buf, StringInfoData::data, DefElem::defname, ereport, errcode(), errdetail(), errhint(), errmsg(), ERROR, initStringInfo(), is_valid_dblink_option(), _PQconninfoOption::keyword, StringInfoData::len, lfirst, PG_GETARG_DATUM, PG_GETARG_OID, PG_RETURN_VOID, PQconndefaults(), and untransformRelOptions().

Referenced by dblink_get_notify().

{
    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_option(options, def->defname, context))
        {
            /*
             * Unknown option, or invalid option for the context specified, so
             * complain about it.  Provide a hint with list of valid options
             * for the context.
             */
            StringInfoData buf;
            const PQconninfoOption *opt;

            initStringInfo(&buf);
            for (opt = options; opt->keyword; opt++)
            {
                if (is_valid_dblink_option(options, opt->keyword, context))
                    appendStringInfo(&buf, "%s%s",
                                     (buf.len > 0) ? ", " : "",
                                     opt->keyword);
            }
            ereport(ERROR,
                    (errcode(ERRCODE_FDW_OPTION_NAME_NOT_FOUND),
                     errmsg("invalid option \"%s\"", def->defname),
                     errhint("Valid options in this context are: %s",
                             buf.data)));
        }
    }

    PG_RETURN_VOID();
}

dblink_fetch()

Datum dblink_fetch

(

PG_FUNCTION_ARGS

)

Definition at line 604 of file dblink.c.

References appendStringInfo(), buf, remoteConn::conn, StringInfoData::data, dblink_init(), dblink_record(), dblink_res_error(), ereport, errcode(), errmsg(), ERROR, get_fn_expr_argtype(), getConnectionByName(), initStringInfo(), materializeResult(), pconn, PG_FUNCTION_INFO_V1(), PG_GETARG_BOOL, PG_GETARG_INT32, PG_GETARG_TEXT_PP, PG_NARGS, PGRES_COMMAND_OK, PGRES_TUPLES_OK, PQclear(), PQexec(), PQresultStatus(), prepTuplestoreResult(), and text_to_cstring().

Referenced by dblink_close().

{
    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 = PQexec(conn, buf.data);
    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;
}

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 180 of file dblink.c.

References AcquireExternalFD(), remoteConn::conn, CONNECTION_BAD, connstr, dblink_connstr_check(), dblink_security_check(), ereport, errcode(), errdetail(), errdetail_internal(), errhint(), errmsg(), ERROR, get_connect_string(), getConnectionByName(), GetDatabaseEncoding(), GetDatabaseEncodingName(), pchomp(), PQclientEncoding(), PQconnectdb(), PQerrorMessage(), PQfinish(), PQsetClientEncoding(), PQstatus(), and ReleaseExternalFD().

Referenced by dblink_exec(), and dblink_record_internal().

{
    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);

        /*
         * We must obey fd.c's limit on non-virtual file descriptors.  Assume
         * that a PGconn represents one long-lived FD.  (Doing this here also
         * ensures that VFDs are closed if needed to make room.)
         */
        if (!AcquireExternalFD())
        {
#ifndef WIN32                   /* can't write #if within ereport() macro */
            ereport(ERROR,
                    (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
                     errmsg("could not establish connection"),
                     errdetail("There are too many open files on the local server."),
                     errhint("Raise the server's max_files_per_process and/or \"ulimit -n\" limits.")));
#else
            ereport(ERROR,
                    (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
                     errmsg("could not establish connection"),
                     errdetail("There are too many open files on the local server."),
                     errhint("Raise the server's max_files_per_process setting.")));
#endif
        }

        /* OK to make connection */
        conn = PQconnectdb(connstr);

        if (PQstatus(conn) == CONNECTION_BAD)
        {
            char       *msg = pchomp(PQerrorMessage(conn));

            PQfinish(conn);
            ReleaseExternalFD();
            ereport(ERROR,
                    (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
                     errmsg("could not establish connection"),
                     errdetail_internal("%s", msg)));
        }
        dblink_security_check(conn, rconn);
        if (PQclientEncoding(conn) != GetDatabaseEncoding())
            PQsetClientEncoding(conn, GetDatabaseEncodingName());
        freeconn = true;
        conname = NULL;
    }

    *conn_p = conn;
    *conname_p = conname;
    *freeconn_p = freeconn;
}

dblink_get_connections()

Datum dblink_get_connections

(

PG_FUNCTION_ARGS

)

Definition at line 1324 of file dblink.c.

References accumArrayResult(), CStringGetTextDatum, CurrentMemoryContext, dblink_is_busy(), hash_seq_init(), hash_seq_search(), makeArrayResult(), remoteConnHashEnt::name, PG_FUNCTION_INFO_V1(), PG_RETURN_ARRAYTYPE_P, PG_RETURN_NULL, remoteConnHash, and status().

Referenced by storeRow().

{
    HASH_SEQ_STATUS status;
    remoteConnHashEnt *hentry;
    ArrayBuildState *astate = NULL;

    if (remoteConnHash)
    {
        hash_seq_init(&status, remoteConnHash);
        while ((hentry = (remoteConnHashEnt *) hash_seq_search(&status)) != NULL)
        {
            /* stash away current value */
            astate = accumArrayResult(astate,
                                      CStringGetTextDatum(hentry->name),
                                      false, TEXTOID, CurrentMemoryContext);
        }
    }

    if (astate)
        PG_RETURN_ARRAYTYPE_P(makeArrayResult(astate,
                                              CurrentMemoryContext));
    else
        PG_RETURN_NULL();
}

dblink_get_named_conn()

static PGconn* dblink_get_named_conn ( const char *  conname )

static

Definition at line 252 of file dblink.c.

References remoteConn::conn, and getConnectionByName().

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

{
    remoteConn *rconn = getConnectionByName(conname);

    if (rconn)
        return rconn->conn;

    dblink_conn_not_avail(conname);
    return NULL;                /* keep compiler quiet */
}

dblink_get_notify()

Datum dblink_get_notify

(

PG_FUNCTION_ARGS

)

Definition at line 1928 of file dblink.c.

References pgNotify::be_pid, remoteConn::conn, CreateTemplateTupleDesc(), CStringGetTextDatum, dblink_fdw_validator(), dblink_get_named_conn(), dblink_init(), DBLINK_NOTIFY_COLS, ReturnSetInfo::econtext, ExprContext::ecxt_per_query_memory, pgNotify::extra, Int32GetDatum, MemoryContextSwitchTo(), pconn, PG_FUNCTION_INFO_V1(), PG_GETARG_TEXT_PP, PG_NARGS, PQconsumeInput(), PQfreemem(), PQnotifies(), prepTuplestoreResult(), pgNotify::relname, ReturnSetInfo::setDesc, ReturnSetInfo::setResult, text_to_cstring(), TupleDescInitEntry(), tuplestore_begin_heap(), tuplestore_donestoring, tuplestore_putvalues(), and work_mem.

{
    PGconn     *conn;
    PGnotify   *notify;
    ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
    TupleDesc   tupdesc;
    Tuplestorestate *tupstore;
    MemoryContext per_query_ctx;
    MemoryContext oldcontext;

    prepTuplestoreResult(fcinfo);

    dblink_init();
    if (PG_NARGS() == 1)
        conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
    else
        conn = pconn->conn;

    /* create the tuplestore in per-query memory */
    per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
    oldcontext = MemoryContextSwitchTo(per_query_ctx);

    tupdesc = CreateTemplateTupleDesc(DBLINK_NOTIFY_COLS);
    TupleDescInitEntry(tupdesc, (AttrNumber) 1, "notify_name",
                       TEXTOID, -1, 0);
    TupleDescInitEntry(tupdesc, (AttrNumber) 2, "be_pid",
                       INT4OID, -1, 0);
    TupleDescInitEntry(tupdesc, (AttrNumber) 3, "extra",
                       TEXTOID, -1, 0);

    tupstore = tuplestore_begin_heap(true, false, work_mem);
    rsinfo->setResult = tupstore;
    rsinfo->setDesc = tupdesc;

    MemoryContextSwitchTo(oldcontext);

    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(tupstore, tupdesc, values, nulls);

        PQfreemem(notify);
        PQconsumeInput(conn);
    }

    /* clean up and return the tuplestore */
    tuplestore_donestoring(tupstore);

    return (Datum) 0;
}

dblink_get_pkey()

Datum dblink_get_pkey

(

PG_FUNCTION_ARGS

)

Definition at line 1542 of file dblink.c.

References AccessShareLock, ACL_SELECT, FuncCallContext::attinmeta, BuildTupleFromCStrings(), FuncCallContext::call_cntr, CreateTemplateTupleDesc(), dblink_build_sql_insert(), get_pkey_attnames(), get_rel_from_relname(), HeapTupleGetDatum, FuncCallContext::max_calls, MemoryContextSwitchTo(), FuncCallContext::multi_call_memory_ctx, palloc(), PG_FUNCTION_INFO_V1(), PG_GETARG_TEXT_PP, psprintf(), relation_close(), SRF_FIRSTCALL_INIT, SRF_IS_FIRSTCALL, SRF_PERCALL_SETUP, SRF_RETURN_DONE, SRF_RETURN_NEXT, TupleDescGetAttInMetadata(), TupleDescInitEntry(), FuncCallContext::user_fctx, and values.

Referenced by dblink_exec().

{
    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);

        /*
         * 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 = (char **) palloc(2 * sizeof(char *));
        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);
    }
}

dblink_get_result()

Datum dblink_get_result

(

PG_FUNCTION_ARGS

)

Definition at line 730 of file dblink.c.

References dblink_record_internal().

Referenced by dblink_send_query().

{
    return dblink_record_internal(fcinfo, true);
}

dblink_init()

static void dblink_init ( void  )

static

Definition at line 264 of file dblink.c.

References remoteConn::conn, dblink_connect(), MemoryContextAlloc(), remoteConn::newXactForCursor, remoteConn::openCursorCount, pconn, PG_FUNCTION_INFO_V1(), and TopMemoryContext.

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().

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

dblink_is_busy()

Datum dblink_is_busy

(

PG_FUNCTION_ARGS

)

Definition at line 1359 of file dblink.c.

References remoteConn::conn, dblink_cancel_query(), dblink_get_named_conn(), dblink_init(), PG_FUNCTION_INFO_V1(), PG_GETARG_TEXT_PP, PG_RETURN_INT32, PQconsumeInput(), PQisBusy(), and text_to_cstring().

Referenced by dblink_get_connections().

{
    PGconn     *conn;

    dblink_init();
    conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));

    PQconsumeInput(conn);
    PG_RETURN_INT32(PQisBusy(conn));
}

dblink_open()

Datum dblink_open

(

PG_FUNCTION_ARGS

)

Definition at line 424 of file dblink.c.

References appendStringInfo(), buf, remoteConn::conn, cstring_to_text(), StringInfoData::data, dblink_close(), dblink_init(), dblink_res_error(), get_fn_expr_argtype(), getConnectionByName(), initStringInfo(), remoteConn::newXactForCursor, remoteConn::openCursorCount, pconn, PG_FUNCTION_INFO_V1(), PG_GETARG_BOOL, PG_GETARG_TEXT_PP, PG_NARGS, PG_RETURN_TEXT_P, PGRES_COMMAND_OK, PQclear(), PQexec(), PQresultStatus(), PQTRANS_IDLE, PQtransactionStatus(), and text_to_cstring().

Referenced by dblink_disconnect().

{
    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 = PQexec(conn, "BEGIN");
        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 = PQexec(conn, buf.data);
    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"));
}

dblink_record()

Datum dblink_record

(

PG_FUNCTION_ARGS

)

Definition at line 698 of file dblink.c.

References dblink_record_internal(), dblink_send_query(), and PG_FUNCTION_INFO_V1().

Referenced by dblink_fetch().

{
    return dblink_record_internal(fcinfo, false);
}

dblink_record_internal()

static Datum dblink_record_internal ( FunctionCallInfo  fcinfo, bool  is_async  )

static

Definition at line 736 of file dblink.c.

References remoteConn::conn, dblink_get_conn(), dblink_get_named_conn(), dblink_init(), dblink_res_error(), elog, ERROR, FunctionCallInfoBaseData::flinfo, get_fn_expr_argtype(), 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, PQfinish(), PQgetResult(), PQresultStatus(), prepTuplestoreResult(), ReleaseExternalFD(), and text_to_cstring().

Referenced by dblink_get_result(), and dblink_record().

{
    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 = PQgetResult(conn);

            /* 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)
        {
            PQfinish(conn);
            ReleaseExternalFD();
        }
    }
    PG_END_TRY();

    return (Datum) 0;
}

dblink_res_error()

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

static

Definition at line 2736 of file dblink.c.

References ereport, errcode(), errcontext, errdetail_internal(), errhint(), errmsg(), errmsg_internal(), ERROR, 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(), vsnprintf, and xpstrdup().

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

{
    int         level;
    char       *pg_diag_sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
    char       *pg_diag_message_primary = PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY);
    char       *pg_diag_message_detail = PQresultErrorField(res, PG_DIAG_MESSAGE_DETAIL);
    char       *pg_diag_message_hint = PQresultErrorField(res, PG_DIAG_MESSAGE_HINT);
    char       *pg_diag_context = PQresultErrorField(res, PG_DIAG_CONTEXT);
    int         sqlstate;
    char       *message_primary;
    char       *message_detail;
    char       *message_hint;
    char       *message_context;
    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;

    message_primary = xpstrdup(pg_diag_message_primary);
    message_detail = xpstrdup(pg_diag_message_detail);
    message_hint = xpstrdup(pg_diag_message_hint);
    message_context = xpstrdup(pg_diag_context);

    /*
     * If we don't get a message from the PGresult, try the PGconn.  This is
     * needed because for connection-level failures, PQexec may just return
     * NULL, not a PGresult at all.
     */
    if (message_primary == NULL)
        message_primary = pchomp(PQerrorMessage(conn));

    /*
     * Now that we've copied all the data we need out of the PGresult, it's
     * safe to free it.  We must do this to avoid PGresult leakage.  We're
     * leaking all the strings too, but those are in palloc'd memory that will
     * get cleaned up eventually.
     */
    if (res)
        PQclear(res);

    /*
     * 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 ? 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))));
}

dblink_security_check()

static void dblink_security_check ( PGconn *  conn, remoteConn *  rconn  )

static

Definition at line 2667 of file dblink.c.

References ereport, errcode(), errdetail(), errhint(), errmsg(), ERROR, pfree(), PQconnectionUsedPassword(), PQfinish(), ReleaseExternalFD(), and superuser().

Referenced by dblink_connect(), and dblink_get_conn().

{
    if (!superuser())
    {
        if (!PQconnectionUsedPassword(conn))
        {
            PQfinish(conn);
            ReleaseExternalFD();
            if (rconn)
                pfree(rconn);

            ereport(ERROR,
                    (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
                     errmsg("password is required"),
                     errdetail("Non-superuser cannot connect if the server does not request a password."),
                     errhint("Target server's authentication method must be changed.")));
        }
    }
}

dblink_send_query()

Datum dblink_send_query

(

PG_FUNCTION_ARGS

)

Definition at line 705 of file dblink.c.

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

Referenced by dblink_record().

{
    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);
}

deleteConnection()

static void deleteConnection ( const char *  name )

static

Definition at line 2645 of file dblink.c.

References createConnHash(), ereport, errcode(), errmsg(), ERROR, HASH_REMOVE, hash_search(), sort-test::key, pstrdup(), remoteConnHash, and truncate_identifier().

Referenced by dblink_disconnect().

{
    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")));

}

escape_param_str()

static char * escape_param_str ( const char *  from )

static

Definition at line 2908 of file dblink.c.

References appendStringInfoChar(), buf, StringInfoData::data, and initStringInfo().

Referenced by get_connect_string().

{
    const char *cp;
    StringInfoData buf;

    initStringInfo(&buf);

    for (cp = str; *cp; cp++)
    {
        if (*cp == '\\' || *cp == '\'')
            appendStringInfoChar(&buf, '\\');
        appendStringInfoChar(&buf, *cp);
    }

    return buf.data;
}

generate_relation_name()

static char * generate_relation_name ( Relation  rel )

static

Definition at line 2565 of file dblink.c.

References 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().

{
    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;
}

get_attnum_pk_pos()

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

static

Definition at line 2412 of file dblink.c.

References i.

Referenced by get_sql_insert(), and get_sql_update().

{
    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;
}

get_connect_string()

static char * get_connect_string ( const char *  servername )

static

Definition at line 2818 of file dblink.c.

References ACL_USAGE, aclcheck_error(), ACLCHECK_OK, appendStringInfo(), DefElem::arg, buf, StringInfoData::data, DefElem::defname, ereport, errcode(), errdetail(), errmsg(), ERROR, escape_param_str(), ForeignServer::fdwid, GetForeignDataWrapper(), GetForeignServerByName(), GetUserId(), GetUserMapping(), initStringInfo(), is_valid_dblink_option(), lfirst, OBJECT_FOREIGN_SERVER, ForeignDataWrapper::options, ForeignServer::options, UserMapping::options, pg_foreign_server_aclcheck(), PQconndefaults(), pstrdup(), ForeignServer::serverid, ForeignServer::servername, strVal, and truncate_identifier().

Referenced by dblink_connect(), and dblink_get_conn().

{
    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 = pg_foreign_server_aclcheck(serverid, userid, ACL_USAGE);
        if (aclresult != ACLCHECK_OK)
            aclcheck_error(aclresult, OBJECT_FOREIGN_SERVER, foreign_server->servername);

        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;
}

get_pkey_attnames()

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

static

Definition at line 2078 of file dblink.c.

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

Referenced by dblink_get_pkey().

{
    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 = (char **) palloc(*indnkeyatts * sizeof(char *));

                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;
}

get_rel_from_relname()

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

static

Definition at line 2540 of file dblink.c.

References aclcheck_error(), ACLCHECK_OK, get_relkind_objtype(), GetUserId(), makeRangeVarFromNameList(), pg_class_aclcheck(), RelationData::rd_rel, RelationGetRelationName, RelationGetRelid, table_openrv(), and textToQualifiedNameList().

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

{
    RangeVar   *relvar;
    Relation    rel;
    AclResult   aclresult;

    relvar = makeRangeVarFromNameList(textToQualifiedNameList(relname_text));
    rel = table_openrv(relvar, lockmode);

    aclresult = pg_class_aclcheck(RelationGetRelid(rel), GetUserId(),
                                  aclmode);
    if (aclresult != ACLCHECK_OK)
        aclcheck_error(aclresult, get_relkind_objtype(rel->rd_rel->relkind),
                       RelationGetRelationName(rel));

    return rel;
}

get_sql_delete()

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

static

Definition at line 2269 of file dblink.c.

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

Referenced by dblink_build_sql_delete().

{
    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;
}

get_sql_insert()

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

static

Definition at line 2189 of file dblink.c.

References appendStringInfo(), appendStringInfoChar(), appendStringInfoString(), buf, StringInfoData::data, ereport, errcode(), errmsg(), ERROR, generate_relation_name(), get_attnum_pk_pos(), get_tuple_of_interest(), i, initStringInfo(), sort-test::key, 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().

{
    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;
}

get_sql_update()

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

static

Definition at line 2306 of file dblink.c.

References appendStringInfo(), appendStringInfoString(), buf, StringInfoData::data, ereport, errcode(), errmsg(), ERROR, generate_relation_name(), get_attnum_pk_pos(), get_tuple_of_interest(), i, initStringInfo(), sort-test::key, 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().

{
    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;
}

get_text_array_contents()

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

static

Definition at line 2133 of file dblink.c.

References ARR_DATA_PTR, ARR_DIMS, ARR_ELEMTYPE, ARR_NDIM, ARR_NULLBITMAP, ArrayGetNItems(), Assert, att_addlength_pointer, att_align_nominal, get_typlenbyvalalign(), i, palloc(), PointerGetDatum, TextDatumGetCString, typalign, and values.

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

{
    int         ndim = ARR_NDIM(array);
    int        *dims = ARR_DIMS(array);
    int         nitems;
    int16       typlen;
    bool        typbyval;
    char        typalign;
    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);

    values = (char **) palloc(nitems * sizeof(char *));

    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_align_nominal(ptr, typalign);
        }

        /* advance bitmap pointer if any */
        if (bitmap)
        {
            bitmask <<= 1;
            if (bitmask == 0x100)
            {
                bitmap++;
                bitmask = 1;
            }
        }
    }

    return values;
}

get_tuple_of_interest()

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

static

Definition at line 2427 of file dblink.c.

References appendStringInfo(), appendStringInfoString(), buf, StringInfoData::data, elog, ereport, errcode(), errmsg(), ERROR, 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().

{
    char       *relname;
    TupleDesc   tupdesc;
    int         natts;
    StringInfoData buf;
    int         ret;
    HeapTuple   tuple;
    int         i;

    /*
     * Connect to SPI manager
     */
    if ((ret = SPI_connect()) < 0)
        /* internal error */
        elog(ERROR, "SPI connect failure - returned %d", ret);

    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;
}

getConnectionByName()

static remoteConn * getConnectionByName ( const char *  name )

static

Definition at line 2583 of file dblink.c.

References createConnHash(), HASH_FIND, hash_search(), sort-test::key, pstrdup(), remoteConnHashEnt::rconn, remoteConnHash, and truncate_identifier().

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

{
    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)
        return hentry->rconn;

    return NULL;
}

is_valid_dblink_option()

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

static

Definition at line 3012 of file dblink.c.

References _PQconninfoOption::dispchar, and _PQconninfoOption::keyword.

Referenced by dblink_fdw_validator(), and get_connect_string().

{
    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;

    /*
     * 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;
}

materializeQueryResult()

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

static

Definition at line 1029 of file dblink.c.

References FunctionCallInfoBaseData::resultinfo.

Referenced by dblink_record_internal().

{
    ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
    PGresult   *volatile res = NULL;
    volatile storeInfo sinfo = {0};

    /* prepTuplestoreResult must have been called previously */
    Assert(rsinfo->returnMode == SFRM_Materialize);

    sinfo.fcinfo = fcinfo;

    PG_TRY();
    {
        /* 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 will clear the passed PGresult, so we need
             * this ugly dance to avoid doing so twice during error exit
             */
            PGresult   *res1 = res;

            res = NULL;
            dblink_res_error(conn, conname, res1, 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);
            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);
            res = NULL;
        }
        else
        {
            Assert(PQresultStatus(res) == PGRES_TUPLES_OK);
            /* storeRow should have created a tuplestore */
            Assert(rsinfo->setResult != NULL);

            PQclear(res);
            res = NULL;
        }

        /* clean up data conversion short-lived memory context */
        if (sinfo.tmpcontext != NULL)
            MemoryContextDelete(sinfo.tmpcontext);
        sinfo.tmpcontext = NULL;

        PQclear(sinfo.last_res);
        sinfo.last_res = NULL;
        PQclear(sinfo.cur_res);
        sinfo.cur_res = NULL;
    }
    PG_CATCH();
    {
        /* be sure to release any libpq result we collected */
        PQclear(res);
        PQclear(sinfo.last_res);
        PQclear(sinfo.cur_res);
        /* and clear out any pending data in libpq */
        while ((res = PQgetResult(conn)) != NULL)
            PQclear(res);
        PG_RE_THROW();
    }
    PG_END_TRY();
}

materializeResult()

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

static

Definition at line 887 of file dblink.c.

References applyRemoteGucs(), Assert, BuildTupleFromCStrings(), CreateTemplateTupleDesc(), CreateTupleDescCopy(), ReturnSetInfo::econtext, ExprContext::ecxt_per_query_memory, elog, ereport, errcode(), errmsg(), ERROR, get_call_result_type(), i, MemoryContextSwitchTo(), TupleDescData::natts, palloc(), PG_END_TRY, PG_FINALLY, PG_TRY, PGRES_COMMAND_OK, PGRES_TUPLES_OK, PQclear(), PQcmdStatus(), PQgetisnull(), PQgetvalue(), PQnfields(), PQntuples(), PQresultStatus(), restoreLocalGucs(), FunctionCallInfoBaseData::resultinfo, ReturnSetInfo::returnMode, ReturnSetInfo::setDesc, ReturnSetInfo::setResult, SFRM_Materialize, TupleDescGetAttInMetadata(), TupleDescInitEntry(), tuplestore_begin_heap(), tuplestore_donestoring, tuplestore_puttuple(), TYPEFUNC_COMPOSITE, TYPEFUNC_RECORD, values, and work_mem.

Referenced by dblink_fetch(), and dblink_record_internal().

{
    ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;

    /* prepTuplestoreResult must have been called previously */
    Assert(rsinfo->returnMode == SFRM_Materialize);

    PG_TRY();
    {
        TupleDesc   tupdesc;
        bool        is_sql_cmd;
        int         ntuples;
        int         nfields;

        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);
            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 = (char **) palloc(nfields * sizeof(char *));

            /* 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);

            /* clean up and return the tuplestore */
            tuplestore_donestoring(tupstore);
        }
    }
    PG_FINALLY();
    {
        /* be sure to release the libpq result */
        PQclear(res);
    }
    PG_END_TRY();
}

pg_attribute_noreturn()

static void pg_attribute_noreturn ( )

static

Definition at line 155 of file dblink.c.

References elog, ereport, errcode(), errmsg(), ERROR, pchomp(), PQclear(), and PQerrorMessage().

{
    char       *msg = pchomp(PQerrorMessage(conn));

    if (res)
        PQclear(res);
    elog(ERROR, "%s: %s", p2, msg);
}

PG_FUNCTION_INFO_V1() [1/20]

PG_FUNCTION_INFO_V1

(

dblink_connect

)

Referenced by dblink_build_sql_delete(), dblink_build_sql_insert(), dblink_build_sql_update(), dblink_cancel_query(), dblink_close(), dblink_connect(), dblink_disconnect(), dblink_error_message(), dblink_exec(), dblink_fetch(), dblink_get_connections(), dblink_get_notify(), dblink_get_pkey(), dblink_init(), dblink_is_busy(), dblink_open(), dblink_record(), dblink_send_query(), and storeRow().

PG_FUNCTION_INFO_V1() [2/20]

PG_FUNCTION_INFO_V1

(

dblink_disconnect

)

PG_FUNCTION_INFO_V1() [3/20]

PG_FUNCTION_INFO_V1

(

dblink_open

)

PG_FUNCTION_INFO_V1() [4/20]

PG_FUNCTION_INFO_V1

(

dblink_close

)

PG_FUNCTION_INFO_V1() [5/20]

PG_FUNCTION_INFO_V1

(

dblink_fetch

)

PG_FUNCTION_INFO_V1() [6/20]

PG_FUNCTION_INFO_V1

(

dblink_record

)

PG_FUNCTION_INFO_V1() [7/20]

PG_FUNCTION_INFO_V1

(

dblink_send_query

)

PG_FUNCTION_INFO_V1() [8/20]

PG_FUNCTION_INFO_V1

(

dblink_get_result

)

PG_FUNCTION_INFO_V1() [9/20]

PG_FUNCTION_INFO_V1

(

dblink_get_connections

)

PG_FUNCTION_INFO_V1() [10/20]

PG_FUNCTION_INFO_V1

(

dblink_is_busy

)

PG_FUNCTION_INFO_V1() [11/20]

PG_FUNCTION_INFO_V1

(

dblink_cancel_query

)

PG_FUNCTION_INFO_V1() [12/20]

PG_FUNCTION_INFO_V1

(

dblink_error_message

)

PG_FUNCTION_INFO_V1() [13/20]

PG_FUNCTION_INFO_V1

(

dblink_exec

)

PG_FUNCTION_INFO_V1() [14/20]

PG_FUNCTION_INFO_V1

(

dblink_get_pkey

)

PG_FUNCTION_INFO_V1() [15/20]

PG_FUNCTION_INFO_V1

(

dblink_build_sql_insert

)

PG_FUNCTION_INFO_V1() [16/20]

PG_FUNCTION_INFO_V1

(

dblink_build_sql_delete

)

PG_FUNCTION_INFO_V1() [17/20]

PG_FUNCTION_INFO_V1

(

dblink_build_sql_update

)

PG_FUNCTION_INFO_V1() [18/20]

PG_FUNCTION_INFO_V1

(

dblink_current_query

)

PG_FUNCTION_INFO_V1() [19/20]

PG_FUNCTION_INFO_V1

(

dblink_get_notify

)

PG_FUNCTION_INFO_V1() [20/20]

PG_FUNCTION_INFO_V1

(

dblink_fdw_validator

)

prepTuplestoreResult()

static void prepTuplestoreResult ( FunctionCallInfo  fcinfo )

static

Definition at line 859 of file dblink.c.

References ReturnSetInfo::allowedModes, ereport, errcode(), errmsg(), ERROR, IsA, FunctionCallInfoBaseData::resultinfo, ReturnSetInfo::returnMode, ReturnSetInfo::setDesc, ReturnSetInfo::setResult, and SFRM_Materialize.

Referenced by dblink_fetch(), dblink_get_notify(), and dblink_record_internal().

{
    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;
}

quote_ident_cstr()

static char * quote_ident_cstr ( char *  rawstr )

static

Definition at line 2397 of file dblink.c.

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

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

{
    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;
}

restoreLocalGucs()

static void restoreLocalGucs ( int  nestlevel )

static

Definition at line 3115 of file dblink.c.

References AtEOXact_GUC().

Referenced by materializeResult(), and storeQueryResult().

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

storeQueryResult()

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

static

Definition at line 1144 of file dblink.c.

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

{
    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 = PQgetResult(conn);
        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;
}

storeRow()

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

static

Definition at line 1212 of file dblink.c.

References Assert, storeInfo::attinmeta, BuildTupleFromCStrings(), CreateTupleDescCopy(), storeInfo::cstrs, dblink_get_connections(), ReturnSetInfo::econtext, ExprContext::ecxt_per_query_memory, elog, ereport, errcode(), errmsg(), ERROR, storeInfo::fcinfo, get_call_result_type(), i, MemoryContextReset(), MemoryContextSwitchTo(), TupleDescData::natts, palloc(), pfree(), PG_FUNCTION_INFO_V1(), PQgetisnull(), PQgetvalue(), PQnfields(), PQntuples(), FunctionCallInfoBaseData::resultinfo, ReturnSetInfo::setDesc, ReturnSetInfo::setResult, storeInfo::tmpcontext, TupleDescGetAttInMetadata(), storeInfo::tuplestore, tuplestore_begin_heap(), tuplestore_end(), tuplestore_puttuple(), TYPEFUNC_COMPOSITE, TYPEFUNC_RECORD, and work_mem.

Referenced by storeQueryResult().

{
    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 = (char **) palloc(nfields * sizeof(char *));
    }

    /* 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);
}

validate_pkattnums()

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

static

Definition at line 2941 of file dblink.c.

References int2vector::dim1, ereport, errcode(), errmsg(), ERROR, i, Min, TupleDescData::natts, palloc(), RelationData::rd_att, TupleDescAttr, and int2vector::values.

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

{
    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 = (int *) palloc(pknumatts_arg * sizeof(int));
    *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 (TupleDescAttr(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)));
    }
}

xpstrdup()

static char* xpstrdup ( const char *  in )

static

Definition at line 147 of file dblink.c.

References pstrdup().

Referenced by dblink_res_error().

{
    if (in == NULL)
        return NULL;
    return pstrdup(in);
}

Variable Documentation

pconn

remoteConn* pconn = NULL

static

Definition at line 127 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().

PG_MODULE_MAGIC

PG_MODULE_MAGIC

Definition at line 63 of file dblink.c.

remoteConnHash

HTAB* remoteConnHash = NULL

static

Definition at line 128 of file dblink.c.

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