diff options
Diffstat (limited to 'contrib/dblink')
| -rw-r--r-- | contrib/dblink/.gitignore | 4 | ||||
| -rw-r--r-- | contrib/dblink/Makefile | 28 | ||||
| -rw-r--r-- | contrib/dblink/dblink--1.0--1.1.sql | 14 | ||||
| -rw-r--r-- | contrib/dblink/dblink--1.1--1.2.sql | 46 | ||||
| -rw-r--r-- | contrib/dblink/dblink--1.2.sql | 235 | ||||
| -rw-r--r-- | contrib/dblink/dblink.c | 3121 | ||||
| -rw-r--r-- | contrib/dblink/dblink.control | 5 | ||||
| -rw-r--r-- | contrib/dblink/expected/.gitignore | 1 | ||||
| -rw-r--r-- | contrib/dblink/expected/dblink.out | 1181 | ||||
| -rw-r--r-- | contrib/dblink/input/paths.source | 14 | ||||
| -rw-r--r-- | contrib/dblink/output/paths.source | 11 | ||||
| -rw-r--r-- | contrib/dblink/pg_service.conf | 7 | ||||
| -rw-r--r-- | contrib/dblink/sql/.gitignore | 1 | ||||
| -rw-r--r-- | contrib/dblink/sql/dblink.sql | 603 |
14 files changed, 5271 insertions, 0 deletions
diff --git a/contrib/dblink/.gitignore b/contrib/dblink/.gitignore new file mode 100644 index 0000000..5dcb3ff --- /dev/null +++ b/contrib/dblink/.gitignore @@ -0,0 +1,4 @@ +# Generated subdirectories +/log/ +/results/ +/tmp_check/ diff --git a/contrib/dblink/Makefile b/contrib/dblink/Makefile new file mode 100644 index 0000000..b008c8c --- /dev/null +++ b/contrib/dblink/Makefile @@ -0,0 +1,28 @@ +# contrib/dblink/Makefile + +MODULE_big = dblink +OBJS = \ + $(WIN32RES) \ + dblink.o +PG_CPPFLAGS = -I$(libpq_srcdir) +SHLIB_LINK_INTERNAL = $(libpq) + +EXTENSION = dblink +DATA = dblink--1.2.sql dblink--1.1--1.2.sql dblink--1.0--1.1.sql +PGFILEDESC = "dblink - connect to other PostgreSQL databases" + +REGRESS = paths dblink +REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress +EXTRA_CLEAN = sql/paths.sql expected/paths.out + +ifdef USE_PGXS +PG_CONFIG = pg_config +PGXS := $(shell $(PG_CONFIG) --pgxs) +include $(PGXS) +else +SHLIB_PREREQS = submake-libpq +subdir = contrib/dblink +top_builddir = ../.. +include $(top_builddir)/src/Makefile.global +include $(top_srcdir)/contrib/contrib-global.mk +endif diff --git a/contrib/dblink/dblink--1.0--1.1.sql b/contrib/dblink/dblink--1.0--1.1.sql new file mode 100644 index 0000000..67293f0 --- /dev/null +++ b/contrib/dblink/dblink--1.0--1.1.sql @@ -0,0 +1,14 @@ +/* contrib/dblink/dblink--1.0--1.1.sql */ + +-- complain if script is sourced in psql, rather than via ALTER EXTENSION +\echo Use "ALTER EXTENSION dblink UPDATE TO '1.1'" to load this file. \quit + +CREATE FUNCTION dblink_fdw_validator( + options text[], + catalog oid +) +RETURNS void +AS 'MODULE_PATHNAME', 'dblink_fdw_validator' +LANGUAGE C STRICT; + +CREATE FOREIGN DATA WRAPPER dblink_fdw VALIDATOR dblink_fdw_validator; diff --git a/contrib/dblink/dblink--1.1--1.2.sql b/contrib/dblink/dblink--1.1--1.2.sql new file mode 100644 index 0000000..69d58af --- /dev/null +++ b/contrib/dblink/dblink--1.1--1.2.sql @@ -0,0 +1,46 @@ +/* contrib/dblink/dblink--1.1--1.2.sql */ + +-- complain if script is sourced in psql, rather than via ALTER EXTENSION +\echo Use "ALTER EXTENSION dblink UPDATE TO '1.2'" to load this file. \quit + +ALTER FUNCTION dblink_connect(text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_connect(text, text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_connect_u(text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_connect_u(text, text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_disconnect() PARALLEL RESTRICTED; +ALTER FUNCTION dblink_disconnect(text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_open(text, text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_open(text, text, boolean) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_open(text, text, text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_open(text, text, text, boolean) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_fetch(text, int) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_fetch(text, int, boolean) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_fetch(text, text, int) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_fetch(text, text, int, boolean) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_close(text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_close(text, boolean) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_close(text, text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_close(text, text, boolean) PARALLEL RESTRICTED; +ALTER FUNCTION dblink(text, text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink(text, text, boolean) PARALLEL RESTRICTED; +ALTER FUNCTION dblink(text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink(text, boolean) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_exec(text, text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_exec(text, text, boolean) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_exec(text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_exec(text, boolean) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_get_pkey(text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_build_sql_insert(text, int2vector, int, _text, _text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_build_sql_delete(text, int2vector, int, _text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_build_sql_update(text, int2vector, int, _text, _text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_current_query() PARALLEL RESTRICTED; +ALTER FUNCTION dblink_send_query(text, text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_is_busy(text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_get_result(text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_get_result(text, bool) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_get_connections() PARALLEL RESTRICTED; +ALTER FUNCTION dblink_cancel_query(text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_error_message(text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_get_notify() PARALLEL RESTRICTED; +ALTER FUNCTION dblink_get_notify(text) PARALLEL RESTRICTED; +ALTER FUNCTION dblink_fdw_validator(text[], oid) PARALLEL SAFE; diff --git a/contrib/dblink/dblink--1.2.sql b/contrib/dblink/dblink--1.2.sql new file mode 100644 index 0000000..405eccb --- /dev/null +++ b/contrib/dblink/dblink--1.2.sql @@ -0,0 +1,235 @@ +/* contrib/dblink/dblink--1.2.sql */ + +-- complain if script is sourced in psql, rather than via CREATE EXTENSION +\echo Use "CREATE EXTENSION dblink" to load this file. \quit + +-- dblink_connect now restricts non-superusers to password +-- authenticated connections +CREATE FUNCTION dblink_connect (text) +RETURNS text +AS 'MODULE_PATHNAME','dblink_connect' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_connect (text, text) +RETURNS text +AS 'MODULE_PATHNAME','dblink_connect' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +-- dblink_connect_u allows non-superusers to use +-- non-password authenticated connections, but initially +-- privileges are revoked from public +CREATE FUNCTION dblink_connect_u (text) +RETURNS text +AS 'MODULE_PATHNAME','dblink_connect' +LANGUAGE C STRICT PARALLEL RESTRICTED SECURITY DEFINER; + +CREATE FUNCTION dblink_connect_u (text, text) +RETURNS text +AS 'MODULE_PATHNAME','dblink_connect' +LANGUAGE C STRICT PARALLEL RESTRICTED SECURITY DEFINER; + +REVOKE ALL ON FUNCTION dblink_connect_u (text) FROM public; +REVOKE ALL ON FUNCTION dblink_connect_u (text, text) FROM public; + +CREATE FUNCTION dblink_disconnect () +RETURNS text +AS 'MODULE_PATHNAME','dblink_disconnect' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_disconnect (text) +RETURNS text +AS 'MODULE_PATHNAME','dblink_disconnect' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_open (text, text) +RETURNS text +AS 'MODULE_PATHNAME','dblink_open' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_open (text, text, boolean) +RETURNS text +AS 'MODULE_PATHNAME','dblink_open' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_open (text, text, text) +RETURNS text +AS 'MODULE_PATHNAME','dblink_open' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_open (text, text, text, boolean) +RETURNS text +AS 'MODULE_PATHNAME','dblink_open' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_fetch (text, int) +RETURNS setof record +AS 'MODULE_PATHNAME','dblink_fetch' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_fetch (text, int, boolean) +RETURNS setof record +AS 'MODULE_PATHNAME','dblink_fetch' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_fetch (text, text, int) +RETURNS setof record +AS 'MODULE_PATHNAME','dblink_fetch' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_fetch (text, text, int, boolean) +RETURNS setof record +AS 'MODULE_PATHNAME','dblink_fetch' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_close (text) +RETURNS text +AS 'MODULE_PATHNAME','dblink_close' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_close (text, boolean) +RETURNS text +AS 'MODULE_PATHNAME','dblink_close' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_close (text, text) +RETURNS text +AS 'MODULE_PATHNAME','dblink_close' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_close (text, text, boolean) +RETURNS text +AS 'MODULE_PATHNAME','dblink_close' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink (text, text) +RETURNS setof record +AS 'MODULE_PATHNAME','dblink_record' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink (text, text, boolean) +RETURNS setof record +AS 'MODULE_PATHNAME','dblink_record' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink (text) +RETURNS setof record +AS 'MODULE_PATHNAME','dblink_record' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink (text, boolean) +RETURNS setof record +AS 'MODULE_PATHNAME','dblink_record' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_exec (text, text) +RETURNS text +AS 'MODULE_PATHNAME','dblink_exec' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_exec (text, text, boolean) +RETURNS text +AS 'MODULE_PATHNAME','dblink_exec' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_exec (text) +RETURNS text +AS 'MODULE_PATHNAME','dblink_exec' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_exec (text,boolean) +RETURNS text +AS 'MODULE_PATHNAME','dblink_exec' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE TYPE dblink_pkey_results AS (position int, colname text); + +CREATE FUNCTION dblink_get_pkey (text) +RETURNS setof dblink_pkey_results +AS 'MODULE_PATHNAME','dblink_get_pkey' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_build_sql_insert (text, int2vector, int, _text, _text) +RETURNS text +AS 'MODULE_PATHNAME','dblink_build_sql_insert' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_build_sql_delete (text, int2vector, int, _text) +RETURNS text +AS 'MODULE_PATHNAME','dblink_build_sql_delete' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_build_sql_update (text, int2vector, int, _text, _text) +RETURNS text +AS 'MODULE_PATHNAME','dblink_build_sql_update' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_current_query () +RETURNS text +AS 'MODULE_PATHNAME','dblink_current_query' +LANGUAGE C PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_send_query(text, text) +RETURNS int4 +AS 'MODULE_PATHNAME', 'dblink_send_query' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_is_busy(text) +RETURNS int4 +AS 'MODULE_PATHNAME', 'dblink_is_busy' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_get_result(text) +RETURNS SETOF record +AS 'MODULE_PATHNAME', 'dblink_get_result' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_get_result(text, bool) +RETURNS SETOF record +AS 'MODULE_PATHNAME', 'dblink_get_result' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_get_connections() +RETURNS text[] +AS 'MODULE_PATHNAME', 'dblink_get_connections' +LANGUAGE C PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_cancel_query(text) +RETURNS text +AS 'MODULE_PATHNAME', 'dblink_cancel_query' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_error_message(text) +RETURNS text +AS 'MODULE_PATHNAME', 'dblink_error_message' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_get_notify( + OUT notify_name TEXT, + OUT be_pid INT4, + OUT extra TEXT +) +RETURNS setof record +AS 'MODULE_PATHNAME', 'dblink_get_notify' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +CREATE FUNCTION dblink_get_notify( + conname TEXT, + OUT notify_name TEXT, + OUT be_pid INT4, + OUT extra TEXT +) +RETURNS setof record +AS 'MODULE_PATHNAME', 'dblink_get_notify' +LANGUAGE C STRICT PARALLEL RESTRICTED; + +/* New stuff in 1.1 begins here */ + +CREATE FUNCTION dblink_fdw_validator( + options text[], + catalog oid +) +RETURNS void +AS 'MODULE_PATHNAME', 'dblink_fdw_validator' +LANGUAGE C STRICT PARALLEL SAFE; + +CREATE FOREIGN DATA WRAPPER dblink_fdw VALIDATOR dblink_fdw_validator; diff --git a/contrib/dblink/dblink.c b/contrib/dblink/dblink.c new file mode 100644 index 0000000..824d301 --- /dev/null +++ b/contrib/dblink/dblink.c @@ -0,0 +1,3121 @@ +/* + * dblink.c + * + * Functions returning results from a remote database + * + * Joe Conway <mail@joeconway.com> + * And contributors: + * Darko Prenosil <Darko.Prenosil@finteh.hr> + * Shridhar Daithankar <shridhar_daithankar@persistent.co.in> + * + * contrib/dblink/dblink.c + * Copyright (c) 2001-2021, PostgreSQL Global Development Group + * ALL RIGHTS RESERVED; + * + * Permission to use, copy, modify, and distribute this software and its + * documentation for any purpose, without fee, and without a written agreement + * is hereby granted, provided that the above copyright notice and this + * paragraph and the following two paragraphs appear in all copies. + * + * IN NO EVENT SHALL THE AUTHOR OR DISTRIBUTORS BE LIABLE TO ANY PARTY FOR + * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING + * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS + * DOCUMENTATION, EVEN IF THE AUTHOR OR DISTRIBUTORS HAVE BEEN ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + * + * THE AUTHOR AND DISTRIBUTORS SPECIFICALLY DISCLAIMS ANY WARRANTIES, + * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY + * AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS + * ON AN "AS IS" BASIS, AND THE AUTHOR AND DISTRIBUTORS HAS NO OBLIGATIONS TO + * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS. + * + */ +#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" + +PG_MODULE_MAGIC; + +typedef struct remoteConn +{ + PGconn *conn; /* Hold the remote connection */ + int openCursorCount; /* The number of open cursors */ + bool newXactForCursor; /* Opened a transaction for a cursor */ +} remoteConn; + +typedef struct storeInfo +{ + FunctionCallInfo fcinfo; + Tuplestorestate *tuplestore; + AttInMetadata *attinmeta; + MemoryContext tmpcontext; + char **cstrs; + /* temp storage for results to avoid leaks on exception */ + PGresult *last_res; + PGresult *cur_res; +} storeInfo; + +/* + * Internal declarations + */ +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, 6); +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); + +/* Global */ +static remoteConn *pconn = NULL; +static HTAB *remoteConnHash = NULL; + +/* + * Following is list that holds multiple remote connections. + * Calling convention of each dblink function changes to accept + * connection name as the first parameter. The connection list is + * much like ecpg e.g. a mapping between a name and a PGconn object. + */ + +typedef struct remoteConnHashEnt +{ + char name[NAMEDATALEN]; + remoteConn *rconn; +} remoteConnHashEnt; + +/* initial number of connection hashes */ +#define NUMCONN 16 + +static char * +xpstrdup(const char *in) +{ + if (in == NULL) + return NULL; + return pstrdup(in); +} + +static void +pg_attribute_noreturn() +dblink_res_internalerror(PGconn *conn, PGresult *res, const char *p2) +{ + char *msg = pchomp(PQerrorMessage(conn)); + + if (res) + PQclear(res); + elog(ERROR, "%s: %s", p2, msg); +} + +static void +pg_attribute_noreturn() +dblink_conn_not_avail(const char *conname) +{ + if (conname) + ereport(ERROR, + (errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST), + errmsg("connection \"%s\" not available", conname))); + else + ereport(ERROR, + (errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST), + errmsg("connection not available"))); +} + +static void +dblink_get_conn(char *conname_or_str, + PGconn *volatile *conn_p, char **conname_p, volatile bool *freeconn_p) +{ + 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; +} + +static PGconn * +dblink_get_named_conn(const char *conname) +{ + remoteConn *rconn = getConnectionByName(conname); + + if (rconn) + return rconn->conn; + + dblink_conn_not_avail(conname); + return NULL; /* keep compiler quiet */ +} + +static void +dblink_init(void) +{ + if (!pconn) + { + pconn = (remoteConn *) MemoryContextAlloc(TopMemoryContext, sizeof(remoteConn)); + pconn->conn = NULL; + pconn->openCursorCount = 0; + pconn->newXactForCursor = false; + } +} + +/* + * Create a persistent connection to another database + */ +PG_FUNCTION_INFO_V1(dblink_connect); +Datum +dblink_connect(PG_FUNCTION_ARGS) +{ + 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")); +} + +/* + * Clear a persistent connection to another database + */ +PG_FUNCTION_INFO_V1(dblink_disconnect); +Datum +dblink_disconnect(PG_FUNCTION_ARGS) +{ + 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")); +} + +/* + * opens a cursor using a persistent connection + */ +PG_FUNCTION_INFO_V1(dblink_open); +Datum +dblink_open(PG_FUNCTION_ARGS) +{ + 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")); +} + +/* + * closes a cursor + */ +PG_FUNCTION_INFO_V1(dblink_close); +Datum +dblink_close(PG_FUNCTION_ARGS) +{ + 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")); +} + +/* + * Fetch results from an open cursor + */ +PG_FUNCTION_INFO_V1(dblink_fetch); +Datum +dblink_fetch(PG_FUNCTION_ARGS) +{ + 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; +} + +/* + * Note: this is the new preferred version of dblink + */ +PG_FUNCTION_INFO_V1(dblink_record); +Datum +dblink_record(PG_FUNCTION_ARGS) +{ + return dblink_record_internal(fcinfo, false); +} + +PG_FUNCTION_INFO_V1(dblink_send_query); +Datum +dblink_send_query(PG_FUNCTION_ARGS) +{ + 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); +} + +PG_FUNCTION_INFO_V1(dblink_get_result); +Datum +dblink_get_result(PG_FUNCTION_ARGS) +{ + return dblink_record_internal(fcinfo, true); +} + +static Datum +dblink_record_internal(FunctionCallInfo fcinfo, bool is_async) +{ + 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; +} + +/* + * Verify function caller can handle a tuplestore result, and set up for that. + * + * Note: if the caller returns without actually creating a tuplestore, the + * executor will treat the function result as an empty set. + */ +static void +prepTuplestoreResult(FunctionCallInfo fcinfo) +{ + 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; +} + +/* + * Copy the contents of the PGresult into a tuplestore to be returned + * as the result of the current function. + * The PGresult will be released in this function. + */ +static void +materializeResult(FunctionCallInfo fcinfo, PGconn *conn, PGresult *res) +{ + 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(); +} + +/* + * Execute the given SQL command and store its results into a tuplestore + * to be returned as the result of the current function. + * + * This is equivalent to PQexec followed by materializeResult, but we make + * use of libpq's single-row mode to avoid accumulating the whole result + * inside libpq before it gets transferred to the tuplestore. + */ +static void +materializeQueryResult(FunctionCallInfo fcinfo, + PGconn *conn, + const char *conname, + const char *sql, + bool fail) +{ + 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(); +} + +/* + * Execute query, and send any result rows to sinfo->tuplestore. + */ +static PGresult * +storeQueryResult(volatile storeInfo *sinfo, PGconn *conn, const char *sql) +{ + 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; +} + +/* + * Send single row to sinfo->tuplestore. + * + * If "first" is true, create the tuplestore using PGresult's metadata + * (in this case the PGresult might contain either zero or one row). + */ +static void +storeRow(volatile storeInfo *sinfo, PGresult *res, bool first) +{ + 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); +} + +/* + * List all open dblink connections by name. + * Returns an array of all connection names. + * Takes no params + */ +PG_FUNCTION_INFO_V1(dblink_get_connections); +Datum +dblink_get_connections(PG_FUNCTION_ARGS) +{ + 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(); +} + +/* + * Checks if a given remote connection is busy + * + * Returns 1 if the connection is busy, 0 otherwise + * Params: + * text connection_name - name of the connection to check + * + */ +PG_FUNCTION_INFO_V1(dblink_is_busy); +Datum +dblink_is_busy(PG_FUNCTION_ARGS) +{ + PGconn *conn; + + dblink_init(); + conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0))); + + PQconsumeInput(conn); + PG_RETURN_INT32(PQisBusy(conn)); +} + +/* + * Cancels a running request on a connection + * + * Returns text: + * "OK" if the cancel request has been sent correctly, + * an error message otherwise + * + * Params: + * text connection_name - name of the connection to check + * + */ +PG_FUNCTION_INFO_V1(dblink_cancel_query); +Datum +dblink_cancel_query(PG_FUNCTION_ARGS) +{ + 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)); +} + + +/* + * Get error message from a connection + * + * Returns text: + * "OK" if no error, an error message otherwise + * + * Params: + * text connection_name - name of the connection to check + * + */ +PG_FUNCTION_INFO_V1(dblink_error_message); +Datum +dblink_error_message(PG_FUNCTION_ARGS) +{ + 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))); +} + +/* + * Execute an SQL non-SELECT command + */ +PG_FUNCTION_INFO_V1(dblink_exec); +Datum +dblink_exec(PG_FUNCTION_ARGS) +{ + 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_get_pkey + * + * Return list of primary key fields for the supplied relation, + * or NULL if none exists. + */ +PG_FUNCTION_INFO_V1(dblink_get_pkey); +Datum +dblink_get_pkey(PG_FUNCTION_ARGS) +{ + 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_build_sql_insert + * + * Used to generate an SQL insert statement + * based on an existing tuple in a local relation. + * This is useful for selectively replicating data + * to another server via dblink. + * + * API: + * <relname> - name of local table of interest + * <pkattnums> - an int2vector of attnums which will be used + * to identify the local tuple of interest + * <pknumatts> - number of attnums in pkattnums + * <src_pkattvals_arry> - text array of key values which will be used + * to identify the local tuple of interest + * <tgt_pkattvals_arry> - text array of key values which will be used + * to build the string for execution remotely. These are substituted + * for their counterparts in src_pkattvals_arry + */ +PG_FUNCTION_INFO_V1(dblink_build_sql_insert); +Datum +dblink_build_sql_insert(PG_FUNCTION_ARGS) +{ + 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_delete + * + * Used to generate an SQL delete statement. + * This is useful for selectively replicating a + * delete to another server via dblink. + * + * API: + * <relname> - name of remote table of interest + * <pkattnums> - an int2vector of attnums which will be used + * to identify the remote tuple of interest + * <pknumatts> - number of attnums in pkattnums + * <tgt_pkattvals_arry> - text array of key values which will be used + * to build the string for execution remotely. + */ +PG_FUNCTION_INFO_V1(dblink_build_sql_delete); +Datum +dblink_build_sql_delete(PG_FUNCTION_ARGS) +{ + 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_update + * + * Used to generate an SQL update statement + * based on an existing tuple in a local relation. + * This is useful for selectively replicating data + * to another server via dblink. + * + * API: + * <relname> - name of local table of interest + * <pkattnums> - an int2vector of attnums which will be used + * to identify the local tuple of interest + * <pknumatts> - number of attnums in pkattnums + * <src_pkattvals_arry> - text array of key values which will be used + * to identify the local tuple of interest + * <tgt_pkattvals_arry> - text array of key values which will be used + * to build the string for execution remotely. These are substituted + * for their counterparts in src_pkattvals_arry + */ +PG_FUNCTION_INFO_V1(dblink_build_sql_update); +Datum +dblink_build_sql_update(PG_FUNCTION_ARGS) +{ + 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_current_query + * return the current query string + * to allow its use in (among other things) + * rewrite rules + */ +PG_FUNCTION_INFO_V1(dblink_current_query); +Datum +dblink_current_query(PG_FUNCTION_ARGS) +{ + /* This is now just an alias for the built-in function current_query() */ + PG_RETURN_DATUM(current_query(fcinfo)); +} + +/* + * Retrieve async notifications for a connection. + * + * Returns a setof record of notifications, or an empty set if none received. + * Can optionally take a named connection as parameter, but uses the unnamed + * connection per default. + * + */ +#define DBLINK_NOTIFY_COLS 3 + +PG_FUNCTION_INFO_V1(dblink_get_notify); +Datum +dblink_get_notify(PG_FUNCTION_ARGS) +{ + 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; +} + +/* + * Validate the options given to a dblink foreign server or user mapping. + * Raise an error if any option is invalid. + * + * We just check the names of options here, so semantic errors in options, + * such as invalid numeric format, will be detected at the attempt to connect. + */ +PG_FUNCTION_INFO_V1(dblink_fdw_validator); +Datum +dblink_fdw_validator(PG_FUNCTION_ARGS) +{ + 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(); +} + + +/************************************************************* + * internal functions + */ + + +/* + * get_pkey_attnames + * + * Get the primary key attnames for the given relation. + * Return NULL, and set indnkeyatts = 0, if no primary key exists. + */ +static char ** +get_pkey_attnames(Relation rel, int16 *indnkeyatts) +{ + 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; +} + +/* + * Deconstruct a text[] into C-strings (note any NULL elements will be + * returned as NULL pointers) + */ +static char ** +get_text_array_contents(ArrayType *array, int *numitems) +{ + 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; +} + +static char * +get_sql_insert(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals, char **tgt_pkattvals) +{ + 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; +} + +static char * +get_sql_delete(Relation rel, int *pkattnums, int pknumatts, char **tgt_pkattvals) +{ + 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; +} + +static char * +get_sql_update(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals, char **tgt_pkattvals) +{ + 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; +} + +/* + * Return a properly quoted identifier. + * Uses quote_ident in quote.c + */ +static char * +quote_ident_cstr(char *rawstr) +{ + 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; +} + +static int +get_attnum_pk_pos(int *pkattnums, int pknumatts, int key) +{ + 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; +} + +static HeapTuple +get_tuple_of_interest(Relation rel, int *pkattnums, int pknumatts, char **src_pkattvals) +{ + 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; +} + +/* + * Open the relation named by relname_text, acquire specified type of lock, + * verify we have specified permissions. + * Caller must close rel when done with it. + */ +static Relation +get_rel_from_relname(text *relname_text, LOCKMODE lockmode, AclMode aclmode) +{ + 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; +} + +/* + * generate_relation_name - copied from ruleutils.c + * Compute the name to display for a relation + * + * The result includes all necessary quoting and schema-prefixing. + */ +static char * +generate_relation_name(Relation rel) +{ + 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; +} + + +static remoteConn * +getConnectionByName(const char *name) +{ + 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; +} + +static HTAB * +createConnHash(void) +{ + HASHCTL ctl; + + ctl.keysize = NAMEDATALEN; + ctl.entrysize = sizeof(remoteConnHashEnt); + + return hash_create("Remote Con hash", NUMCONN, &ctl, + HASH_ELEM | HASH_STRINGS); +} + +static void +createNewConnection(const char *name, remoteConn *rconn) +{ + 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)); +} + +static void +deleteConnection(const char *name) +{ + 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"))); + +} + +static void +dblink_security_check(PGconn *conn, remoteConn *rconn) +{ + 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."))); + } + } +} + +/* + * For non-superusers, insist that the connstr specify a password. This + * prevents a password from being picked up from .pgpass, a service file, + * the environment, etc. We don't want the postgres user's passwords + * to be accessible to non-superusers. + */ +static void +dblink_connstr_check(const char *connstr) +{ + 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."))); + } +} + +/* + * Report an error received from the remote server + * + * res: the received error result (will be freed) + * fail: true for ERROR ereport, false for NOTICE + * fmt and following args: sprintf-style format and values for errcontext; + * the resulting string should be worded like "while <some action>" + */ +static void +dblink_res_error(PGconn *conn, const char *conname, PGresult *res, + bool fail, const char *fmt,...) +{ + 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 != NULL && message_primary[0] != '\0') ? + errmsg_internal("%s", message_primary) : + errmsg("could not obtain message string for remote error"), + message_detail ? errdetail_internal("%s", message_detail) : 0, + message_hint ? errhint("%s", message_hint) : 0, + message_context ? (errcontext("%s", message_context)) : 0, + conname ? + (errcontext("%s on dblink connection named \"%s\"", + dblink_context_msg, conname)) : + (errcontext("%s on unnamed dblink connection", + dblink_context_msg)))); +} + +/* + * Obtain connection string for a foreign server + */ +static char * +get_connect_string(const char *servername) +{ + 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; +} + +/* + * Escaping libpq connect parameter strings. + * + * Replaces "'" with "\'" and "\" with "\\". + */ +static char * +escape_param_str(const char *str) +{ + const char *cp; + StringInfoData buf; + + initStringInfo(&buf); + + for (cp = str; *cp; cp++) + { + if (*cp == '\\' || *cp == '\'') + appendStringInfoChar(&buf, '\\'); + appendStringInfoChar(&buf, *cp); + } + + return buf.data; +} + +/* + * Validate the PK-attnums argument for dblink_build_sql_insert() and related + * functions, and translate to the internal representation. + * + * The user supplies an int2vector of 1-based logical attnums, plus a count + * argument (the need for the separate count argument is historical, but we + * still check it). We check that each attnum corresponds to a valid, + * non-dropped attribute of the rel. We do *not* prevent attnums from being + * listed twice, though the actual use-case for such things is dubious. + * Note that before Postgres 9.0, the user's attnums were interpreted as + * physical not logical column numbers; this was changed for future-proofing. + * + * The internal representation is a palloc'd int array of 0-based physical + * attnums. + */ +static void +validate_pkattnums(Relation rel, + int2vector *pkattnums_arg, int32 pknumatts_arg, + int **pkattnums, int *pknumatts) +{ + 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))); + } +} + +/* + * Check if the specified connection option is valid. + * + * We basically allow whatever libpq thinks is an option, with these + * restrictions: + * debug options: disallowed + * "client_encoding": disallowed + * "user": valid only in USER MAPPING options + * secure options (eg password): valid only in USER MAPPING options + * others: valid only in FOREIGN SERVER options + * + * We disallow client_encoding because it would be overridden anyway via + * PQclientEncoding; allowing it to be specified would merely promote + * confusion. + */ +static bool +is_valid_dblink_option(const PQconninfoOption *options, const char *option, + Oid context) +{ + 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; +} + +/* + * Copy the remote session's values of GUCs that affect datatype I/O + * and apply them locally in a new GUC nesting level. Returns the new + * nestlevel (which is needed by restoreLocalGucs to undo the settings), + * or -1 if no new nestlevel was needed. + * + * We use the equivalent of a function SET option to allow the settings to + * persist only until the caller calls restoreLocalGucs. If an error is + * thrown in between, guc.c will take care of undoing the settings. + */ +static int +applyRemoteGucs(PGconn *conn) +{ + 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; +} + +/* + * Restore local GUCs after they have been overlaid with remote settings. + */ +static void +restoreLocalGucs(int nestlevel) +{ + /* Do nothing if no new nestlevel was created */ + if (nestlevel > 0) + AtEOXact_GUC(true, nestlevel); +} diff --git a/contrib/dblink/dblink.control b/contrib/dblink/dblink.control new file mode 100644 index 0000000..bdd17d2 --- /dev/null +++ b/contrib/dblink/dblink.control @@ -0,0 +1,5 @@ +# dblink extension +comment = 'connect to other PostgreSQL databases from within a database' +default_version = '1.2' +module_pathname = '$libdir/dblink' +relocatable = true diff --git a/contrib/dblink/expected/.gitignore b/contrib/dblink/expected/.gitignore new file mode 100644 index 0000000..d9c7942 --- /dev/null +++ b/contrib/dblink/expected/.gitignore @@ -0,0 +1 @@ +/paths.out diff --git a/contrib/dblink/expected/dblink.out b/contrib/dblink/expected/dblink.out new file mode 100644 index 0000000..6516d4f --- /dev/null +++ b/contrib/dblink/expected/dblink.out @@ -0,0 +1,1181 @@ +CREATE EXTENSION dblink; +-- want context for notices +\set SHOW_CONTEXT always +CREATE TABLE foo(f1 int, f2 text, f3 text[], primary key (f1,f2)); +INSERT INTO foo VALUES (0,'a','{"a0","b0","c0"}'); +INSERT INTO foo VALUES (1,'b','{"a1","b1","c1"}'); +INSERT INTO foo VALUES (2,'c','{"a2","b2","c2"}'); +INSERT INTO foo VALUES (3,'d','{"a3","b3","c3"}'); +INSERT INTO foo VALUES (4,'e','{"a4","b4","c4"}'); +INSERT INTO foo VALUES (5,'f','{"a5","b5","c5"}'); +INSERT INTO foo VALUES (6,'g','{"a6","b6","c6"}'); +INSERT INTO foo VALUES (7,'h','{"a7","b7","c7"}'); +INSERT INTO foo VALUES (8,'i','{"a8","b8","c8"}'); +INSERT INTO foo VALUES (9,'j','{"a9","b9","c9"}'); +-- misc utilities +-- list the primary key fields +SELECT * +FROM dblink_get_pkey('foo'); + position | colname +----------+--------- + 1 | f1 + 2 | f2 +(2 rows) + +-- build an insert statement based on a local tuple, +-- replacing the primary key values with new ones +SELECT dblink_build_sql_insert('foo','1 2',2,'{"0", "a"}','{"99", "xyz"}'); + dblink_build_sql_insert +----------------------------------------------------------- + INSERT INTO foo(f1,f2,f3) VALUES('99','xyz','{a0,b0,c0}') +(1 row) + +-- too many pk fields, should fail +SELECT dblink_build_sql_insert('foo','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}','{"99", "xyz", "{za0,zb0,zc0}"}'); +ERROR: invalid attribute number 4 +-- build an update statement based on a local tuple, +-- replacing the primary key values with new ones +SELECT dblink_build_sql_update('foo','1 2',2,'{"0", "a"}','{"99", "xyz"}'); + dblink_build_sql_update +---------------------------------------------------------------------------------------- + UPDATE foo SET f1 = '99', f2 = 'xyz', f3 = '{a0,b0,c0}' WHERE f1 = '99' AND f2 = 'xyz' +(1 row) + +-- too many pk fields, should fail +SELECT dblink_build_sql_update('foo','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}','{"99", "xyz", "{za0,zb0,zc0}"}'); +ERROR: invalid attribute number 4 +-- build a delete statement based on a local tuple, +SELECT dblink_build_sql_delete('foo','1 2',2,'{"0", "a"}'); + dblink_build_sql_delete +--------------------------------------------- + DELETE FROM foo WHERE f1 = '0' AND f2 = 'a' +(1 row) + +-- too many pk fields, should fail +SELECT dblink_build_sql_delete('foo','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}'); +ERROR: invalid attribute number 4 +-- repeat the test for table with primary key index with included columns +CREATE TABLE foo_1(f1 int, f2 text, f3 text[], primary key (f1,f2) include (f3)); +INSERT INTO foo_1 VALUES (0,'a','{"a0","b0","c0"}'); +INSERT INTO foo_1 VALUES (1,'b','{"a1","b1","c1"}'); +INSERT INTO foo_1 VALUES (2,'c','{"a2","b2","c2"}'); +INSERT INTO foo_1 VALUES (3,'d','{"a3","b3","c3"}'); +INSERT INTO foo_1 VALUES (4,'e','{"a4","b4","c4"}'); +INSERT INTO foo_1 VALUES (5,'f','{"a5","b5","c5"}'); +INSERT INTO foo_1 VALUES (6,'g','{"a6","b6","c6"}'); +INSERT INTO foo_1 VALUES (7,'h','{"a7","b7","c7"}'); +INSERT INTO foo_1 VALUES (8,'i','{"a8","b8","c8"}'); +INSERT INTO foo_1 VALUES (9,'j','{"a9","b9","c9"}'); +-- misc utilities +-- list the primary key fields +SELECT * +FROM dblink_get_pkey('foo_1'); + position | colname +----------+--------- + 1 | f1 + 2 | f2 +(2 rows) + +-- build an insert statement based on a local tuple, +-- replacing the primary key values with new ones +SELECT dblink_build_sql_insert('foo_1','1 2',2,'{"0", "a"}','{"99", "xyz"}'); + dblink_build_sql_insert +------------------------------------------------------------- + INSERT INTO foo_1(f1,f2,f3) VALUES('99','xyz','{a0,b0,c0}') +(1 row) + +-- too many pk fields, should fail +SELECT dblink_build_sql_insert('foo_1','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}','{"99", "xyz", "{za0,zb0,zc0}"}'); +ERROR: invalid attribute number 4 +-- build an update statement based on a local tuple, +-- replacing the primary key values with new ones +SELECT dblink_build_sql_update('foo_1','1 2',2,'{"0", "a"}','{"99", "xyz"}'); + dblink_build_sql_update +------------------------------------------------------------------------------------------ + UPDATE foo_1 SET f1 = '99', f2 = 'xyz', f3 = '{a0,b0,c0}' WHERE f1 = '99' AND f2 = 'xyz' +(1 row) + +-- too many pk fields, should fail +SELECT dblink_build_sql_update('foo_1','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}','{"99", "xyz", "{za0,zb0,zc0}"}'); +ERROR: invalid attribute number 4 +-- build a delete statement based on a local tuple, +SELECT dblink_build_sql_delete('foo_1','1 2',2,'{"0", "a"}'); + dblink_build_sql_delete +----------------------------------------------- + DELETE FROM foo_1 WHERE f1 = '0' AND f2 = 'a' +(1 row) + +-- too many pk fields, should fail +SELECT dblink_build_sql_delete('foo_1','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}'); +ERROR: invalid attribute number 4 +DROP TABLE foo_1; +-- retest using a quoted and schema qualified table +CREATE SCHEMA "MySchema"; +CREATE TABLE "MySchema"."Foo"(f1 int, f2 text, f3 text[], primary key (f1,f2)); +INSERT INTO "MySchema"."Foo" VALUES (0,'a','{"a0","b0","c0"}'); +-- list the primary key fields +SELECT * +FROM dblink_get_pkey('"MySchema"."Foo"'); + position | colname +----------+--------- + 1 | f1 + 2 | f2 +(2 rows) + +-- build an insert statement based on a local tuple, +-- replacing the primary key values with new ones +SELECT dblink_build_sql_insert('"MySchema"."Foo"','1 2',2,'{"0", "a"}','{"99", "xyz"}'); + dblink_build_sql_insert +------------------------------------------------------------------------ + INSERT INTO "MySchema"."Foo"(f1,f2,f3) VALUES('99','xyz','{a0,b0,c0}') +(1 row) + +-- build an update statement based on a local tuple, +-- replacing the primary key values with new ones +SELECT dblink_build_sql_update('"MySchema"."Foo"','1 2',2,'{"0", "a"}','{"99", "xyz"}'); + dblink_build_sql_update +----------------------------------------------------------------------------------------------------- + UPDATE "MySchema"."Foo" SET f1 = '99', f2 = 'xyz', f3 = '{a0,b0,c0}' WHERE f1 = '99' AND f2 = 'xyz' +(1 row) + +-- build a delete statement based on a local tuple, +SELECT dblink_build_sql_delete('"MySchema"."Foo"','1 2',2,'{"0", "a"}'); + dblink_build_sql_delete +---------------------------------------------------------- + DELETE FROM "MySchema"."Foo" WHERE f1 = '0' AND f2 = 'a' +(1 row) + +CREATE FUNCTION connection_parameters() RETURNS text LANGUAGE SQL AS $f$ + SELECT $$dbname='$$||current_database()||$$' port=$$||current_setting('port'); +$f$; +-- regular old dblink +SELECT * +FROM dblink(connection_parameters(),'SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; + a | b | c +---+---+------------ + 8 | i | {a8,b8,c8} + 9 | j | {a9,b9,c9} +(2 rows) + +-- should generate "connection not available" error +SELECT * +FROM dblink('SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; +ERROR: connection not available +-- The first-level connection's backend will crash on exit given OpenLDAP +-- [2.4.24, 2.4.31]. We won't see evidence of any crash until the victim +-- process terminates and the postmaster responds. If process termination +-- entails writing a core dump, that can take awhile. Wait for the process to +-- vanish. At that point, the postmaster has called waitpid() on the crashed +-- process, and it will accept no new connections until it has reinitialized +-- the cluster. (We can't exploit pg_stat_activity, because the crash happens +-- after the backend updates shared memory to reflect its impending exit.) +DO $pl$ +DECLARE + detail text; +BEGIN + PERFORM wait_pid(crash_pid) + FROM dblink(connection_parameters(), $$ + SELECT pg_backend_pid() FROM dblink( + 'service=test_ldap '||connection_parameters(), + -- This string concatenation is a hack to shoehorn a + -- set_pgservicefile call into the SQL statement. + 'SELECT 1' || set_pgservicefile('pg_service.conf') + ) t(c int) + $$) AS t(crash_pid int); +EXCEPTION WHEN OTHERS THEN + GET STACKED DIAGNOSTICS detail = PG_EXCEPTION_DETAIL; + -- Expected error in a non-LDAP build. + IF NOT detail LIKE 'syntax error in service file%' THEN RAISE; END IF; +END +$pl$; +-- create a persistent connection +SELECT dblink_connect(connection_parameters()); + dblink_connect +---------------- + OK +(1 row) + +-- use the persistent connection +SELECT * +FROM dblink('SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; + a | b | c +---+---+------------ + 8 | i | {a8,b8,c8} + 9 | j | {a9,b9,c9} +(2 rows) + +-- open a cursor with bad SQL and fail_on_error set to false +SELECT dblink_open('rmt_foo_cursor','SELECT * FROM foobar',false); +NOTICE: relation "foobar" does not exist +CONTEXT: while opening cursor "rmt_foo_cursor" on unnamed dblink connection + dblink_open +------------- + ERROR +(1 row) + +-- reset remote transaction state +SELECT dblink_exec('ABORT'); + dblink_exec +------------- + ROLLBACK +(1 row) + +-- open a cursor +SELECT dblink_open('rmt_foo_cursor','SELECT * FROM foo'); + dblink_open +------------- + OK +(1 row) + +-- close the cursor +SELECT dblink_close('rmt_foo_cursor',false); + dblink_close +-------------- + OK +(1 row) + +-- open the cursor again +SELECT dblink_open('rmt_foo_cursor','SELECT * FROM foo'); + dblink_open +------------- + OK +(1 row) + +-- fetch some data +SELECT * +FROM dblink_fetch('rmt_foo_cursor',4) AS t(a int, b text, c text[]); + a | b | c +---+---+------------ + 0 | a | {a0,b0,c0} + 1 | b | {a1,b1,c1} + 2 | c | {a2,b2,c2} + 3 | d | {a3,b3,c3} +(4 rows) + +SELECT * +FROM dblink_fetch('rmt_foo_cursor',4) AS t(a int, b text, c text[]); + a | b | c +---+---+------------ + 4 | e | {a4,b4,c4} + 5 | f | {a5,b5,c5} + 6 | g | {a6,b6,c6} + 7 | h | {a7,b7,c7} +(4 rows) + +-- this one only finds two rows left +SELECT * +FROM dblink_fetch('rmt_foo_cursor',4) AS t(a int, b text, c text[]); + a | b | c +---+---+------------ + 8 | i | {a8,b8,c8} + 9 | j | {a9,b9,c9} +(2 rows) + +-- intentionally botch a fetch +SELECT * +FROM dblink_fetch('rmt_foobar_cursor',4,false) AS t(a int, b text, c text[]); +NOTICE: cursor "rmt_foobar_cursor" does not exist +CONTEXT: while fetching from cursor "rmt_foobar_cursor" on unnamed dblink connection + a | b | c +---+---+--- +(0 rows) + +-- reset remote transaction state +SELECT dblink_exec('ABORT'); + dblink_exec +------------- + ROLLBACK +(1 row) + +-- close the wrong cursor +SELECT dblink_close('rmt_foobar_cursor',false); +NOTICE: cursor "rmt_foobar_cursor" does not exist +CONTEXT: while closing cursor "rmt_foobar_cursor" on unnamed dblink connection + dblink_close +-------------- + ERROR +(1 row) + +-- should generate 'cursor "rmt_foo_cursor" not found' error +SELECT * +FROM dblink_fetch('rmt_foo_cursor',4) AS t(a int, b text, c text[]); +ERROR: cursor "rmt_foo_cursor" does not exist +CONTEXT: while fetching from cursor "rmt_foo_cursor" on unnamed dblink connection +-- this time, 'cursor "rmt_foo_cursor" not found' as a notice +SELECT * +FROM dblink_fetch('rmt_foo_cursor',4,false) AS t(a int, b text, c text[]); +NOTICE: cursor "rmt_foo_cursor" does not exist +CONTEXT: while fetching from cursor "rmt_foo_cursor" on unnamed dblink connection + a | b | c +---+---+--- +(0 rows) + +-- close the persistent connection +SELECT dblink_disconnect(); + dblink_disconnect +------------------- + OK +(1 row) + +-- should generate "connection not available" error +SELECT * +FROM dblink('SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; +ERROR: connection not available +-- put more data into our table, first using arbitrary connection syntax +-- but truncate the actual return value so we can use diff to check for success +SELECT substr(dblink_exec(connection_parameters(),'INSERT INTO foo VALUES(10,''k'',''{"a10","b10","c10"}'')'),1,6); + substr +-------- + INSERT +(1 row) + +-- create a persistent connection +SELECT dblink_connect(connection_parameters()); + dblink_connect +---------------- + OK +(1 row) + +-- put more data into our table, using persistent connection syntax +-- but truncate the actual return value so we can use diff to check for success +SELECT substr(dblink_exec('INSERT INTO foo VALUES(11,''l'',''{"a11","b11","c11"}'')'),1,6); + substr +-------- + INSERT +(1 row) + +-- let's see it +SELECT * +FROM dblink('SELECT * FROM foo') AS t(a int, b text, c text[]); + a | b | c +----+---+--------------- + 0 | a | {a0,b0,c0} + 1 | b | {a1,b1,c1} + 2 | c | {a2,b2,c2} + 3 | d | {a3,b3,c3} + 4 | e | {a4,b4,c4} + 5 | f | {a5,b5,c5} + 6 | g | {a6,b6,c6} + 7 | h | {a7,b7,c7} + 8 | i | {a8,b8,c8} + 9 | j | {a9,b9,c9} + 10 | k | {a10,b10,c10} + 11 | l | {a11,b11,c11} +(12 rows) + +-- bad remote select +SELECT * +FROM dblink('SELECT * FROM foobar',false) AS t(a int, b text, c text[]); +NOTICE: relation "foobar" does not exist +CONTEXT: while executing query on unnamed dblink connection + a | b | c +---+---+--- +(0 rows) + +-- change some data +SELECT dblink_exec('UPDATE foo SET f3[2] = ''b99'' WHERE f1 = 11'); + dblink_exec +------------- + UPDATE 1 +(1 row) + +-- let's see it +SELECT * +FROM dblink('SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE a = 11; + a | b | c +----+---+--------------- + 11 | l | {a11,b99,c11} +(1 row) + +-- botch a change to some other data +SELECT dblink_exec('UPDATE foobar SET f3[2] = ''b99'' WHERE f1 = 11',false); +NOTICE: relation "foobar" does not exist +CONTEXT: while executing command on unnamed dblink connection + dblink_exec +------------- + ERROR +(1 row) + +-- delete some data +SELECT dblink_exec('DELETE FROM foo WHERE f1 = 11'); + dblink_exec +------------- + DELETE 1 +(1 row) + +-- let's see it +SELECT * +FROM dblink('SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE a = 11; + a | b | c +---+---+--- +(0 rows) + +-- close the persistent connection +SELECT dblink_disconnect(); + dblink_disconnect +------------------- + OK +(1 row) + +-- +-- tests for the new named persistent connection syntax +-- +-- should generate "missing "=" after "myconn" in connection info string" error +SELECT * +FROM dblink('myconn','SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; +ERROR: could not establish connection +DETAIL: missing "=" after "myconn" in connection info string +-- create a named persistent connection +SELECT dblink_connect('myconn',connection_parameters()); + dblink_connect +---------------- + OK +(1 row) + +-- use the named persistent connection +SELECT * +FROM dblink('myconn','SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; + a | b | c +----+---+--------------- + 8 | i | {a8,b8,c8} + 9 | j | {a9,b9,c9} + 10 | k | {a10,b10,c10} +(3 rows) + +-- use the named persistent connection, but get it wrong +SELECT * +FROM dblink('myconn','SELECT * FROM foobar',false) AS t(a int, b text, c text[]) +WHERE t.a > 7; +NOTICE: relation "foobar" does not exist +CONTEXT: while executing query on dblink connection named "myconn" + a | b | c +---+---+--- +(0 rows) + +-- create a second named persistent connection +-- should error with "duplicate connection name" +SELECT dblink_connect('myconn',connection_parameters()); +ERROR: duplicate connection name +-- create a second named persistent connection with a new name +SELECT dblink_connect('myconn2',connection_parameters()); + dblink_connect +---------------- + OK +(1 row) + +-- use the second named persistent connection +SELECT * +FROM dblink('myconn2','SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; + a | b | c +----+---+--------------- + 8 | i | {a8,b8,c8} + 9 | j | {a9,b9,c9} + 10 | k | {a10,b10,c10} +(3 rows) + +-- close the second named persistent connection +SELECT dblink_disconnect('myconn2'); + dblink_disconnect +------------------- + OK +(1 row) + +-- open a cursor incorrectly +SELECT dblink_open('myconn','rmt_foo_cursor','SELECT * FROM foobar',false); +NOTICE: relation "foobar" does not exist +CONTEXT: while opening cursor "rmt_foo_cursor" on dblink connection named "myconn" + dblink_open +------------- + ERROR +(1 row) + +-- reset remote transaction state +SELECT dblink_exec('myconn','ABORT'); + dblink_exec +------------- + ROLLBACK +(1 row) + +-- test opening cursor in a transaction +SELECT dblink_exec('myconn','BEGIN'); + dblink_exec +------------- + BEGIN +(1 row) + +-- an open transaction will prevent dblink_open() from opening its own +SELECT dblink_open('myconn','rmt_foo_cursor','SELECT * FROM foo'); + dblink_open +------------- + OK +(1 row) + +-- this should not commit the transaction because the client opened it +SELECT dblink_close('myconn','rmt_foo_cursor'); + dblink_close +-------------- + OK +(1 row) + +-- this should succeed because we have an open transaction +SELECT dblink_exec('myconn','DECLARE xact_test CURSOR FOR SELECT * FROM foo'); + dblink_exec +---------------- + DECLARE CURSOR +(1 row) + +-- commit remote transaction +SELECT dblink_exec('myconn','COMMIT'); + dblink_exec +------------- + COMMIT +(1 row) + +-- test automatic transactions for multiple cursor opens +SELECT dblink_open('myconn','rmt_foo_cursor','SELECT * FROM foo'); + dblink_open +------------- + OK +(1 row) + +-- the second cursor +SELECT dblink_open('myconn','rmt_foo_cursor2','SELECT * FROM foo'); + dblink_open +------------- + OK +(1 row) + +-- this should not commit the transaction +SELECT dblink_close('myconn','rmt_foo_cursor2'); + dblink_close +-------------- + OK +(1 row) + +-- this should succeed because we have an open transaction +SELECT dblink_exec('myconn','DECLARE xact_test CURSOR FOR SELECT * FROM foo'); + dblink_exec +---------------- + DECLARE CURSOR +(1 row) + +-- this should commit the transaction +SELECT dblink_close('myconn','rmt_foo_cursor'); + dblink_close +-------------- + OK +(1 row) + +-- this should fail because there is no open transaction +SELECT dblink_exec('myconn','DECLARE xact_test CURSOR FOR SELECT * FROM foo'); +ERROR: DECLARE CURSOR can only be used in transaction blocks +CONTEXT: while executing command on dblink connection named "myconn" +-- reset remote transaction state +SELECT dblink_exec('myconn','ABORT'); + dblink_exec +------------- + ROLLBACK +(1 row) + +-- open a cursor +SELECT dblink_open('myconn','rmt_foo_cursor','SELECT * FROM foo'); + dblink_open +------------- + OK +(1 row) + +-- fetch some data +SELECT * +FROM dblink_fetch('myconn','rmt_foo_cursor',4) AS t(a int, b text, c text[]); + a | b | c +---+---+------------ + 0 | a | {a0,b0,c0} + 1 | b | {a1,b1,c1} + 2 | c | {a2,b2,c2} + 3 | d | {a3,b3,c3} +(4 rows) + +SELECT * +FROM dblink_fetch('myconn','rmt_foo_cursor',4) AS t(a int, b text, c text[]); + a | b | c +---+---+------------ + 4 | e | {a4,b4,c4} + 5 | f | {a5,b5,c5} + 6 | g | {a6,b6,c6} + 7 | h | {a7,b7,c7} +(4 rows) + +-- this one only finds three rows left +SELECT * +FROM dblink_fetch('myconn','rmt_foo_cursor',4) AS t(a int, b text, c text[]); + a | b | c +----+---+--------------- + 8 | i | {a8,b8,c8} + 9 | j | {a9,b9,c9} + 10 | k | {a10,b10,c10} +(3 rows) + +-- fetch some data incorrectly +SELECT * +FROM dblink_fetch('myconn','rmt_foobar_cursor',4,false) AS t(a int, b text, c text[]); +NOTICE: cursor "rmt_foobar_cursor" does not exist +CONTEXT: while fetching from cursor "rmt_foobar_cursor" on dblink connection named "myconn" + a | b | c +---+---+--- +(0 rows) + +-- reset remote transaction state +SELECT dblink_exec('myconn','ABORT'); + dblink_exec +------------- + ROLLBACK +(1 row) + +-- should generate 'cursor "rmt_foo_cursor" not found' error +SELECT * +FROM dblink_fetch('myconn','rmt_foo_cursor',4) AS t(a int, b text, c text[]); +ERROR: cursor "rmt_foo_cursor" does not exist +CONTEXT: while fetching from cursor "rmt_foo_cursor" on dblink connection named "myconn" +-- close the named persistent connection +SELECT dblink_disconnect('myconn'); + dblink_disconnect +------------------- + OK +(1 row) + +-- should generate "missing "=" after "myconn" in connection info string" error +SELECT * +FROM dblink('myconn','SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; +ERROR: could not establish connection +DETAIL: missing "=" after "myconn" in connection info string +-- create a named persistent connection +SELECT dblink_connect('myconn',connection_parameters()); + dblink_connect +---------------- + OK +(1 row) + +-- put more data into our table, using named persistent connection syntax +-- but truncate the actual return value so we can use diff to check for success +SELECT substr(dblink_exec('myconn','INSERT INTO foo VALUES(11,''l'',''{"a11","b11","c11"}'')'),1,6); + substr +-------- + INSERT +(1 row) + +-- let's see it +SELECT * +FROM dblink('myconn','SELECT * FROM foo') AS t(a int, b text, c text[]); + a | b | c +----+---+--------------- + 0 | a | {a0,b0,c0} + 1 | b | {a1,b1,c1} + 2 | c | {a2,b2,c2} + 3 | d | {a3,b3,c3} + 4 | e | {a4,b4,c4} + 5 | f | {a5,b5,c5} + 6 | g | {a6,b6,c6} + 7 | h | {a7,b7,c7} + 8 | i | {a8,b8,c8} + 9 | j | {a9,b9,c9} + 10 | k | {a10,b10,c10} + 11 | l | {a11,b11,c11} +(12 rows) + +-- change some data +SELECT dblink_exec('myconn','UPDATE foo SET f3[2] = ''b99'' WHERE f1 = 11'); + dblink_exec +------------- + UPDATE 1 +(1 row) + +-- let's see it +SELECT * +FROM dblink('myconn','SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE a = 11; + a | b | c +----+---+--------------- + 11 | l | {a11,b99,c11} +(1 row) + +-- delete some data +SELECT dblink_exec('myconn','DELETE FROM foo WHERE f1 = 11'); + dblink_exec +------------- + DELETE 1 +(1 row) + +-- let's see it +SELECT * +FROM dblink('myconn','SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE a = 11; + a | b | c +---+---+--- +(0 rows) + +-- close the named persistent connection +SELECT dblink_disconnect('myconn'); + dblink_disconnect +------------------- + OK +(1 row) + +-- close the named persistent connection again +-- should get 'connection "myconn" not available' error +SELECT dblink_disconnect('myconn'); +ERROR: connection "myconn" not available +-- test asynchronous queries +SELECT dblink_connect('dtest1', connection_parameters()); + dblink_connect +---------------- + OK +(1 row) + +SELECT * from + dblink_send_query('dtest1', 'select * from foo where f1 < 3') as t1; + t1 +---- + 1 +(1 row) + +SELECT dblink_connect('dtest2', connection_parameters()); + dblink_connect +---------------- + OK +(1 row) + +SELECT * from + dblink_send_query('dtest2', 'select * from foo where f1 > 2 and f1 < 7') as t1; + t1 +---- + 1 +(1 row) + +SELECT dblink_connect('dtest3', connection_parameters()); + dblink_connect +---------------- + OK +(1 row) + +SELECT * from + dblink_send_query('dtest3', 'select * from foo where f1 > 6') as t1; + t1 +---- + 1 +(1 row) + +CREATE TEMPORARY TABLE result AS +(SELECT * from dblink_get_result('dtest1') as t1(f1 int, f2 text, f3 text[])) +UNION +(SELECT * from dblink_get_result('dtest2') as t2(f1 int, f2 text, f3 text[])) +UNION +(SELECT * from dblink_get_result('dtest3') as t3(f1 int, f2 text, f3 text[])) +ORDER by f1; +-- dblink_get_connections returns an array with elements in a machine-dependent +-- ordering, so we must resort to unnesting and sorting for a stable result +create function unnest(anyarray) returns setof anyelement +language sql strict immutable as $$ +select $1[i] from generate_series(array_lower($1,1), array_upper($1,1)) as i +$$; +SELECT * FROM unnest(dblink_get_connections()) ORDER BY 1; + unnest +-------- + dtest1 + dtest2 + dtest3 +(3 rows) + +SELECT dblink_is_busy('dtest1'); + dblink_is_busy +---------------- + 0 +(1 row) + +SELECT dblink_disconnect('dtest1'); + dblink_disconnect +------------------- + OK +(1 row) + +SELECT dblink_disconnect('dtest2'); + dblink_disconnect +------------------- + OK +(1 row) + +SELECT dblink_disconnect('dtest3'); + dblink_disconnect +------------------- + OK +(1 row) + +SELECT * from result; + f1 | f2 | f3 +----+----+--------------- + 0 | a | {a0,b0,c0} + 1 | b | {a1,b1,c1} + 2 | c | {a2,b2,c2} + 3 | d | {a3,b3,c3} + 4 | e | {a4,b4,c4} + 5 | f | {a5,b5,c5} + 6 | g | {a6,b6,c6} + 7 | h | {a7,b7,c7} + 8 | i | {a8,b8,c8} + 9 | j | {a9,b9,c9} + 10 | k | {a10,b10,c10} +(11 rows) + +SELECT dblink_connect('dtest1', connection_parameters()); + dblink_connect +---------------- + OK +(1 row) + +SELECT * from + dblink_send_query('dtest1', 'select * from foo where f1 < 3') as t1; + t1 +---- + 1 +(1 row) + +SELECT dblink_cancel_query('dtest1'); + dblink_cancel_query +--------------------- + OK +(1 row) + +SELECT dblink_error_message('dtest1'); + dblink_error_message +---------------------- + OK +(1 row) + +SELECT dblink_disconnect('dtest1'); + dblink_disconnect +------------------- + OK +(1 row) + +-- test foreign data wrapper functionality +CREATE ROLE regress_dblink_user; +DO $d$ + BEGIN + EXECUTE $$CREATE SERVER fdtest FOREIGN DATA WRAPPER dblink_fdw + OPTIONS (dbname '$$||current_database()||$$', + port '$$||current_setting('port')||$$' + )$$; + END; +$d$; +CREATE USER MAPPING FOR public SERVER fdtest + OPTIONS (server 'localhost'); -- fail, can't specify server here +ERROR: invalid option "server" +HINT: Valid options in this context are: user, password, sslpassword +CREATE USER MAPPING FOR public SERVER fdtest OPTIONS (user :'USER'); +GRANT USAGE ON FOREIGN SERVER fdtest TO regress_dblink_user; +GRANT EXECUTE ON FUNCTION dblink_connect_u(text, text) TO regress_dblink_user; +SET SESSION AUTHORIZATION regress_dblink_user; +-- should fail +SELECT dblink_connect('myconn', 'fdtest'); +ERROR: password is required +DETAIL: Non-superusers must provide a password in the connection string. +-- should succeed +SELECT dblink_connect_u('myconn', 'fdtest'); + dblink_connect_u +------------------ + OK +(1 row) + +SELECT * FROM dblink('myconn','SELECT * FROM foo') AS t(a int, b text, c text[]); + a | b | c +----+---+--------------- + 0 | a | {a0,b0,c0} + 1 | b | {a1,b1,c1} + 2 | c | {a2,b2,c2} + 3 | d | {a3,b3,c3} + 4 | e | {a4,b4,c4} + 5 | f | {a5,b5,c5} + 6 | g | {a6,b6,c6} + 7 | h | {a7,b7,c7} + 8 | i | {a8,b8,c8} + 9 | j | {a9,b9,c9} + 10 | k | {a10,b10,c10} +(11 rows) + +\c - - +REVOKE USAGE ON FOREIGN SERVER fdtest FROM regress_dblink_user; +REVOKE EXECUTE ON FUNCTION dblink_connect_u(text, text) FROM regress_dblink_user; +DROP USER regress_dblink_user; +DROP USER MAPPING FOR public SERVER fdtest; +DROP SERVER fdtest; +-- test asynchronous notifications +SELECT dblink_connect(connection_parameters()); + dblink_connect +---------------- + OK +(1 row) + +--should return listen +SELECT dblink_exec('LISTEN regression'); + dblink_exec +------------- + LISTEN +(1 row) + +--should return listen +SELECT dblink_exec('LISTEN foobar'); + dblink_exec +------------- + LISTEN +(1 row) + +SELECT dblink_exec('NOTIFY regression'); + dblink_exec +------------- + NOTIFY +(1 row) + +SELECT dblink_exec('NOTIFY foobar'); + dblink_exec +------------- + NOTIFY +(1 row) + +SELECT notify_name, be_pid = (select t.be_pid from dblink('select pg_backend_pid()') as t(be_pid int)) AS is_self_notify, extra from dblink_get_notify(); + notify_name | is_self_notify | extra +-------------+----------------+------- + regression | t | + foobar | t | +(2 rows) + +SELECT * from dblink_get_notify(); + notify_name | be_pid | extra +-------------+--------+------- +(0 rows) + +SELECT dblink_disconnect(); + dblink_disconnect +------------------- + OK +(1 row) + +-- test dropped columns in dblink_build_sql_insert, dblink_build_sql_update +CREATE TEMP TABLE test_dropped +( + col1 INT NOT NULL DEFAULT 111, + id SERIAL PRIMARY KEY, + col2 INT NOT NULL DEFAULT 112, + col2b INT NOT NULL DEFAULT 113 +); +INSERT INTO test_dropped VALUES(default); +ALTER TABLE test_dropped + DROP COLUMN col1, + DROP COLUMN col2, + ADD COLUMN col3 VARCHAR(10) NOT NULL DEFAULT 'foo', + ADD COLUMN col4 INT NOT NULL DEFAULT 42; +SELECT dblink_build_sql_insert('test_dropped', '1', 1, + ARRAY['1'::TEXT], ARRAY['2'::TEXT]); + dblink_build_sql_insert +--------------------------------------------------------------------------- + INSERT INTO test_dropped(id,col2b,col3,col4) VALUES('2','113','foo','42') +(1 row) + +SELECT dblink_build_sql_update('test_dropped', '1', 1, + ARRAY['1'::TEXT], ARRAY['2'::TEXT]); + dblink_build_sql_update +------------------------------------------------------------------------------------------- + UPDATE test_dropped SET id = '2', col2b = '113', col3 = 'foo', col4 = '42' WHERE id = '2' +(1 row) + +SELECT dblink_build_sql_delete('test_dropped', '1', 1, + ARRAY['2'::TEXT]); + dblink_build_sql_delete +----------------------------------------- + DELETE FROM test_dropped WHERE id = '2' +(1 row) + +-- test local mimicry of remote GUC values that affect datatype I/O +SET datestyle = ISO, MDY; +SET intervalstyle = postgres; +SET timezone = UTC; +SELECT dblink_connect('myconn',connection_parameters()); + dblink_connect +---------------- + OK +(1 row) + +SELECT dblink_exec('myconn', 'SET datestyle = GERMAN, DMY;'); + dblink_exec +------------- + SET +(1 row) + +-- single row synchronous case +SELECT * +FROM dblink('myconn', + 'SELECT * FROM (VALUES (''12.03.2013 00:00:00+00'')) t') + AS t(a timestamptz); + a +------------------------ + 2013-03-12 00:00:00+00 +(1 row) + +-- multi-row synchronous case +SELECT * +FROM dblink('myconn', + 'SELECT * FROM + (VALUES (''12.03.2013 00:00:00+00''), + (''12.03.2013 00:00:00+00'')) t') + AS t(a timestamptz); + a +------------------------ + 2013-03-12 00:00:00+00 + 2013-03-12 00:00:00+00 +(2 rows) + +-- single-row asynchronous case +SELECT * +FROM dblink_send_query('myconn', + 'SELECT * FROM + (VALUES (''12.03.2013 00:00:00+00'')) t'); + dblink_send_query +------------------- + 1 +(1 row) + +CREATE TEMPORARY TABLE result AS +(SELECT * from dblink_get_result('myconn') as t(t timestamptz)) +UNION ALL +(SELECT * from dblink_get_result('myconn') as t(t timestamptz)); +SELECT * FROM result; + t +------------------------ + 2013-03-12 00:00:00+00 +(1 row) + +DROP TABLE result; +-- multi-row asynchronous case +SELECT * +FROM dblink_send_query('myconn', + 'SELECT * FROM + (VALUES (''12.03.2013 00:00:00+00''), + (''12.03.2013 00:00:00+00'')) t'); + dblink_send_query +------------------- + 1 +(1 row) + +CREATE TEMPORARY TABLE result AS +(SELECT * from dblink_get_result('myconn') as t(t timestamptz)) +UNION ALL +(SELECT * from dblink_get_result('myconn') as t(t timestamptz)) +UNION ALL +(SELECT * from dblink_get_result('myconn') as t(t timestamptz)); +SELECT * FROM result; + t +------------------------ + 2013-03-12 00:00:00+00 + 2013-03-12 00:00:00+00 +(2 rows) + +DROP TABLE result; +-- Try an ambiguous interval +SELECT dblink_exec('myconn', 'SET intervalstyle = sql_standard;'); + dblink_exec +------------- + SET +(1 row) + +SELECT * +FROM dblink('myconn', + 'SELECT * FROM (VALUES (''-1 2:03:04'')) i') + AS i(i interval); + i +------------------- + -1 days -02:03:04 +(1 row) + +-- Try swapping to another format to ensure the GUCs are tracked +-- properly through a change. +CREATE TEMPORARY TABLE result (t timestamptz); +SELECT dblink_exec('myconn', 'SET datestyle = ISO, MDY;'); + dblink_exec +------------- + SET +(1 row) + +INSERT INTO result + SELECT * + FROM dblink('myconn', + 'SELECT * FROM (VALUES (''03.12.2013 00:00:00+00'')) t') + AS t(a timestamptz); +SELECT dblink_exec('myconn', 'SET datestyle = GERMAN, DMY;'); + dblink_exec +------------- + SET +(1 row) + +INSERT INTO result + SELECT * + FROM dblink('myconn', + 'SELECT * FROM (VALUES (''12.03.2013 00:00:00+00'')) t') + AS t(a timestamptz); +SELECT * FROM result; + t +------------------------ + 2013-03-12 00:00:00+00 + 2013-03-12 00:00:00+00 +(2 rows) + +DROP TABLE result; +-- Check error throwing in dblink_fetch +SELECT dblink_open('myconn','error_cursor', + 'SELECT * FROM (VALUES (''1''), (''not an int'')) AS t(text);'); + dblink_open +------------- + OK +(1 row) + +SELECT * +FROM dblink_fetch('myconn','error_cursor', 1) AS t(i int); + i +--- + 1 +(1 row) + +SELECT * +FROM dblink_fetch('myconn','error_cursor', 1) AS t(i int); +ERROR: invalid input syntax for type integer: "not an int" +-- Make sure that the local settings have retained their values in spite +-- of shenanigans on the connection. +SHOW datestyle; + DateStyle +----------- + ISO, MDY +(1 row) + +SHOW intervalstyle; + IntervalStyle +--------------- + postgres +(1 row) + +-- Clean up GUC-setting tests +SELECT dblink_disconnect('myconn'); + dblink_disconnect +------------------- + OK +(1 row) + +RESET datestyle; +RESET intervalstyle; +RESET timezone; diff --git a/contrib/dblink/input/paths.source b/contrib/dblink/input/paths.source new file mode 100644 index 0000000..881a653 --- /dev/null +++ b/contrib/dblink/input/paths.source @@ -0,0 +1,14 @@ +-- Initialization that requires path substitution. + +CREATE FUNCTION setenv(text, text) + RETURNS void + AS '@libdir@/regress@DLSUFFIX@', 'regress_setenv' + LANGUAGE C STRICT; + +CREATE FUNCTION wait_pid(int) + RETURNS void + AS '@libdir@/regress@DLSUFFIX@' + LANGUAGE C STRICT; + +CREATE FUNCTION set_pgservicefile(text) RETURNS void LANGUAGE SQL + AS $$SELECT setenv('PGSERVICEFILE', '@abs_srcdir@/' || $1)$$; diff --git a/contrib/dblink/output/paths.source b/contrib/dblink/output/paths.source new file mode 100644 index 0000000..8ed95e1 --- /dev/null +++ b/contrib/dblink/output/paths.source @@ -0,0 +1,11 @@ +-- Initialization that requires path substitution. +CREATE FUNCTION setenv(text, text) + RETURNS void + AS '@libdir@/regress@DLSUFFIX@', 'regress_setenv' + LANGUAGE C STRICT; +CREATE FUNCTION wait_pid(int) + RETURNS void + AS '@libdir@/regress@DLSUFFIX@' + LANGUAGE C STRICT; +CREATE FUNCTION set_pgservicefile(text) RETURNS void LANGUAGE SQL + AS $$SELECT setenv('PGSERVICEFILE', '@abs_srcdir@/' || $1)$$; diff --git a/contrib/dblink/pg_service.conf b/contrib/dblink/pg_service.conf new file mode 100644 index 0000000..92201f0 --- /dev/null +++ b/contrib/dblink/pg_service.conf @@ -0,0 +1,7 @@ +# pg_service.conf for minimally exercising libpq use of LDAP. + +# Having failed to reach an LDAP server, libpq essentially ignores the +# "service=test_ldap" in its connection string. Contact the "discard" +# service; the test works whether or not it answers. +[test_ldap] +ldap://127.0.0.1:9/base?attribute?one?filter diff --git a/contrib/dblink/sql/.gitignore b/contrib/dblink/sql/.gitignore new file mode 100644 index 0000000..d175078 --- /dev/null +++ b/contrib/dblink/sql/.gitignore @@ -0,0 +1 @@ +/paths.sql diff --git a/contrib/dblink/sql/dblink.sql b/contrib/dblink/sql/dblink.sql new file mode 100644 index 0000000..3e96b98 --- /dev/null +++ b/contrib/dblink/sql/dblink.sql @@ -0,0 +1,603 @@ +CREATE EXTENSION dblink; + +-- want context for notices +\set SHOW_CONTEXT always + +CREATE TABLE foo(f1 int, f2 text, f3 text[], primary key (f1,f2)); +INSERT INTO foo VALUES (0,'a','{"a0","b0","c0"}'); +INSERT INTO foo VALUES (1,'b','{"a1","b1","c1"}'); +INSERT INTO foo VALUES (2,'c','{"a2","b2","c2"}'); +INSERT INTO foo VALUES (3,'d','{"a3","b3","c3"}'); +INSERT INTO foo VALUES (4,'e','{"a4","b4","c4"}'); +INSERT INTO foo VALUES (5,'f','{"a5","b5","c5"}'); +INSERT INTO foo VALUES (6,'g','{"a6","b6","c6"}'); +INSERT INTO foo VALUES (7,'h','{"a7","b7","c7"}'); +INSERT INTO foo VALUES (8,'i','{"a8","b8","c8"}'); +INSERT INTO foo VALUES (9,'j','{"a9","b9","c9"}'); + +-- misc utilities + +-- list the primary key fields +SELECT * +FROM dblink_get_pkey('foo'); + +-- build an insert statement based on a local tuple, +-- replacing the primary key values with new ones +SELECT dblink_build_sql_insert('foo','1 2',2,'{"0", "a"}','{"99", "xyz"}'); +-- too many pk fields, should fail +SELECT dblink_build_sql_insert('foo','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}','{"99", "xyz", "{za0,zb0,zc0}"}'); + +-- build an update statement based on a local tuple, +-- replacing the primary key values with new ones +SELECT dblink_build_sql_update('foo','1 2',2,'{"0", "a"}','{"99", "xyz"}'); +-- too many pk fields, should fail +SELECT dblink_build_sql_update('foo','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}','{"99", "xyz", "{za0,zb0,zc0}"}'); + +-- build a delete statement based on a local tuple, +SELECT dblink_build_sql_delete('foo','1 2',2,'{"0", "a"}'); +-- too many pk fields, should fail +SELECT dblink_build_sql_delete('foo','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}'); + +-- repeat the test for table with primary key index with included columns +CREATE TABLE foo_1(f1 int, f2 text, f3 text[], primary key (f1,f2) include (f3)); +INSERT INTO foo_1 VALUES (0,'a','{"a0","b0","c0"}'); +INSERT INTO foo_1 VALUES (1,'b','{"a1","b1","c1"}'); +INSERT INTO foo_1 VALUES (2,'c','{"a2","b2","c2"}'); +INSERT INTO foo_1 VALUES (3,'d','{"a3","b3","c3"}'); +INSERT INTO foo_1 VALUES (4,'e','{"a4","b4","c4"}'); +INSERT INTO foo_1 VALUES (5,'f','{"a5","b5","c5"}'); +INSERT INTO foo_1 VALUES (6,'g','{"a6","b6","c6"}'); +INSERT INTO foo_1 VALUES (7,'h','{"a7","b7","c7"}'); +INSERT INTO foo_1 VALUES (8,'i','{"a8","b8","c8"}'); +INSERT INTO foo_1 VALUES (9,'j','{"a9","b9","c9"}'); + +-- misc utilities + +-- list the primary key fields +SELECT * +FROM dblink_get_pkey('foo_1'); + +-- build an insert statement based on a local tuple, +-- replacing the primary key values with new ones +SELECT dblink_build_sql_insert('foo_1','1 2',2,'{"0", "a"}','{"99", "xyz"}'); +-- too many pk fields, should fail +SELECT dblink_build_sql_insert('foo_1','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}','{"99", "xyz", "{za0,zb0,zc0}"}'); + +-- build an update statement based on a local tuple, +-- replacing the primary key values with new ones +SELECT dblink_build_sql_update('foo_1','1 2',2,'{"0", "a"}','{"99", "xyz"}'); +-- too many pk fields, should fail +SELECT dblink_build_sql_update('foo_1','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}','{"99", "xyz", "{za0,zb0,zc0}"}'); + +-- build a delete statement based on a local tuple, +SELECT dblink_build_sql_delete('foo_1','1 2',2,'{"0", "a"}'); +-- too many pk fields, should fail +SELECT dblink_build_sql_delete('foo_1','1 2 3 4',4,'{"0", "a", "{a0,b0,c0}"}'); + +DROP TABLE foo_1; + +-- retest using a quoted and schema qualified table +CREATE SCHEMA "MySchema"; +CREATE TABLE "MySchema"."Foo"(f1 int, f2 text, f3 text[], primary key (f1,f2)); +INSERT INTO "MySchema"."Foo" VALUES (0,'a','{"a0","b0","c0"}'); + +-- list the primary key fields +SELECT * +FROM dblink_get_pkey('"MySchema"."Foo"'); + +-- build an insert statement based on a local tuple, +-- replacing the primary key values with new ones +SELECT dblink_build_sql_insert('"MySchema"."Foo"','1 2',2,'{"0", "a"}','{"99", "xyz"}'); + +-- build an update statement based on a local tuple, +-- replacing the primary key values with new ones +SELECT dblink_build_sql_update('"MySchema"."Foo"','1 2',2,'{"0", "a"}','{"99", "xyz"}'); + +-- build a delete statement based on a local tuple, +SELECT dblink_build_sql_delete('"MySchema"."Foo"','1 2',2,'{"0", "a"}'); + +CREATE FUNCTION connection_parameters() RETURNS text LANGUAGE SQL AS $f$ + SELECT $$dbname='$$||current_database()||$$' port=$$||current_setting('port'); +$f$; + +-- regular old dblink +SELECT * +FROM dblink(connection_parameters(),'SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; + +-- should generate "connection not available" error +SELECT * +FROM dblink('SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; + +-- The first-level connection's backend will crash on exit given OpenLDAP +-- [2.4.24, 2.4.31]. We won't see evidence of any crash until the victim +-- process terminates and the postmaster responds. If process termination +-- entails writing a core dump, that can take awhile. Wait for the process to +-- vanish. At that point, the postmaster has called waitpid() on the crashed +-- process, and it will accept no new connections until it has reinitialized +-- the cluster. (We can't exploit pg_stat_activity, because the crash happens +-- after the backend updates shared memory to reflect its impending exit.) +DO $pl$ +DECLARE + detail text; +BEGIN + PERFORM wait_pid(crash_pid) + FROM dblink(connection_parameters(), $$ + SELECT pg_backend_pid() FROM dblink( + 'service=test_ldap '||connection_parameters(), + -- This string concatenation is a hack to shoehorn a + -- set_pgservicefile call into the SQL statement. + 'SELECT 1' || set_pgservicefile('pg_service.conf') + ) t(c int) + $$) AS t(crash_pid int); +EXCEPTION WHEN OTHERS THEN + GET STACKED DIAGNOSTICS detail = PG_EXCEPTION_DETAIL; + -- Expected error in a non-LDAP build. + IF NOT detail LIKE 'syntax error in service file%' THEN RAISE; END IF; +END +$pl$; + +-- create a persistent connection +SELECT dblink_connect(connection_parameters()); + +-- use the persistent connection +SELECT * +FROM dblink('SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; + +-- open a cursor with bad SQL and fail_on_error set to false +SELECT dblink_open('rmt_foo_cursor','SELECT * FROM foobar',false); + +-- reset remote transaction state +SELECT dblink_exec('ABORT'); + +-- open a cursor +SELECT dblink_open('rmt_foo_cursor','SELECT * FROM foo'); + +-- close the cursor +SELECT dblink_close('rmt_foo_cursor',false); + +-- open the cursor again +SELECT dblink_open('rmt_foo_cursor','SELECT * FROM foo'); + +-- fetch some data +SELECT * +FROM dblink_fetch('rmt_foo_cursor',4) AS t(a int, b text, c text[]); + +SELECT * +FROM dblink_fetch('rmt_foo_cursor',4) AS t(a int, b text, c text[]); + +-- this one only finds two rows left +SELECT * +FROM dblink_fetch('rmt_foo_cursor',4) AS t(a int, b text, c text[]); + +-- intentionally botch a fetch +SELECT * +FROM dblink_fetch('rmt_foobar_cursor',4,false) AS t(a int, b text, c text[]); + +-- reset remote transaction state +SELECT dblink_exec('ABORT'); + +-- close the wrong cursor +SELECT dblink_close('rmt_foobar_cursor',false); + +-- should generate 'cursor "rmt_foo_cursor" not found' error +SELECT * +FROM dblink_fetch('rmt_foo_cursor',4) AS t(a int, b text, c text[]); + +-- this time, 'cursor "rmt_foo_cursor" not found' as a notice +SELECT * +FROM dblink_fetch('rmt_foo_cursor',4,false) AS t(a int, b text, c text[]); + +-- close the persistent connection +SELECT dblink_disconnect(); + +-- should generate "connection not available" error +SELECT * +FROM dblink('SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; + +-- put more data into our table, first using arbitrary connection syntax +-- but truncate the actual return value so we can use diff to check for success +SELECT substr(dblink_exec(connection_parameters(),'INSERT INTO foo VALUES(10,''k'',''{"a10","b10","c10"}'')'),1,6); + +-- create a persistent connection +SELECT dblink_connect(connection_parameters()); + +-- put more data into our table, using persistent connection syntax +-- but truncate the actual return value so we can use diff to check for success +SELECT substr(dblink_exec('INSERT INTO foo VALUES(11,''l'',''{"a11","b11","c11"}'')'),1,6); + +-- let's see it +SELECT * +FROM dblink('SELECT * FROM foo') AS t(a int, b text, c text[]); + +-- bad remote select +SELECT * +FROM dblink('SELECT * FROM foobar',false) AS t(a int, b text, c text[]); + +-- change some data +SELECT dblink_exec('UPDATE foo SET f3[2] = ''b99'' WHERE f1 = 11'); + +-- let's see it +SELECT * +FROM dblink('SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE a = 11; + +-- botch a change to some other data +SELECT dblink_exec('UPDATE foobar SET f3[2] = ''b99'' WHERE f1 = 11',false); + +-- delete some data +SELECT dblink_exec('DELETE FROM foo WHERE f1 = 11'); + +-- let's see it +SELECT * +FROM dblink('SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE a = 11; + +-- close the persistent connection +SELECT dblink_disconnect(); + +-- +-- tests for the new named persistent connection syntax +-- + +-- should generate "missing "=" after "myconn" in connection info string" error +SELECT * +FROM dblink('myconn','SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; + +-- create a named persistent connection +SELECT dblink_connect('myconn',connection_parameters()); + +-- use the named persistent connection +SELECT * +FROM dblink('myconn','SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; + +-- use the named persistent connection, but get it wrong +SELECT * +FROM dblink('myconn','SELECT * FROM foobar',false) AS t(a int, b text, c text[]) +WHERE t.a > 7; + +-- create a second named persistent connection +-- should error with "duplicate connection name" +SELECT dblink_connect('myconn',connection_parameters()); + +-- create a second named persistent connection with a new name +SELECT dblink_connect('myconn2',connection_parameters()); + +-- use the second named persistent connection +SELECT * +FROM dblink('myconn2','SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; + +-- close the second named persistent connection +SELECT dblink_disconnect('myconn2'); + +-- open a cursor incorrectly +SELECT dblink_open('myconn','rmt_foo_cursor','SELECT * FROM foobar',false); + +-- reset remote transaction state +SELECT dblink_exec('myconn','ABORT'); + +-- test opening cursor in a transaction +SELECT dblink_exec('myconn','BEGIN'); + +-- an open transaction will prevent dblink_open() from opening its own +SELECT dblink_open('myconn','rmt_foo_cursor','SELECT * FROM foo'); + +-- this should not commit the transaction because the client opened it +SELECT dblink_close('myconn','rmt_foo_cursor'); + +-- this should succeed because we have an open transaction +SELECT dblink_exec('myconn','DECLARE xact_test CURSOR FOR SELECT * FROM foo'); + +-- commit remote transaction +SELECT dblink_exec('myconn','COMMIT'); + +-- test automatic transactions for multiple cursor opens +SELECT dblink_open('myconn','rmt_foo_cursor','SELECT * FROM foo'); + +-- the second cursor +SELECT dblink_open('myconn','rmt_foo_cursor2','SELECT * FROM foo'); + +-- this should not commit the transaction +SELECT dblink_close('myconn','rmt_foo_cursor2'); + +-- this should succeed because we have an open transaction +SELECT dblink_exec('myconn','DECLARE xact_test CURSOR FOR SELECT * FROM foo'); + +-- this should commit the transaction +SELECT dblink_close('myconn','rmt_foo_cursor'); + +-- this should fail because there is no open transaction +SELECT dblink_exec('myconn','DECLARE xact_test CURSOR FOR SELECT * FROM foo'); + +-- reset remote transaction state +SELECT dblink_exec('myconn','ABORT'); + +-- open a cursor +SELECT dblink_open('myconn','rmt_foo_cursor','SELECT * FROM foo'); + +-- fetch some data +SELECT * +FROM dblink_fetch('myconn','rmt_foo_cursor',4) AS t(a int, b text, c text[]); + +SELECT * +FROM dblink_fetch('myconn','rmt_foo_cursor',4) AS t(a int, b text, c text[]); + +-- this one only finds three rows left +SELECT * +FROM dblink_fetch('myconn','rmt_foo_cursor',4) AS t(a int, b text, c text[]); + +-- fetch some data incorrectly +SELECT * +FROM dblink_fetch('myconn','rmt_foobar_cursor',4,false) AS t(a int, b text, c text[]); + +-- reset remote transaction state +SELECT dblink_exec('myconn','ABORT'); + +-- should generate 'cursor "rmt_foo_cursor" not found' error +SELECT * +FROM dblink_fetch('myconn','rmt_foo_cursor',4) AS t(a int, b text, c text[]); + +-- close the named persistent connection +SELECT dblink_disconnect('myconn'); + +-- should generate "missing "=" after "myconn" in connection info string" error +SELECT * +FROM dblink('myconn','SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE t.a > 7; + +-- create a named persistent connection +SELECT dblink_connect('myconn',connection_parameters()); + +-- put more data into our table, using named persistent connection syntax +-- but truncate the actual return value so we can use diff to check for success +SELECT substr(dblink_exec('myconn','INSERT INTO foo VALUES(11,''l'',''{"a11","b11","c11"}'')'),1,6); + +-- let's see it +SELECT * +FROM dblink('myconn','SELECT * FROM foo') AS t(a int, b text, c text[]); + +-- change some data +SELECT dblink_exec('myconn','UPDATE foo SET f3[2] = ''b99'' WHERE f1 = 11'); + +-- let's see it +SELECT * +FROM dblink('myconn','SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE a = 11; + +-- delete some data +SELECT dblink_exec('myconn','DELETE FROM foo WHERE f1 = 11'); + +-- let's see it +SELECT * +FROM dblink('myconn','SELECT * FROM foo') AS t(a int, b text, c text[]) +WHERE a = 11; + +-- close the named persistent connection +SELECT dblink_disconnect('myconn'); + +-- close the named persistent connection again +-- should get 'connection "myconn" not available' error +SELECT dblink_disconnect('myconn'); + +-- test asynchronous queries +SELECT dblink_connect('dtest1', connection_parameters()); +SELECT * from + dblink_send_query('dtest1', 'select * from foo where f1 < 3') as t1; + +SELECT dblink_connect('dtest2', connection_parameters()); +SELECT * from + dblink_send_query('dtest2', 'select * from foo where f1 > 2 and f1 < 7') as t1; + +SELECT dblink_connect('dtest3', connection_parameters()); +SELECT * from + dblink_send_query('dtest3', 'select * from foo where f1 > 6') as t1; + +CREATE TEMPORARY TABLE result AS +(SELECT * from dblink_get_result('dtest1') as t1(f1 int, f2 text, f3 text[])) +UNION +(SELECT * from dblink_get_result('dtest2') as t2(f1 int, f2 text, f3 text[])) +UNION +(SELECT * from dblink_get_result('dtest3') as t3(f1 int, f2 text, f3 text[])) +ORDER by f1; + +-- dblink_get_connections returns an array with elements in a machine-dependent +-- ordering, so we must resort to unnesting and sorting for a stable result +create function unnest(anyarray) returns setof anyelement +language sql strict immutable as $$ +select $1[i] from generate_series(array_lower($1,1), array_upper($1,1)) as i +$$; + +SELECT * FROM unnest(dblink_get_connections()) ORDER BY 1; + +SELECT dblink_is_busy('dtest1'); + +SELECT dblink_disconnect('dtest1'); +SELECT dblink_disconnect('dtest2'); +SELECT dblink_disconnect('dtest3'); + +SELECT * from result; + +SELECT dblink_connect('dtest1', connection_parameters()); +SELECT * from + dblink_send_query('dtest1', 'select * from foo where f1 < 3') as t1; + +SELECT dblink_cancel_query('dtest1'); +SELECT dblink_error_message('dtest1'); +SELECT dblink_disconnect('dtest1'); + +-- test foreign data wrapper functionality +CREATE ROLE regress_dblink_user; +DO $d$ + BEGIN + EXECUTE $$CREATE SERVER fdtest FOREIGN DATA WRAPPER dblink_fdw + OPTIONS (dbname '$$||current_database()||$$', + port '$$||current_setting('port')||$$' + )$$; + END; +$d$; + +CREATE USER MAPPING FOR public SERVER fdtest + OPTIONS (server 'localhost'); -- fail, can't specify server here +CREATE USER MAPPING FOR public SERVER fdtest OPTIONS (user :'USER'); + +GRANT USAGE ON FOREIGN SERVER fdtest TO regress_dblink_user; +GRANT EXECUTE ON FUNCTION dblink_connect_u(text, text) TO regress_dblink_user; + +SET SESSION AUTHORIZATION regress_dblink_user; +-- should fail +SELECT dblink_connect('myconn', 'fdtest'); +-- should succeed +SELECT dblink_connect_u('myconn', 'fdtest'); +SELECT * FROM dblink('myconn','SELECT * FROM foo') AS t(a int, b text, c text[]); + +\c - - +REVOKE USAGE ON FOREIGN SERVER fdtest FROM regress_dblink_user; +REVOKE EXECUTE ON FUNCTION dblink_connect_u(text, text) FROM regress_dblink_user; +DROP USER regress_dblink_user; +DROP USER MAPPING FOR public SERVER fdtest; +DROP SERVER fdtest; + +-- test asynchronous notifications +SELECT dblink_connect(connection_parameters()); + +--should return listen +SELECT dblink_exec('LISTEN regression'); +--should return listen +SELECT dblink_exec('LISTEN foobar'); + +SELECT dblink_exec('NOTIFY regression'); +SELECT dblink_exec('NOTIFY foobar'); + +SELECT notify_name, be_pid = (select t.be_pid from dblink('select pg_backend_pid()') as t(be_pid int)) AS is_self_notify, extra from dblink_get_notify(); + +SELECT * from dblink_get_notify(); + +SELECT dblink_disconnect(); + +-- test dropped columns in dblink_build_sql_insert, dblink_build_sql_update +CREATE TEMP TABLE test_dropped +( + col1 INT NOT NULL DEFAULT 111, + id SERIAL PRIMARY KEY, + col2 INT NOT NULL DEFAULT 112, + col2b INT NOT NULL DEFAULT 113 +); + +INSERT INTO test_dropped VALUES(default); + +ALTER TABLE test_dropped + DROP COLUMN col1, + DROP COLUMN col2, + ADD COLUMN col3 VARCHAR(10) NOT NULL DEFAULT 'foo', + ADD COLUMN col4 INT NOT NULL DEFAULT 42; + +SELECT dblink_build_sql_insert('test_dropped', '1', 1, + ARRAY['1'::TEXT], ARRAY['2'::TEXT]); + +SELECT dblink_build_sql_update('test_dropped', '1', 1, + ARRAY['1'::TEXT], ARRAY['2'::TEXT]); + +SELECT dblink_build_sql_delete('test_dropped', '1', 1, + ARRAY['2'::TEXT]); + +-- test local mimicry of remote GUC values that affect datatype I/O +SET datestyle = ISO, MDY; +SET intervalstyle = postgres; +SET timezone = UTC; +SELECT dblink_connect('myconn',connection_parameters()); +SELECT dblink_exec('myconn', 'SET datestyle = GERMAN, DMY;'); + +-- single row synchronous case +SELECT * +FROM dblink('myconn', + 'SELECT * FROM (VALUES (''12.03.2013 00:00:00+00'')) t') + AS t(a timestamptz); + +-- multi-row synchronous case +SELECT * +FROM dblink('myconn', + 'SELECT * FROM + (VALUES (''12.03.2013 00:00:00+00''), + (''12.03.2013 00:00:00+00'')) t') + AS t(a timestamptz); + +-- single-row asynchronous case +SELECT * +FROM dblink_send_query('myconn', + 'SELECT * FROM + (VALUES (''12.03.2013 00:00:00+00'')) t'); +CREATE TEMPORARY TABLE result AS +(SELECT * from dblink_get_result('myconn') as t(t timestamptz)) +UNION ALL +(SELECT * from dblink_get_result('myconn') as t(t timestamptz)); +SELECT * FROM result; +DROP TABLE result; + +-- multi-row asynchronous case +SELECT * +FROM dblink_send_query('myconn', + 'SELECT * FROM + (VALUES (''12.03.2013 00:00:00+00''), + (''12.03.2013 00:00:00+00'')) t'); +CREATE TEMPORARY TABLE result AS +(SELECT * from dblink_get_result('myconn') as t(t timestamptz)) +UNION ALL +(SELECT * from dblink_get_result('myconn') as t(t timestamptz)) +UNION ALL +(SELECT * from dblink_get_result('myconn') as t(t timestamptz)); +SELECT * FROM result; +DROP TABLE result; + +-- Try an ambiguous interval +SELECT dblink_exec('myconn', 'SET intervalstyle = sql_standard;'); +SELECT * +FROM dblink('myconn', + 'SELECT * FROM (VALUES (''-1 2:03:04'')) i') + AS i(i interval); + +-- Try swapping to another format to ensure the GUCs are tracked +-- properly through a change. +CREATE TEMPORARY TABLE result (t timestamptz); + +SELECT dblink_exec('myconn', 'SET datestyle = ISO, MDY;'); +INSERT INTO result + SELECT * + FROM dblink('myconn', + 'SELECT * FROM (VALUES (''03.12.2013 00:00:00+00'')) t') + AS t(a timestamptz); + +SELECT dblink_exec('myconn', 'SET datestyle = GERMAN, DMY;'); +INSERT INTO result + SELECT * + FROM dblink('myconn', + 'SELECT * FROM (VALUES (''12.03.2013 00:00:00+00'')) t') + AS t(a timestamptz); + +SELECT * FROM result; + +DROP TABLE result; + +-- Check error throwing in dblink_fetch +SELECT dblink_open('myconn','error_cursor', + 'SELECT * FROM (VALUES (''1''), (''not an int'')) AS t(text);'); +SELECT * +FROM dblink_fetch('myconn','error_cursor', 1) AS t(i int); +SELECT * +FROM dblink_fetch('myconn','error_cursor', 1) AS t(i int); + +-- Make sure that the local settings have retained their values in spite +-- of shenanigans on the connection. +SHOW datestyle; +SHOW intervalstyle; + +-- Clean up GUC-setting tests +SELECT dblink_disconnect('myconn'); +RESET datestyle; +RESET intervalstyle; +RESET timezone; |