pg_background contrib module proposal

diff --git a/contrib/pg_background/Makefile b/contrib/pg_background/Makefile
index c4e717d..085fbff 100644
--- a/contrib/pg_background/Makefile
+++ b/contrib/pg_background/Makefile
@@ -6,6 +6,8 @@ OBJS = pg_background.o
 EXTENSION = pg_background
 DATA = pg_background--1.0.sql
 
+REGRESS = pg_background
+
 ifdef USE_PGXS
 PG_CONFIG = pg_config
 PGXS := $(shell $(PG_CONFIG) --pgxs)
diff --git a/contrib/pg_background/expected/pg_background.out b/contrib/pg_background/expected/pg_background.out
new file mode 100644
index 0000000..dbc344e
--- /dev/null
+++ b/contrib/pg_background/expected/pg_background.out
@@ -0,0 +1,26 @@
+CREATE EXTENSION pg_background;
+--run 5 workers which wait about 1 second
+CREATE TABLE input as SELECT x FROM generate_series(1,5,1) x ORDER BY x DESC;
+CREATE TABLE output(place int,value int);
+CREATE sequence s start 1;
+CREATE TABLE handles as SELECT pg_background_launch('select pg_sleep('||x||'); insert into output values (nextval(''s''),'||x||');') h FROM input;
+SELECT (SELECT * FROM pg_background_result(h) as (x text) limit 1) FROM handles;
+    x     
+----------
+ SELECT 1
+ SELECT 1
+ SELECT 1
+ SELECT 1
+ SELECT 1
+(5 rows)
+
+SELECT * FROM output ORDER BY place;
+ place | value 
+-------+-------
+     1 |     1
+     2 |     2
+     3 |     3
+     4 |     4
+     5 |     5
+(5 rows)
+
diff --git a/contrib/pg_background/sql/pg_background.sql b/contrib/pg_background/sql/pg_background.sql
new file mode 100644
index 0000000..d7cbd44
--- /dev/null
+++ b/contrib/pg_background/sql/pg_background.sql
@@ -0,0 +1,9 @@
+CREATE EXTENSION pg_background;
+
+--run 5 workers which wait about 1 second
+CREATE TABLE input as SELECT x FROM generate_series(1,5,1) x ORDER BY x DESC;
+CREATE TABLE output(place int,value int);
+CREATE sequence s start 1;
+CREATE TABLE handles as SELECT pg_background_launch('select pg_sleep('||x||'); insert into output values (nextval(''s''),'||x||');') h FROM input;
+SELECT (SELECT * FROM pg_background_result(h) as (x text) limit 1) FROM handles;
+SELECT * FROM output ORDER BY place;

diff --git a/contrib/pg_background/Makefile b/contrib/pg_background/Makefile
index c4e717d..085fbff 100644
--- a/contrib/pg_background/Makefile
+++ b/contrib/pg_background/Makefile
@@ -6,6 +6,8 @@ OBJS = pg_background.o
 EXTENSION = pg_background
 DATA = pg_background--1.0.sql
 
+REGRESS = pg_background
+
 ifdef USE_PGXS
 PG_CONFIG = pg_config
 PGXS := $(shell $(PG_CONFIG) --pgxs)
diff --git a/contrib/pg_background/expected/pg_background.out b/contrib/pg_background/expected/pg_background.out
new file mode 100644
index 0000000..a6613ce
--- /dev/null
+++ b/contrib/pg_background/expected/pg_background.out
@@ -0,0 +1,31 @@
+CREATE EXTENSION pg_background;
+--run 6 workers which wait .0, .1, .2, .3, .4, .5 seconds respectively
+CREATE TABLE input as SELECT x FROM generate_series(0,.5,0.1) x ORDER BY x DESC;
+CREATE TABLE output(place int,value float);
+--sequence for indication of who's finished on which place
+CREATE sequence s start 1;
+CREATE TABLE handles as SELECT pg_background_launch(format('select pg_sleep(%s); insert into output values (nextval(''s''),%s);',x,x)) h FROM input;
+--wait until everyone finishes
+SELECT (SELECT * FROM pg_background_result(h) as (x text) limit 1) FROM handles;
+    x     
+----------
+ SELECT 1
+ SELECT 1
+ SELECT 1
+ SELECT 1
+ SELECT 1
+ SELECT 1
+(6 rows)
+
+--output results
+SELECT * FROM output ORDER BY place;
+ place | value 
+-------+-------
+     1 |     0
+     2 |   0.1
+     3 |   0.2
+     4 |   0.3
+     5 |   0.4
+     6 |   0.5
+(6 rows)
+
diff --git a/contrib/pg_background/sql/pg_background.sql b/contrib/pg_background/sql/pg_background.sql
new file mode 100644
index 0000000..770f0b8
--- /dev/null
+++ b/contrib/pg_background/sql/pg_background.sql
@@ -0,0 +1,12 @@
+CREATE EXTENSION pg_background;
+
+--run 6 workers which wait .0, .1, .2, .3, .4, .5 seconds respectively
+CREATE TABLE input as SELECT x FROM generate_series(0,.5,0.1) x ORDER BY x DESC;
+CREATE TABLE output(place int,value float);
+--sequence for indication of who's finished on which place
+CREATE sequence s start 1;
+CREATE TABLE handles as SELECT pg_background_launch(format('select pg_sleep(%s); insert into output values (nextval(''s''),%s);',x,x)) h FROM input;
+--wait until everyone finishes
+SELECT (SELECT * FROM pg_background_result(h) as (x text) limit 1) FROM handles;
+--output results
+SELECT * FROM output ORDER BY place;

diff --git a/contrib/Makefile b/contrib/Makefile
index 25263c0..04ec28a 100644
--- a/contrib/Makefile
+++ b/contrib/Makefile
@@ -29,6 +29,7 @@ SUBDIRS = \
 		oid2name	\
 		pageinspect	\
 		passwordcheck	\
+		pg_background \
 		pg_buffercache	\
 		pg_freespacemap \
 		pg_prewarm	\
diff --git a/contrib/pg_background/Makefile b/contrib/pg_background/Makefile
new file mode 100644
index 0000000..c4e717d
--- /dev/null
+++ b/contrib/pg_background/Makefile
@@ -0,0 +1,18 @@
+# contrib/pg_background/Makefile
+
+MODULE_big = pg_background
+OBJS = pg_background.o
+
+EXTENSION = pg_background
+DATA = pg_background--1.0.sql
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = contrib/pg_background
+top_builddir = ../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/contrib/pg_background/pg_background--1.0.sql b/contrib/pg_background/pg_background--1.0.sql
new file mode 100644
index 0000000..bc8a881
--- /dev/null
+++ b/contrib/pg_background/pg_background--1.0.sql
@@ -0,0 +1,30 @@
+/* contrib/pg_background/pg_background--1.0.sql */
+
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "CREATE EXTENSION pg_background" to load this file. \quit
+
+CREATE FUNCTION pg_background_launch()
+    RETURNS pg_catalog.int4 STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION pg_background_run(pid pg_catalog.int4, sql pg_catalog.text)
+    RETURNS pg_catalog.void STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION pg_background_result(pid pg_catalog.int4)
+    RETURNS SETOF pg_catalog.record STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION pg_background_detach(pid pg_catalog.int4)
+    RETURNS pg_catalog.void STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
+
+REVOKE ALL ON FUNCTION pg_background_launch()
+	FROM public;
+REVOKE ALL ON FUNCTION pg_background_run(pid pg_catalog.int4,
+										 sql pg_catalog.text)
+	FROM public;
+REVOKE ALL ON FUNCTION pg_background_result(pg_catalog.int4)
+	FROM public;
+REVOKE ALL ON FUNCTION pg_background_detach(pg_catalog.int4)
+	FROM public;
diff --git a/contrib/pg_background/pg_background.c b/contrib/pg_background/pg_background.c
new file mode 100644
index 0000000..8d26b38
--- /dev/null
+++ b/contrib/pg_background/pg_background.c
@@ -0,0 +1,316 @@
+/*--------------------------------------------------------------------------
+ *
+ * pg_background.c
+ *		Run SQL commands using a background worker.
+ *
+ * Copyright (C) 2016, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		contrib/pg_background/pg_background.c
+ *
+ * -------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/htup_details.h"
+#include "catalog/pg_type.h"
+#include "funcapi.h"
+#include "miscadmin.h"
+#include "tcop/bgsession.h"
+#include "utils/acl.h"
+#include "utils/builtins.h"
+
+/*  Private state maintained by the launching backend for IPC. */
+typedef struct pg_background_worker_info
+{
+	pid_t				pid;
+	Oid					current_user_id;
+	BackgroundSession  *session;
+	uint32				result_count; /* TODO: uint32? */
+} pg_background_worker_info;
+
+/* Private state maintained across calls to pg_background_result. */
+typedef struct pg_background_result_state
+{
+	pg_background_worker_info *info;
+	BackgroundSessionResult *result;
+} pg_background_result_state;
+
+static HTAB *worker_hash;
+
+static void remove_worker_info(pid_t pid);
+static pg_background_worker_info *find_worker_info(pid_t pid);
+static void save_worker_info(pid_t pid, BackgroundSession *session);
+static void check_rights(pg_background_worker_info *info);
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(pg_background_launch);
+PG_FUNCTION_INFO_V1(pg_background_run);
+PG_FUNCTION_INFO_V1(pg_background_result);
+PG_FUNCTION_INFO_V1(pg_background_detach);
+
+/*
+ * Start a dynamic background worker.
+ */
+Datum
+pg_background_launch(PG_FUNCTION_ARGS)
+{
+	BackgroundSession *session;
+	pid_t		pid;
+
+	session = BackgroundSessionNew(&pid);
+
+	/* Save worker info */
+	save_worker_info(pid, session);
+
+	/*  Return the worker's PID. */
+	PG_RETURN_INT32(pid);
+}
+
+/*
+ * Run a user-specified SQL command.
+ */
+Datum
+pg_background_run(PG_FUNCTION_ARGS)
+{
+	int32		pid = PG_GETARG_INT32(0);
+	char	   *sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
+	pg_background_worker_info *info;
+
+	/* See if we have a connection to the specified PID. */
+	if ((info = find_worker_info(pid)) == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("PID %d is not attached to this session", pid)));
+	check_rights(info);
+
+	/* Execute give SQL query */
+	BackgroundSessionExecuteSQL(info->session, sql);
+	info->result_count++;
+
+	PG_RETURN_VOID();
+}
+
+
+/*
+ * Retrieve the results of a background query previously launched in this
+ * session.
+ */
+Datum
+pg_background_result(PG_FUNCTION_ARGS)
+{
+	int32		pid = PG_GETARG_INT32(0);
+	FuncCallContext *funcctx;
+	pg_background_result_state *state;
+	TupleDesc   tupdesc;
+	BackgroundSessionResult *result;
+
+	/* First-time setup. */
+	if (SRF_IS_FIRSTCALL())
+	{
+		MemoryContext	oldcontext;
+		pg_background_worker_info *info;
+
+		funcctx = SRF_FIRSTCALL_INIT();
+		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+		/* See if we have a connection to the specified PID. */
+		if ((info = find_worker_info(pid)) == NULL)
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+					 errmsg("PID %d is not attached to this session", pid)));
+		check_rights(info);
+
+		/* Can't read results twice. */
+		if (info->result_count <= 0)
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("results for PID %d have already been consumed", pid)));
+		info->result_count--;
+
+		/* Set up tuple-descriptor based on column definition list. */
+		if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
+			ereport(ERROR,
+					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+					 errmsg("function returning record called in context "
+							"that cannot accept type record"),
+					 errhint("Try calling the function in the FROM clause "
+							 "using a column definition list.")));
+		result = BackgroundSessionFetchResult(info->session);
+		funcctx->tuple_desc = BlessTupleDesc(tupdesc);
+
+		if (result->tupdesc && tupdesc->natts != result->tupdesc->natts)
+			ereport(ERROR,
+					(errcode(ERRCODE_DATATYPE_MISMATCH),
+					 errmsg("remote query result rowtype does not match the specified FROM clause rowtype")));
+
+		/* Cache state that will be needed on every call. */
+		state = palloc0(sizeof(pg_background_result_state));
+		state->info = info;
+		state->result = result;
+
+		funcctx->user_fctx = state;
+		MemoryContextSwitchTo(oldcontext);
+	}
+
+	funcctx = SRF_PERCALL_SETUP();
+	tupdesc = funcctx->tuple_desc;
+	state = funcctx->user_fctx;
+	result = state->result;
+
+	if (result->tupdesc)
+	{
+		if (result->tuples != NIL)
+		{
+			HeapTuple tuple = (HeapTuple) linitial(result->tuples);
+			Datum *values = (Datum *) palloc(tupdesc->natts * sizeof(Datum));
+			bool *isnull = (bool *) palloc(tupdesc->natts * sizeof(bool));
+
+			/*
+			 * Perform conversion of a tuple according to column
+			 * defination list given in FROM clause
+			 */
+			heap_deform_tuple(tuple, result->tupdesc, values, isnull);
+			tuple = heap_form_tuple(tupdesc, values, isnull);
+			result->tuples = list_delete_first(result->tuples);
+
+			SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
+		}
+	}
+	else	/* If no data rows, return the command tags instead. */
+	{
+		if (tupdesc->natts != 1 || tupdesc->attrs[0]->atttypid != TEXTOID)
+			ereport(ERROR,
+					(errcode(ERRCODE_DATATYPE_MISMATCH),
+					 errmsg("remote query did not return a result set, but result rowtype is not a single text column")));
+
+		if (result->command != NULL)
+		{
+			bool	isnull = false;
+			Datum	value = PointerGetDatum(cstring_to_text(result->command));
+			HeapTuple	tuple = heap_form_tuple(tupdesc, &value, &isnull);
+
+			result->command = NULL;
+
+			SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
+		}
+	}
+
+	SRF_RETURN_DONE(funcctx);
+}
+
+/*
+ * Detach from the dynamic shared memory segment used for communication with
+ * a background worker.  This prevents the worker from stalling waiting for
+ * us to read its results.
+ */
+Datum
+pg_background_detach(PG_FUNCTION_ARGS)
+{
+	int32		pid = PG_GETARG_INT32(0);
+	pg_background_worker_info *info;
+
+	info = find_worker_info(pid);
+	if (info == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("PID %d is not attached to this session", pid)));
+
+	check_rights(info);
+	remove_worker_info(pid);
+	BackgroundSessionEnd(info->session);
+
+	PG_RETURN_VOID();
+}
+
+static void
+remove_worker_info(pid_t pid)
+{
+	bool	found;
+
+	/* Remove the hashtable entry. */
+	hash_search(worker_hash, (void *) &pid, HASH_REMOVE, &found);
+	if (!found)
+		elog(ERROR, "pg_background worker_hash table corrupted");
+}
+
+/*
+ * Find the background worker information for the worker with a given PID.
+ */
+static pg_background_worker_info *
+find_worker_info(pid_t pid)
+{
+	pg_background_worker_info *info = NULL;
+
+	if (worker_hash != NULL)
+		info = hash_search(worker_hash, (void *) &pid, HASH_FIND, NULL);
+
+	return info;
+}
+
+/*
+ * Save worker info.
+ */
+static void
+save_worker_info(pid_t pid, BackgroundSession *session)
+{
+	pg_background_worker_info *info;
+	Oid current_user_id;
+	int	sec_context;
+
+	/* If the hash table hasn't been set up yet, do that now. */
+	if (worker_hash == NULL)
+	{
+		HASHCTL	ctl;
+
+		ctl.keysize = sizeof(pid_t);
+		ctl.entrysize = sizeof(pg_background_worker_info);
+		worker_hash = hash_create("pg_background worker_hash", 8, &ctl,
+								  HASH_ELEM);
+	}
+
+	/* Get current authentication information. */
+	GetUserIdAndSecContext(&current_user_id, &sec_context);
+
+	/*
+	 * In the unlikely event that there's an older worker with this PID,
+	 * just detach it - unless it has a different user ID than the
+	 * currently-active one, in which case someone might be trying to pull
+	 * a fast one.  Let's kill the backend to make sure we don't break
+	 * anyone's expectations.
+	 */
+	if ((info = find_worker_info(pid)) != NULL)
+	{
+		if (current_user_id != info->current_user_id)
+			ereport(FATAL,
+				(errcode(ERRCODE_DUPLICATE_OBJECT),
+			 errmsg("background worker with PID \"%d\" already exists",
+						pid)));
+	}
+
+	/* Create a new entry for this worker. */
+	info = hash_search(worker_hash, (void *) &pid, HASH_ENTER, NULL);
+	info->session = session;
+	info->result_count = 0;
+	info->current_user_id = current_user_id;
+}
+
+/*
+ * Check whether the current user has rights to manipulate the background
+ * worker with the given PID.
+ */
+static void
+check_rights(pg_background_worker_info *info)
+{
+	Oid	current_user_id;
+	int	sec_context;
+
+	GetUserIdAndSecContext(&current_user_id, &sec_context);
+	if (!has_privs_of_role(current_user_id, info->current_user_id))
+		ereport(ERROR,
+				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+			 errmsg("permission denied for background worker with PID \"%d\"",
+						info->pid)));
+}
diff --git a/contrib/pg_background/pg_background.control b/contrib/pg_background/pg_background.control
new file mode 100644
index 0000000..733d0e1
--- /dev/null
+++ b/contrib/pg_background/pg_background.control
@@ -0,0 +1,4 @@
+comment = 'Run SQL queries in the background'
+default_version = '1.0'
+module_pathname = '$libdir/pg_background'
+relocatable = true
diff --git a/src/backend/tcop/bgsession.c b/src/backend/tcop/bgsession.c
index 2cc6438..143a244 100644
--- a/src/backend/tcop/bgsession.c
+++ b/src/backend/tcop/bgsession.c
@@ -115,9 +115,16 @@ static void invalid_protocol_message(char msgtype) pg_attribute_noreturn();
 BackgroundSession *
 BackgroundSessionStart(void)
 {
+	pid_t *pid = 0;
+
+	return BackgroundSessionNew(pid);
+}
+
+BackgroundSession *
+BackgroundSessionNew(pid_t *pid)
+{
 	ResourceOwner oldowner;
 	BackgroundWorker worker;
-	pid_t		pid;
 	BackgroundSession *session;
 	shm_toc_estimator e;
 	Size		segsize;
@@ -131,8 +138,11 @@ BackgroundSessionStart(void)
 	BgwHandleStatus bgwstatus;
 	StringInfoData msg;
 	char		msgtype;
+	MemoryContext   oldcontext;
 
+	oldcontext = MemoryContextSwitchTo(TopMemoryContext);
 	session = palloc(sizeof(*session));
+	MemoryContextSwitchTo(oldcontext);
 
 	session->resowner = ResourceOwnerCreate(NULL, "background session");
 
@@ -184,8 +194,10 @@ BackgroundSessionStart(void)
 	shm_toc_insert(toc, BGSESSION_KEY_RESPONSE_QUEUE, response_mq);
 	shm_mq_set_receiver(response_mq, MyProc);
 
+	oldcontext = MemoryContextSwitchTo(TopMemoryContext);
 	session->command_qh = shm_mq_attach(command_mq, seg, NULL);
 	session->response_qh = shm_mq_attach(response_mq, seg, NULL);
+	MemoryContextSwitchTo(oldcontext);
 
 	worker.bgw_flags =
 		BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
@@ -196,16 +208,18 @@ BackgroundSessionStart(void)
 	worker.bgw_main_arg = UInt32GetDatum(dsm_segment_handle(seg));
 	worker.bgw_notify_pid = MyProcPid;
 
+	oldcontext = MemoryContextSwitchTo(TopMemoryContext);
 	if (!RegisterDynamicBackgroundWorker(&worker, &session->worker_handle))
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_RESOURCES),
 				 errmsg("could not register background process"),
 				 errhint("You might need to increase max_worker_processes.")));
+	MemoryContextSwitchTo(oldcontext);
 
 	shm_mq_set_handle(session->command_qh, session->worker_handle);
 	shm_mq_set_handle(session->response_qh, session->worker_handle);
 
-	bgwstatus = WaitForBackgroundWorkerStartup(session->worker_handle, &pid);
+	bgwstatus = WaitForBackgroundWorkerStartup(session->worker_handle, pid);
 	if (bgwstatus != BGWH_STARTED)
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_RESOURCES),
@@ -256,24 +270,41 @@ BackgroundSessionEnd(BackgroundSession *session)
 	dsm_detach(session->seg);
 	ResourceOwnerRelease(session->resowner, RESOURCE_RELEASE_BEFORE_LOCKS, false, false);
 	ResourceOwnerDelete(session->resowner);
-	pfree(session);
+//TODO:	pfree(session);
 }
 
-
 BackgroundSessionResult *
 BackgroundSessionExecute(BackgroundSession *session, const char *sql)
 {
-	StringInfoData msg;
-	char		msgtype;
 	BackgroundSessionResult *result;
 
+	BackgroundSessionExecuteSQL(session, sql);
+	result = BackgroundSessionFetchResult(session);
+
+	return result;
+}
+
+void
+BackgroundSessionExecuteSQL(BackgroundSession *session, const char *sql)
+{
+	StringInfoData msg;
+
 	pq_redirect_to_shm_mq(session->seg, session->command_qh);
 	pq_beginmessage(&msg, 'Q');
 	pq_sendstring(&msg, sql);
 	pq_endmessage(&msg);
 	pq_stop_redirect_to_shm_mq();
+}
+
+BackgroundSessionResult *
+BackgroundSessionFetchResult(BackgroundSession *session)
+{
+	StringInfoData msg;
+	char		msgtype;
+	BackgroundSessionResult *result;
 
 	result = palloc0(sizeof(*result));
+	result->command = NULL;
 
 	do
 	{
@@ -322,7 +353,6 @@ BackgroundSessionExecute(BackgroundSession *session, const char *sql)
 	return result;
 }
 
-
 BackgroundSessionPreparedStatement *
 BackgroundSessionPrepare(BackgroundSession *session, const char *sql, int nargs,
 						 Oid argtypes[], const char *argnames[])
diff --git a/src/include/tcop/bgsession.h b/src/include/tcop/bgsession.h
index 70dad45..f38cb29 100644
--- a/src/include/tcop/bgsession.h
+++ b/src/include/tcop/bgsession.h
@@ -18,9 +18,11 @@ typedef struct BackgroundSessionResult
 } BackgroundSessionResult;
 
 BackgroundSession *BackgroundSessionStart(void);
+BackgroundSession *BackgroundSessionNew(pid_t *pid);
 void BackgroundSessionEnd(BackgroundSession *session);
 BackgroundSessionResult *BackgroundSessionExecute(BackgroundSession *session, const char *sql);
+void BackgroundSessionExecuteSQL(BackgroundSession *session, const char *sql);
+BackgroundSessionResult * BackgroundSessionFetchResult(BackgroundSession *session);
 BackgroundSessionPreparedStatement *BackgroundSessionPrepare(BackgroundSession *session, const char *sql, int nargs, Oid argtypes[], const char *argnames[]);
 BackgroundSessionResult *BackgroundSessionExecutePrepared(BackgroundSessionPreparedStatement *stmt, int nargs, Datum values[], bool nulls[]);
-
 #endif /* BGSESSION_H */

From f0fd108b415abf7bc425d6ca94e8f2f5bf3a827f Mon Sep 17 00:00:00 2001
From: Amul Sul <sulamul@gmail.com>
Date: Fri, 6 Jan 2017 18:13:13 +0530
Subject: [PATCH 1/2] Changes to Peter Eisentraut's bgsession v2 patch

---
 src/backend/tcop/bgsession.c     | 26 +++++++++-----------------
 src/include/tcop/bgsession.h     | 19 +++++++++++++++----
 src/pl/plpython/plpy_bgsession.c |  2 +-
 3 files changed, 25 insertions(+), 22 deletions(-)

diff --git a/src/backend/tcop/bgsession.c b/src/backend/tcop/bgsession.c
index 486819b..26dee53 100644
--- a/src/backend/tcop/bgsession.c
+++ b/src/backend/tcop/bgsession.c
@@ -60,14 +60,11 @@
 #include "nodes/pg_list.h"
 #include "pgstat.h"
 #include "postmaster/bgworker.h"
-#include "storage/dsm.h"
-#include "storage/shm_mq.h"
 #include "storage/shm_toc.h"
 #include "tcop/bgsession.h"
 #include "tcop/tcopprot.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
-#include "utils/resowner.h"
 
 /* Table-of-contents constants for our dynamic shared memory segment. */
 #define BGSESSION_MAGIC					0x50674267
@@ -89,16 +86,6 @@ typedef struct background_session_fixed_data
 	int sec_context;
 } background_session_fixed_data;
 
-struct BackgroundSession
-{
-	ResourceOwner resowner;
-	dsm_segment *seg;
-	BackgroundWorkerHandle *worker_handle;
-	shm_mq_handle *command_qh;
-	shm_mq_handle *response_qh;
-	int		transaction_status;
-};
-
 struct BackgroundSessionPreparedStatement
 {
 	BackgroundSession *session;
@@ -117,11 +104,10 @@ static void invalid_protocol_message(char msgtype) pg_attribute_noreturn();
 
 
 BackgroundSession *
-BackgroundSessionStart(void)
+BackgroundSessionStart(MemoryContext context)
 {
 	ResourceOwner oldowner;
 	BackgroundWorker worker;
-	pid_t		pid;
 	BackgroundSession *session;
 	shm_toc_estimator e;
 	Size		segsize;
@@ -135,8 +121,9 @@ BackgroundSessionStart(void)
 	BgwHandleStatus bgwstatus;
 	StringInfoData msg;
 	char		msgtype;
+	MemoryContext   oldcontext;
 
-	session = palloc(sizeof(*session));
+	session = MemoryContextAlloc(context, sizeof(*session));;
 
 	session->resowner = ResourceOwnerCreate(NULL, "background session");
 
@@ -188,8 +175,10 @@ BackgroundSessionStart(void)
 	shm_toc_insert(toc, BGSESSION_KEY_RESPONSE_QUEUE, response_mq);
 	shm_mq_set_receiver(response_mq, MyProc);
 
+	oldcontext = MemoryContextSwitchTo(context);
 	session->command_qh = shm_mq_attach(command_mq, seg, NULL);
 	session->response_qh = shm_mq_attach(response_mq, seg, NULL);
+	MemoryContextSwitchTo(oldcontext);
 
 	worker.bgw_flags =
 		BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
@@ -200,16 +189,19 @@ BackgroundSessionStart(void)
 	worker.bgw_main_arg = UInt32GetDatum(dsm_segment_handle(seg));
 	worker.bgw_notify_pid = MyProcPid;
 
+	oldcontext = MemoryContextSwitchTo(context);
 	if (!RegisterDynamicBackgroundWorker(&worker, &session->worker_handle))
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_RESOURCES),
 				 errmsg("could not register background process"),
 				 errhint("You might need to increase max_worker_processes.")));
+	MemoryContextSwitchTo(oldcontext);
 
 	shm_mq_set_handle(session->command_qh, session->worker_handle);
 	shm_mq_set_handle(session->response_qh, session->worker_handle);
 
-	bgwstatus = WaitForBackgroundWorkerStartup(session->worker_handle, &pid);
+	bgwstatus = WaitForBackgroundWorkerStartup(session->worker_handle,
+											   &session->pid);
 	if (bgwstatus != BGWH_STARTED)
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_RESOURCES),
diff --git a/src/include/tcop/bgsession.h b/src/include/tcop/bgsession.h
index 71415a6..65dfb56 100644
--- a/src/include/tcop/bgsession.h
+++ b/src/include/tcop/bgsession.h
@@ -3,13 +3,24 @@
 
 #include "access/tupdesc.h"
 #include "nodes/pg_list.h"
-
-struct BackgroundSession;
-typedef struct BackgroundSession BackgroundSession;
+#include "storage/dsm.h"
+#include "storage/shm_mq.h"
+#include "utils/resowner.h"
 
 struct BackgroundSessionPreparedStatement;
 typedef struct BackgroundSessionPreparedStatement BackgroundSessionPreparedStatement;
 
+typedef struct BackgroundSession
+{
+	pid_t pid;
+	ResourceOwner resowner;
+	dsm_segment *seg;
+	BackgroundWorkerHandle *worker_handle;
+	shm_mq_handle *command_qh;
+	shm_mq_handle *response_qh;
+	int		transaction_status;
+} BackgroundSession;
+
 typedef struct BackgroundSessionResult
 {
 	TupleDesc	tupdesc;
@@ -17,7 +28,7 @@ typedef struct BackgroundSessionResult
 	const char *command;
 } BackgroundSessionResult;
 
-BackgroundSession *BackgroundSessionStart(void);
+BackgroundSession *BackgroundSessionStart(MemoryContext oldcontext);
 void BackgroundSessionEnd(BackgroundSession *session);
 
 void BackgroundSessionSend(BackgroundSession *session, const char *sql);
diff --git a/src/pl/plpython/plpy_bgsession.c b/src/pl/plpython/plpy_bgsession.c
index 68f7207..7e6e906 100644
--- a/src/pl/plpython/plpy_bgsession.c
+++ b/src/pl/plpython/plpy_bgsession.c
@@ -110,7 +110,7 @@ PLyBackgroundSession_new(PyTypeObject *type, PyObject *args, PyObject *kw)
 	PyObject   *result = type->tp_alloc(type, 0);
 	PLyBackgroundSession_Object *bgsession = (PLyBackgroundSession_Object *) result;
 
-	bgsession->bgsession = BackgroundSessionStart();
+	bgsession->bgsession = BackgroundSessionStart(CurrentMemoryContext);
 
 	return result;
 }
-- 
2.6.2

From a5811a850ebc66b6b6267afe341c3929cbb57b17 Mon Sep 17 00:00:00 2001
From: Amul Sul <sulamul@gmail.com>
Date: Fri, 6 Jan 2017 18:13:37 +0530
Subject: [PATCH 2/2] pg_background as client of BackgroundSession-v1

---
 contrib/Makefile                                 |   1 +
 contrib/pg_background/Makefile                   |  20 ++
 contrib/pg_background/expected/pg_background.out |  75 +++++
 contrib/pg_background/pg_background--1.0.sql     |  37 +++
 contrib/pg_background/pg_background.c            | 397 +++++++++++++++++++++++
 contrib/pg_background/pg_background.control      |   4 +
 contrib/pg_background/sql/pg_background.sql      |  30 ++
 7 files changed, 564 insertions(+)
 create mode 100644 contrib/pg_background/Makefile
 create mode 100644 contrib/pg_background/expected/pg_background.out
 create mode 100644 contrib/pg_background/pg_background--1.0.sql
 create mode 100644 contrib/pg_background/pg_background.c
 create mode 100644 contrib/pg_background/pg_background.control
 create mode 100644 contrib/pg_background/sql/pg_background.sql

diff --git a/contrib/Makefile b/contrib/Makefile
index 25263c0..04ec28a 100644
--- a/contrib/Makefile
+++ b/contrib/Makefile
@@ -29,6 +29,7 @@ SUBDIRS = \
 		oid2name	\
 		pageinspect	\
 		passwordcheck	\
+		pg_background \
 		pg_buffercache	\
 		pg_freespacemap \
 		pg_prewarm	\
diff --git a/contrib/pg_background/Makefile b/contrib/pg_background/Makefile
new file mode 100644
index 0000000..085fbff
--- /dev/null
+++ b/contrib/pg_background/Makefile
@@ -0,0 +1,20 @@
+# contrib/pg_background/Makefile
+
+MODULE_big = pg_background
+OBJS = pg_background.o
+
+EXTENSION = pg_background
+DATA = pg_background--1.0.sql
+
+REGRESS = pg_background
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = contrib/pg_background
+top_builddir = ../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/contrib/pg_background/expected/pg_background.out b/contrib/pg_background/expected/pg_background.out
new file mode 100644
index 0000000..23679e6
--- /dev/null
+++ b/contrib/pg_background/expected/pg_background.out
@@ -0,0 +1,75 @@
+CREATE EXTENSION pg_background;
+--launch 6 workers which wait  .0, .1, .2, .3, .4, .5 seconds respectively
+CREATE TABLE input AS
+	SELECT pg_background_start() pid, x, row_number() OVER (ORDER BY x) n
+	FROM generate_series(0,.5,0.1) x
+	ORDER BY x DESC;
+CREATE TABLE output(place int,value float);
+--display active background workers
+SELECT count(*) as active_background_workers FROM pg_background_show();
+ active_background_workers 
+---------------------------
+                         6
+(1 row)
+
+--run sql query
+SELECT n as worker_num
+FROM input JOIN LATERAL pg_background_run(input.pid,
+										  format($$
+												 SELECT pg_sleep(%s)::text;
+												 INSERT INTO output VALUES (%s, %s);
+												 $$,
+												 x, n, x
+												)
+										 ) ON (true);
+ worker_num 
+------------
+          6
+          5
+          4
+          3
+          2
+          1
+(6 rows)
+
+--wait until everyone finishes
+SELECT n as worker_num
+FROM input JOIN LATERAL pg_background_result(input.pid) AS (x TEXT) ON (true);
+ worker_num 
+------------
+          6
+          5
+          4
+          3
+          2
+          1
+(6 rows)
+
+--output results
+SELECT * FROM output ORDER BY place;
+ place | value 
+-------+-------
+     1 |     0
+     2 |   0.1
+     3 |   0.2
+     4 |   0.3
+     5 |   0.4
+     6 |   0.5
+(6 rows)
+
+--cleanup
+SELECT n as worker_num
+FROM input JOIN LATERAL pg_background_close(input.pid) ON (true);
+ worker_num 
+------------
+          6
+          5
+          4
+          3
+          2
+          1
+(6 rows)
+
+DROP TABLE input;
+DROP TABLE output;
+DROP EXTENSION pg_background;
diff --git a/contrib/pg_background/pg_background--1.0.sql b/contrib/pg_background/pg_background--1.0.sql
new file mode 100644
index 0000000..62a9d18
--- /dev/null
+++ b/contrib/pg_background/pg_background--1.0.sql
@@ -0,0 +1,37 @@
+/* contrib/pg_background/pg_background--1.0.sql */
+
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "CREATE EXTENSION pg_background" to load this file. \quit
+
+CREATE FUNCTION pg_background_show()
+    RETURNS TABLE(background_worker_id pg_catalog.int4,
+				  num_of_results_waiting_to_read pg_catalog.int4) STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION pg_background_start()
+    RETURNS pg_catalog.int4 STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION pg_background_run(pid pg_catalog.int4, sql pg_catalog.text)
+    RETURNS pg_catalog.void STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION pg_background_result(pid pg_catalog.int4)
+    RETURNS SETOF pg_catalog.record STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION pg_background_close(pid pg_catalog.int4)
+    RETURNS pg_catalog.void STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
+
+REVOKE ALL ON FUNCTION pg_background_show()
+	FROM public;
+REVOKE ALL ON FUNCTION pg_background_start()
+	FROM public;
+REVOKE ALL ON FUNCTION pg_background_run(pid pg_catalog.int4,
+										 sql pg_catalog.text)
+	FROM public;
+REVOKE ALL ON FUNCTION pg_background_result(pg_catalog.int4)
+	FROM public;
+REVOKE ALL ON FUNCTION pg_background_close(pg_catalog.int4)
+	FROM public;
diff --git a/contrib/pg_background/pg_background.c b/contrib/pg_background/pg_background.c
new file mode 100644
index 0000000..e5a5b35
--- /dev/null
+++ b/contrib/pg_background/pg_background.c
@@ -0,0 +1,397 @@
+/*--------------------------------------------------------------------------
+ *
+ * pg_background.c
+ *		Run SQL commands using a background worker.
+ *
+ * Copyright (C) 2017, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		contrib/pg_background/pg_background.c
+ *
+ * -------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/htup_details.h"
+#include "catalog/pg_type.h"
+#include "funcapi.h"
+#include "miscadmin.h"
+#include "tcop/bgsession.h"
+#include "utils/acl.h"
+#include "utils/builtins.h"
+#include "utils/memutils.h"
+
+/*  Private state maintained by the launching backend for IPC. */
+typedef struct pg_background_worker_info
+{
+	int32				pid;
+	Oid					current_user_id;
+	BackgroundSession  *session;
+	uint32				result_count;
+} pg_background_worker_info;
+
+/* Private state maintained across calls to pg_background_result. */
+typedef struct pg_background_result_state
+{
+	pg_background_worker_info  *info;
+	BackgroundSessionResult	   *result;
+} pg_background_result_state;
+
+static HTAB *worker_hash = NULL;
+
+static void remove_worker_info(int32 pid);
+static pg_background_worker_info *find_worker_info(int32 pid);
+static void save_worker_info(int32 pid, BackgroundSession *session);
+static void check_rights(pg_background_worker_info *info);
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(pg_background_show);
+PG_FUNCTION_INFO_V1(pg_background_start);
+PG_FUNCTION_INFO_V1(pg_background_run);
+PG_FUNCTION_INFO_V1(pg_background_result);
+PG_FUNCTION_INFO_V1(pg_background_close);
+
+/*
+ * Display the list of background worker previously launched in this session.
+ */
+Datum
+pg_background_show(PG_FUNCTION_ARGS)
+{
+	FuncCallContext			   *funcctx;
+	HASH_SEQ_STATUS			   *hash_seq;
+	TupleDesc					tupdesc;
+	pg_background_worker_info  *info;
+
+	/* First-time setup. */
+	if (SRF_IS_FIRSTCALL())
+	{
+		MemoryContext	oldcontext;
+
+		funcctx = SRF_FIRSTCALL_INIT();
+		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+		/* Construct a tuple descriptor for the result rows. */
+		tupdesc = CreateTemplateTupleDesc(2, false);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 1, "pid",
+						   INT4OID, -1, 0);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 2, "results",
+						   INT4OID, -1, 0);
+
+		funcctx->tuple_desc = BlessTupleDesc(tupdesc);
+
+		if (worker_hash)
+		{
+			hash_seq = palloc(sizeof(HASH_SEQ_STATUS));
+
+			hash_seq_init(hash_seq, worker_hash);
+			funcctx->user_fctx = hash_seq;
+		}
+
+		MemoryContextSwitchTo(oldcontext);
+	}
+
+	funcctx = SRF_PERCALL_SETUP();
+	hash_seq = (HASH_SEQ_STATUS *) funcctx->user_fctx;
+
+	while (hash_seq && (info = hash_seq_search(hash_seq)) != NULL)
+	{
+		Datum		values[2];
+		bool		nulls[2];
+		HeapTuple	tuple;
+
+		values[0] = Int32GetDatum(info->pid);
+		nulls[0] = false;
+		values[1] = Int32GetDatum(info->result_count);
+		nulls[1] = false;
+
+		/* Build and return the tuple. */
+		tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
+		SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
+	}
+
+	SRF_RETURN_DONE(funcctx);
+}
+
+/*
+ * Start a dynamic background worker.
+ */
+Datum
+pg_background_start(PG_FUNCTION_ARGS)
+{
+	BackgroundSession  *session;
+	int32				pid;
+
+	session = BackgroundSessionStart(TopMemoryContext);
+	pid = session->pid;
+
+	/* Save worker info */
+	save_worker_info(pid, session);
+
+	/*  Return the worker's PID. */
+	PG_RETURN_INT32(pid);
+}
+
+/*
+ * Run a user-specified SQL command.
+ */
+Datum
+pg_background_run(PG_FUNCTION_ARGS)
+{
+	char	   *sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
+	int32		pid = PG_GETARG_INT32(0);
+	pg_background_worker_info *info;
+
+	/* See if we have a connection to the specified PID. */
+	if ((info = find_worker_info(pid)) == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("PID %d is not attached to this session", pid)));
+	check_rights(info);
+
+	/* Execute give SQL query */
+	BackgroundSessionSend(info->session, sql);
+	info->result_count++;
+
+	PG_RETURN_VOID();
+}
+
+/*
+ * Retrieve the results of a background query previously launched in this
+ * session.
+ */
+Datum
+pg_background_result(PG_FUNCTION_ARGS)
+{
+	int32						pid = PG_GETARG_INT32(0);
+	FuncCallContext			   *funcctx;
+	pg_background_result_state *state;
+	TupleDesc					tupdesc;
+	BackgroundSessionResult	   *result;
+
+	/* First-time setup. */
+	if (SRF_IS_FIRSTCALL())
+	{
+		MemoryContext				oldcontext;
+		pg_background_worker_info  *info;
+
+		funcctx = SRF_FIRSTCALL_INIT();
+		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+		/* See if we have a connection to the specified PID. */
+		if ((info = find_worker_info(pid)) == NULL)
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+					 errmsg("PID %d is not attached to this session", pid)));
+		check_rights(info);
+
+		/* Can't read results twice. */
+		if (info->result_count <= 0)
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("results for PID %d have already been consumed", pid)));
+		info->result_count--;
+
+		/* Set up tuple-descriptor based on column definition list. */
+		if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
+			ereport(ERROR,
+					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+					 errmsg("function returning record called in context "
+							"that cannot accept type record"),
+					 errhint("Try calling the function in the FROM clause "
+							 "using a column definition list.")));
+		result = BackgroundSessionGetResult(info->session);
+		funcctx->tuple_desc = BlessTupleDesc(tupdesc);
+
+		if (result->tupdesc)
+		{
+			bool datatype_mismatch = false;
+
+			if (tupdesc->natts != result->tupdesc->natts)
+				datatype_mismatch = true;
+			else
+			{
+				int i;
+				for (i = 0; i < tupdesc->natts; ++i)
+					if(tupdesc->attrs[i]->atttypid !=
+					   result->tupdesc->attrs[i]->atttypid)
+					{
+						datatype_mismatch = true;
+						break;
+					}
+			}
+
+			if (datatype_mismatch)
+				ereport(ERROR,
+						(errcode(ERRCODE_DATATYPE_MISMATCH),
+						 errmsg("remote query result rowtype does not match the specified FROM clause rowtype")));
+		}
+
+		/* Cache state that will be needed on every call. */
+		state = palloc0(sizeof(pg_background_result_state));
+		state->info = info;
+		state->result = result;
+
+		funcctx->user_fctx = state;
+		MemoryContextSwitchTo(oldcontext);
+	}
+
+	funcctx = SRF_PERCALL_SETUP();
+	tupdesc = funcctx->tuple_desc;
+	state = funcctx->user_fctx;
+	result = state->result;
+
+	if (result->tupdesc)
+	{
+		if (result->tuples != NIL && funcctx->call_cntr > 0)
+			result->tuples = list_delete_first(result->tuples);
+
+		if (result->tuples != NIL)
+		{
+			HeapTuple tuple = (HeapTuple) linitial(result->tuples);
+
+			/*
+			 * Set the tuple type ID information fields correctly because
+			 * BackgroundSessionFetchResult returns it as an anonymous record
+			 * type.
+			 */
+			HeapTupleHeaderSetTypeId(tuple->t_data, tupdesc->tdtypeid);
+			HeapTupleHeaderSetTypMod(tuple->t_data, tupdesc->tdtypmod);
+
+			SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
+		}
+	}
+	else	/* If no data rows, return the command tags instead. */
+	{
+		if (tupdesc->natts != 1 || tupdesc->attrs[0]->atttypid != TEXTOID)
+			ereport(ERROR,
+					(errcode(ERRCODE_DATATYPE_MISMATCH),
+					 errmsg("remote query did not return a result set, but result rowtype is not a single text column")));
+
+		if (result->command != NULL)
+		{
+			bool	isnull = false;
+			Datum	value = PointerGetDatum(cstring_to_text(result->command));
+			HeapTuple	tuple = heap_form_tuple(tupdesc, &value, &isnull);
+
+			result->command = NULL;
+
+			SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
+		}
+	}
+
+	SRF_RETURN_DONE(funcctx);
+}
+
+/*
+ * End background session and remove hashtable entry.
+ */
+Datum
+pg_background_close(PG_FUNCTION_ARGS)
+{
+	int32						pid = PG_GETARG_INT32(0);
+	pg_background_worker_info  *info;
+
+	info = find_worker_info(pid);
+	if (info == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("PID %d is not attached to this session", pid)));
+
+	check_rights(info);
+	remove_worker_info(pid);
+	BackgroundSessionEnd(info->session);
+
+	PG_RETURN_VOID();
+}
+
+static void
+remove_worker_info(int32 pid)
+{
+	bool	found;
+
+	/* Remove the hashtable entry. */
+	hash_search(worker_hash, (void *) &pid, HASH_REMOVE, &found);
+	if (!found)
+		elog(ERROR, "pg_background worker_hash table corrupted");
+}
+
+/*
+ * Find the background worker information for the worker with a given PID.
+ */
+static pg_background_worker_info *
+find_worker_info(int32 pid)
+{
+	pg_background_worker_info *info = NULL;
+
+	if (worker_hash != NULL)
+		info = hash_search(worker_hash, (void *) &pid, HASH_FIND, NULL);
+
+	return info;
+}
+
+/*
+ * Save worker info.
+ */
+static void
+save_worker_info(int32 pid, BackgroundSession *session)
+{
+	pg_background_worker_info  *info;
+	Oid							current_user_id;
+	int							sec_context;
+
+	/* If the hash table hasn't been set up yet, do that now. */
+	if (worker_hash == NULL)
+	{
+		HASHCTL	ctl;
+
+		ctl.keysize = sizeof(int32);
+		ctl.entrysize = sizeof(pg_background_worker_info);
+		worker_hash = hash_create("pg_background worker_hash", 8, &ctl,
+								  HASH_ELEM);
+	}
+
+	/* Get current authentication information. */
+	GetUserIdAndSecContext(&current_user_id, &sec_context);
+
+	/*
+	 * In the unlikely event that there's an older worker with this PID,
+	 * just detach it - unless it has a different user ID than the
+	 * currently-active one, in which case someone might be trying to pull
+	 * a fast one.  Let's kill the backend to make sure we don't break
+	 * anyone's expectations.
+	 */
+	if ((info = find_worker_info(pid)) != NULL)
+	{
+		if (current_user_id != info->current_user_id)
+			ereport(FATAL,
+				(errcode(ERRCODE_DUPLICATE_OBJECT),
+			 errmsg("background worker with PID \"%d\" already exists",
+						pid)));
+	}
+
+	/* Create a new entry for this worker. */
+	info = hash_search(worker_hash, (void *) &pid, HASH_ENTER, NULL);
+	info->session = session;
+	info->result_count = 0;
+	info->current_user_id = current_user_id;
+}
+
+/*
+ * Check whether the current user has rights to manipulate the background
+ * worker with the given PID.
+ */
+static void
+check_rights(pg_background_worker_info *info)
+{
+	Oid	current_user_id;
+	int	sec_context;
+
+	GetUserIdAndSecContext(&current_user_id, &sec_context);
+	if (!has_privs_of_role(current_user_id, info->current_user_id))
+		ereport(ERROR,
+				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+			 errmsg("permission denied for background worker with PID \"%d\"",
+						info->pid)));
+}
diff --git a/contrib/pg_background/pg_background.control b/contrib/pg_background/pg_background.control
new file mode 100644
index 0000000..733d0e1
--- /dev/null
+++ b/contrib/pg_background/pg_background.control
@@ -0,0 +1,4 @@
+comment = 'Run SQL queries in the background'
+default_version = '1.0'
+module_pathname = '$libdir/pg_background'
+relocatable = true
diff --git a/contrib/pg_background/sql/pg_background.sql b/contrib/pg_background/sql/pg_background.sql
new file mode 100644
index 0000000..be49dd5
--- /dev/null
+++ b/contrib/pg_background/sql/pg_background.sql
@@ -0,0 +1,30 @@
+CREATE EXTENSION pg_background;
+--launch 6 workers which wait  .0, .1, .2, .3, .4, .5 seconds respectively
+CREATE TABLE input AS
+	SELECT pg_background_start() pid, x, row_number() OVER (ORDER BY x) n
+	FROM generate_series(0,.5,0.1) x
+	ORDER BY x DESC;
+CREATE TABLE output(place int,value float);
+--display active background workers
+SELECT count(*) as active_background_workers FROM pg_background_show();
+--run sql query
+SELECT n as worker_num
+FROM input JOIN LATERAL pg_background_run(input.pid,
+										  format($$
+												 SELECT pg_sleep(%s)::text;
+												 INSERT INTO output VALUES (%s, %s);
+												 $$,
+												 x, n, x
+												)
+										 ) ON (true);
+--wait until everyone finishes
+SELECT n as worker_num
+FROM input JOIN LATERAL pg_background_result(input.pid) AS (x TEXT) ON (true);
+--output results
+SELECT * FROM output ORDER BY place;
+--cleanup
+SELECT n as worker_num
+FROM input JOIN LATERAL pg_background_close(input.pid) ON (true);
+DROP TABLE input;
+DROP TABLE output;
+DROP EXTENSION pg_background;
-- 
2.6.2

From fcf10d1741a236012c0bab327f535fc5277071d9 Mon Sep 17 00:00:00 2001
From: Amul Sul <sulamul@gmail.com>
Date: Sat, 7 Jan 2017 18:51:51 +0530
Subject: [PATCH 1/2] Changes to Peter Eisentraut's bgsession v2 patch

---
 src/backend/tcop/bgsession.c | 17 ++---------------
 src/include/tcop/bgsession.h | 17 ++++++++++++++---
 2 files changed, 16 insertions(+), 18 deletions(-)

diff --git a/src/backend/tcop/bgsession.c b/src/backend/tcop/bgsession.c
index 486819b..510e740 100644
--- a/src/backend/tcop/bgsession.c
+++ b/src/backend/tcop/bgsession.c
@@ -60,14 +60,11 @@
 #include "nodes/pg_list.h"
 #include "pgstat.h"
 #include "postmaster/bgworker.h"
-#include "storage/dsm.h"
-#include "storage/shm_mq.h"
 #include "storage/shm_toc.h"
 #include "tcop/bgsession.h"
 #include "tcop/tcopprot.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
-#include "utils/resowner.h"
 
 /* Table-of-contents constants for our dynamic shared memory segment. */
 #define BGSESSION_MAGIC					0x50674267
@@ -89,16 +86,6 @@ typedef struct background_session_fixed_data
 	int sec_context;
 } background_session_fixed_data;
 
-struct BackgroundSession
-{
-	ResourceOwner resowner;
-	dsm_segment *seg;
-	BackgroundWorkerHandle *worker_handle;
-	shm_mq_handle *command_qh;
-	shm_mq_handle *response_qh;
-	int		transaction_status;
-};
-
 struct BackgroundSessionPreparedStatement
 {
 	BackgroundSession *session;
@@ -121,7 +108,6 @@ BackgroundSessionStart(void)
 {
 	ResourceOwner oldowner;
 	BackgroundWorker worker;
-	pid_t		pid;
 	BackgroundSession *session;
 	shm_toc_estimator e;
 	Size		segsize;
@@ -209,7 +195,8 @@ BackgroundSessionStart(void)
 	shm_mq_set_handle(session->command_qh, session->worker_handle);
 	shm_mq_set_handle(session->response_qh, session->worker_handle);
 
-	bgwstatus = WaitForBackgroundWorkerStartup(session->worker_handle, &pid);
+	bgwstatus = WaitForBackgroundWorkerStartup(session->worker_handle,
+											   &session->pid);
 	if (bgwstatus != BGWH_STARTED)
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_RESOURCES),
diff --git a/src/include/tcop/bgsession.h b/src/include/tcop/bgsession.h
index 71415a6..008fa28 100644
--- a/src/include/tcop/bgsession.h
+++ b/src/include/tcop/bgsession.h
@@ -3,13 +3,24 @@
 
 #include "access/tupdesc.h"
 #include "nodes/pg_list.h"
-
-struct BackgroundSession;
-typedef struct BackgroundSession BackgroundSession;
+#include "storage/dsm.h"
+#include "storage/shm_mq.h"
+#include "utils/resowner.h"
 
 struct BackgroundSessionPreparedStatement;
 typedef struct BackgroundSessionPreparedStatement BackgroundSessionPreparedStatement;
 
+typedef struct BackgroundSession
+{
+	pid_t pid;
+	ResourceOwner resowner;
+	dsm_segment *seg;
+	BackgroundWorkerHandle *worker_handle;
+	shm_mq_handle *command_qh;
+	shm_mq_handle *response_qh;
+	int		transaction_status;
+} BackgroundSession;
+
 typedef struct BackgroundSessionResult
 {
 	TupleDesc	tupdesc;
-- 
2.6.2

From e72e4fa60d9f489c33411f83e9e34dc8b202474a Mon Sep 17 00:00:00 2001
From: Amul Sul <sulamul@gmail.com>
Date: Sat, 7 Jan 2017 19:20:07 +0530
Subject: [PATCH 2/2] pg_background as client of BackgroundSession-v2

---
 contrib/Makefile                                 |   1 +
 contrib/pg_background/Makefile                   |  20 ++
 contrib/pg_background/expected/pg_background.out |  75 +++++
 contrib/pg_background/pg_background--1.0.sql     |  37 +++
 contrib/pg_background/pg_background.c            | 400 +++++++++++++++++++++++
 contrib/pg_background/pg_background.control      |   4 +
 contrib/pg_background/sql/pg_background.sql      |  30 ++
 7 files changed, 567 insertions(+)
 create mode 100644 contrib/pg_background/Makefile
 create mode 100644 contrib/pg_background/expected/pg_background.out
 create mode 100644 contrib/pg_background/pg_background--1.0.sql
 create mode 100644 contrib/pg_background/pg_background.c
 create mode 100644 contrib/pg_background/pg_background.control
 create mode 100644 contrib/pg_background/sql/pg_background.sql

diff --git a/contrib/Makefile b/contrib/Makefile
index 25263c0..04ec28a 100644
--- a/contrib/Makefile
+++ b/contrib/Makefile
@@ -29,6 +29,7 @@ SUBDIRS = \
 		oid2name	\
 		pageinspect	\
 		passwordcheck	\
+		pg_background \
 		pg_buffercache	\
 		pg_freespacemap \
 		pg_prewarm	\
diff --git a/contrib/pg_background/Makefile b/contrib/pg_background/Makefile
new file mode 100644
index 0000000..085fbff
--- /dev/null
+++ b/contrib/pg_background/Makefile
@@ -0,0 +1,20 @@
+# contrib/pg_background/Makefile
+
+MODULE_big = pg_background
+OBJS = pg_background.o
+
+EXTENSION = pg_background
+DATA = pg_background--1.0.sql
+
+REGRESS = pg_background
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = contrib/pg_background
+top_builddir = ../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/contrib/pg_background/expected/pg_background.out b/contrib/pg_background/expected/pg_background.out
new file mode 100644
index 0000000..23679e6
--- /dev/null
+++ b/contrib/pg_background/expected/pg_background.out
@@ -0,0 +1,75 @@
+CREATE EXTENSION pg_background;
+--launch 6 workers which wait  .0, .1, .2, .3, .4, .5 seconds respectively
+CREATE TABLE input AS
+	SELECT pg_background_start() pid, x, row_number() OVER (ORDER BY x) n
+	FROM generate_series(0,.5,0.1) x
+	ORDER BY x DESC;
+CREATE TABLE output(place int,value float);
+--display active background workers
+SELECT count(*) as active_background_workers FROM pg_background_show();
+ active_background_workers 
+---------------------------
+                         6
+(1 row)
+
+--run sql query
+SELECT n as worker_num
+FROM input JOIN LATERAL pg_background_run(input.pid,
+										  format($$
+												 SELECT pg_sleep(%s)::text;
+												 INSERT INTO output VALUES (%s, %s);
+												 $$,
+												 x, n, x
+												)
+										 ) ON (true);
+ worker_num 
+------------
+          6
+          5
+          4
+          3
+          2
+          1
+(6 rows)
+
+--wait until everyone finishes
+SELECT n as worker_num
+FROM input JOIN LATERAL pg_background_result(input.pid) AS (x TEXT) ON (true);
+ worker_num 
+------------
+          6
+          5
+          4
+          3
+          2
+          1
+(6 rows)
+
+--output results
+SELECT * FROM output ORDER BY place;
+ place | value 
+-------+-------
+     1 |     0
+     2 |   0.1
+     3 |   0.2
+     4 |   0.3
+     5 |   0.4
+     6 |   0.5
+(6 rows)
+
+--cleanup
+SELECT n as worker_num
+FROM input JOIN LATERAL pg_background_close(input.pid) ON (true);
+ worker_num 
+------------
+          6
+          5
+          4
+          3
+          2
+          1
+(6 rows)
+
+DROP TABLE input;
+DROP TABLE output;
+DROP EXTENSION pg_background;
diff --git a/contrib/pg_background/pg_background--1.0.sql b/contrib/pg_background/pg_background--1.0.sql
new file mode 100644
index 0000000..62a9d18
--- /dev/null
+++ b/contrib/pg_background/pg_background--1.0.sql
@@ -0,0 +1,37 @@
+/* contrib/pg_background/pg_background--1.0.sql */
+
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "CREATE EXTENSION pg_background" to load this file. \quit
+
+CREATE FUNCTION pg_background_show()
+    RETURNS TABLE(background_worker_id pg_catalog.int4,
+				  num_of_results_waiting_to_read pg_catalog.int4) STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION pg_background_start()
+    RETURNS pg_catalog.int4 STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION pg_background_run(pid pg_catalog.int4, sql pg_catalog.text)
+    RETURNS pg_catalog.void STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION pg_background_result(pid pg_catalog.int4)
+    RETURNS SETOF pg_catalog.record STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION pg_background_close(pid pg_catalog.int4)
+    RETURNS pg_catalog.void STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
+
+REVOKE ALL ON FUNCTION pg_background_show()
+	FROM public;
+REVOKE ALL ON FUNCTION pg_background_start()
+	FROM public;
+REVOKE ALL ON FUNCTION pg_background_run(pid pg_catalog.int4,
+										 sql pg_catalog.text)
+	FROM public;
+REVOKE ALL ON FUNCTION pg_background_result(pg_catalog.int4)
+	FROM public;
+REVOKE ALL ON FUNCTION pg_background_close(pg_catalog.int4)
+	FROM public;
diff --git a/contrib/pg_background/pg_background.c b/contrib/pg_background/pg_background.c
new file mode 100644
index 0000000..d8aefab
--- /dev/null
+++ b/contrib/pg_background/pg_background.c
@@ -0,0 +1,400 @@
+/*--------------------------------------------------------------------------
+ *
+ * pg_background.c
+ *		Run SQL commands using a background worker.
+ *
+ * Copyright (C) 2017, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		contrib/pg_background/pg_background.c
+ *
+ * -------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/htup_details.h"
+#include "catalog/pg_type.h"
+#include "funcapi.h"
+#include "miscadmin.h"
+#include "tcop/bgsession.h"
+#include "utils/acl.h"
+#include "utils/builtins.h"
+#include "utils/memutils.h"
+
+/*  Private state maintained by the launching backend for IPC. */
+typedef struct pg_background_worker_info
+{
+	int32				pid;
+	Oid					current_user_id;
+	BackgroundSession  *session;
+	uint32				result_count;
+} pg_background_worker_info;
+
+/* Private state maintained across calls to pg_background_result. */
+typedef struct pg_background_result_state
+{
+	pg_background_worker_info  *info;
+	BackgroundSessionResult	   *result;
+} pg_background_result_state;
+
+static HTAB *worker_hash = NULL;
+
+static void remove_worker_info(int32 pid);
+static pg_background_worker_info *find_worker_info(int32 pid);
+static void save_worker_info(int32 pid, BackgroundSession *session);
+static void check_rights(pg_background_worker_info *info);
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(pg_background_show);
+PG_FUNCTION_INFO_V1(pg_background_start);
+PG_FUNCTION_INFO_V1(pg_background_run);
+PG_FUNCTION_INFO_V1(pg_background_result);
+PG_FUNCTION_INFO_V1(pg_background_close);
+
+/*
+ * Display the list of background worker previously launched in this session.
+ */
+Datum
+pg_background_show(PG_FUNCTION_ARGS)
+{
+	FuncCallContext			   *funcctx;
+	HASH_SEQ_STATUS			   *hash_seq;
+	TupleDesc					tupdesc;
+	pg_background_worker_info  *info;
+
+	/* First-time setup. */
+	if (SRF_IS_FIRSTCALL())
+	{
+		MemoryContext	oldcontext;
+
+		funcctx = SRF_FIRSTCALL_INIT();
+		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+		/* Construct a tuple descriptor for the result rows. */
+		tupdesc = CreateTemplateTupleDesc(2, false);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 1, "pid",
+						   INT4OID, -1, 0);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 2, "results",
+						   INT4OID, -1, 0);
+
+		funcctx->tuple_desc = BlessTupleDesc(tupdesc);
+
+		if (worker_hash)
+		{
+			hash_seq = palloc(sizeof(HASH_SEQ_STATUS));
+
+			hash_seq_init(hash_seq, worker_hash);
+			funcctx->user_fctx = hash_seq;
+		}
+
+		MemoryContextSwitchTo(oldcontext);
+	}
+
+	funcctx = SRF_PERCALL_SETUP();
+	hash_seq = (HASH_SEQ_STATUS *) funcctx->user_fctx;
+
+	while (hash_seq && (info = hash_seq_search(hash_seq)) != NULL)
+	{
+		Datum		values[2];
+		bool		nulls[2];
+		HeapTuple	tuple;
+
+		values[0] = Int32GetDatum(info->pid);
+		nulls[0] = false;
+		values[1] = Int32GetDatum(info->result_count);
+		nulls[1] = false;
+
+		/* Build and return the tuple. */
+		tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
+		SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
+	}
+
+	SRF_RETURN_DONE(funcctx);
+}
+
+/*
+ * Start a dynamic background worker.
+ */
+Datum
+pg_background_start(PG_FUNCTION_ARGS)
+{
+	BackgroundSession  *session;
+	int32				pid;
+	MemoryContext		oldcontext;
+
+	oldcontext = MemoryContextSwitchTo(TopMemoryContext);
+	session = BackgroundSessionStart();
+	MemoryContextSwitchTo(oldcontext);
+	pid = session->pid;
+
+	/* Save worker info */
+	save_worker_info(pid, session);
+
+	/*  Return the worker's PID. */
+	PG_RETURN_INT32(pid);
+}
+
+/*
+ * Run a user-specified SQL command.
+ */
+Datum
+pg_background_run(PG_FUNCTION_ARGS)
+{
+	char	   *sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
+	int32		pid = PG_GETARG_INT32(0);
+	pg_background_worker_info *info;
+
+	/* See if we have a connection to the specified PID. */
+	if ((info = find_worker_info(pid)) == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("PID %d is not attached to this session", pid)));
+	check_rights(info);
+
+	/* Execute give SQL query */
+	BackgroundSessionSend(info->session, sql);
+	info->result_count++;
+
+	PG_RETURN_VOID();
+}
+
+/*
+ * Retrieve the results of a background query previously launched in this
+ * session.
+ */
+Datum
+pg_background_result(PG_FUNCTION_ARGS)
+{
+	int32						pid = PG_GETARG_INT32(0);
+	FuncCallContext			   *funcctx;
+	pg_background_result_state *state;
+	TupleDesc					tupdesc;
+	BackgroundSessionResult	   *result;
+
+	/* First-time setup. */
+	if (SRF_IS_FIRSTCALL())
+	{
+		MemoryContext				oldcontext;
+		pg_background_worker_info  *info;
+
+		funcctx = SRF_FIRSTCALL_INIT();
+		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+		/* See if we have a connection to the specified PID. */
+		if ((info = find_worker_info(pid)) == NULL)
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+					 errmsg("PID %d is not attached to this session", pid)));
+		check_rights(info);
+
+		/* Can't read results twice. */
+		if (info->result_count <= 0)
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("results for PID %d have already been consumed", pid)));
+		info->result_count--;
+
+		/* Set up tuple-descriptor based on column definition list. */
+		if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
+			ereport(ERROR,
+					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+					 errmsg("function returning record called in context "
+							"that cannot accept type record"),
+					 errhint("Try calling the function in the FROM clause "
+							 "using a column definition list.")));
+		result = BackgroundSessionGetResult(info->session);
+		funcctx->tuple_desc = BlessTupleDesc(tupdesc);
+
+		if (result->tupdesc)
+		{
+			bool datatype_mismatch = false;
+
+			if (tupdesc->natts != result->tupdesc->natts)
+				datatype_mismatch = true;
+			else
+			{
+				int i;
+				for (i = 0; i < tupdesc->natts; ++i)
+					if(tupdesc->attrs[i]->atttypid !=
+					   result->tupdesc->attrs[i]->atttypid)
+					{
+						datatype_mismatch = true;
+						break;
+					}
+			}
+
+			if (datatype_mismatch)
+				ereport(ERROR,
+						(errcode(ERRCODE_DATATYPE_MISMATCH),
+						 errmsg("remote query result rowtype does not match the specified FROM clause rowtype")));
+		}
+
+		/* Cache state that will be needed on every call. */
+		state = palloc0(sizeof(pg_background_result_state));
+		state->info = info;
+		state->result = result;
+
+		funcctx->user_fctx = state;
+		MemoryContextSwitchTo(oldcontext);
+	}
+
+	funcctx = SRF_PERCALL_SETUP();
+	tupdesc = funcctx->tuple_desc;
+	state = funcctx->user_fctx;
+	result = state->result;
+
+	if (result->tupdesc)
+	{
+		if (result->tuples != NIL && funcctx->call_cntr > 0)
+			result->tuples = list_delete_first(result->tuples);
+
+		if (result->tuples != NIL)
+		{
+			HeapTuple tuple = (HeapTuple) linitial(result->tuples);
+
+			/*
+			 * Set the tuple type ID information fields correctly because
+			 * BackgroundSessionFetchResult returns it as an anonymous record
+			 * type.
+			 */
+			HeapTupleHeaderSetTypeId(tuple->t_data, tupdesc->tdtypeid);
+			HeapTupleHeaderSetTypMod(tuple->t_data, tupdesc->tdtypmod);
+
+			SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
+		}
+	}
+	else	/* If no data rows, return the command tags instead. */
+	{
+		if (tupdesc->natts != 1 || tupdesc->attrs[0]->atttypid != TEXTOID)
+			ereport(ERROR,
+					(errcode(ERRCODE_DATATYPE_MISMATCH),
+					 errmsg("remote query did not return a result set, but result rowtype is not a single text column")));
+
+		if (result->command != NULL)
+		{
+			bool	isnull = false;
+			Datum	value = PointerGetDatum(cstring_to_text(result->command));
+			HeapTuple	tuple = heap_form_tuple(tupdesc, &value, &isnull);
+
+			result->command = NULL;
+
+			SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
+		}
+	}
+
+	SRF_RETURN_DONE(funcctx);
+}
+
+/*
+ * End background session and remove hashtable entry.
+ */
+Datum
+pg_background_close(PG_FUNCTION_ARGS)
+{
+	int32						pid = PG_GETARG_INT32(0);
+	pg_background_worker_info  *info;
+
+	info = find_worker_info(pid);
+	if (info == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("PID %d is not attached to this session", pid)));
+
+	check_rights(info);
+	remove_worker_info(pid);
+	BackgroundSessionEnd(info->session);
+
+	PG_RETURN_VOID();
+}
+
+static void
+remove_worker_info(int32 pid)
+{
+	bool	found;
+
+	/* Remove the hashtable entry. */
+	hash_search(worker_hash, (void *) &pid, HASH_REMOVE, &found);
+	if (!found)
+		elog(ERROR, "pg_background worker_hash table corrupted");
+}
+
+/*
+ * Find the background worker information for the worker with a given PID.
+ */
+static pg_background_worker_info *
+find_worker_info(int32 pid)
+{
+	pg_background_worker_info *info = NULL;
+
+	if (worker_hash != NULL)
+		info = hash_search(worker_hash, (void *) &pid, HASH_FIND, NULL);
+
+	return info;
+}
+
+/*
+ * Save worker info.
+ */
+static void
+save_worker_info(int32 pid, BackgroundSession *session)
+{
+	pg_background_worker_info  *info;
+	Oid							current_user_id;
+	int							sec_context;
+
+	/* If the hash table hasn't been set up yet, do that now. */
+	if (worker_hash == NULL)
+	{
+		HASHCTL	ctl;
+
+		ctl.keysize = sizeof(int32);
+		ctl.entrysize = sizeof(pg_background_worker_info);
+		worker_hash = hash_create("pg_background worker_hash", 8, &ctl,
+								  HASH_ELEM);
+	}
+
+	/* Get current authentication information. */
+	GetUserIdAndSecContext(&current_user_id, &sec_context);
+
+	/*
+	 * In the unlikely event that there's an older worker with this PID,
+	 * just detach it - unless it has a different user ID than the
+	 * currently-active one, in which case someone might be trying to pull
+	 * a fast one.  Let's kill the backend to make sure we don't break
+	 * anyone's expectations.
+	 */
+	if ((info = find_worker_info(pid)) != NULL)
+	{
+		if (current_user_id != info->current_user_id)
+			ereport(FATAL,
+				(errcode(ERRCODE_DUPLICATE_OBJECT),
+			 errmsg("background worker with PID \"%d\" already exists",
+						pid)));
+	}
+
+	/* Create a new entry for this worker. */
+	info = hash_search(worker_hash, (void *) &pid, HASH_ENTER, NULL);
+	info->session = session;
+	info->result_count = 0;
+	info->current_user_id = current_user_id;
+}
+
+/*
+ * Check whether the current user has rights to manipulate the background
+ * worker with the given PID.
+ */
+static void
+check_rights(pg_background_worker_info *info)
+{
+	Oid	current_user_id;
+	int	sec_context;
+
+	GetUserIdAndSecContext(&current_user_id, &sec_context);
+	if (!has_privs_of_role(current_user_id, info->current_user_id))
+		ereport(ERROR,
+				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+			 errmsg("permission denied for background worker with PID \"%d\"",
+						info->pid)));
+}
diff --git a/contrib/pg_background/pg_background.control b/contrib/pg_background/pg_background.control
new file mode 100644
index 0000000..733d0e1
--- /dev/null
+++ b/contrib/pg_background/pg_background.control
@@ -0,0 +1,4 @@
+comment = 'Run SQL queries in the background'
+default_version = '1.0'
+module_pathname = '$libdir/pg_background'
+relocatable = true
diff --git a/contrib/pg_background/sql/pg_background.sql b/contrib/pg_background/sql/pg_background.sql
new file mode 100644
index 0000000..be49dd5
--- /dev/null
+++ b/contrib/pg_background/sql/pg_background.sql
@@ -0,0 +1,30 @@
+CREATE EXTENSION pg_background;
+--launch 6 workers which wait  .0, .1, .2, .3, .4, .5 seconds respectively
+CREATE TABLE input AS
+	SELECT pg_background_start() pid, x, row_number() OVER (ORDER BY x) n
+	FROM generate_series(0,.5,0.1) x
+	ORDER BY x DESC;
+CREATE TABLE output(place int,value float);
+--display active background workers
+SELECT count(*) as active_background_workers FROM pg_background_show();
+--run sql query
+SELECT n as worker_num
+FROM input JOIN LATERAL pg_background_run(input.pid,
+										  format($$
+												 SELECT pg_sleep(%s)::text;
+												 INSERT INTO output VALUES (%s, %s);
+												 $$,
+												 x, n, x
+												)
+										 ) ON (true);
+--wait until everyone finishes
+SELECT n as worker_num
+FROM input JOIN LATERAL pg_background_result(input.pid) AS (x TEXT) ON (true);
+--output results
+SELECT * FROM output ORDER BY place;
+--cleanup
+SELECT n as worker_num
+FROM input JOIN LATERAL pg_background_close(input.pid) ON (true);
+DROP TABLE input;
+DROP TABLE output;
+DROP EXTENSION pg_background;
-- 
2.6.2