From 16f77685f7701790191b4f8d9a3cbce60604d397 Mon Sep 17 00:00:00 2001
From: Peter Eisentraut <peter_e@gmx.net>
Date: Tue, 28 Feb 2017 14:58:07 -0500
Subject: [PATCH v2 2/2] dblink: Add background sessions support

Add function dblink_connect_self(), which creates a dblink connection
via the background session API, instead of libpq.  After that, it can be
used transparently as a dblink connection, with some restrictions.

This allows the typical use case of using dblink for self-connections
without having to deal with access control issues of a libpq connection.
---
 contrib/dblink/Makefile             |   1 +
 contrib/dblink/dblink--1.2--1.3.sql |  14 +
 contrib/dblink/dblink.c             | 557 ++++++++++++++++++++++--------------
 contrib/dblink/dblink.control       |   2 +-
 contrib/dblink/expected/dblink.out  |  95 ++++++
 contrib/dblink/sql/dblink.sql       |  20 ++
 doc/src/sgml/dblink.sgml            |  37 +++
 7 files changed, 516 insertions(+), 210 deletions(-)
 create mode 100644 contrib/dblink/dblink--1.2--1.3.sql

diff --git a/contrib/dblink/Makefile b/contrib/dblink/Makefile
index 5189758dab..106039da3b 100644
--- a/contrib/dblink/Makefile
+++ b/contrib/dblink/Makefile
@@ -7,6 +7,7 @@ SHLIB_LINK = $(libpq)
 
 EXTENSION = dblink
 DATA = dblink--1.2.sql dblink--1.1--1.2.sql dblink--1.0--1.1.sql \
+	dblink--1.2--1.3.sql \
 	dblink--unpackaged--1.0.sql
 PGFILEDESC = "dblink - connect to other PostgreSQL databases"
 
diff --git a/contrib/dblink/dblink--1.2--1.3.sql b/contrib/dblink/dblink--1.2--1.3.sql
new file mode 100644
index 0000000000..b863613a3c
--- /dev/null
+++ b/contrib/dblink/dblink--1.2--1.3.sql
@@ -0,0 +1,14 @@
+/* contrib/dblink/dblink--1.2--1.3.sql */
+
+-- complain if script is sourced in psql, rather than via ALTER EXTENSION
+\echo Use "ALTER EXTENSION dblink UPDATE TO '1.3'" to load this file. \quit
+
+CREATE FUNCTION dblink_connect_self ()
+RETURNS text
+AS 'MODULE_PATHNAME','dblink_connect_self'
+LANGUAGE C STRICT PARALLEL RESTRICTED;
+
+CREATE FUNCTION dblink_connect_self (text)
+RETURNS text
+AS 'MODULE_PATHNAME','dblink_connect_self'
+LANGUAGE C STRICT PARALLEL RESTRICTED;
diff --git a/contrib/dblink/dblink.c b/contrib/dblink/dblink.c
index 67d6699066..6cb1a1e2e6 100644
--- a/contrib/dblink/dblink.c
+++ b/contrib/dblink/dblink.c
@@ -51,6 +51,7 @@
 #include "mb/pg_wchar.h"
 #include "miscadmin.h"
 #include "parser/scansup.h"
+#include "tcop/bgsession.h"
 #include "utils/acl.h"
 #include "utils/builtins.h"
 #include "utils/fmgroids.h"
@@ -66,6 +67,7 @@ PG_MODULE_MAGIC;
 typedef struct remoteConn
 {
 	PGconn	   *conn;			/* Hold the remote connection */
+	BackgroundSession *bgconn;
 	int			openCursorCount;	/* The number of open cursors */
 	bool		newXactForCursor;		/* Opened a transaction for a cursor */
 } remoteConn;
@@ -89,6 +91,7 @@ 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 materializeBgResult(FunctionCallInfo fcinfo, BackgroundSession *bgconn, BackgroundSessionResult *res);
 static void materializeQueryResult(FunctionCallInfo fcinfo,
 					   PGconn *conn,
 					   const char *conname,
@@ -111,7 +114,7 @@ static HeapTuple get_tuple_of_interest(Relation rel, int *pkattnums, int pknumat
 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_security_check(remoteConn *rconn);
 static void dblink_res_error(PGconn *conn, const char *conname, PGresult *res,
 							 const char *dblink_context_msg, bool fail);
 static char *get_connect_string(const char *servername);
@@ -176,22 +179,21 @@ dblink_conn_not_avail(const char *conname)
 
 static void
 dblink_get_conn(char *conname_or_str,
-				PGconn * volatile *conn_p, char **conname_p, volatile bool *freeconn_p)
+				remoteConn * volatile *rconn_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;
+		PGconn	   *conn;
 
 		connstr = get_connect_string(conname_or_str);
 		if (connstr == NULL)
@@ -207,38 +209,52 @@ dblink_get_conn(char *conname_or_str,
 					 errmsg("could not establish connection"),
 					 errdetail_internal("%s", msg)));
 		}
-		dblink_security_check(conn, rconn);
 		if (PQclientEncoding(conn) != GetDatabaseEncoding())
 			PQsetClientEncoding(conn, GetDatabaseEncodingName());
+		rconn = palloc0(sizeof(*rconn));
+		rconn->conn = conn;
 		freeconn = true;
 		conname = NULL;
+		dblink_security_check(rconn);
 	}
 
-	*conn_p = conn;
+	*rconn_p = rconn;
 	*conname_p = conname;
 	*freeconn_p = freeconn;
 }
 
-static PGconn *
+static void
+dblink_finish_conn(remoteConn *rconn)
+{
+	if (rconn->conn)
+	{
+		PQfinish(rconn->conn);
+		rconn->conn = NULL;
+	}
+	if (rconn->bgconn)
+	{
+		BackgroundSessionEnd(rconn->bgconn);
+		rconn->bgconn = NULL;
+	}
+}
+
+static remoteConn *
 dblink_get_named_conn(const char *conname)
 {
 	remoteConn *rconn = getConnectionByName(conname);
 	if (rconn)
-		return rconn->conn;
+		return rconn;
 	else
 		dblink_conn_not_avail(conname);
 }
 
 static void
-dblink_init(void)
+dblink_bgsession_not_supported(const remoteConn *rconn)
 {
-	if (!pconn)
-	{
-		pconn = (remoteConn *) MemoryContextAlloc(TopMemoryContext, sizeof(remoteConn));
-		pconn->conn = NULL;
-		pconn->openCursorCount = 0;
-		pconn->newXactForCursor = FALSE;
-	}
+	if (!rconn->conn)
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("not supported with background session connection")));
 }
 
 /*
@@ -255,8 +271,6 @@ dblink_connect(PG_FUNCTION_ARGS)
 	PGconn	   *conn = NULL;
 	remoteConn *rconn = NULL;
 
-	dblink_init();
-
 	if (PG_NARGS() == 2)
 	{
 		conname_or_str = text_to_cstring(PG_GETARG_TEXT_PP(1));
@@ -265,8 +279,7 @@ dblink_connect(PG_FUNCTION_ARGS)
 	else if (PG_NARGS() == 1)
 		conname_or_str = text_to_cstring(PG_GETARG_TEXT_PP(0));
 
-	if (connname)
-		rconn = (remoteConn *) MemoryContextAlloc(TopMemoryContext,
+	rconn = (remoteConn *) MemoryContextAllocZero(TopMemoryContext,
 												  sizeof(remoteConn));
 
 	/* first check for valid foreign data server */
@@ -292,19 +305,46 @@ dblink_connect(PG_FUNCTION_ARGS)
 	}
 
 	/* check password actually used if not superuser */
-	dblink_security_check(conn, rconn);
+	dblink_security_check(rconn);
 
 	/* attempt to set client encoding to match server encoding, if needed */
 	if (PQclientEncoding(conn) != GetDatabaseEncoding())
 		PQsetClientEncoding(conn, GetDatabaseEncodingName());
 
+	rconn->conn = conn;
+
 	if (connname)
-	{
-		rconn->conn = conn;
 		createNewConnection(connname, rconn);
-	}
 	else
-		pconn->conn = conn;
+		pconn = rconn;
+
+	PG_RETURN_TEXT_P(cstring_to_text("OK"));
+}
+
+PG_FUNCTION_INFO_V1(dblink_connect_self);
+Datum
+dblink_connect_self(PG_FUNCTION_ARGS)
+{
+	char	   *connname = NULL;
+	BackgroundSession *conn = NULL;
+	remoteConn *rconn = NULL;
+	MemoryContext oldcontext;
+
+	if (PG_NARGS() == 1)
+		connname = text_to_cstring(PG_GETARG_TEXT_PP(0));
+
+	oldcontext = MemoryContextSwitchTo(TopMemoryContext);
+	conn = BackgroundSessionStart();
+	MemoryContextSwitchTo(oldcontext);
+
+	rconn = (remoteConn *) MemoryContextAllocZero(TopMemoryContext,
+												  sizeof(remoteConn));
+	rconn->bgconn = conn;
+
+	if (connname)
+		createNewConnection(connname, rconn);
+	else
+		pconn = rconn;
 
 	PG_RETURN_TEXT_P(cstring_to_text("OK"));
 }
@@ -318,31 +358,25 @@ 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;
+		rconn = pconn;
 
-	if (!conn)
+	if (!rconn)
 		dblink_conn_not_avail(conname);
 
-	PQfinish(conn);
-	if (rconn)
-	{
+	dblink_finish_conn(rconn);
+	pfree(rconn);
+
+	if (conname)
 		deleteConnection(conname);
-		pfree(rconn);
-	}
 	else
-		pconn->conn = NULL;
+		pconn = NULL;
 
 	PG_RETURN_TEXT_P(cstring_to_text("OK"));
 }
@@ -363,7 +397,6 @@ dblink_open(PG_FUNCTION_ARGS)
 	remoteConn *rconn = NULL;
 	bool		fail = true;	/* default to backward compatible behavior */
 
-	dblink_init();
 	initStringInfo(&buf);
 
 	if (PG_NARGS() == 2)
@@ -401,10 +434,12 @@ dblink_open(PG_FUNCTION_ARGS)
 		rconn = getConnectionByName(conname);
 	}
 
-	if (!rconn || !rconn->conn)
+	if (!rconn)
 		dblink_conn_not_avail(conname);
-	else
-		conn = rconn->conn;
+
+	dblink_bgsession_not_supported(rconn);
+
+	conn = rconn->conn;
 
 	/* If we are not in a transaction, start one */
 	if (PQtransactionStatus(conn) == PQTRANS_IDLE)
@@ -454,7 +489,6 @@ dblink_close(PG_FUNCTION_ARGS)
 	remoteConn *rconn = NULL;
 	bool		fail = true;	/* default to backward compatible behavior */
 
-	dblink_init();
 	initStringInfo(&buf);
 
 	if (PG_NARGS() == 1)
@@ -488,10 +522,12 @@ dblink_close(PG_FUNCTION_ARGS)
 		rconn = getConnectionByName(conname);
 	}
 
-	if (!rconn || !rconn->conn)
+	if (!rconn)
 		dblink_conn_not_avail(conname);
-	else
-		conn = rconn->conn;
+
+	dblink_bgsession_not_supported(rconn);
+
+	conn = rconn->conn;
 
 	appendStringInfo(&buf, "CLOSE %s", curname);
 
@@ -543,8 +579,6 @@ dblink_fetch(PG_FUNCTION_ARGS)
 
 	prepTuplestoreResult(fcinfo);
 
-	dblink_init();
-
 	if (PG_NARGS() == 4)
 	{
 		/* text,text,int,bool */
@@ -554,8 +588,6 @@ dblink_fetch(PG_FUNCTION_ARGS)
 		fail = PG_GETARG_BOOL(3);
 
 		rconn = getConnectionByName(conname);
-		if (rconn)
-			conn = rconn->conn;
 	}
 	else if (PG_NARGS() == 3)
 	{
@@ -565,7 +597,7 @@ dblink_fetch(PG_FUNCTION_ARGS)
 			curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
 			howmany = PG_GETARG_INT32(1);
 			fail = PG_GETARG_BOOL(2);
-			conn = pconn->conn;
+			rconn = pconn;
 		}
 		else
 		{
@@ -574,8 +606,6 @@ dblink_fetch(PG_FUNCTION_ARGS)
 			howmany = PG_GETARG_INT32(2);
 
 			rconn = getConnectionByName(conname);
-			if (rconn)
-				conn = rconn->conn;
 		}
 	}
 	else if (PG_NARGS() == 2)
@@ -583,12 +613,16 @@ dblink_fetch(PG_FUNCTION_ARGS)
 		/* text,int */
 		curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
 		howmany = PG_GETARG_INT32(1);
-		conn = pconn->conn;
+		rconn = pconn;
 	}
 
-	if (!conn)
+	if (!rconn)
 		dblink_conn_not_avail(conname);
 
+	dblink_bgsession_not_supported(rconn);
+
+	conn = rconn->conn;
+
 	initStringInfo(&buf);
 	appendStringInfo(&buf, "FETCH %d FROM %s", howmany, curname);
 
@@ -633,23 +667,30 @@ PG_FUNCTION_INFO_V1(dblink_send_query);
 Datum
 dblink_send_query(PG_FUNCTION_ARGS)
 {
-	PGconn	   *conn;
+	remoteConn *rconn;
 	char	   *sql;
 	int			retval;
 
 	if (PG_NARGS() == 2)
 	{
-		conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
+		rconn = 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)));
+	if (rconn->conn)
+	{
+		retval = PQsendQuery(rconn->conn, sql);
+		if (retval != 1)
+			elog(NOTICE, "could not send query: %s", pchomp(PQerrorMessage(rconn->conn)));
+	}
+	else
+	{
+		BackgroundSessionSend(rconn->bgconn, sql);
+		retval = 1;
+	}
 
 	PG_RETURN_INT32(retval);
 }
@@ -664,13 +705,11 @@ dblink_get_result(PG_FUNCTION_ARGS)
 static Datum
 dblink_record_internal(FunctionCallInfo fcinfo, bool is_async)
 {
-	PGconn	   *volatile conn = NULL;
+	remoteConn *volatile rconn = NULL;
 	volatile bool freeconn = false;
 
 	prepTuplestoreResult(fcinfo);
 
-	dblink_init();
-
 	PG_TRY();
 	{
 		char	   *sql = NULL;
@@ -682,7 +721,7 @@ dblink_record_internal(FunctionCallInfo fcinfo, bool is_async)
 			if (PG_NARGS() == 3)
 			{
 				/* text,text,bool */
-				dblink_get_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)), &conn, &conname, &freeconn);
+				dblink_get_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)), &rconn, &conname, &freeconn);
 				sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
 				fail = PG_GETARG_BOOL(2);
 			}
@@ -691,20 +730,20 @@ dblink_record_internal(FunctionCallInfo fcinfo, bool is_async)
 				/* text,text or text,bool */
 				if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
 				{
-					conn = pconn->conn;
+					rconn = pconn;
 					sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
 					fail = PG_GETARG_BOOL(1);
 				}
 				else
 				{
-					dblink_get_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)), &conn, &conname, &freeconn);
+					dblink_get_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)), &rconn, &conname, &freeconn);
 					sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
 				}
 			}
 			else if (PG_NARGS() == 1)
 			{
 				/* text */
-				conn = pconn->conn;
+				rconn = pconn;
 				sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
 			}
 			else
@@ -717,61 +756,75 @@ dblink_record_internal(FunctionCallInfo fcinfo, bool is_async)
 			if (PG_NARGS() == 2)
 			{
 				/* text,bool */
-				conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
+				rconn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
 				fail = PG_GETARG_BOOL(1);
 			}
 			else if (PG_NARGS() == 1)
 			{
 				/* text */
-				conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
+				rconn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
 			}
 			else
 				/* shouldn't happen */
 				elog(ERROR, "wrong number of arguments");
 		}
 
-		if (!conn)
+		if (!rconn)
 			dblink_conn_not_avail(conname);
 
 		if (!is_async)
 		{
-			/* synchronous query, use efficient tuple collection method */
-			materializeQueryResult(fcinfo, conn, conname, sql, fail);
+			if (rconn->conn)
+				/* synchronous query, use efficient tuple collection method */
+				materializeQueryResult(fcinfo, rconn->conn, conname, sql, fail);
+			else if (rconn->bgconn)
+			{
+				BackgroundSessionResult *res = BackgroundSessionExecute(rconn->bgconn, sql);
+				materializeBgResult(fcinfo, rconn->bgconn, res);
+			}
 		}
 		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 (rconn->conn)
 			{
-				if (PQresultStatus(res) != PGRES_COMMAND_OK &&
-					PQresultStatus(res) != PGRES_TUPLES_OK)
-				{
-					dblink_res_error(conn, conname, res,
-									 "could not execute query", fail);
-					/* if fail isn't set, we'll return an empty query result */
-				}
-				else
+				PGresult   *res = PQgetResult(rconn->conn);
+
+				/* NULL means we're all done with the async results */
+				if (res)
 				{
-					materializeResult(fcinfo, conn, res);
+					if (PQresultStatus(res) != PGRES_COMMAND_OK &&
+						PQresultStatus(res) != PGRES_TUPLES_OK)
+					{
+						dblink_res_error(rconn->conn, conname, res,
+										 "could not execute query", fail);
+						/* if fail isn't set, we'll return an empty query result */
+					}
+					else
+					{
+						materializeResult(fcinfo, rconn->conn, res);
+					}
 				}
 			}
+			else
+			{
+				BackgroundSessionResult *res = BackgroundSessionGetResult(rconn->bgconn);
+				materializeBgResult(fcinfo, rconn->bgconn, res);
+			}
 		}
 	}
 	PG_CATCH();
 	{
 		/* if needed, close the connection to the database */
 		if (freeconn)
-			PQfinish(conn);
+			dblink_finish_conn(rconn);
 		PG_RE_THROW();
 	}
 	PG_END_TRY();
 
 	/* if needed, close the connection to the database */
 	if (freeconn)
-		PQfinish(conn);
+		dblink_finish_conn(rconn);
 
 	return (Datum) 0;
 }
@@ -806,6 +859,45 @@ prepTuplestoreResult(FunctionCallInfo fcinfo)
 }
 
 /*
+ * get a tuple descriptor for our result type
+ */
+static TupleDesc
+dblink_get_call_result_type(FunctionCallInfo fcinfo, int nfields)
+{
+	TupleDesc	tupdesc;
+
+	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;
+	}
+
+	/*
+	 * 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")));
+
+	/* make sure we have a persistent copy of the tupdesc */
+	return CreateTupleDescCopy(tupdesc);
+}
+
+/*
  * 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.
@@ -844,41 +936,11 @@ materializeResult(FunctionCallInfo fcinfo, PGconn *conn, PGresult *res)
 			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);
+			tupdesc = dblink_get_call_result_type(fcinfo, nfields);
 		}
 
-		/*
-		 * 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;
@@ -948,6 +1010,95 @@ materializeResult(FunctionCallInfo fcinfo, PGconn *conn, PGresult *res)
 	PG_END_TRY();
 }
 
+static void
+materializeBgResult(FunctionCallInfo fcinfo, BackgroundSession *bgconn, BackgroundSessionResult *res)
+{
+	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
+	TupleDesc	tupdesc;
+	bool		is_sql_cmd;
+	int			nfields;
+	AttInMetadata *attinmeta;
+	HeapTuple	outtuple;
+	Tuplestorestate *tupstore;
+	MemoryContext oldcontext;
+	char	  **values;
+
+	/* prepTuplestoreResult must have been called previously */
+	Assert(rsinfo->returnMode == SFRM_Materialize);
+
+	if (!res->tupdesc)
+	{
+		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, false);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 1, "status",
+						   TEXTOID, -1, 0);
+		nfields = 1;
+	}
+	else
+	{
+		is_sql_cmd = false;
+		nfields = res->tupdesc->natts;
+		tupdesc = dblink_get_call_result_type(fcinfo, nfields);
+	}
+
+	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 = (char **) palloc(nfields * sizeof(char *));
+
+	if (is_sql_cmd)
+	{
+		values[0] = (char *) res->command;
+		outtuple = BuildTupleFromCStrings(attinmeta, values);
+		tuplestore_puttuple(tupstore, outtuple);
+	}
+	else
+	{
+		int			j;
+
+		for (j = 0; j < res->ntuples; j++)
+		{
+			HeapTuple intuple = res->tuples[j];
+			int			i;
+
+			for (i = 0; i < nfields; i++)
+			{
+				bool		isnull;
+				Datum		val;
+
+				val = heap_getattr(intuple, i+1, res->tupdesc, &isnull);
+				if (isnull)
+					values[i] = NULL;
+				else
+				{
+					Oid		typOutput;
+					bool	typIsVarlena;
+
+					getTypeOutputInfo(res->tupdesc->attrs[i]->atttypid, &typOutput, &typIsVarlena);
+					values[i] = DatumGetCString(OidFunctionCall1(typOutput, val));
+				}
+			}
+
+			outtuple = BuildTupleFromCStrings(attinmeta, values);
+			tuplestore_puttuple(tupstore, outtuple);
+		}
+	}
+
+	/* clean up and return the tuplestore */
+	tuplestore_donestoring(tupstore);
+}
+
 /*
  * Execute the given SQL command and store its results into a tuplestore
  * to be returned as the result of the current function.
@@ -1165,34 +1316,7 @@ storeRow(volatile storeInfo *sinfo, PGresult *res, bool first)
 			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")));
+		tupdesc = dblink_get_call_result_type(sinfo->fcinfo, nfields);
 
 		/* Prepare attinmeta for later data conversions */
 		sinfo->attinmeta = TupleDescGetAttInMetadata(tupdesc);
@@ -1292,13 +1416,14 @@ PG_FUNCTION_INFO_V1(dblink_is_busy);
 Datum
 dblink_is_busy(PG_FUNCTION_ARGS)
 {
-	PGconn	   *conn;
+	remoteConn	   *rconn;
 
-	dblink_init();
-	conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
+	rconn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
 
-	PQconsumeInput(conn);
-	PG_RETURN_INT32(PQisBusy(conn));
+	dblink_bgsession_not_supported(rconn);
+
+	PQconsumeInput(rconn->conn);
+	PG_RETURN_INT32(PQisBusy(rconn->conn));
 }
 
 /*
@@ -1317,13 +1442,13 @@ Datum
 dblink_cancel_query(PG_FUNCTION_ARGS)
 {
 	int			res;
-	PGconn	   *conn;
+	remoteConn *rconn;
 	PGcancel   *cancel;
 	char		errbuf[256];
 
-	dblink_init();
-	conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
-	cancel = PQgetCancel(conn);
+	rconn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
+	dblink_bgsession_not_supported(rconn);
+	cancel = PQgetCancel(rconn->conn);
 
 	res = PQcancel(cancel, errbuf, 256);
 	PQfreeCancel(cancel);
@@ -1350,12 +1475,12 @@ Datum
 dblink_error_message(PG_FUNCTION_ARGS)
 {
 	char	   *msg;
-	PGconn	   *conn;
+	remoteConn *rconn;
 
-	dblink_init();
-	conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
+	rconn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
+	dblink_bgsession_not_supported(rconn);
 
-	msg = PQerrorMessage(conn);
+	msg = PQerrorMessage(rconn->conn);
 	if (msg == NULL || msg[0] == '\0')
 		PG_RETURN_TEXT_P(cstring_to_text("OK"));
 	else
@@ -1370,14 +1495,11 @@ Datum
 dblink_exec(PG_FUNCTION_ARGS)
 {
 	text	   *volatile sql_cmd_status = NULL;
-	PGconn	   *volatile conn = NULL;
+	remoteConn *volatile rconn = 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 */
@@ -1385,7 +1507,7 @@ dblink_exec(PG_FUNCTION_ARGS)
 		if (PG_NARGS() == 3)
 		{
 			/* must be text,text,bool */
-			dblink_get_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)), &conn, &conname, &freeconn);
+			dblink_get_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)), &rconn, &conname, &freeconn);
 			sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
 			fail = PG_GETARG_BOOL(2);
 		}
@@ -1394,72 +1516,89 @@ dblink_exec(PG_FUNCTION_ARGS)
 			/* might be text,text or text,bool */
 			if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
 			{
-				conn = pconn->conn;
+				rconn = pconn;
 				sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
 				fail = PG_GETARG_BOOL(1);
 			}
 			else
 			{
-				dblink_get_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)), &conn, &conname, &freeconn);
+				dblink_get_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)), &rconn, &conname, &freeconn);
 				sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
 			}
 		}
 		else if (PG_NARGS() == 1)
 		{
 			/* must be single text argument */
-			conn = pconn->conn;
+			rconn = pconn;
 			sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
 		}
 		else
 			/* shouldn't happen */
 			elog(ERROR, "wrong number of arguments");
 
-		if (!conn)
+		if (!rconn)
 			dblink_conn_not_avail(conname);
 
-		res = PQexec(conn, sql);
-		if (!res ||
-			(PQresultStatus(res) != PGRES_COMMAND_OK &&
-			 PQresultStatus(res) != PGRES_TUPLES_OK))
+		if (rconn->conn)
 		{
-			dblink_res_error(conn, conname, res,
-							 "could not execute command", fail);
+			PGresult   *res;
 
-			/*
-			 * 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);
+			res = PQexec(rconn->conn, sql);
+
+			if (!res ||
+				(PQresultStatus(res) != PGRES_COMMAND_OK &&
+				 PQresultStatus(res) != PGRES_TUPLES_OK))
+			{
+				dblink_res_error(rconn->conn, conname, res,
+								 "could not execute command", fail);
+
+				/*
+				 * 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")));
+			}
 		}
 		else
 		{
-			PQclear(res);
-			ereport(ERROR,
-				  (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
-				   errmsg("statement returning results not allowed")));
+			BackgroundSessionResult *res;
+
+			res = BackgroundSessionExecute(rconn->bgconn, sql);
+			if (res->tupdesc)
+				ereport(ERROR,
+						(errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
+						 errmsg("statement returning results not allowed")));
+			sql_cmd_status = cstring_to_text(res->command);
 		}
 	}
 	PG_CATCH();
 	{
 		/* if needed, close the connection to the database */
 		if (freeconn)
-			PQfinish(conn);
+			dblink_finish_conn(rconn);
 		PG_RE_THROW();
 	}
 	PG_END_TRY();
 
 	/* if needed, close the connection to the database */
 	if (freeconn)
-		PQfinish(conn);
+		dblink_finish_conn(rconn);
 
 	PG_RETURN_TEXT_P(sql_cmd_status);
 }
@@ -1866,7 +2005,7 @@ PG_FUNCTION_INFO_V1(dblink_get_notify);
 Datum
 dblink_get_notify(PG_FUNCTION_ARGS)
 {
-	PGconn	   *conn;
+	remoteConn *rconn;
 	PGnotify   *notify;
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	TupleDesc	tupdesc;
@@ -1876,11 +2015,12 @@ dblink_get_notify(PG_FUNCTION_ARGS)
 
 	prepTuplestoreResult(fcinfo);
 
-	dblink_init();
 	if (PG_NARGS() == 1)
-		conn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
+		rconn = dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
 	else
-		conn = pconn->conn;
+		rconn = pconn;
+
+	dblink_bgsession_not_supported(rconn);
 
 	/* create the tuplestore in per-query memory */
 	per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
@@ -1900,8 +2040,8 @@ dblink_get_notify(PG_FUNCTION_ARGS)
 
 	MemoryContextSwitchTo(oldcontext);
 
-	PQconsumeInput(conn);
-	while ((notify = PQnotifies(conn)) != NULL)
+	PQconsumeInput(rconn->conn);
+	while ((notify = PQnotifies(rconn->conn)) != NULL)
 	{
 		Datum		values[DBLINK_NOTIFY_COLS];
 		bool		nulls[DBLINK_NOTIFY_COLS];
@@ -1924,7 +2064,7 @@ dblink_get_notify(PG_FUNCTION_ARGS)
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 
 		PQfreemem(notify);
-		PQconsumeInput(conn);
+		PQconsumeInput(rconn->conn);
 	}
 
 	/* clean up and return the tuplestore */
@@ -2557,7 +2697,7 @@ createNewConnection(const char *name, remoteConn *rconn)
 
 	if (found)
 	{
-		PQfinish(rconn->conn);
+		dblink_finish_conn(rconn);
 		pfree(rconn);
 
 		ereport(ERROR,
@@ -2592,15 +2732,14 @@ deleteConnection(const char *name)
 }
 
 static void
-dblink_security_check(PGconn *conn, remoteConn *rconn)
+dblink_security_check(remoteConn *rconn)
 {
-	if (!superuser())
+	if (!superuser() && rconn->conn)
 	{
-		if (!PQconnectionUsedPassword(conn))
+		if (!PQconnectionUsedPassword(rconn->conn))
 		{
-			PQfinish(conn);
-			if (rconn)
-				pfree(rconn);
+			PQfinish(rconn->conn);
+			pfree(rconn);
 
 			ereport(ERROR,
 				  (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
diff --git a/contrib/dblink/dblink.control b/contrib/dblink/dblink.control
index bdd17d28a4..816d19f448 100644
--- a/contrib/dblink/dblink.control
+++ b/contrib/dblink/dblink.control
@@ -1,5 +1,5 @@
 # dblink extension
 comment = 'connect to other PostgreSQL databases from within a database'
-default_version = '1.2'
+default_version = '1.3'
 module_pathname = '$libdir/dblink'
 relocatable = true
diff --git a/contrib/dblink/expected/dblink.out b/contrib/dblink/expected/dblink.out
index 4b6d26e574..6faa525f0c 100644
--- a/contrib/dblink/expected/dblink.out
+++ b/contrib/dblink/expected/dblink.out
@@ -1124,3 +1124,98 @@ SELECT dblink_disconnect('myconn');
 RESET datestyle;
 RESET intervalstyle;
 RESET timezone;
+-- background sessions
+SELECT * FROM foo;
+ 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_self();
+ dblink_connect_self 
+---------------------
+ OK
+(1 row)
+
+SELECT dblink_exec('SELECT * FROM foo');
+ERROR:  statement returning results not allowed
+SELECT dblink_exec('DELETE FROM foo WHERE f1 > 5');
+ dblink_exec 
+-------------
+ DELETE 5
+(1 row)
+
+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 rows)
+
+SELECT * FROM dblink('DELETE FROM foo WHERE f1 > 5') AS (status text);
+  status  
+----------
+ DELETE 0
+(1 row)
+
+SELECT dblink_disconnect();
+ dblink_disconnect 
+-------------------
+ OK
+(1 row)
+
+SELECT * FROM foo;
+ 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 rows)
+
+-- asynchronous case
+SELECT dblink_connect_self('myconn');
+ dblink_connect_self 
+---------------------
+ OK
+(1 row)
+
+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)
+
+SELECT * from dblink_get_result('myconn') as t(t timestamptz);
+              t               
+------------------------------
+ Mon Dec 02 16:00:00 2013 PST
+ Mon Dec 02 16:00:00 2013 PST
+(2 rows)
+
+SELECT dblink_disconnect('myconn');
+ dblink_disconnect 
+-------------------
+ OK
+(1 row)
+
diff --git a/contrib/dblink/sql/dblink.sql b/contrib/dblink/sql/dblink.sql
index 681cf6a6e8..7f66a98aa5 100644
--- a/contrib/dblink/sql/dblink.sql
+++ b/contrib/dblink/sql/dblink.sql
@@ -563,3 +563,23 @@ CREATE TEMPORARY TABLE result (t timestamptz);
 RESET datestyle;
 RESET intervalstyle;
 RESET timezone;
+
+-- background sessions
+SELECT * FROM foo;
+SELECT dblink_connect_self();
+SELECT dblink_exec('SELECT * FROM foo');
+SELECT dblink_exec('DELETE FROM foo WHERE f1 > 5');
+SELECT * FROM dblink('SELECT * FROM foo') AS t(a int, b text, c text[]);
+SELECT * FROM dblink('DELETE FROM foo WHERE f1 > 5') AS (status text);
+SELECT dblink_disconnect();
+SELECT * FROM foo;
+
+-- asynchronous case
+SELECT dblink_connect_self('myconn');
+SELECT *
+FROM dblink_send_query('myconn',
+    'SELECT * FROM
+     (VALUES (''12.03.2013 00:00:00+00''),
+             (''12.03.2013 00:00:00+00'')) t');
+SELECT * from dblink_get_result('myconn') as t(t timestamptz);
+SELECT dblink_disconnect('myconn');
diff --git a/doc/src/sgml/dblink.sgml b/doc/src/sgml/dblink.sgml
index f19c6b19f5..4b070ddb23 100644
--- a/doc/src/sgml/dblink.sgml
+++ b/doc/src/sgml/dblink.sgml
@@ -239,6 +239,43 @@ <title>Description</title>
   </refsect1>
  </refentry>
 
+ <refentry id="CONTRIB-DBLINK-CONNECT-SELF">
+  <indexterm>
+   <primary>dblink_connect_self</primary>
+  </indexterm>
+
+  <refmeta>
+   <refentrytitle>dblink_connect_self</refentrytitle>
+   <manvolnum>3</manvolnum>
+  </refmeta>
+
+  <refnamediv>
+   <refname>dblink_connect_self</refname>
+   <refpurpose>opens a persistent connection to the same database</refpurpose>
+  </refnamediv>
+
+  <refsynopsisdiv>
+<synopsis>
+dblink_connect_self() returns text
+dblink_connect_self(text connname) returns text
+</synopsis>
+  </refsynopsisdiv>
+
+  <refsect1>
+   <title>Description</title>
+
+   <para>
+    <function>dblink_connect_self()</> opens a connection to the current
+    database and as the current user, using the background session API
+    (<xref linkend="bgsession">).
+   </para>
+
+   <para>
+    For further details see <function>dblink_connect()</>.
+   </para>
+  </refsect1>
+ </refentry>
+
  <refentry id="CONTRIB-DBLINK-DISCONNECT">
   <indexterm>
    <primary>dblink_disconnect</primary>
-- 
2.12.0

