diff --git a/src/backend/access/common/printtup.c b/src/backend/access/common/printtup.c
index baed981..639451a 100644
--- a/src/backend/access/common/printtup.c
+++ b/src/backend/access/common/printtup.c
@@ -26,9 +26,9 @@
 
 static void printtup_startup(DestReceiver *self, int operation,
 				 TupleDesc typeinfo);
-static void printtup(TupleTableSlot *slot, DestReceiver *self);
-static void printtup_20(TupleTableSlot *slot, DestReceiver *self);
-static void printtup_internal_20(TupleTableSlot *slot, DestReceiver *self);
+static bool printtup(TupleTableSlot *slot, DestReceiver *self);
+static bool printtup_20(TupleTableSlot *slot, DestReceiver *self);
+static bool printtup_internal_20(TupleTableSlot *slot, DestReceiver *self);
 static void printtup_shutdown(DestReceiver *self);
 static void printtup_destroy(DestReceiver *self);
 
@@ -299,7 +299,7 @@ printtup_prepare_info(DR_printtup *myState, TupleDesc typeinfo, int numAttrs)
  *		printtup --- print a tuple in protocol 3.0
  * ----------------
  */
-static void
+static bool
 printtup(TupleTableSlot *slot, DestReceiver *self)
 {
 	TupleDesc	typeinfo = slot->tts_tupleDescriptor;
@@ -376,13 +376,15 @@ printtup(TupleTableSlot *slot, DestReceiver *self)
 	/* Return to caller's context, and flush row's temporary memory */
 	MemoryContextSwitchTo(oldcontext);
 	MemoryContextReset(myState->tmpcontext);
+
+	return true;
 }
 
 /* ----------------
  *		printtup_20 --- print a tuple in protocol 2.0
  * ----------------
  */
-static void
+static bool
 printtup_20(TupleTableSlot *slot, DestReceiver *self)
 {
 	TupleDesc	typeinfo = slot->tts_tupleDescriptor;
@@ -452,6 +454,8 @@ printtup_20(TupleTableSlot *slot, DestReceiver *self)
 	/* Return to caller's context, and flush row's temporary memory */
 	MemoryContextSwitchTo(oldcontext);
 	MemoryContextReset(myState->tmpcontext);
+
+	return true;
 }
 
 /* ----------------
@@ -528,7 +532,7 @@ debugStartup(DestReceiver *self, int operation, TupleDesc typeinfo)
  *		debugtup - print one tuple for an interactive backend
  * ----------------
  */
-void
+bool
 debugtup(TupleTableSlot *slot, DestReceiver *self)
 {
 	TupleDesc	typeinfo = slot->tts_tupleDescriptor;
@@ -553,6 +557,8 @@ debugtup(TupleTableSlot *slot, DestReceiver *self)
 		printatt((unsigned) i + 1, typeinfo->attrs[i], value);
 	}
 	printf("\t----\n");
+
+	return true;
 }
 
 /* ----------------
@@ -564,7 +570,7 @@ debugtup(TupleTableSlot *slot, DestReceiver *self)
  * This is largely same as printtup_20, except we use binary formatting.
  * ----------------
  */
-static void
+static bool
 printtup_internal_20(TupleTableSlot *slot, DestReceiver *self)
 {
 	TupleDesc	typeinfo = slot->tts_tupleDescriptor;
@@ -636,4 +642,6 @@ printtup_internal_20(TupleTableSlot *slot, DestReceiver *self)
 	/* Return to caller's context, and flush row's temporary memory */
 	MemoryContextSwitchTo(oldcontext);
 	MemoryContextReset(myState->tmpcontext);
+
+	return true;
 }
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index 8db1b35..b55c4dc 100644
--- a/src/backend/commands/copy.c
+++ b/src/backend/commands/copy.c
@@ -4414,7 +4414,7 @@ copy_dest_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
 /*
  * copy_dest_receive --- receive one tuple
  */
-static void
+static bool
 copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
 {
 	DR_copy    *myState = (DR_copy *) self;
@@ -4426,6 +4426,8 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
 	/* And send the data */
 	CopyOneRowTo(cstate, InvalidOid, slot->tts_values, slot->tts_isnull);
 	myState->processed++;
+
+	return true;
 }
 
 /*
diff --git a/src/backend/commands/createas.c b/src/backend/commands/createas.c
index 41183f6..418b0f6 100644
--- a/src/backend/commands/createas.c
+++ b/src/backend/commands/createas.c
@@ -62,7 +62,7 @@ typedef struct
 static ObjectAddress CreateAsReladdr = {InvalidOid, InvalidOid, 0};
 
 static void intorel_startup(DestReceiver *self, int operation, TupleDesc typeinfo);
-static void intorel_receive(TupleTableSlot *slot, DestReceiver *self);
+static bool intorel_receive(TupleTableSlot *slot, DestReceiver *self);
 static void intorel_shutdown(DestReceiver *self);
 static void intorel_destroy(DestReceiver *self);
 
@@ -482,7 +482,7 @@ intorel_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
 /*
  * intorel_receive --- receive one tuple
  */
-static void
+static bool
 intorel_receive(TupleTableSlot *slot, DestReceiver *self)
 {
 	DR_intorel *myState = (DR_intorel *) self;
@@ -507,6 +507,8 @@ intorel_receive(TupleTableSlot *slot, DestReceiver *self)
 				myState->bistate);
 
 	/* We know this is a newly created relation, so there are no indexes */
+
+	return true;
 }
 
 /*
diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c
index f0d9e94..e99ad56 100644
--- a/src/backend/commands/explain.c
+++ b/src/backend/commands/explain.c
@@ -20,6 +20,7 @@
 #include "commands/defrem.h"
 #include "commands/prepare.h"
 #include "executor/hashjoin.h"
+#include "executor/nodeFunnel.h"
 #include "foreign/fdwapi.h"
 #include "nodes/nodeFuncs.h"
 #include "optimizer/clauses.h"
@@ -730,6 +731,7 @@ ExplainPreScanNode(PlanState *planstate, Bitmapset **rels_used)
 	{
 		case T_SeqScan:
 		case T_SampleScan:
+		case T_Funnel:
 		case T_IndexScan:
 		case T_IndexOnlyScan:
 		case T_BitmapHeapScan:
@@ -853,6 +855,9 @@ ExplainNode(PlanState *planstate, List *ancestors,
 		case T_SampleScan:
 			pname = sname = "Sample Scan";
 			break;
+		case T_Funnel:
+			pname = sname = "Funnel";
+			break;
 		case T_IndexScan:
 			pname = sname = "Index Scan";
 			break;
@@ -1003,6 +1008,7 @@ ExplainNode(PlanState *planstate, List *ancestors,
 	{
 		case T_SeqScan:
 		case T_SampleScan:
+		case T_Funnel:
 		case T_BitmapHeapScan:
 		case T_TidScan:
 		case T_SubqueryScan:
@@ -1147,6 +1153,15 @@ ExplainNode(PlanState *planstate, List *ancestors,
 	}
 
 	/*
+	 * Aggregate instrumentation information of all the backend workers for
+	 * Funnel node.  Though we already accumulate this information when last
+	 * tuple is fetched from Funnel node, this is to cover cases when we don't
+	 * fetch all tuples from a node such as for Limit node.
+	 */
+	if (es->analyze && IsA(plan, Funnel))
+		DestroyParallelSetupAndAccumStats((FunnelState *) planstate);
+
+	/*
 	 * We have to forcibly clean up the instrumentation state because we
 	 * haven't done ExecutorEnd yet.  This is pretty grotty ...
 	 *
@@ -1276,6 +1291,14 @@ ExplainNode(PlanState *planstate, List *ancestors,
 				show_instrumentation_count("Rows Removed by Filter", 1,
 										   planstate, es);
 			break;
+		case T_Funnel:
+			show_scan_qual(plan->qual, "Filter", planstate, ancestors, es);
+			if (plan->qual)
+				show_instrumentation_count("Rows Removed by Filter", 1,
+										   planstate, es);
+			ExplainPropertyInteger("Number of Workers",
+								   ((Funnel *) plan)->num_workers, es);
+			break;
 		case T_FunctionScan:
 			if (es->verbose)
 			{
@@ -2335,6 +2358,7 @@ ExplainTargetRel(Plan *plan, Index rti, ExplainState *es)
 	{
 		case T_SeqScan:
 		case T_SampleScan:
+		case T_Funnel:
 		case T_IndexScan:
 		case T_IndexOnlyScan:
 		case T_BitmapHeapScan:
diff --git a/src/backend/commands/matview.c b/src/backend/commands/matview.c
index 5492e59..750a59c 100644
--- a/src/backend/commands/matview.c
+++ b/src/backend/commands/matview.c
@@ -56,7 +56,7 @@ typedef struct
 static int	matview_maintenance_depth = 0;
 
 static void transientrel_startup(DestReceiver *self, int operation, TupleDesc typeinfo);
-static void transientrel_receive(TupleTableSlot *slot, DestReceiver *self);
+static bool transientrel_receive(TupleTableSlot *slot, DestReceiver *self);
 static void transientrel_shutdown(DestReceiver *self);
 static void transientrel_destroy(DestReceiver *self);
 static void refresh_matview_datafill(DestReceiver *dest, Query *query,
@@ -422,7 +422,7 @@ transientrel_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
 /*
  * transientrel_receive --- receive one tuple
  */
-static void
+static bool
 transientrel_receive(TupleTableSlot *slot, DestReceiver *self)
 {
 	DR_transientrel *myState = (DR_transientrel *) self;
@@ -441,6 +441,8 @@ transientrel_receive(TupleTableSlot *slot, DestReceiver *self)
 				myState->bistate);
 
 	/* We know this is a newly created relation, so there are no indexes */
+
+	return true;
 }
 
 /*
diff --git a/src/backend/executor/Makefile b/src/backend/executor/Makefile
index 249534b..fb34864 100644
--- a/src/backend/executor/Makefile
+++ b/src/backend/executor/Makefile
@@ -13,11 +13,11 @@ top_builddir = ../../..
 include $(top_builddir)/src/Makefile.global
 
 OBJS = execAmi.o execCurrent.o execGrouping.o execIndexing.o execJunk.o \
-       execMain.o execProcnode.o execQual.o execScan.o execTuples.o \
+       execMain.o execParallel.o execProcnode.o execQual.o execScan.o execTuples.o \
        execUtils.o functions.o instrument.o nodeAppend.o nodeAgg.o \
        nodeBitmapAnd.o nodeBitmapOr.o \
-       nodeBitmapHeapscan.o nodeBitmapIndexscan.o nodeCustom.o nodeHash.o \
-       nodeHashjoin.o nodeIndexscan.o nodeIndexonlyscan.o \
+       nodeBitmapHeapscan.o nodeBitmapIndexscan.o nodeCustom.o nodeFunnel.o \
+       nodeHash.o nodeHashjoin.o nodeIndexscan.o nodeIndexonlyscan.o \
        nodeLimit.o nodeLockRows.o \
        nodeMaterial.o nodeMergeAppend.o nodeMergejoin.o nodeModifyTable.o \
        nodeNestloop.o nodeFunctionscan.o nodeRecursiveunion.o nodeResult.o \
diff --git a/src/backend/executor/execAmi.c b/src/backend/executor/execAmi.c
index 93e1e9a..4915151 100644
--- a/src/backend/executor/execAmi.c
+++ b/src/backend/executor/execAmi.c
@@ -24,6 +24,7 @@
 #include "executor/nodeCustom.h"
 #include "executor/nodeForeignscan.h"
 #include "executor/nodeFunctionscan.h"
+#include "executor/nodeFunnel.h"
 #include "executor/nodeGroup.h"
 #include "executor/nodeGroup.h"
 #include "executor/nodeHash.h"
@@ -160,6 +161,10 @@ ExecReScan(PlanState *node)
 			ExecReScanSampleScan((SampleScanState *) node);
 			break;
 
+		case T_FunnelState:
+			ExecReScanFunnel((FunnelState *) node);
+			break;
+
 		case T_IndexScanState:
 			ExecReScanIndexScan((IndexScanState *) node);
 			break;
@@ -467,6 +472,9 @@ ExecSupportsBackwardScan(Plan *node)
 			/* Simplify life for tablesample methods by disallowing this */
 			return false;
 
+		case T_Funnel:
+			return false;
+
 		case T_IndexScan:
 			return IndexSupportsBackwardScan(((IndexScan *) node)->indexid) &&
 				TargetListSupportsBackwardScan(node->targetlist);
diff --git a/src/backend/executor/execCurrent.c b/src/backend/executor/execCurrent.c
index bcd287f..650fcc5 100644
--- a/src/backend/executor/execCurrent.c
+++ b/src/backend/executor/execCurrent.c
@@ -262,6 +262,7 @@ search_plan_tree(PlanState *node, Oid table_oid)
 			 */
 		case T_SeqScanState:
 		case T_SampleScanState:
+		case T_FunnelState:
 		case T_IndexScanState:
 		case T_IndexOnlyScanState:
 		case T_BitmapHeapScanState:
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 85ff46b..66e015b 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -45,9 +45,11 @@
 #include "commands/matview.h"
 #include "commands/trigger.h"
 #include "executor/execdebug.h"
+#include "executor/execParallel.h"
 #include "foreign/fdwapi.h"
 #include "mb/pg_wchar.h"
 #include "miscadmin.h"
+#include "nodes/nodeFuncs.h"
 #include "optimizer/clauses.h"
 #include "parser/parsetree.h"
 #include "storage/bufmgr.h"
@@ -354,7 +356,11 @@ standard_ExecutorRun(QueryDesc *queryDesc,
 		(*dest->rShutdown) (dest);
 
 	if (queryDesc->totaltime)
+	{
+		/* Accumulate stats from parallel workers before stopping node */
+		(void) ExecParallelBufferUsageAccum((Node *) queryDesc->planstate);
 		InstrStopNode(queryDesc->totaltime, estate->es_processed);
+	}
 
 	MemoryContextSwitchTo(oldcontext);
 }
@@ -1581,7 +1587,15 @@ ExecutePlan(EState *estate,
 		 * practice, this is probably always the case at this point.)
 		 */
 		if (sendTuples)
-			(*dest->receiveSlot) (slot, dest);
+		{
+			/*
+			 * If we are not able to send the tuple, we assume the destination
+			 * has closed and no more tuples can be sent. If that's the case,
+			 * end the loop.
+			 */
+			if (!((*dest->receiveSlot) (slot, dest)))
+				break;
+		}
 
 		/*
 		 * Count tuples processed, if this is a SELECT.  (For other operation
diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c
new file mode 100644
index 0000000..726bc7f
--- /dev/null
+++ b/src/backend/executor/execParallel.c
@@ -0,0 +1,542 @@
+/*-------------------------------------------------------------------------
+ *
+ * execParallel.c
+ *	  Support routines for setting up backend workers for parallel execution.
+ *
+ * Portions Copyright (c) 1996-2015, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/executor/execParallel.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "executor/execParallel.h"
+#include "executor/nodeFunnel.h"
+#include "nodes/nodeFuncs.h"
+#include "optimizer/planmain.h"
+#include "optimizer/planner.h"
+#include "tcop/tcopprot.h"
+
+
+#define PARALLEL_TUPLE_QUEUE_SIZE					65536
+
+static void ParallelQueryMain(dsm_segment *seg, shm_toc *toc);
+static void EstimateParallelSupportInfoSpace(ParallelContext *pcxt, ParamListInfo params,
+								 List *serialized_param_exec_vals,
+								 int instOptions, Size *params_size,
+								 Size *params_exec_size);
+static void StoreParallelSupportInfo(ParallelContext *pcxt, ParamListInfo params,
+						 List *serialized_param_exec_vals,
+						 int instOptions, Size params_size,
+						 Size params_exec_size,
+						 char **inst_options_space,
+						 char **buffer_usage_space);
+static void EstimatePlannedStmtSpace(ParallelContext *pcxt, PlanState *planstate,
+						 char *plannedstmt_str, Size *plannedstmt_len,
+						 Size *pscan_size);
+static void StorePlannedStmt(ParallelContext *pcxt, PlanState *planstate,
+				 char *plannedstmt_str, Size plannedstmt_size,
+				 Size pscan_size);
+static void EstimateResponseQueueSpace(ParallelContext *pcxt);
+static void StoreResponseQueue(ParallelContext *pcxt,
+				   shm_mq_handle ***responseqp);
+static void
+			ExecParallelGetPlannedStmt(shm_toc *toc, PlannedStmt **plannedstmt);
+static void GetParallelSupportInfo(shm_toc *toc, ParamListInfo *params,
+					   List **serialized_param_exec_vals,
+					   int *inst_options, char **instrument,
+					   char **buffer_usage);
+static void SetupResponseQueue(dsm_segment *seg, shm_toc *toc, shm_mq **mq,
+				   shm_mq_handle **responseq);
+
+
+/*
+ * This is required for parallel plan execution to fetch the information
+ * from dsm.
+ */
+static shm_toc *parallel_shm_toc = NULL;
+
+typedef struct parallel_estimate_ctx
+{
+	ParallelContext *context;
+	Size	   *psize;
+
+} parallel_estimate_ctx;
+
+static bool ExecParallelEstimate(Node *node,
+					 parallel_estimate_ctx *pestcontext);
+static bool ExecParallelInitializeDSM(Node *node,
+						  parallel_estimate_ctx *pestcontext);
+
+/*
+ * EstimateParallelSupportInfoSpace
+ *
+ * Estimate the amount of space required to record information of bind
+ * parameters, PARAM_EXEC parameters and instrumentation information that
+ * need to be retrieved from parallel workers.
+ */
+void
+EstimateParallelSupportInfoSpace(ParallelContext *pcxt, ParamListInfo params,
+								 List *serialized_param_exec_vals,
+								 int instOptions, Size *params_size,
+								 Size *params_exec_size)
+{
+	*params_size = EstimateBoundParametersSpace(params);
+	shm_toc_estimate_chunk(&pcxt->estimator, *params_size);
+
+	*params_exec_size = EstimateExecParametersSpace(serialized_param_exec_vals);
+	shm_toc_estimate_chunk(&pcxt->estimator, *params_exec_size);
+
+	/*
+	 * We expect each worker to populate the BufferUsage and instrumentation
+	 * structure allocated by master backend and then master backend will
+	 * aggregate all the usage along with it's own, so account it for each
+	 * worker.
+	 */
+	shm_toc_estimate_chunk(&pcxt->estimator,
+						   sizeof(BufferUsage) * pcxt->nworkers);
+
+	/* account for instrumentation options. */
+	shm_toc_estimate_chunk(&pcxt->estimator, sizeof(int));
+
+	if (instOptions)
+	{
+		shm_toc_estimate_chunk(&pcxt->estimator,
+							   sizeof(Instrumentation) * pcxt->nworkers);
+		/* keys for parallel support information. */
+		shm_toc_estimate_keys(&pcxt->estimator, 1);
+	}
+
+	/* keys for parallel support information. */
+	shm_toc_estimate_keys(&pcxt->estimator, 4);
+}
+
+/*
+ * StoreParallelSupportInfo
+ *
+ * Sets up the bind parameters, PARAM_EXEC parameters and instrumentation
+ * information required for parallel execution.
+ */
+void
+StoreParallelSupportInfo(ParallelContext *pcxt, ParamListInfo params,
+						 List *serialized_param_exec_vals,
+						 int instOptions, Size params_size,
+						 Size params_exec_size,
+						 char **inst_options_space,
+						 char **buffer_usage_space)
+{
+	char	   *paramsdata;
+	char	   *paramsexecdata;
+	int		   *inst_options;
+
+	/* Store PARAM_EXEC aka bind parameters in dynamic shared memory */
+	paramsdata = shm_toc_allocate(pcxt->toc, params_size);
+	SerializeBoundParams(params, params_size, paramsdata);
+	shm_toc_insert(pcxt->toc, PARALLEL_KEY_PARAMS, paramsdata);
+
+	/*
+	 * Store PARAM_EXEC parameters in dynamic shared memory.  This is used for
+	 * evaluation plan->initPlan params.
+	 */
+	paramsexecdata = shm_toc_allocate(pcxt->toc, params_exec_size);
+	SerializeExecParams(serialized_param_exec_vals, params_exec_size, paramsexecdata);
+	shm_toc_insert(pcxt->toc, PARALLEL_KEY_PARAMS_EXEC, paramsexecdata);
+
+	/*
+	 * Allocate space for BufferUsage information to be filled by each worker.
+	 */
+	*buffer_usage_space =
+		shm_toc_allocate(pcxt->toc, sizeof(BufferUsage) * pcxt->nworkers);
+	shm_toc_insert(pcxt->toc, PARALLEL_KEY_BUFF_USAGE, *buffer_usage_space);
+
+	/* Store instrument options in dynamic shared memory. */
+	inst_options = shm_toc_allocate(pcxt->toc, sizeof(int));
+	*inst_options = instOptions;
+	shm_toc_insert(pcxt->toc, PARALLEL_KEY_INST_OPTIONS, inst_options);
+
+	/*
+	 * Allocate space for instrumentation information to be filled by each
+	 * worker.
+	 */
+	if (instOptions)
+	{
+		*inst_options_space =
+			shm_toc_allocate(pcxt->toc, sizeof(Instrumentation) * pcxt->nworkers);
+		shm_toc_insert(pcxt->toc, PARALLEL_KEY_INST_INFO, *inst_options_space);
+	}
+}
+
+/*
+ * EstimatePlannedStmtSpace
+ *
+ * Estimate the amount of space required to record information of planned
+ * statement and parallel node specific information that need to be copied
+ * to parallel workers.
+ */
+void
+EstimatePlannedStmtSpace(ParallelContext *pcxt, PlanState *planstate,
+						 char *plannedstmt_str, Size *plannedstmt_len,
+						 Size *pscan_size)
+{
+	parallel_estimate_ctx *pestcontext;
+
+	pestcontext = palloc(sizeof(parallel_estimate_ctx));
+	pestcontext->context = pcxt;
+	pestcontext->psize = pscan_size;
+
+	/* Estimate space for planned statement. */
+	*plannedstmt_len = strlen(plannedstmt_str) + 1;
+	shm_toc_estimate_chunk(&pcxt->estimator, *plannedstmt_len);
+
+	/* keys for planned statement information. */
+	shm_toc_estimate_keys(&pcxt->estimator, 1);
+
+	(void) ExecParallelEstimate((Node *) planstate, pestcontext);
+}
+
+/*
+ * StorePlannedStmt
+ *
+ * Sets up the planned statement and node specific information.
+ */
+void
+StorePlannedStmt(ParallelContext *pcxt, PlanState *planstate,
+				 char *plannedstmt_str, Size plannedstmt_size,
+				 Size pscan_size)
+{
+	char	   *plannedstmtdata;
+	parallel_estimate_ctx *pestcontext;
+
+	pestcontext = palloc(sizeof(parallel_estimate_ctx));
+	pestcontext->context = pcxt;
+	pestcontext->psize = &pscan_size;
+
+	/* Store planned statement in dynamic shared memory. */
+	plannedstmtdata = shm_toc_allocate(pcxt->toc, plannedstmt_size);
+	memcpy(plannedstmtdata, plannedstmt_str, plannedstmt_size);
+	shm_toc_insert(pcxt->toc, PARALLEL_KEY_PLANNEDSTMT, plannedstmtdata);
+
+	(void) ExecParallelInitializeDSM((Node *) planstate, pestcontext);
+}
+
+/*
+ * EstimateResponseQueueSpace
+ *
+ * Estimate the amount of space required to record information of tuple
+ * queues that need to be established between parallel workers and master
+ * backend.
+ */
+void
+EstimateResponseQueueSpace(ParallelContext *pcxt)
+{
+	/* Estimate space for tuple queues. */
+	shm_toc_estimate_chunk(&pcxt->estimator,
+						   (Size) PARALLEL_TUPLE_QUEUE_SIZE * pcxt->nworkers);
+
+	/* keys for response queue. */
+	shm_toc_estimate_keys(&pcxt->estimator, 1);
+}
+
+/*
+ * StoreResponseQueue
+ *
+ * It sets up the response queues for backend workers to return tuples
+ * to the main backend and start the workers.
+ */
+void
+StoreResponseQueue(ParallelContext *pcxt,
+				   shm_mq_handle ***responseqp)
+{
+	shm_mq	   *mq;
+	char	   *tuple_queue_space;
+	int			i;
+
+	/* Allocate memory for shared memory queue handles. */
+	*responseqp = (shm_mq_handle **) palloc(pcxt->nworkers * sizeof(shm_mq_handle *));
+
+	/*
+	 * Establish one message queue per worker in dynamic shared memory. These
+	 * queues should be used to transmit tuple data.
+	 */
+	tuple_queue_space =
+		shm_toc_allocate(pcxt->toc, PARALLEL_TUPLE_QUEUE_SIZE * pcxt->nworkers);
+	for (i = 0; i < pcxt->nworkers; ++i)
+	{
+		mq = shm_mq_create(tuple_queue_space + i * PARALLEL_TUPLE_QUEUE_SIZE,
+						   (Size) PARALLEL_TUPLE_QUEUE_SIZE);
+
+		shm_mq_set_receiver(mq, MyProc);
+
+		/*
+		 * Attach the queue before launching a worker, so that we'll
+		 * automatically detach the queue if we error out.  Otherwise, the
+		 * worker might sit there trying to write the queue long after we've
+		 * gone away.
+		 */
+		(*responseqp)[i] = shm_mq_attach(mq, pcxt->seg, NULL);
+	}
+	shm_toc_insert(pcxt->toc, PARALLEL_KEY_TUPLE_QUEUE, tuple_queue_space);
+}
+
+/*
+ * ExecParallelEstimate
+ *
+ * Estimate the amount of space required to record information of
+ * parallel node that need to be copied to parallel workers.
+ */
+static bool
+ExecParallelEstimate(Node *node, parallel_estimate_ctx *pestcontext)
+{
+	if (node == NULL)
+		return false;
+
+	switch (nodeTag(node))
+	{
+		default:
+			break;
+	}
+
+	return planstate_tree_walker((PlanState *) node, ExecParallelEstimate, pestcontext);
+}
+
+/*
+ *	ExecParallelInitializeDSM
+ *
+ *		Store the information of parallel node in dsm.
+ */
+static bool
+ExecParallelInitializeDSM(Node *node, parallel_estimate_ctx *pestcontext)
+{
+	if (node == NULL)
+		return false;
+
+	switch (nodeTag(node))
+	{
+		default:
+			break;
+	}
+
+	return planstate_tree_walker((PlanState *) node, ExecParallelInitializeDSM, pestcontext);
+}
+
+/*
+ * InitializeParallelWorkers
+ *
+ * Sets up the required infrastructure for backend workers to perform
+ * execution and return results to the main backend.
+ */
+void
+InitializeParallelWorkers(PlanState *planstate,
+						  List *serialized_param_exec_vals,
+						  EState *estate,
+						  char **inst_options_space,
+						  char **buffer_usage_space,
+						  shm_mq_handle ***responseqp,
+						  ParallelContext **pcxtp,
+						  int nWorkers)
+{
+	Size		params_size,
+				params_exec_size,
+				pscan_size,
+				plannedstmt_size;
+	char	   *plannedstmt_str;
+	PlannedStmt *plannedstmt;
+	ParallelContext *pcxt;
+
+	pcxt = CreateParallelContext(ParallelQueryMain, nWorkers);
+
+	plannedstmt = create_parallel_worker_plannedstmt(planstate->plan,
+													 estate->es_range_table,
+										 estate->es_plannedstmt->nParamExec);
+	plannedstmt_str = nodeToString(plannedstmt);
+
+	EstimatePlannedStmtSpace(pcxt, planstate, plannedstmt_str,
+							 &plannedstmt_size, &pscan_size);
+	EstimateParallelSupportInfoSpace(pcxt, estate->es_param_list_info,
+									 serialized_param_exec_vals,
+									 estate->es_instrument, &params_size,
+									 &params_exec_size);
+	EstimateResponseQueueSpace(pcxt);
+
+	InitializeParallelDSM(pcxt);
+
+	StorePlannedStmt(pcxt, planstate, plannedstmt_str,
+					 plannedstmt_size, pscan_size);
+	StoreParallelSupportInfo(pcxt, estate->es_param_list_info,
+							 serialized_param_exec_vals,
+							 estate->es_instrument,
+							 params_size,
+							 params_exec_size,
+							 inst_options_space,
+							 buffer_usage_space);
+	StoreResponseQueue(pcxt, responseqp);
+
+	/* Return results to caller. */
+	*pcxtp = pcxt;
+}
+
+/*
+ * GetParallelSupportInfo
+ *
+ * Look up based on keys in dynamic shared memory segment and get the
+ * bind parameters, PARAM_EXEC parameters and instrumentation information
+ * required to perform parallel operation.
+ */
+static void
+GetParallelSupportInfo(shm_toc *toc, ParamListInfo *params,
+					   List **serialized_param_exec_vals,
+					   int *inst_options, char **instrument,
+					   char **buffer_usage)
+{
+	char	   *paramsdata;
+	char	   *paramsexecdata;
+	char	   *inst_options_space;
+	char	   *buffer_usage_space;
+	int		   *instoptions;
+
+	paramsdata = shm_toc_lookup(toc, PARALLEL_KEY_PARAMS);
+	*params = RestoreBoundParams(paramsdata);
+
+	paramsexecdata = shm_toc_lookup(toc, PARALLEL_KEY_PARAMS_EXEC);
+	*serialized_param_exec_vals = RestoreExecParams(paramsexecdata);
+
+	instoptions = shm_toc_lookup(toc, PARALLEL_KEY_INST_OPTIONS);
+	*inst_options = *instoptions;
+	if (*inst_options)
+	{
+		inst_options_space = shm_toc_lookup(toc, PARALLEL_KEY_INST_INFO);
+		*instrument = (inst_options_space +
+					   ParallelWorkerNumber * sizeof(Instrumentation));
+	}
+
+	buffer_usage_space = shm_toc_lookup(toc, PARALLEL_KEY_BUFF_USAGE);
+	*buffer_usage = (buffer_usage_space + ParallelWorkerNumber * sizeof(BufferUsage));
+}
+
+/*
+ * ExecParallelGetPlannedStmt
+ *
+ * Look up based on keys in dynamic shared memory segment and get the
+ * planned statement required to perform parallel operation.
+ */
+void
+ExecParallelGetPlannedStmt(shm_toc *toc, PlannedStmt **plannedstmt)
+{
+	char	   *plannedstmtdata;
+
+	plannedstmtdata = shm_toc_lookup(toc, PARALLEL_KEY_PLANNEDSTMT);
+
+	*plannedstmt = (PlannedStmt *) stringToNode(plannedstmtdata);
+
+	/* Fill in opfuncid values if missing */
+	fix_node_funcids((*plannedstmt)->planTree);
+}
+
+/*
+ * SetupResponseQueue
+ *
+ * Look up based on keys in dynamic shared memory segment and get the
+ * tuple queue information for a particular worker, attach to the queue
+ * and redirect all futher responses from worker backend via that queue.
+ */
+void
+SetupResponseQueue(dsm_segment *seg, shm_toc *toc, shm_mq **mq,
+				   shm_mq_handle **responseq)
+{
+	char	   *tuple_queue_space;
+
+	tuple_queue_space = shm_toc_lookup(toc, PARALLEL_KEY_TUPLE_QUEUE);
+	*mq = (shm_mq *) (tuple_queue_space +
+					  ParallelWorkerNumber * PARALLEL_TUPLE_QUEUE_SIZE);
+
+	shm_mq_set_sender(*mq, MyProc);
+	*responseq = shm_mq_attach(*mq, seg, NULL);
+}
+
+/*
+ * GetParallelShmToc
+ */
+shm_toc *
+GetParallelShmToc(void)
+{
+	return parallel_shm_toc;
+}
+
+/*
+ * ParallelQueryMain
+ *
+ * Execute the operation to return the tuples or other information to
+ * parallelism driving node.
+ */
+void
+ParallelQueryMain(dsm_segment *seg, shm_toc *toc)
+{
+	shm_mq	   *mq;
+	shm_mq_handle *responseq;
+	PlannedStmt *plannedstmt;
+	ParamListInfo params;
+	List	   *serialized_param_exec_vals;
+	int			inst_options;
+	char	   *instrument = NULL;
+	char	   *buffer_usage = NULL;
+	ParallelStmt *parallelstmt;
+
+	SetupResponseQueue(seg, toc, &mq, &responseq);
+
+	ExecParallelGetPlannedStmt(toc, &plannedstmt);
+	GetParallelSupportInfo(toc, &params, &serialized_param_exec_vals,
+						   &inst_options, &instrument, &buffer_usage);
+
+	parallelstmt = palloc(sizeof(ParallelStmt));
+
+	parallelstmt->plannedstmt = plannedstmt;
+	parallelstmt->params = params;
+	parallelstmt->serialized_param_exec_vals = serialized_param_exec_vals;
+	parallelstmt->inst_options = inst_options;
+	parallelstmt->instrument = instrument;
+	parallelstmt->buffer_usage = buffer_usage;
+	parallelstmt->responseq = responseq;
+
+	parallel_shm_toc = toc;
+
+	/* Execute the worker command. */
+	exec_parallel_stmt(parallelstmt);
+
+	/*
+	 * Once we are done with sending tuples, detach from shared memory message
+	 * queue used to send tuples.
+	 */
+	shm_mq_detach(mq);
+}
+
+/*
+ * ExecParallelBufferUsageAccum
+ *
+ * Recursively accumulate the stats for all the funnel nodes in a plan
+ * state tree.
+ */
+bool
+ExecParallelBufferUsageAccum(Node *node)
+{
+	if (node == NULL)
+		return false;
+
+	switch (nodeTag(node))
+	{
+		case T_FunnelState:
+			{
+				DestroyParallelSetupAndAccumStats((FunnelState *) node);
+				return true;
+			}
+			break;
+		default:
+			break;
+	}
+
+	return planstate_tree_walker((PlanState *) node, ExecParallelBufferUsageAccum, NULL);
+}
diff --git a/src/backend/executor/execProcnode.c b/src/backend/executor/execProcnode.c
index 03c2feb..c181bf2 100644
--- a/src/backend/executor/execProcnode.c
+++ b/src/backend/executor/execProcnode.c
@@ -100,6 +100,7 @@
 #include "executor/nodeMergejoin.h"
 #include "executor/nodeModifyTable.h"
 #include "executor/nodeNestloop.h"
+#include "executor/nodeFunnel.h"
 #include "executor/nodeRecursiveunion.h"
 #include "executor/nodeResult.h"
 #include "executor/nodeSamplescan.h"
@@ -196,6 +197,11 @@ ExecInitNode(Plan *node, EState *estate, int eflags)
 													  estate, eflags);
 			break;
 
+		case T_Funnel:
+			result = (PlanState *) ExecInitFunnel((Funnel *) node,
+												  estate, eflags);
+			break;
+
 		case T_IndexScan:
 			result = (PlanState *) ExecInitIndexScan((IndexScan *) node,
 													 estate, eflags);
@@ -416,6 +422,10 @@ ExecProcNode(PlanState *node)
 			result = ExecSampleScan((SampleScanState *) node);
 			break;
 
+		case T_FunnelState:
+			result = ExecFunnel((FunnelState *) node);
+			break;
+
 		case T_IndexScanState:
 			result = ExecIndexScan((IndexScanState *) node);
 			break;
@@ -658,6 +668,10 @@ ExecEndNode(PlanState *node)
 			ExecEndSampleScan((SampleScanState *) node);
 			break;
 
+		case T_FunnelState:
+			ExecEndFunnel((FunnelState *) node);
+			break;
+
 		case T_IndexScanState:
 			ExecEndIndexScan((IndexScanState *) node);
 			break;
diff --git a/src/backend/executor/execTuples.c b/src/backend/executor/execTuples.c
index a05d8b1..d5619bd 100644
--- a/src/backend/executor/execTuples.c
+++ b/src/backend/executor/execTuples.c
@@ -1313,7 +1313,7 @@ do_tup_output(TupOutputState *tstate, Datum *values, bool *isnull)
 	ExecStoreVirtualTuple(slot);
 
 	/* send the tuple to the receiver */
-	(*tstate->dest->receiveSlot) (slot, tstate->dest);
+	(void) (*tstate->dest->receiveSlot) (slot, tstate->dest);
 
 	/* clean up */
 	ExecClearTuple(slot);
diff --git a/src/backend/executor/execUtils.c b/src/backend/executor/execUtils.c
index a735815..3213684 100644
--- a/src/backend/executor/execUtils.c
+++ b/src/backend/executor/execUtils.c
@@ -964,3 +964,27 @@ ShutdownExprContext(ExprContext *econtext, bool isCommit)
 
 	MemoryContextSwitchTo(oldcontext);
 }
+
+/*
+ * Populate the values of PARAM_EXEC parameters.
+ *
+ * This is used by worker backends to fill in the values of PARAM_EXEC
+ * parameters after fetching the same from dynamic shared memory.
+ * This needs to be called before ExecutorRun.
+ */
+void
+PopulateParamExecParams(QueryDesc *queryDesc,
+						List *serialized_param_exec_vals)
+{
+	ListCell   *lparam;
+
+	foreach(lparam, serialized_param_exec_vals)
+	{
+		SerializedParamExecData *param_val = (SerializedParamExecData *) lfirst(lparam);
+
+		queryDesc->estate->es_param_exec_vals[param_val->paramid].value =
+			param_val->value;
+		queryDesc->estate->es_param_exec_vals[param_val->paramid].isnull =
+			param_val->isnull;
+	}
+}
diff --git a/src/backend/executor/functions.c b/src/backend/executor/functions.c
index 812a610..863bd64 100644
--- a/src/backend/executor/functions.c
+++ b/src/backend/executor/functions.c
@@ -167,7 +167,7 @@ static Datum postquel_get_single_result(TupleTableSlot *slot,
 static void sql_exec_error_callback(void *arg);
 static void ShutdownSQLFunction(Datum arg);
 static void sqlfunction_startup(DestReceiver *self, int operation, TupleDesc typeinfo);
-static void sqlfunction_receive(TupleTableSlot *slot, DestReceiver *self);
+static bool sqlfunction_receive(TupleTableSlot *slot, DestReceiver *self);
 static void sqlfunction_shutdown(DestReceiver *self);
 static void sqlfunction_destroy(DestReceiver *self);
 
@@ -1903,7 +1903,7 @@ sqlfunction_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
 /*
  * sqlfunction_receive --- receive one tuple
  */
-static void
+static bool
 sqlfunction_receive(TupleTableSlot *slot, DestReceiver *self)
 {
 	DR_sqlfunction *myState = (DR_sqlfunction *) self;
@@ -1913,6 +1913,8 @@ sqlfunction_receive(TupleTableSlot *slot, DestReceiver *self)
 
 	/* Store the filtered tuple into the tuplestore */
 	tuplestore_puttupleslot(myState->tstore, slot);
+
+	return true;
 }
 
 /*
diff --git a/src/backend/executor/instrument.c b/src/backend/executor/instrument.c
index f5351eb..639eb04 100644
--- a/src/backend/executor/instrument.c
+++ b/src/backend/executor/instrument.c
@@ -19,9 +19,6 @@
 
 BufferUsage pgBufferUsage;
 
-static void BufferUsageAccumDiff(BufferUsage *dst,
-					 const BufferUsage *add, const BufferUsage *sub);
-
 
 /* Allocate new instrumentation structure(s) */
 Instrumentation *
@@ -127,8 +124,29 @@ InstrEndLoop(Instrumentation *instr)
 	instr->tuplecount = 0;
 }
 
+/*
+ * Aggregate the instrumentation information.  This is used to aggregate
+ * the information of worker backends.  We only need to sum the buffer
+ * usage and tuple count statistics as for other timing related statistics
+ * it is sufficient to have the master backends information.
+ */
+void
+InstrAggNode(Instrumentation *instr1, Instrumentation *instr2)
+{
+	/* count the returned tuples */
+	instr1->tuplecount += instr2->tuplecount;
+
+	instr1->nfiltered1 += instr2->nfiltered1;
+	instr1->nfiltered2 += instr2->nfiltered2;
+
+	/* Add delta of buffer usage since entry to node's totals */
+	if (instr1->need_bufusage)
+		BufferUsageAdd(&instr1->bufusage, &instr2->bufusage);
+
+}
+
 /* dst += add - sub */
-static void
+void
 BufferUsageAccumDiff(BufferUsage *dst,
 					 const BufferUsage *add,
 					 const BufferUsage *sub)
@@ -148,3 +166,21 @@ BufferUsageAccumDiff(BufferUsage *dst,
 	INSTR_TIME_ACCUM_DIFF(dst->blk_write_time,
 						  add->blk_write_time, sub->blk_write_time);
 }
+
+/* dst += add */
+void
+BufferUsageAdd(BufferUsage *dst, const BufferUsage *add)
+{
+	dst->shared_blks_hit += add->shared_blks_hit;
+	dst->shared_blks_read += add->shared_blks_read;
+	dst->shared_blks_dirtied += add->shared_blks_dirtied;
+	dst->shared_blks_written += add->shared_blks_written;
+	dst->local_blks_hit += add->local_blks_hit;
+	dst->local_blks_read += add->local_blks_read;
+	dst->local_blks_dirtied += add->local_blks_dirtied;
+	dst->local_blks_written += add->local_blks_written;
+	dst->temp_blks_read += add->temp_blks_read;
+	dst->temp_blks_written += add->temp_blks_written;
+	INSTR_TIME_ADD(dst->blk_read_time, add->blk_read_time);
+	INSTR_TIME_ADD(dst->blk_write_time, add->blk_write_time);
+}
diff --git a/src/backend/executor/nodeFunnel.c b/src/backend/executor/nodeFunnel.c
new file mode 100644
index 0000000..81f4074
--- /dev/null
+++ b/src/backend/executor/nodeFunnel.c
@@ -0,0 +1,399 @@
+/*-------------------------------------------------------------------------
+ *
+ * nodeFunnel.c
+ *	  Support routines for scanning a relation via multiple workers.
+ *
+ * Portions Copyright (c) 1996-2015, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/executor/nodeFunnel.c
+ *
+ *-------------------------------------------------------------------------
+ */
+/*
+ * INTERFACE ROUTINES
+ *		ExecFunnel				scans a relation using worker backends.
+ *		ExecInitFunnel			creates and initializes a funnel node.
+ *		ExecEndFunnel			releases any storage allocated.
+ *		ExecReScanFunnel		Re-initialize the workers and rescans a relation via them.
+ */
+#include "postgres.h"
+
+#include "access/relscan.h"
+#include "executor/execdebug.h"
+#include "executor/execParallel.h"
+#include "executor/nodeFunnel.h"
+#include "executor/nodeSubplan.h"
+#include "utils/rel.h"
+
+
+static TupleTableSlot *funnel_getnext(FunnelState *funnelstate);
+static void ExecAccumulateInstInfo(FunnelState *node);
+static void ExecAccumulateBufUsageInfo(FunnelState *node);
+
+
+/* ----------------------------------------------------------------
+ *		ExecInitFunnel
+ * ----------------------------------------------------------------
+ */
+FunnelState *
+ExecInitFunnel(Funnel * node, EState *estate, int eflags)
+{
+	FunnelState *funnelstate;
+
+	/* Funnel node doesn't have innerPlan node. */
+	Assert(innerPlan(node) == NULL);
+
+	/*
+	 * create state structure
+	 */
+	funnelstate = makeNode(FunnelState);
+	funnelstate->ss.ps.plan = (Plan *) node;
+	funnelstate->ss.ps.state = estate;
+	funnelstate->fs_workersReady = false;
+
+	/*
+	 * Miscellaneous initialization
+	 *
+	 * create expression context for node
+	 */
+	ExecAssignExprContext(estate, &funnelstate->ss.ps);
+
+	/*
+	 * initialize child expressions
+	 */
+	funnelstate->ss.ps.targetlist = (List *)
+		ExecInitExpr((Expr *) node->scan.plan.targetlist,
+					 (PlanState *) funnelstate);
+	funnelstate->ss.ps.qual = (List *)
+		ExecInitExpr((Expr *) node->scan.plan.qual,
+					 (PlanState *) funnelstate);
+
+	/*
+	 * tuple table initialization
+	 */
+	ExecInitResultTupleSlot(estate, &funnelstate->ss.ps);
+	ExecInitScanTupleSlot(estate, &funnelstate->ss);
+
+	/*
+	 * now initialize outer plan
+	 */
+	outerPlanState(funnelstate) = ExecInitNode(outerPlan(node), estate, eflags);
+
+
+	funnelstate->ss.ps.ps_TupFromTlist = false;
+
+	/*
+	 * Initialize result tuple type and projection info.
+	 */
+	ExecAssignResultTypeFromTL(&funnelstate->ss.ps);
+	ExecAssignProjectionInfo(&funnelstate->ss.ps, NULL);
+
+	return funnelstate;
+}
+
+/* ----------------------------------------------------------------
+ *		ExecFunnel(node)
+ *
+ *		Scans the relation via multiple workers and returns
+ *		the next qualifying tuple.
+ * ----------------------------------------------------------------
+ */
+TupleTableSlot *
+ExecFunnel(FunnelState *node)
+{
+	int			i;
+	TupleTableSlot *slot;
+
+	/*
+	 * Initialize the parallel context and workers on first execution. We do
+	 * this on first execution rather than during node initialization, as it
+	 * needs to allocate large dynamic segement, so it is better to do if it
+	 * is really needed.
+	 */
+	if (!node->pcxt)
+	{
+		EState	   *estate = node->ss.ps.state;
+		ExprContext *econtext = node->ss.ps.ps_ExprContext;
+		bool		any_worker_launched = false;
+		List	   *serialized_param_exec;
+
+		/*
+		 * Evaluate the InitPlan and pass the PARAM_EXEC params, so that
+		 * values can be shared with worker backend.  This is different from
+		 * the way InitPlans are evaluated (lazy evaluation) at other places
+		 * as instead of sharing the InitPlan to all the workers and let them
+		 * execute, we pass the values which can be directly used by worker
+		 * backends.
+		 */
+		serialized_param_exec = ExecAndFormSerializeParamExec(econtext,
+									   node->ss.ps.plan->lefttree->allParam);
+
+		/* Initialize the workers required to execute funnel node. */
+		InitializeParallelWorkers(node->ss.ps.lefttree,
+								  serialized_param_exec,
+								  estate,
+								  &node->inst_options_space,
+								  &node->buffer_usage_space,
+								  &node->responseq,
+								  &node->pcxt,
+							   ((Funnel *) (node->ss.ps.plan))->num_workers);
+
+		outerPlanState(node)->toc = node->pcxt->toc;
+
+		/*
+		 * Register backend workers. If the required number of workers are not
+		 * available then we perform the scan with available workers and if
+		 * there are no more workers available, then the funnel node will just
+		 * scan locally.
+		 */
+		LaunchParallelWorkers(node->pcxt);
+
+		node->funnel = CreateTupleQueueFunnel();
+
+		for (i = 0; i < node->pcxt->nworkers; ++i)
+		{
+			if (node->pcxt->worker[i].bgwhandle)
+			{
+				shm_mq_set_handle((node->responseq)[i], node->pcxt->worker[i].bgwhandle);
+				RegisterTupleQueueOnFunnel(node->funnel, (node->responseq)[i]);
+				any_worker_launched = true;
+			}
+		}
+
+		if (any_worker_launched)
+			node->fs_workersReady = true;
+	}
+
+	slot = funnel_getnext(node);
+
+	if (TupIsNull(slot))
+	{
+		/*
+		 * Destroy the parallel context once we complete fetching all the
+		 * tuples, this will ensure that if in the same statement we need to
+		 * have Funnel node for multiple parts of statement, it won't
+		 * accumulate lot of dsm segments and workers can be made available to
+		 * use by other parts of statement.
+		 */
+		DestroyParallelSetupAndAccumStats(node);
+	}
+	return slot;
+}
+
+/* ----------------------------------------------------------------
+ *		ExecEndFunnel
+ *
+ *		frees any storage allocated through C routines.
+ * ----------------------------------------------------------------
+ */
+void
+ExecEndFunnel(FunnelState *node)
+{
+	Relation	relation;
+
+	relation = node->ss.ss_currentRelation;
+
+	/*
+	 * Free the exprcontext
+	 */
+	ExecFreeExprContext(&node->ss.ps);
+
+	/*
+	 * clean out the tuple table
+	 */
+	ExecClearTuple(node->ss.ps.ps_ResultTupleSlot);
+
+	ExecEndNode(outerPlanState(node));
+
+	DestroyParallelSetupAndAccumStats(node);
+}
+
+/*
+ * funnel_getnext
+ *
+ * Get the next tuple from shared memory queue.  This function
+ * is reponsible for fetching tuples from all the queues associated
+ * with worker backends used in funnel scan and if there is no
+ * data available from queues or no worker is available, it does
+ * fetch the data from local node.
+ */
+TupleTableSlot *
+funnel_getnext(FunnelState *funnelstate)
+{
+	PlanState  *outerPlan;
+	TupleTableSlot *outerTupleSlot;
+	TupleTableSlot *slot;
+	HeapTuple	tup;
+
+	/*
+	 * We can use projection info of funnel for the tuples received from
+	 * worker backends as currently for all cases worker backends sends the
+	 * projected tuple as required by funnel node.
+	 */
+	slot = funnelstate->ss.ps.ps_ProjInfo->pi_slot;
+
+	while ((!funnelstate->all_workers_done && funnelstate->fs_workersReady) ||
+		   !funnelstate->local_scan_done)
+	{
+		if (!funnelstate->all_workers_done && funnelstate->fs_workersReady)
+		{
+			/* wait only if local scan is done */
+			tup = TupleQueueFunnelNext(funnelstate->funnel,
+									   !funnelstate->local_scan_done,
+									   &funnelstate->all_workers_done);
+
+			if (HeapTupleIsValid(tup))
+			{
+				ExecStoreTuple(tup,		/* tuple to store */
+							   slot,	/* slot to store in */
+							   InvalidBuffer,	/* buffer associated with this
+												 * tuple */
+							   true);	/* pfree this pointer if not from heap */
+
+				return slot;
+			}
+		}
+		if (!funnelstate->local_scan_done)
+		{
+			outerPlan = outerPlanState(funnelstate);
+
+			outerTupleSlot = ExecProcNode(outerPlan);
+
+			if (!TupIsNull(outerTupleSlot))
+				return outerTupleSlot;
+
+			funnelstate->local_scan_done = true;
+		}
+	}
+
+	return ExecClearTuple(slot);
+}
+
+/* ----------------------------------------------------------------
+ *		DestroyParallelSetupAndAccumStats
+ *
+ *		Destroy the setup for parallel workers.  Collect all the
+ *		stats after workers are stopped, else some work done by
+ *		workers won't be accounted.
+ * ----------------------------------------------------------------
+ */
+void
+DestroyParallelSetupAndAccumStats(FunnelState *node)
+{
+	if (node->pcxt)
+	{
+		/*
+		 * Ensure all workers have finished before destroying the parallel
+		 * context to ensure a clean exit.
+		 */
+		if (node->fs_workersReady)
+		{
+			DestroyTupleQueueFunnel(node->funnel);
+			node->funnel = NULL;
+			WaitForParallelWorkersToFinish(node->pcxt);
+		}
+
+		/*
+		 * Aggregate the buffer usage stats from all workers.  This is
+		 * required by external modules like pg_stat_statements.
+		 */
+		ExecAccumulateBufUsageInfo(node);
+
+		/*
+		 * Aggregate instrumentation information of all the backend workers
+		 * for Funnel node.  This has to be done before we destroy the
+		 * parallel context.
+		 */
+		if (node->ss.ps.state->es_instrument)
+			ExecAccumulateInstInfo(node);
+
+		/* destroy parallel context. */
+		DestroyParallelContext(node->pcxt);
+		node->pcxt = NULL;
+
+		node->fs_workersReady = false;
+		node->all_workers_done = false;
+		node->local_scan_done = false;
+	}
+}
+
+/* ----------------------------------------------------------------
+ *		ExecAccumulateInstInfo
+ *
+ *		Accumulate instrumentation information of all the workers
+ * ----------------------------------------------------------------
+ */
+void
+ExecAccumulateInstInfo(FunnelState *node)
+{
+	int			i;
+	Instrumentation *instrument_worker;
+	int			nworkers;
+	char	   *inst_info_workers;
+
+	if (node->pcxt)
+	{
+		nworkers = node->pcxt->nworkers;
+		inst_info_workers = node->inst_options_space;
+		for (i = 0; i < nworkers; i++)
+		{
+			instrument_worker = (Instrumentation *) (inst_info_workers + (i * sizeof(Instrumentation)));
+			InstrAggNode(node->ss.ps.instrument, instrument_worker);
+		}
+	}
+}
+
+/* ----------------------------------------------------------------
+ *		ExecAccumulateBufUsageInfo
+ *
+ *		Accumulate buffer usage information of all the workers
+ * ----------------------------------------------------------------
+ */
+void
+ExecAccumulateBufUsageInfo(FunnelState *node)
+{
+	int			i;
+	int			nworkers;
+	BufferUsage *buffer_usage_worker;
+	char	   *buffer_usage;
+
+	if (node->pcxt)
+	{
+		nworkers = node->pcxt->nworkers;
+		buffer_usage = node->buffer_usage_space;
+
+		for (i = 0; i < nworkers; i++)
+		{
+			buffer_usage_worker = &((BufferUsage *) buffer_usage)[i];
+			BufferUsageAdd(&pgBufferUsage, buffer_usage_worker);
+		}
+	}
+}
+
+/* ----------------------------------------------------------------
+ *						Join Support
+ * ----------------------------------------------------------------
+ */
+
+/* ----------------------------------------------------------------
+ *		ExecReScanFunnel
+ *
+ *		Re-initialize the workers and rescans a relation via them.
+ * ----------------------------------------------------------------
+ */
+void
+ExecReScanFunnel(FunnelState *node)
+{
+	/*
+	 * Re-initialize the parallel context and workers to perform rescan of
+	 * relation.  We want to gracefully shutdown all the workers so that they
+	 * should be able to propagate any error or other information to master
+	 * backend before dying.
+	 */
+	DestroyParallelSetupAndAccumStats(node);
+
+	ExecReScan(node->ss.ps.lefttree);
+}
diff --git a/src/backend/executor/nodeNestloop.c b/src/backend/executor/nodeNestloop.c
index e66bcda..50c2f39 100644
--- a/src/backend/executor/nodeNestloop.c
+++ b/src/backend/executor/nodeNestloop.c
@@ -144,6 +144,7 @@ ExecNestLoop(NestLoopState *node)
 			{
 				NestLoopParam *nlp = (NestLoopParam *) lfirst(lc);
 				int			paramno = nlp->paramno;
+				TupleDesc	tdesc = outerTupleSlot->tts_tupleDescriptor;
 				ParamExecData *prm;
 
 				prm = &(econtext->ecxt_param_exec_vals[paramno]);
@@ -154,6 +155,7 @@ ExecNestLoop(NestLoopState *node)
 				prm->value = slot_getattr(outerTupleSlot,
 										  nlp->paramval->varattno,
 										  &(prm->isnull));
+				prm->ptype = tdesc->attrs[nlp->paramval->varattno - 1]->atttypid;
 				/* Flag parameter value as changed */
 				innerPlan->chgParam = bms_add_member(innerPlan->chgParam,
 													 paramno);
diff --git a/src/backend/executor/nodeSubplan.c b/src/backend/executor/nodeSubplan.c
index 9eb4d63..4c33bcf 100644
--- a/src/backend/executor/nodeSubplan.c
+++ b/src/backend/executor/nodeSubplan.c
@@ -30,11 +30,14 @@
 #include <math.h>
 
 #include "access/htup_details.h"
+#include "catalog/pg_type.h"
 #include "executor/executor.h"
 #include "executor/nodeSubplan.h"
 #include "nodes/makefuncs.h"
+#include "nodes/nodeFuncs.h"
 #include "optimizer/clauses.h"
 #include "utils/array.h"
+#include "utils/datum.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 
@@ -281,12 +284,14 @@ ExecScanSubPlan(SubPlanState *node,
 	forboth(l, subplan->parParam, pvar, node->args)
 	{
 		int			paramid = lfirst_int(l);
+		ExprState  *exprstate = (ExprState *) lfirst(pvar);
 		ParamExecData *prm = &(econtext->ecxt_param_exec_vals[paramid]);
 
-		prm->value = ExecEvalExprSwitchContext((ExprState *) lfirst(pvar),
+		prm->value = ExecEvalExprSwitchContext(exprstate,
 											   econtext,
 											   &(prm->isnull),
 											   NULL);
+		prm->ptype = exprType((Node *) exprstate->expr);
 		planstate->chgParam = bms_add_member(planstate->chgParam, paramid);
 	}
 
@@ -399,6 +404,7 @@ ExecScanSubPlan(SubPlanState *node,
 			prmdata = &(econtext->ecxt_param_exec_vals[paramid]);
 			Assert(prmdata->execPlan == NULL);
 			prmdata->value = slot_getattr(slot, col, &(prmdata->isnull));
+			prmdata->ptype = tdesc->attrs[col - 1]->atttypid;
 			col++;
 		}
 
@@ -551,6 +557,7 @@ buildSubPlanHash(SubPlanState *node, ExprContext *econtext)
 		 !TupIsNull(slot);
 		 slot = ExecProcNode(planstate))
 	{
+		TupleDesc	tdesc = slot->tts_tupleDescriptor;
 		int			col = 1;
 		ListCell   *plst;
 		bool		isnew;
@@ -568,6 +575,7 @@ buildSubPlanHash(SubPlanState *node, ExprContext *econtext)
 			Assert(prmdata->execPlan == NULL);
 			prmdata->value = slot_getattr(slot, col,
 										  &(prmdata->isnull));
+			prmdata->ptype = tdesc->attrs[col - 1]->atttypid;
 			col++;
 		}
 		slot = ExecProject(node->projRight, NULL);
@@ -954,6 +962,7 @@ ExecSetParamPlan(SubPlanState *node, ExprContext *econtext)
 	ListCell   *l;
 	bool		found = false;
 	ArrayBuildStateAny *astate = NULL;
+	Oid			ptype;
 
 	if (subLinkType == ANY_SUBLINK ||
 		subLinkType == ALL_SUBLINK)
@@ -961,6 +970,8 @@ ExecSetParamPlan(SubPlanState *node, ExprContext *econtext)
 	if (subLinkType == CTE_SUBLINK)
 		elog(ERROR, "CTE subplans should not be executed via ExecSetParamPlan");
 
+	ptype = exprType((Node *) node->xprstate.expr);
+
 	/* Initialize ArrayBuildStateAny in caller's context, if needed */
 	if (subLinkType == ARRAY_SUBLINK)
 		astate = initArrayResultAny(subplan->firstColType,
@@ -983,12 +994,14 @@ ExecSetParamPlan(SubPlanState *node, ExprContext *econtext)
 	forboth(l, subplan->parParam, pvar, node->args)
 	{
 		int			paramid = lfirst_int(l);
+		ExprState  *exprstate = (ExprState *) lfirst(pvar);
 		ParamExecData *prm = &(econtext->ecxt_param_exec_vals[paramid]);
 
-		prm->value = ExecEvalExprSwitchContext((ExprState *) lfirst(pvar),
+		prm->value = ExecEvalExprSwitchContext(exprstate,
 											   econtext,
 											   &(prm->isnull),
 											   NULL);
+		prm->ptype = exprType((Node *) exprstate->expr);
 		planstate->chgParam = bms_add_member(planstate->chgParam, paramid);
 	}
 
@@ -1011,6 +1024,7 @@ ExecSetParamPlan(SubPlanState *node, ExprContext *econtext)
 
 			prm->execPlan = NULL;
 			prm->value = BoolGetDatum(true);
+			prm->ptype = ptype;
 			prm->isnull = false;
 			found = true;
 			break;
@@ -1062,6 +1076,7 @@ ExecSetParamPlan(SubPlanState *node, ExprContext *econtext)
 			prm->execPlan = NULL;
 			prm->value = heap_getattr(node->curTuple, i, tdesc,
 									  &(prm->isnull));
+			prm->ptype = tdesc->attrs[i - 1]->atttypid;
 			i++;
 		}
 	}
@@ -1084,6 +1099,7 @@ ExecSetParamPlan(SubPlanState *node, ExprContext *econtext)
 											true);
 		prm->execPlan = NULL;
 		prm->value = node->curArray;
+		prm->ptype = ptype;
 		prm->isnull = false;
 	}
 	else if (!found)
@@ -1096,6 +1112,7 @@ ExecSetParamPlan(SubPlanState *node, ExprContext *econtext)
 
 			prm->execPlan = NULL;
 			prm->value = BoolGetDatum(false);
+			prm->ptype = ptype;
 			prm->isnull = false;
 		}
 		else
@@ -1108,6 +1125,7 @@ ExecSetParamPlan(SubPlanState *node, ExprContext *econtext)
 
 				prm->execPlan = NULL;
 				prm->value = (Datum) 0;
+				prm->ptype = VOIDOID;
 				prm->isnull = true;
 			}
 		}
@@ -1238,3 +1256,47 @@ ExecAlternativeSubPlan(AlternativeSubPlanState *node,
 					   isNull,
 					   isDone);
 }
+
+/*
+ * ExecAndFormSerializeParamExec
+ *
+ * Execute the subplan stored in PARAM_EXEC param if it is not executed
+ * till now and form the serialized structure required for passing to
+ * worker backend.
+ */
+List *
+ExecAndFormSerializeParamExec(ExprContext *econtext, Bitmapset *params)
+{
+	List	   *lparam = NIL;
+	SerializedParamExecData *sparamdata;
+	ParamExecData *prm;
+	int			paramid;
+
+	paramid = -1;
+	while ((paramid = bms_next_member(params, paramid)) >= 0)
+	{
+		/*
+		 * PARAM_EXEC params (internal executor parameters) are stored in the
+		 * ecxt_param_exec_vals array, and can be accessed by array index.
+		 */
+		sparamdata = palloc0(sizeof(SerializedParamExecData));
+
+		prm = &(econtext->ecxt_param_exec_vals[paramid]);
+		if (prm->execPlan != NULL)
+		{
+			/* Parameter not evaluated yet, so go do it */
+			ExecSetParamPlan(prm->execPlan, econtext);
+			/* ExecSetParamPlan should have processed this param... */
+			Assert(prm->execPlan == NULL);
+		}
+
+		sparamdata->paramid = paramid;
+		sparamdata->ptype = prm->ptype;
+		sparamdata->value = prm->value;
+		sparamdata->isnull = prm->isnull;
+
+		lparam = lappend(lparam, sparamdata);
+	}
+
+	return lparam;
+}
diff --git a/src/backend/executor/spi.c b/src/backend/executor/spi.c
index 300401e..a60f228 100644
--- a/src/backend/executor/spi.c
+++ b/src/backend/executor/spi.c
@@ -1774,7 +1774,7 @@ spi_dest_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
  *		store tuple retrieved by Executor into SPITupleTable
  *		of current SPI procedure
  */
-void
+bool
 spi_printtup(TupleTableSlot *slot, DestReceiver *self)
 {
 	SPITupleTable *tuptable;
@@ -1809,6 +1809,8 @@ spi_printtup(TupleTableSlot *slot, DestReceiver *self)
 	(tuptable->free)--;
 
 	MemoryContextSwitchTo(oldcxt);
+
+	return true;
 }
 
 /*
diff --git a/src/backend/executor/tqueue.c b/src/backend/executor/tqueue.c
index d0edf4e..4e27b30 100644
--- a/src/backend/executor/tqueue.c
+++ b/src/backend/executor/tqueue.c
@@ -41,14 +41,24 @@ struct TupleQueueFunnel
 /*
  * Receive a tuple.
  */
-static void
+static bool
 tqueueReceiveSlot(TupleTableSlot *slot, DestReceiver *self)
 {
 	TQueueDestReceiver *tqueue = (TQueueDestReceiver *) self;
 	HeapTuple	tuple;
+	shm_mq_result result;
 
 	tuple = ExecMaterializeSlot(slot);
-	shm_mq_send(tqueue->handle, tuple->t_len, tuple->t_data, false);
+	result = shm_mq_send(tqueue->handle, tuple->t_len, tuple->t_data, false);
+
+	if (result == SHM_MQ_DETACHED)
+		return false;
+	else if (result != SHM_MQ_SUCCESS)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("unable to send tuples")));
+
+	return true;
 }
 
 /*
@@ -82,7 +92,7 @@ tqueueDestroyReceiver(DestReceiver *self)
  * Create a DestReceiver that writes tuples to a tuple queue.
  */
 DestReceiver *
-CreateTupleQueueDestReceiver(shm_mq_handle *handle)
+CreateTupleQueueDestReceiver(void)
 {
 	TQueueDestReceiver *self;
 
@@ -93,12 +103,25 @@ CreateTupleQueueDestReceiver(shm_mq_handle *handle)
 	self->pub.rShutdown = tqueueShutdownReceiver;
 	self->pub.rDestroy = tqueueDestroyReceiver;
 	self->pub.mydest = DestTupleQueue;
-	self->handle = handle;
+
+	/* private fields will be set by SetTupleQueueDestReceiverParams */
 
 	return (DestReceiver *) self;
 }
 
 /*
+ * Set parameters for a TupleQueueDestReceiver
+ */
+void
+SetTupleQueueDestReceiverParams(DestReceiver *self,
+								shm_mq_handle *handle)
+{
+	TQueueDestReceiver *myState = (TQueueDestReceiver *) self;
+
+	myState->handle = handle;
+}
+
+/*
  * Create a tuple queue funnel.
  */
 TupleQueueFunnel *
diff --git a/src/backend/executor/tstoreReceiver.c b/src/backend/executor/tstoreReceiver.c
index c1fdeb7..b0862ae 100644
--- a/src/backend/executor/tstoreReceiver.c
+++ b/src/backend/executor/tstoreReceiver.c
@@ -37,8 +37,8 @@ typedef struct
 } TStoreState;
 
 
-static void tstoreReceiveSlot_notoast(TupleTableSlot *slot, DestReceiver *self);
-static void tstoreReceiveSlot_detoast(TupleTableSlot *slot, DestReceiver *self);
+static bool tstoreReceiveSlot_notoast(TupleTableSlot *slot, DestReceiver *self);
+static bool tstoreReceiveSlot_detoast(TupleTableSlot *slot, DestReceiver *self);
 
 
 /*
@@ -90,19 +90,21 @@ tstoreStartupReceiver(DestReceiver *self, int operation, TupleDesc typeinfo)
  * Receive a tuple from the executor and store it in the tuplestore.
  * This is for the easy case where we don't have to detoast.
  */
-static void
+static bool
 tstoreReceiveSlot_notoast(TupleTableSlot *slot, DestReceiver *self)
 {
 	TStoreState *myState = (TStoreState *) self;
 
 	tuplestore_puttupleslot(myState->tstore, slot);
+
+	return true;
 }
 
 /*
  * Receive a tuple from the executor and store it in the tuplestore.
  * This is for the case where we have to detoast any toasted values.
  */
-static void
+static bool
 tstoreReceiveSlot_detoast(TupleTableSlot *slot, DestReceiver *self)
 {
 	TStoreState *myState = (TStoreState *) self;
@@ -152,6 +154,8 @@ tstoreReceiveSlot_detoast(TupleTableSlot *slot, DestReceiver *self)
 	/* And release any temporary detoasted values */
 	for (i = 0; i < nfree; i++)
 		pfree(DatumGetPointer(myState->tofree[i]));
+
+	return true;
 }
 
 /*
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 62355aa..ec605bd 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -382,6 +382,27 @@ _copySampleScan(const SampleScan *from)
 }
 
 /*
+ * _copyFunnel
+ */
+static Funnel *
+_copyFunnel(const Funnel *from)
+{
+	Funnel	   *newnode = makeNode(Funnel);
+
+	/*
+	 * copy node superclass fields
+	 */
+	CopyScanFields((const Scan *) from, (Scan *) newnode);
+
+	/*
+	 * copy remainder of node
+	 */
+	COPY_SCALAR_FIELD(num_workers);
+
+	return newnode;
+}
+
+/*
  * _copyIndexScan
  */
 static IndexScan *
@@ -4240,6 +4261,9 @@ copyObject(const void *from)
 		case T_SampleScan:
 			retval = _copySampleScan(from);
 			break;
+		case T_Funnel:
+			retval = _copyFunnel(from);
+			break;
 		case T_IndexScan:
 			retval = _copyIndexScan(from);
 			break;
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index c91273c..bc9b481 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -458,6 +458,16 @@ _outSampleScan(StringInfo str, const SampleScan *node)
 }
 
 static void
+_outFunnel(StringInfo str, const Funnel *node)
+{
+	WRITE_NODE_TYPE("FUNNEL");
+
+	_outScanInfo(str, (const Scan *) node);
+
+	WRITE_UINT_FIELD(num_workers);
+}
+
+static void
 _outIndexScan(StringInfo str, const IndexScan *node)
 {
 	WRITE_NODE_TYPE("INDEXSCAN");
@@ -3008,6 +3018,9 @@ _outNode(StringInfo str, const void *obj)
 			case T_SampleScan:
 				_outSampleScan(str, obj);
 				break;
+			case T_Funnel:
+				_outFunnel(str, obj);
+				break;
 			case T_IndexScan:
 				_outIndexScan(str, obj);
 				break;
diff --git a/src/backend/nodes/params.c b/src/backend/nodes/params.c
index fb803f8..5751386 100644
--- a/src/backend/nodes/params.c
+++ b/src/backend/nodes/params.c
@@ -16,9 +16,22 @@
 #include "postgres.h"
 
 #include "nodes/params.h"
+#include "storage/shmem.h"
 #include "utils/datum.h"
 #include "utils/lsyscache.h"
 
+/*
+ * for each bind parameter, pass this structure followed by value
+ * except for pass-by-value parameters.
+ */
+typedef struct SerializedParamExternData
+{
+	Datum		value;			/* pass-by-val are directly stored */
+	Size		length;			/* length of parameter value */
+	bool		isnull;			/* is it NULL? */
+	uint16		pflags;			/* flag bits, same as in original Param */
+	Oid			ptype;			/* parameter's datatype, or 0 */
+}	SerializedParamExternData;
 
 /*
  * Copy a ParamListInfo structure.
@@ -73,3 +86,357 @@ copyParamList(ParamListInfo from)
 
 	return retval;
 }
+
+/*
+ * Estimate the amount of space required to serialize the bound
+ * parameters.
+ */
+Size
+EstimateBoundParametersSpace(ParamListInfo paramInfo)
+{
+	Size		size;
+	int			i;
+
+	/* Add space required for saving numParams */
+	size = sizeof(int);
+
+	if (paramInfo)
+	{
+		/* Add space required for saving the param data */
+		for (i = 0; i < paramInfo->numParams; i++)
+		{
+			/*
+			 * for each parameter, calculate the size of fixed part of
+			 * parameter (SerializedParamExternData) and length of parameter
+			 * value.
+			 */
+			ParamExternData *oprm;
+			int16		typLen;
+			bool		typByVal;
+			Size		length;
+
+			length = sizeof(SerializedParamExternData);
+
+			oprm = &paramInfo->params[i];
+
+			get_typlenbyval(oprm->ptype, &typLen, &typByVal);
+
+			/*
+			 * pass-by-value parameters are directly stored in
+			 * SerializedParamExternData, so no need of additional space for
+			 * them.
+			 */
+			if (!(typByVal || oprm->isnull))
+			{
+				length += datumGetSize(oprm->value, typByVal, typLen);
+				size = add_size(size, length);
+
+				/* Allow space for terminating zero-byte */
+				size = add_size(size, 1);
+			}
+			else
+				size = add_size(size, length);
+		}
+	}
+
+	return size;
+}
+
+/*
+ * Serialize the bind parameters into the memory, beginning at start_address.
+ * maxsize should be at least as large as the value returned by
+ * EstimateBoundParametersSpace.
+ */
+void
+SerializeBoundParams(ParamListInfo paramInfo, Size maxsize, char *start_address)
+{
+	char	   *curptr;
+	SerializedParamExternData *retval;
+	int			i;
+
+	/*
+	 * First, we store the number of bind parameters, if there is no bind
+	 * parameter then no need to store any more information.
+	 */
+	if (paramInfo && paramInfo->numParams > 0)
+		*(int *) start_address = paramInfo->numParams;
+	else
+	{
+		*(int *) start_address = 0;
+		return;
+	}
+	curptr = start_address + sizeof(int);
+
+
+	for (i = 0; i < paramInfo->numParams; i++)
+	{
+		ParamExternData *oprm;
+		int16		typLen;
+		bool		typByVal;
+		Size		datumlength,
+					length;
+		const char *s;
+
+		Assert(curptr <= start_address + maxsize);
+		retval = (SerializedParamExternData *) curptr;
+		oprm = &paramInfo->params[i];
+
+		retval->isnull = oprm->isnull;
+		retval->pflags = oprm->pflags;
+		retval->ptype = oprm->ptype;
+		retval->value = oprm->value;
+
+		curptr = curptr + sizeof(SerializedParamExternData);
+
+		if (retval->isnull)
+			continue;
+
+		get_typlenbyval(oprm->ptype, &typLen, &typByVal);
+
+		if (!typByVal)
+		{
+			datumlength = datumGetSize(oprm->value, typByVal, typLen);
+			s = (char *) DatumGetPointer(oprm->value);
+			memcpy(curptr, s, datumlength);
+			length = datumlength;
+			curptr[length] = '\0';
+			retval->length = length;
+			curptr += length + 1;
+		}
+	}
+}
+
+/*
+ * RestoreBoundParams
+ *		Restore bind parameters from the specified address.
+ *
+ * The params are palloc'd in CurrentMemoryContext.
+ */
+ParamListInfo
+RestoreBoundParams(char *start_address)
+{
+	ParamListInfo retval;
+	Size		size;
+	int			num_params,
+				i;
+	char	   *curptr;
+
+	num_params = *(int *) start_address;
+
+	if (num_params <= 0)
+		return NULL;
+
+	size = offsetof(ParamListInfoData, params) +
+		num_params * sizeof(ParamExternData);
+	retval = (ParamListInfo) palloc(size);
+	retval->paramFetch = NULL;
+	retval->paramFetchArg = NULL;
+	retval->parserSetup = NULL;
+	retval->parserSetupArg = NULL;
+	retval->numParams = num_params;
+
+	curptr = start_address + sizeof(int);
+
+	for (i = 0; i < num_params; i++)
+	{
+		SerializedParamExternData *nprm;
+		char	   *s;
+		int16		typLen;
+		bool		typByVal;
+
+		nprm = (SerializedParamExternData *) curptr;
+
+		/* copy the parameter info */
+		retval->params[i].isnull = nprm->isnull;
+		retval->params[i].pflags = nprm->pflags;
+		retval->params[i].ptype = nprm->ptype;
+		retval->params[i].value = nprm->value;
+
+		curptr = curptr + sizeof(SerializedParamExternData);
+
+		if (nprm->isnull)
+			continue;
+
+		get_typlenbyval(nprm->ptype, &typLen, &typByVal);
+
+		if (!typByVal)
+		{
+			s = palloc(nprm->length + 1);
+			memcpy(s, curptr, nprm->length + 1);
+			retval->params[i].value = CStringGetDatum(s);
+
+			curptr += nprm->length + 1;
+		}
+	}
+
+	return retval;
+}
+
+/*
+ * Estimate the amount of space required to serialize the PARAM_EXEC
+ * parameters.
+ */
+Size
+EstimateExecParametersSpace(List *serialized_param_exec_vals)
+{
+	Size		size;
+	ListCell   *lparam;
+
+	/*
+	 * Add space required for saving number of PARAM_EXEC parameters that
+	 * needs to be serialized.
+	 */
+	size = sizeof(int);
+
+	foreach(lparam, serialized_param_exec_vals)
+	{
+		int16		typLen;
+		bool		typByVal;
+		Size		length;
+		SerializedParamExecData *param_val = (SerializedParamExecData *) lfirst(lparam);
+
+		length = sizeof(SerializedParamExecData);
+
+		get_typlenbyval(param_val->ptype, &typLen, &typByVal);
+
+		/*
+		 * pass-by-value parameters are directly stored in
+		 * SerializedParamExternData, so no need of additional space for them.
+		 */
+		if (!(typByVal || param_val->isnull))
+		{
+			length += datumGetSize(param_val->value, typByVal, typLen);
+			size = add_size(size, length);
+
+			/* Allow space for terminating zero-byte */
+			size = add_size(size, 1);
+		}
+		else
+			size = add_size(size, length);
+	}
+
+	return size;
+}
+
+/*
+ * Serialize the PARAM_EXEC parameters into the memory, beginning at
+ * start_address.  maxsize should be at least as large as the value
+ * returned by EstimateExecParametersSpace.
+ */
+void
+SerializeExecParams(List *serialized_param_exec_vals, Size maxsize,
+					char *start_address)
+{
+	char	   *curptr;
+	SerializedParamExecData *retval;
+	ListCell   *lparam;
+
+	/*
+	 * First, we store the number of PARAM_EXEC parameters that needs to be
+	 * serialized.
+	 */
+	if (serialized_param_exec_vals)
+		*(int *) start_address = list_length(serialized_param_exec_vals);
+	else
+	{
+		*(int *) start_address = 0;
+		return;
+	}
+
+	curptr = start_address + sizeof(int);
+
+	foreach(lparam, serialized_param_exec_vals)
+	{
+		int16		typLen;
+		bool		typByVal;
+		Size		datumlength,
+					length;
+		const char *s;
+		SerializedParamExecData *param_val = (SerializedParamExecData *) lfirst(lparam);
+
+		retval = (SerializedParamExecData *) curptr;
+
+		retval->paramid = param_val->paramid;
+		retval->value = param_val->value;
+		retval->isnull = param_val->isnull;
+		retval->ptype = param_val->ptype;
+
+		curptr = curptr + sizeof(SerializedParamExecData);
+
+		if (retval->isnull)
+			continue;
+
+		get_typlenbyval(retval->ptype, &typLen, &typByVal);
+
+		if (!typByVal)
+		{
+			datumlength = datumGetSize(retval->value, typByVal, typLen);
+			s = (char *) DatumGetPointer(retval->value);
+			memcpy(curptr, s, datumlength);
+			length = datumlength;
+			curptr[length] = '\0';
+			retval->length = length;
+			curptr += length + 1;
+		}
+	}
+}
+
+/*
+ * RestoreExecParams
+ *		Restore PARAM_EXEC parameters from the specified address.
+ *
+ * The params are palloc'd in CurrentMemoryContext.
+ */
+List *
+RestoreExecParams(char *start_address)
+{
+	List	   *lparamexecvals = NIL;
+	int			num_params,
+				i;
+	char	   *curptr;
+
+	num_params = *(int *) start_address;
+
+	if (num_params <= 0)
+		return NULL;
+
+	curptr = start_address + sizeof(int);
+
+	for (i = 0; i < num_params; i++)
+	{
+		SerializedParamExecData *nprm;
+		SerializedParamExecData *outparam;
+		char	   *s;
+		int16		typLen;
+		bool		typByVal;
+
+		nprm = (SerializedParamExecData *) curptr;
+
+		outparam = palloc0(sizeof(SerializedParamExecData));
+
+		/* copy the parameter info */
+		outparam->isnull = nprm->isnull;
+		outparam->value = nprm->value;
+		outparam->paramid = nprm->paramid;
+
+		curptr = curptr + sizeof(SerializedParamExecData);
+
+		if (nprm->isnull)
+			continue;
+
+		get_typlenbyval(nprm->ptype, &typLen, &typByVal);
+
+		if (!typByVal)
+		{
+			s = palloc(nprm->length + 1);
+			memcpy(s, curptr, nprm->length + 1);
+			outparam->value = CStringGetDatum(s);
+
+			curptr += nprm->length + 1;
+		}
+
+		lparamexecvals = lappend(lparamexecvals, outparam);
+	}
+
+	return lparamexecvals;
+}
diff --git a/src/backend/optimizer/path/costsize.c b/src/backend/optimizer/path/costsize.c
index d107d76..0a40f6c 100644
--- a/src/backend/optimizer/path/costsize.c
+++ b/src/backend/optimizer/path/costsize.c
@@ -11,6 +11,8 @@
  *	cpu_tuple_cost		Cost of typical CPU time to process a tuple
  *	cpu_index_tuple_cost  Cost of typical CPU time to process an index tuple
  *	cpu_operator_cost	Cost of CPU time to execute an operator or function
+ *	cpu_tuple_comm_cost Cost of CPU time to pass a tuple from worker to master backend
+ *	parallel_setup_cost Cost of setting up shared memory for parallelism
  *
  * We expect that the kernel will typically do some amount of read-ahead
  * optimization; this in conjunction with seek costs means that seq_page_cost
@@ -102,11 +104,15 @@ double		random_page_cost = DEFAULT_RANDOM_PAGE_COST;
 double		cpu_tuple_cost = DEFAULT_CPU_TUPLE_COST;
 double		cpu_index_tuple_cost = DEFAULT_CPU_INDEX_TUPLE_COST;
 double		cpu_operator_cost = DEFAULT_CPU_OPERATOR_COST;
+double		cpu_tuple_comm_cost = DEFAULT_CPU_TUPLE_COMM_COST;
+double		parallel_setup_cost = DEFAULT_PARALLEL_SETUP_COST;
 
 int			effective_cache_size = DEFAULT_EFFECTIVE_CACHE_SIZE;
 
 Cost		disable_cost = 1.0e10;
 
+int			parallel_seqscan_degree = 0;
+
 bool		enable_seqscan = true;
 bool		enable_indexscan = true;
 bool		enable_indexonlyscan = true;
@@ -290,6 +296,42 @@ cost_samplescan(Path *path, PlannerInfo *root,
 }
 
 /*
+ * cost_funnel
+ *	  Determines and returns the cost of funnel path.
+ *
+ * 'baserel' is the relation to be scanned
+ * 'param_info' is the ParamPathInfo if this is a parameterized path, else NULL
+ */
+void
+cost_funnel(FunnelPath *path, PlannerInfo *root,
+			RelOptInfo *baserel, ParamPathInfo *param_info)
+{
+	Cost		startup_cost = 0;
+	Cost		run_cost = 0;
+
+	/* Should only be applied to base relations */
+	Assert(baserel->relid > 0);
+	Assert(baserel->rtekind == RTE_RELATION);
+
+	/* Mark the path with the correct row estimate */
+	if (param_info)
+		path->path.rows = param_info->ppi_rows;
+	else
+		path->path.rows = baserel->rows;
+
+	startup_cost = path->subpath->startup_cost;
+
+	run_cost = path->subpath->total_cost - path->subpath->startup_cost;
+
+	/* Parallel setup and communication cost. */
+	startup_cost += parallel_setup_cost;
+	run_cost += cpu_tuple_comm_cost * baserel->tuples;
+
+	path->path.startup_cost = startup_cost;
+	path->path.total_cost = (startup_cost + run_cost);
+}
+
+/*
  * cost_index
  *	  Determines and returns the cost of scanning a relation using an index.
  *
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index 404c6f5..e987922 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -60,6 +60,8 @@ static SeqScan *create_seqscan_plan(PlannerInfo *root, Path *best_path,
 					List *tlist, List *scan_clauses);
 static SampleScan *create_samplescan_plan(PlannerInfo *root, Path *best_path,
 					   List *tlist, List *scan_clauses);
+static Funnel *create_funnel_plan(PlannerInfo *root,
+				   FunnelPath *best_path);
 static Scan *create_indexscan_plan(PlannerInfo *root, IndexPath *best_path,
 					  List *tlist, List *scan_clauses, bool indexonly);
 static BitmapHeapScan *create_bitmap_scan_plan(PlannerInfo *root,
@@ -104,6 +106,9 @@ static void copy_plan_costsize(Plan *dest, Plan *src);
 static SeqScan *make_seqscan(List *qptlist, List *qpqual, Index scanrelid);
 static SampleScan *make_samplescan(List *qptlist, List *qpqual, Index scanrelid,
 				TableSampleClause *tsc);
+static Funnel *make_funnel(List *qptlist, List *qpqual,
+			Index scanrelid, int nworkers,
+			Plan *subplan);
 static IndexScan *make_indexscan(List *qptlist, List *qpqual, Index scanrelid,
 			   Oid indexid, List *indexqual, List *indexqualorig,
 			   List *indexorderby, List *indexorderbyorig,
@@ -273,6 +278,10 @@ create_plan_recurse(PlannerInfo *root, Path *best_path)
 			plan = create_unique_plan(root,
 									  (UniquePath *) best_path);
 			break;
+		case T_Funnel:
+			plan = (Plan *) create_funnel_plan(root,
+											   (FunnelPath *) best_path);
+			break;
 		default:
 			elog(ERROR, "unrecognized node type: %d",
 				 (int) best_path->pathtype);
@@ -560,6 +569,7 @@ disuse_physical_tlist(PlannerInfo *root, Plan *plan, Path *path)
 	{
 		case T_SeqScan:
 		case T_SampleScan:
+		case T_Funnel:
 		case T_IndexScan:
 		case T_IndexOnlyScan:
 		case T_BitmapHeapScan:
@@ -1194,6 +1204,66 @@ create_samplescan_plan(PlannerInfo *root, Path *best_path,
 }
 
 /*
+ * create_funnel_plan
+ *
+ * Returns a funnel plan for the base relation scanned by
+ * 'best_path'.
+ */
+static Funnel *
+create_funnel_plan(PlannerInfo *root, FunnelPath *best_path)
+{
+	Funnel	   *funnel_plan;
+	Plan	   *subplan;
+	List	   *tlist;
+	RelOptInfo *rel = best_path->path.parent;
+	Index		scan_relid = best_path->path.parent->relid;
+
+	/*
+	 * For table scans, rather than using the relation targetlist (which is
+	 * only those Vars actually needed by the query), we prefer to generate a
+	 * tlist containing all Vars in order.  This will allow the executor to
+	 * optimize away projection of the table tuples, if possible.  (Note that
+	 * planner.c may replace the tlist we generate here, forcing projection to
+	 * occur.)
+	 */
+	if (use_physical_tlist(root, rel))
+	{
+		tlist = build_physical_tlist(root, rel);
+		/* if fail because of dropped cols, use regular method */
+		if (tlist == NIL)
+			tlist = build_path_tlist(root, &best_path->path);
+	}
+	else
+	{
+		tlist = build_path_tlist(root, &best_path->path);
+	}
+
+	/* it should be a base rel... */
+	Assert(scan_relid > 0);
+	Assert(best_path->path.parent->rtekind == RTE_RELATION);
+
+	subplan = create_plan_recurse(root, best_path->subpath);
+
+	/*
+	 * quals for subplan and top level plan are same as either all the quals
+	 * are pushed to subplan (partialseqscan plan) or parallel plan won't be
+	 * choosen.
+	 */
+	funnel_plan = make_funnel(tlist,
+							  subplan->qual,
+							  scan_relid,
+							  best_path->num_workers,
+							  subplan);
+
+	copy_path_costsize(&funnel_plan->scan.plan, &best_path->path);
+
+	/* use parallel mode for parallel plans. */
+	root->glob->parallelModeNeeded = true;
+
+	return funnel_plan;
+}
+
+/*
  * create_indexscan_plan
  *	  Returns an indexscan plan for the base relation scanned by 'best_path'
  *	  with restriction clauses 'scan_clauses' and targetlist 'tlist'.
@@ -3462,6 +3532,27 @@ make_samplescan(List *qptlist,
 	return node;
 }
 
+static Funnel *
+make_funnel(List *qptlist,
+			List *qpqual,
+			Index scanrelid,
+			int nworkers,
+			Plan *subplan)
+{
+	Funnel	   *node = makeNode(Funnel);
+	Plan	   *plan = &node->scan.plan;
+
+	/* cost should be inserted by caller */
+	plan->targetlist = qptlist;
+	plan->qual = qpqual;
+	plan->lefttree = subplan;
+	plan->righttree = NULL;
+	node->scan.scanrelid = scanrelid;
+	node->num_workers = nworkers;
+
+	return node;
+}
+
 static IndexScan *
 make_indexscan(List *qptlist,
 			   List *qpqual,
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index 06be922..efad32b 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -342,6 +342,53 @@ standard_planner(Query *parse, int cursorOptions, ParamListInfo boundParams)
 	return result;
 }
 
+PlannedStmt *
+create_parallel_worker_plannedstmt(Plan *plan,
+								   List *rangetable,
+								   int num_exec_params)
+{
+	PlannedStmt *result;
+	ListCell   *tlist;
+
+	/*
+	 * Avoid removing junk entries in worker as those are required by upper
+	 * nodes in master backend.
+	 */
+	foreach(tlist, plan->targetlist)
+	{
+		TargetEntry *tle = (TargetEntry *) lfirst(tlist);
+
+		tle->resjunk = false;
+	}
+
+	/* build the PlannedStmt result */
+	result = makeNode(PlannedStmt);
+
+	result->commandType = CMD_SELECT;
+	result->queryId = 0;
+	result->hasReturning = 0;
+	result->hasModifyingCTE = 0;
+	result->canSetTag = 1;
+	result->transientPlan = 0;
+	result->planTree = plan;
+	result->rtable = rangetable;
+	result->resultRelations = NIL;
+	result->utilityStmt = NULL;
+	result->subplans = NIL;
+	result->rewindPlanIDs = NULL;
+	result->rowMarks = NIL;
+	result->nParamExec = num_exec_params;
+
+	/*
+	 * Don't bother to set parameters used for invalidation as worker backend
+	 * plans are not saved, so can't be invalidated.
+	 */
+	result->relationOids = NIL;
+	result->invalItems = NIL;
+	result->hasRowSecurity = false;
+
+	return result;
+}
 
 /*--------------------
  * subquery_planner
diff --git a/src/backend/optimizer/plan/setrefs.c b/src/backend/optimizer/plan/setrefs.c
index daeb584..2cff5a9 100644
--- a/src/backend/optimizer/plan/setrefs.c
+++ b/src/backend/optimizer/plan/setrefs.c
@@ -465,6 +465,26 @@ set_plan_refs(PlannerInfo *root, Plan *plan, int rtoffset)
 					fix_scan_expr(root, (Node *) splan->tablesample, rtoffset);
 			}
 			break;
+		case T_Funnel:
+			{
+				Funnel	   *splan = (Funnel *) plan;
+
+				/*
+				 * target list for leftree of funnel plan should be same as
+				 * for funnel scan as both nodes need to produce same
+				 * projection. We don't want to do this assignment after
+				 * fixing references as that will be done separately for
+				 * lefttree node.
+				 */
+				splan->scan.plan.lefttree->targetlist = splan->scan.plan.targetlist;
+
+				splan->scan.scanrelid += rtoffset;
+				splan->scan.plan.targetlist =
+					fix_scan_list(root, splan->scan.plan.targetlist, rtoffset);
+				splan->scan.plan.qual =
+					fix_scan_list(root, splan->scan.plan.qual, rtoffset);
+			}
+			break;
 		case T_IndexScan:
 			{
 				IndexScan  *splan = (IndexScan *) plan;
@@ -2265,6 +2285,40 @@ fix_opfuncids_walker(Node *node, void *context)
 }
 
 /*
+ * fix_node_funcids
+ *		Set the opfuncid (procedure OID) in an OpExpr node,
+ *		for plan tree.
+ *
+ * We need it mainly to fix the opfuncid in nodes of plantree
+ * after reading the planned statement by worker backend.
+ * Currently the support of nodes that could be executed by
+ * worker backend are limited, we can enhance this API based
+ * on it's usage in future.
+ */
+void
+fix_node_funcids(Plan *node)
+{
+	/*
+	 * do nothing when we get to the end of a leaf on tree.
+	 */
+	if (node == NULL)
+		return;
+
+	fix_opfuncids((Node *) node->qual);
+	fix_opfuncids((Node *) node->targetlist);
+
+	switch (nodeTag(node))
+	{
+		default:
+			elog(ERROR, "unrecognized node type: %d", (int) nodeTag(node));
+			break;
+	}
+
+	fix_node_funcids(node->lefttree);
+	fix_node_funcids(node->righttree);
+}
+
+/*
  * set_opfuncid
  *		Set the opfuncid (procedure OID) in an OpExpr node,
  *		if it hasn't been set already.
diff --git a/src/backend/optimizer/plan/subselect.c b/src/backend/optimizer/plan/subselect.c
index d0bc412..073a7f5 100644
--- a/src/backend/optimizer/plan/subselect.c
+++ b/src/backend/optimizer/plan/subselect.c
@@ -2243,6 +2243,10 @@ finalize_plan(PlannerInfo *root, Plan *plan, Bitmapset *valid_params,
 			context.paramids = bms_add_members(context.paramids, scan_params);
 			break;
 
+		case T_Funnel:
+			context.paramids = bms_add_members(context.paramids, scan_params);
+			break;
+
 		case T_IndexScan:
 			finalize_primnode((Node *) ((IndexScan *) plan)->indexqual,
 							  &context);
diff --git a/src/backend/optimizer/util/pathnode.c b/src/backend/optimizer/util/pathnode.c
index 935bc2b..c886075 100644
--- a/src/backend/optimizer/util/pathnode.c
+++ b/src/backend/optimizer/util/pathnode.c
@@ -732,6 +732,32 @@ create_samplescan_path(PlannerInfo *root, RelOptInfo *rel, Relids required_outer
 }
 
 /*
+ * create_funnel_path
+ *
+ *	  Creates a path corresponding to a funnel scan, returning the
+ *	  pathnode.
+ */
+FunnelPath *
+create_funnel_path(PlannerInfo *root, RelOptInfo *rel, Path *subpath,
+				   Relids required_outer, int nworkers)
+{
+	FunnelPath *pathnode = makeNode(FunnelPath);
+
+	pathnode->path.pathtype = T_Funnel;
+	pathnode->path.parent = rel;
+	pathnode->path.param_info = get_baserel_parampathinfo(root, rel,
+														  required_outer);
+	pathnode->path.pathkeys = NIL;		/* Funnel has unordered result */
+
+	pathnode->subpath = subpath;
+	pathnode->num_workers = nworkers;
+
+	cost_funnel(pathnode, root, rel, pathnode->path.param_info);
+
+	return pathnode;
+}
+
+/*
  * create_index_path
  *	  Creates a path node for an index scan.
  *
diff --git a/src/backend/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c
index baa43b2..58fc6d3 100644
--- a/src/backend/postmaster/postmaster.c
+++ b/src/backend/postmaster/postmaster.c
@@ -103,6 +103,7 @@
 #include "miscadmin.h"
 #include "pg_getopt.h"
 #include "pgstat.h"
+#include "optimizer/cost.h"
 #include "postmaster/autovacuum.h"
 #include "postmaster/bgworker_internals.h"
 #include "postmaster/fork_process.h"
diff --git a/src/backend/tcop/dest.c b/src/backend/tcop/dest.c
index d645751..57014ee 100644
--- a/src/backend/tcop/dest.c
+++ b/src/backend/tcop/dest.c
@@ -45,9 +45,10 @@
  *		dummy DestReceiver functions
  * ----------------
  */
-static void
+static bool
 donothingReceive(TupleTableSlot *slot, DestReceiver *self)
 {
+	return true;
 }
 
 static void
@@ -132,7 +133,7 @@ CreateDestReceiver(CommandDest dest)
 			return CreateTransientRelDestReceiver(InvalidOid);
 
 		case DestTupleQueue:
-			return CreateTupleQueueDestReceiver(NULL);
+			return CreateTupleQueueDestReceiver();
 	}
 
 	/* should never get here */
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index d1f43c5..43a8114 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -42,6 +42,8 @@
 #include "catalog/pg_type.h"
 #include "commands/async.h"
 #include "commands/prepare.h"
+#include "executor/execParallel.h"
+#include "executor/tqueue.h"
 #include "libpq/libpq.h"
 #include "libpq/pqformat.h"
 #include "libpq/pqsignal.h"
@@ -1192,6 +1194,93 @@ exec_simple_query(const char *query_string)
 }
 
 /*
+ * exec_parallel_stmt
+ *
+ * Execute the plan for backend worker.
+ */
+void
+exec_parallel_stmt(ParallelStmt *parallelstmt)
+{
+	DestReceiver *receiver;
+	QueryDesc  *queryDesc;
+	MemoryContext oldcontext;
+	MemoryContext plancontext;
+	BufferUsage bufusage_start;
+	BufferUsage bufusage_end = {0};
+
+	set_ps_display("SELECT", false);
+
+	/*
+	 * Unlike exec_simple_query(), in backend worker we won't allow
+	 * transaction control statements, so we can allow plancontext to be
+	 * created in TopTransaction context.
+	 */
+	plancontext = AllocSetContextCreate(CurrentMemoryContext,
+										"worker plan",
+										ALLOCSET_DEFAULT_MINSIZE,
+										ALLOCSET_DEFAULT_INITSIZE,
+										ALLOCSET_DEFAULT_MAXSIZE);
+
+	oldcontext = MemoryContextSwitchTo(plancontext);
+
+	receiver = CreateDestReceiver(DestTupleQueue);
+	SetTupleQueueDestReceiverParams(receiver, parallelstmt->responseq);
+
+	/* Create a QueryDesc for the query */
+	queryDesc = CreateQueryDesc(parallelstmt->plannedstmt, "",
+								GetActiveSnapshot(), InvalidSnapshot,
+								receiver, parallelstmt->params,
+								parallelstmt->inst_options);
+
+	PushActiveSnapshot(queryDesc->snapshot);
+
+	/* call ExecutorStart to prepare the plan for execution */
+	ExecutorStart(queryDesc, 0);
+
+	PopulateParamExecParams(queryDesc, parallelstmt->serialized_param_exec_vals);
+
+	bufusage_start = pgBufferUsage;
+
+	/* run the plan */
+	ExecutorRun(queryDesc, ForwardScanDirection, 0L);
+
+	/*
+	 * Calculate the buffer usage for this statement run, it is required by
+	 * plugins like pg_stat_statements to report the total usage for statement
+	 * execution.
+	 */
+	BufferUsageAccumDiff(&bufusage_end,
+						 &pgBufferUsage, &bufusage_start);
+
+	/* run cleanup too */
+	ExecutorFinish(queryDesc);
+
+	/* copy buffer usage into shared memory. */
+	memcpy(parallelstmt->buffer_usage,
+		   &bufusage_end,
+		   sizeof(BufferUsage));
+
+	/*
+	 * copy intrumentation information into shared memory if requested by
+	 * master backend.
+	 */
+	if (parallelstmt->inst_options)
+		memcpy(parallelstmt->instrument,
+			   queryDesc->planstate->instrument,
+			   sizeof(Instrumentation));
+
+	ExecutorEnd(queryDesc);
+
+	PopActiveSnapshot();
+
+	FreeQueryDesc(queryDesc);
+
+	(*receiver->rDestroy) (receiver);
+
+	MemoryContextSwitchTo(oldcontext);
+}
+
+/*
  * exec_parse_message
  *
  * Execute a "Parse" protocol message.
diff --git a/src/backend/tcop/pquery.c b/src/backend/tcop/pquery.c
index 0df86a2..5eab231 100644
--- a/src/backend/tcop/pquery.c
+++ b/src/backend/tcop/pquery.c
@@ -1117,7 +1117,13 @@ RunFromStore(Portal portal, ScanDirection direction, long count,
 			if (!ok)
 				break;
 
-			(*dest->receiveSlot) (slot, dest);
+			/*
+			 * If we are not able to send the tuple, we assume the destination
+			 * has closed and no more tuples can be sent. If that's the case,
+			 * end the loop.
+			 */
+			if (!((*dest->receiveSlot) (slot, dest)))
+				break;
 
 			ExecClearTuple(slot);
 
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 17053af..c30298b 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -588,6 +588,8 @@ const char *const config_group_names[] =
 	gettext_noop("Statistics / Query and Index Statistics Collector"),
 	/* AUTOVACUUM */
 	gettext_noop("Autovacuum"),
+	/* PARALLEL_QUERY */
+	gettext_noop("parallel_seqscan_degree"),
 	/* CLIENT_CONN */
 	gettext_noop("Client Connection Defaults"),
 	/* CLIENT_CONN_STATEMENT */
@@ -2535,6 +2537,16 @@ static struct config_int ConfigureNamesInt[] =
 	},
 
 	{
+		{"parallel_seqscan_degree", PGC_SUSET, PARALLEL_QUERY,
+			gettext_noop("Sets the maximum number of simultaneously running backend worker processes."),
+			NULL
+		},
+		&parallel_seqscan_degree,
+		0, 0, MAX_BACKENDS,
+		NULL, NULL, NULL
+	},
+
+	{
 		{"autovacuum_work_mem", PGC_SIGHUP, RESOURCES_MEM,
 			gettext_noop("Sets the maximum memory to be used by each autovacuum worker process."),
 			NULL,
@@ -2711,6 +2723,26 @@ static struct config_real ConfigureNamesReal[] =
 		DEFAULT_CPU_OPERATOR_COST, 0, DBL_MAX,
 		NULL, NULL, NULL
 	},
+	{
+		{"cpu_tuple_comm_cost", PGC_USERSET, QUERY_TUNING_COST,
+			gettext_noop("Sets the planner's estimate of the cost of "
+				  "passing each tuple (row) from worker to master backend."),
+			NULL
+		},
+		&cpu_tuple_comm_cost,
+		DEFAULT_CPU_TUPLE_COMM_COST, 0, DBL_MAX,
+		NULL, NULL, NULL
+	},
+	{
+		{"parallel_setup_cost", PGC_USERSET, QUERY_TUNING_COST,
+			gettext_noop("Sets the planner's estimate of the cost of "
+				  "setting up environment (shared memory) for parallelism."),
+			NULL
+		},
+		&parallel_setup_cost,
+		DEFAULT_PARALLEL_SETUP_COST, 0, DBL_MAX,
+		NULL, NULL, NULL
+	},
 
 	{
 		{"cursor_tuple_fraction", PGC_USERSET, QUERY_TUNING_OTHER,
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 8c65287..1ddf317 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -290,6 +290,8 @@
 #cpu_tuple_cost = 0.01			# same scale as above
 #cpu_index_tuple_cost = 0.005		# same scale as above
 #cpu_operator_cost = 0.0025		# same scale as above
+#cpu_tuple_comm_cost = 0.1		# same scale as above
+#parallel_setup_cost = 1000.0	# same scale as above
 #effective_cache_size = 4GB
 
 # - Genetic Query Optimizer -
@@ -502,6 +504,11 @@
 					# autovacuum, -1 means use
 					# vacuum_cost_limit
 
+#------------------------------------------------------------------------------
+# PARALLEL_QUERY PARAMETERS
+#------------------------------------------------------------------------------
+
+#parallel_seqscan_degree = 0		# max number of worker backend subprocesses
 
 #------------------------------------------------------------------------------
 # CLIENT CONNECTION DEFAULTS
diff --git a/src/include/access/printtup.h b/src/include/access/printtup.h
index 46c4148..92ec882 100644
--- a/src/include/access/printtup.h
+++ b/src/include/access/printtup.h
@@ -25,11 +25,11 @@ extern void SendRowDescriptionMessage(TupleDesc typeinfo, List *targetlist,
 
 extern void debugStartup(DestReceiver *self, int operation,
 			 TupleDesc typeinfo);
-extern void debugtup(TupleTableSlot *slot, DestReceiver *self);
+extern bool debugtup(TupleTableSlot *slot, DestReceiver *self);
 
 /* XXX these are really in executor/spi.c */
 extern void spi_dest_startup(DestReceiver *self, int operation,
 				 TupleDesc typeinfo);
-extern void spi_printtup(TupleTableSlot *slot, DestReceiver *self);
+extern bool spi_printtup(TupleTableSlot *slot, DestReceiver *self);
 
 #endif   /* PRINTTUP_H */
diff --git a/src/include/executor/execParallel.h b/src/include/executor/execParallel.h
new file mode 100644
index 0000000..9bcf9af
--- /dev/null
+++ b/src/include/executor/execParallel.h
@@ -0,0 +1,61 @@
+/*--------------------------------------------------------------------
+ * execParallel.h
+ *		POSTGRES parallel execution interface
+ *
+ * Portions Copyright (c) 1996-2015, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ *		src/include/executor/execParallel.h
+ *--------------------------------------------------------------------
+ */
+#ifndef EXECPARALLEL_H
+#define EXECPARALLEL_H
+
+/*---------------------------------------------------------------------
+ * External module API.
+ *---------------------------------------------------------------------
+ */
+
+#include "libpq/pqmq.h"
+#include "nodes/execnodes.h"
+#include "nodes/parsenodes.h"
+#include "nodes/plannodes.h"
+
+/* Table-of-contents constants for our dynamic shared memory segment. */
+#define PARALLEL_KEY_PLANNEDSTMT	0
+#define PARALLEL_KEY_PARAMS			1
+#define PARALLEL_KEY_PARAMS_EXEC	2
+#define PARALLEL_KEY_BUFF_USAGE		3
+#define PARALLEL_KEY_INST_OPTIONS	4
+#define PARALLEL_KEY_INST_INFO		5
+#define PARALLEL_KEY_TUPLE_QUEUE	6
+#define PARALLEL_KEY_SCAN			7
+
+extern int	parallel_seqscan_degree;
+
+/* worker statement required for parallel execution. */
+typedef struct ParallelStmt
+{
+	PlannedStmt *plannedstmt;
+	ParamListInfo params;
+	List	   *serialized_param_exec_vals;
+	shm_mq_handle *responseq;
+	int			inst_options;
+	char	   *instrument;
+	char	   *buffer_usage;
+} ParallelStmt;
+
+extern void InitializeParallelWorkers(PlanState *planstate,
+						  List *serialized_param_exec_vals,
+						  EState *estate,
+						  char **inst_options_space,
+						  char **buffer_usage_space,
+						  shm_mq_handle ***responseqp,
+						  ParallelContext **pcxtp,
+						  int nWorkers);
+extern shm_toc *GetParallelShmToc(void);
+extern bool ExecParallelBufferUsageAccum(Node *node);
+extern void ExecAssociateBufferStatsToDSM(BufferUsage *buf_usage,
+							  ParallelStmt *parallel_stmt);
+#endif   /* EXECPARALLEL_H */
diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h
index 226f905..cb05f52 100644
--- a/src/include/executor/executor.h
+++ b/src/include/executor/executor.h
@@ -273,6 +273,8 @@ extern TupleDesc ExecCleanTypeFromTL(List *targetList, bool hasoid);
 extern TupleDesc ExecTypeFromExprList(List *exprList);
 extern void ExecTypeSetColNames(TupleDesc typeInfo, List *namesList);
 extern void UpdateChangedParamSet(PlanState *node, Bitmapset *newchg);
+extern void PopulateParamExecParams(QueryDesc *queryDesc,
+						List *serialized_param_exec_vals);
 
 typedef struct TupOutputState
 {
diff --git a/src/include/executor/instrument.h b/src/include/executor/instrument.h
index c9a2129..5e1b04e 100644
--- a/src/include/executor/instrument.h
+++ b/src/include/executor/instrument.h
@@ -69,5 +69,12 @@ extern Instrumentation *InstrAlloc(int n, int instrument_options);
 extern void InstrStartNode(Instrumentation *instr);
 extern void InstrStopNode(Instrumentation *instr, double nTuples);
 extern void InstrEndLoop(Instrumentation *instr);
+extern void InstrAggNode(Instrumentation *instr1, Instrumentation *instr2);
+extern void
+			InstrAggBufferUsage(BufferUsage *buffer_usage_dst, BufferUsage *buffer_usage_add);
+extern void BufferUsageAccumDiff(BufferUsage *dst,
+					 const BufferUsage *add,
+					 const BufferUsage *sub);
+extern void BufferUsageAdd(BufferUsage *dst, const BufferUsage *add);
 
 #endif   /* INSTRUMENT_H */
diff --git a/src/include/executor/nodeFunnel.h b/src/include/executor/nodeFunnel.h
new file mode 100644
index 0000000..fdb51d4
--- /dev/null
+++ b/src/include/executor/nodeFunnel.h
@@ -0,0 +1,25 @@
+/*-------------------------------------------------------------------------
+ *
+ * nodeFunnel.h
+ *		prototypes for nodeFunnel.c
+ *
+ *
+ * Portions Copyright (c) 1996-2015, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/executor/nodeFunnel.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef NODEFUNNEL_H
+#define NODEFUNNEL_H
+
+#include "nodes/execnodes.h"
+
+extern FunnelState *ExecInitFunnel(Funnel *node, EState *estate, int eflags);
+extern TupleTableSlot *ExecFunnel(FunnelState *node);
+extern void ExecEndFunnel(FunnelState *node);
+extern void DestroyParallelSetupAndAccumStats(FunnelState *node);
+extern void ExecReScanFunnel(FunnelState *node);
+
+#endif   /* NODEFUNNEL_H */
diff --git a/src/include/executor/nodeSubplan.h b/src/include/executor/nodeSubplan.h
index 3732ad4..ded4b19 100644
--- a/src/include/executor/nodeSubplan.h
+++ b/src/include/executor/nodeSubplan.h
@@ -24,4 +24,7 @@ extern void ExecReScanSetParamPlan(SubPlanState *node, PlanState *parent);
 
 extern void ExecSetParamPlan(SubPlanState *node, ExprContext *econtext);
 
+extern List *
+			ExecAndFormSerializeParamExec(ExprContext *econtext, Bitmapset *params);
+
 #endif   /* NODESUBPLAN_H */
diff --git a/src/include/executor/tqueue.h b/src/include/executor/tqueue.h
index 6f8eb73..d63416e 100644
--- a/src/include/executor/tqueue.h
+++ b/src/include/executor/tqueue.h
@@ -18,7 +18,9 @@
 #include "tcop/dest.h"
 
 /* Use this to send tuples to a shm_mq. */
-extern DestReceiver *CreateTupleQueueDestReceiver(shm_mq_handle *handle);
+extern DestReceiver *CreateTupleQueueDestReceiver(void);
+extern void SetTupleQueueDestReceiverParams(DestReceiver *self,
+								shm_mq_handle *handle);
 
 /* Use these to receive tuples from a shm_mq. */
 typedef struct TupleQueueFunnel TupleQueueFunnel;
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 4ae2f3e..6f6128a 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -16,7 +16,9 @@
 
 #include "access/genam.h"
 #include "access/heapam.h"
+#include "access/parallel.h"
 #include "executor/instrument.h"
+#include "executor/tqueue.h"
 #include "lib/pairingheap.h"
 #include "nodes/params.h"
 #include "nodes/plannodes.h"
@@ -1049,6 +1051,13 @@ typedef struct PlanState
 	Bitmapset  *chgParam;		/* set of IDs of changed Params */
 
 	/*
+	 * At execution time, parallel scan descriptor is initialized and stored
+	 * in dynamic shared memory segment by master backend and parallel workers
+	 * retrieve it from shared memory.
+	 */
+	shm_toc    *toc;
+
+	/*
 	 * Other run-time state needed by most if not all node types.
 	 */
 	TupleTableSlot *ps_ResultTupleSlot; /* slot for my result tuples */
@@ -1273,6 +1282,35 @@ typedef struct SampleScanState
 } SampleScanState;
 
 /*
+ * FunnelState extends ScanState by storing additional information
+ * related to parallel workers.
+ *		pcxt				parallel context for managing generic state information
+ *							required for parallelism.
+ *		responseq			shared memory queues to receive data from workers.
+ *		funnel				maintains the runtime information about queue's used to
+ *							receive data from parallel workers.
+ *		inst_options_space	to accumulate instrumentation information from all
+ *							parallel workers.
+ *		buffer_usage_space	to accumulate buffer usage information from all
+ *							parallel workers.
+ *		fs_workersReady		indicates that workers are launched.
+ *		all_workers_done	indicates that all the data from workers has been received.
+ *		local_scan_done		indicates that local scan is compleleted.
+ */
+typedef struct FunnelState
+{
+	ScanState	ss;				/* its first field is NodeTag */
+	ParallelContext *pcxt;
+	shm_mq_handle **responseq;
+	TupleQueueFunnel *funnel;
+	char	   *inst_options_space;
+	char	   *buffer_usage_space;
+	bool		fs_workersReady;
+	bool		all_workers_done;
+	bool		local_scan_done;
+} FunnelState;
+
+/*
  * These structs store information about index quals that don't have simple
  * constant right-hand sides.  See comments for ExecIndexBuildScanKeys()
  * for discussion.
diff --git a/src/include/nodes/nodeFuncs.h b/src/include/nodes/nodeFuncs.h
index 36b5dac..4f8be76 100644
--- a/src/include/nodes/nodeFuncs.h
+++ b/src/include/nodes/nodeFuncs.h
@@ -13,6 +13,7 @@
 #ifndef NODEFUNCS_H
 #define NODEFUNCS_H
 
+#include "access/parallel.h"
 #include "nodes/parsenodes.h"
 
 
diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index 274480e..48f7160 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -52,6 +52,7 @@ typedef enum NodeTag
 	T_Scan,
 	T_SeqScan,
 	T_SampleScan,
+	T_Funnel,
 	T_IndexScan,
 	T_IndexOnlyScan,
 	T_BitmapIndexScan,
@@ -99,6 +100,7 @@ typedef enum NodeTag
 	T_ScanState,
 	T_SeqScanState,
 	T_SampleScanState,
+	T_FunnelState,
 	T_IndexScanState,
 	T_IndexOnlyScanState,
 	T_BitmapIndexScanState,
@@ -223,6 +225,7 @@ typedef enum NodeTag
 	T_IndexOptInfo,
 	T_ParamPathInfo,
 	T_Path,
+	T_FunnelPath,
 	T_IndexPath,
 	T_BitmapHeapPath,
 	T_BitmapAndPath,
diff --git a/src/include/nodes/params.h b/src/include/nodes/params.h
index a0f7dd0..05b9042 100644
--- a/src/include/nodes/params.h
+++ b/src/include/nodes/params.h
@@ -14,6 +14,8 @@
 #ifndef PARAMS_H
 #define PARAMS_H
 
+#include "nodes/pg_list.h"
+
 /* To avoid including a pile of parser headers, reference ParseState thus: */
 struct ParseState;
 
@@ -96,11 +98,46 @@ typedef struct ParamExecData
 {
 	void	   *execPlan;		/* should be "SubPlanState *" */
 	Datum		value;
+
+	/*
+	 * parameter's datatype, or 0.  This is required so that datum value can
+	 * be read and used for other purposes like passing it to worker backend
+	 * via shared memory.  This is required only for evaluation of initPlan's,
+	 * however for consistency we set this for Subplan as well.  We left it
+	 * for other cases like CTE or RecursiveUnion cases where this structure
+	 * is not used for evaluation of subplans.
+	 */
+	Oid			ptype;
 	bool		isnull;
 } ParamExecData;
 
+/*
+ * This structure is used to pass PARAM_EXEC parameters to backend
+ * workers.  For each PARAM_EXEC parameter, pass this structure
+ * followed by value except for pass-by-value parameters.
+ */
+typedef struct SerializedParamExecData
+{
+	int			paramid;		/* parameter id of this param */
+	Size		length;			/* length of parameter value */
+	Oid			ptype;			/* parameter's datatype, or 0 */
+	Datum		value;
+	bool		isnull;
+}	SerializedParamExecData;
+
 
 /* Functions found in src/backend/nodes/params.c */
 extern ParamListInfo copyParamList(ParamListInfo from);
 
+extern Size
+			EstimateBoundParametersSpace(ParamListInfo params);
+extern void
+			SerializeBoundParams(ParamListInfo params, Size maxsize, char *start_address);
+extern ParamListInfo RestoreBoundParams(char *start_address);
+extern Size
+			EstimateExecParametersSpace(List *serialized_param_exec_vals);
+extern void SerializeExecParams(List *serialized_param_exec_vals, Size maxsize,
+					char *start_address);
+List *
+			RestoreExecParams(char *start_address);
 #endif   /* PARAMS_H */
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index cc259f1..69302af 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -296,6 +296,16 @@ typedef struct SampleScan
 	struct TableSampleClause *tablesample;
 } SampleScan;
 
+/* ------------
+ *		Funnel node
+ * ------------
+ */
+typedef struct Funnel
+{
+	Scan		scan;
+	int			num_workers;
+} Funnel;
+
 /* ----------------
  *		index scan node
  *
diff --git a/src/include/nodes/relation.h b/src/include/nodes/relation.h
index 79bed33..0ec1a45 100644
--- a/src/include/nodes/relation.h
+++ b/src/include/nodes/relation.h
@@ -761,6 +761,13 @@ typedef struct Path
 	/* pathkeys is a List of PathKey nodes; see above */
 } Path;
 
+typedef struct FunnelPath
+{
+	Path		path;
+	Path	   *subpath;		/* path for each worker */
+	int			num_workers;
+} FunnelPath;
+
 /* Macro for extracting a path's parameterization relids; beware double eval */
 #define PATH_REQ_OUTER(path)  \
 	((path)->param_info ? (path)->param_info->ppi_req_outer : (Relids) NULL)
diff --git a/src/include/optimizer/cost.h b/src/include/optimizer/cost.h
index dd43e45..8ca7d0f 100644
--- a/src/include/optimizer/cost.h
+++ b/src/include/optimizer/cost.h
@@ -26,6 +26,13 @@
 #define DEFAULT_CPU_TUPLE_COST	0.01
 #define DEFAULT_CPU_INDEX_TUPLE_COST 0.005
 #define DEFAULT_CPU_OPERATOR_COST  0.0025
+#define DEFAULT_CPU_TUPLE_COMM_COST 0.1
+/*
+ * XXX - We have kept reasonably high value for default parallel
+ * setup cost. In future we might want to change this value based
+ * on results.
+ */
+#define DEFAULT_PARALLEL_SETUP_COST  1000.0
 
 #define DEFAULT_EFFECTIVE_CACHE_SIZE  524288	/* measured in pages */
 
@@ -48,8 +55,11 @@ extern PGDLLIMPORT double random_page_cost;
 extern PGDLLIMPORT double cpu_tuple_cost;
 extern PGDLLIMPORT double cpu_index_tuple_cost;
 extern PGDLLIMPORT double cpu_operator_cost;
+extern PGDLLIMPORT double cpu_tuple_comm_cost;
+extern PGDLLIMPORT double parallel_setup_cost;
 extern PGDLLIMPORT int effective_cache_size;
 extern Cost disable_cost;
+extern int	parallel_seqscan_degree;
 extern bool enable_seqscan;
 extern bool enable_indexscan;
 extern bool enable_indexonlyscan;
@@ -70,6 +80,8 @@ extern void cost_seqscan(Path *path, PlannerInfo *root, RelOptInfo *baserel,
 			 ParamPathInfo *param_info);
 extern void cost_samplescan(Path *path, PlannerInfo *root, RelOptInfo *baserel,
 				ParamPathInfo *param_info);
+extern void cost_funnel(FunnelPath *path, PlannerInfo *root,
+			RelOptInfo *baserel, ParamPathInfo *param_info);
 extern void cost_index(IndexPath *path, PlannerInfo *root,
 		   double loop_count);
 extern void cost_bitmap_heap_scan(Path *path, PlannerInfo *root, RelOptInfo *baserel,
diff --git a/src/include/optimizer/pathnode.h b/src/include/optimizer/pathnode.h
index 161644c..29b344f 100644
--- a/src/include/optimizer/pathnode.h
+++ b/src/include/optimizer/pathnode.h
@@ -34,6 +34,9 @@ extern Path *create_seqscan_path(PlannerInfo *root, RelOptInfo *rel,
 					Relids required_outer);
 extern Path *create_samplescan_path(PlannerInfo *root, RelOptInfo *rel,
 					   Relids required_outer);
+extern FunnelPath *create_funnel_path(PlannerInfo *root,
+				   RelOptInfo *rel, Path *subpath, Relids required_outer,
+				   int nworkers);
 extern IndexPath *create_index_path(PlannerInfo *root,
 				  IndexOptInfo *index,
 				  List *indexclauses,
diff --git a/src/include/optimizer/planmain.h b/src/include/optimizer/planmain.h
index 52b077a..67a8582 100644
--- a/src/include/optimizer/planmain.h
+++ b/src/include/optimizer/planmain.h
@@ -133,6 +133,7 @@ extern bool query_is_distinct_for(Query *query, List *colnos, List *opids);
  */
 extern Plan *set_plan_references(PlannerInfo *root, Plan *plan);
 extern void fix_opfuncids(Node *node);
+extern void fix_node_funcids(Plan *node);
 extern void set_opfuncid(OpExpr *opexpr);
 extern void set_sa_opfuncid(ScalarArrayOpExpr *opexpr);
 extern void record_plan_function_dependency(PlannerInfo *root, Oid funcid);
diff --git a/src/include/optimizer/planner.h b/src/include/optimizer/planner.h
index b10a504..623223c 100644
--- a/src/include/optimizer/planner.h
+++ b/src/include/optimizer/planner.h
@@ -14,6 +14,7 @@
 #ifndef PLANNER_H
 #define PLANNER_H
 
+#include "nodes/parsenodes.h"
 #include "nodes/plannodes.h"
 #include "nodes/relation.h"
 
@@ -29,6 +30,8 @@ extern PlannedStmt *planner(Query *parse, int cursorOptions,
 		ParamListInfo boundParams);
 extern PlannedStmt *standard_planner(Query *parse, int cursorOptions,
 				 ParamListInfo boundParams);
+extern PlannedStmt *create_parallel_worker_plannedstmt(Plan *plan,
+								   List *rangetable, int num_exec_params);
 
 extern Plan *subquery_planner(PlannerGlobal *glob, Query *parse,
 				 PlannerInfo *parent_root,
diff --git a/src/include/tcop/dest.h b/src/include/tcop/dest.h
index b560672..91acd60 100644
--- a/src/include/tcop/dest.h
+++ b/src/include/tcop/dest.h
@@ -104,7 +104,9 @@ typedef enum
  *		pointers that the executor must call.
  *
  * Note: the receiveSlot routine must be passed a slot containing a TupleDesc
- * identical to the one given to the rStartup routine.
+ * identical to the one given to the rStartup routine.  It returns bool where
+ * a "true" value means "continue processing" and a "false" value means
+ * "stop early, just as if we'd reached the end of the scan".
  * ----------------
  */
 typedef struct _DestReceiver DestReceiver;
@@ -112,7 +114,7 @@ typedef struct _DestReceiver DestReceiver;
 struct _DestReceiver
 {
 	/* Called for each tuple to be output: */
-	void		(*receiveSlot) (TupleTableSlot *slot,
+	bool		(*receiveSlot) (TupleTableSlot *slot,
 											DestReceiver *self);
 	/* Per-executor-run initialization and shutdown: */
 	void		(*rStartup) (DestReceiver *self,
diff --git a/src/include/tcop/tcopprot.h b/src/include/tcop/tcopprot.h
index 96c5b8b..6f319c1 100644
--- a/src/include/tcop/tcopprot.h
+++ b/src/include/tcop/tcopprot.h
@@ -19,6 +19,7 @@
 #ifndef TCOPPROT_H
 #define TCOPPROT_H
 
+#include "executor/execParallel.h"
 #include "nodes/params.h"
 #include "nodes/parsenodes.h"
 #include "nodes/plannodes.h"
@@ -84,5 +85,6 @@ extern void set_debug_options(int debug_flag,
 extern bool set_plan_disabling_options(const char *arg,
 						   GucContext context, GucSource source);
 extern const char *get_stats_option_name(const char *arg);
+extern void exec_parallel_stmt(ParallelStmt *parallelscan);
 
 #endif   /* TCOPPROT_H */
diff --git a/src/include/utils/guc_tables.h b/src/include/utils/guc_tables.h
index 7a58ddb..3505d31 100644
--- a/src/include/utils/guc_tables.h
+++ b/src/include/utils/guc_tables.h
@@ -85,6 +85,7 @@ enum config_group
 	STATS_MONITORING,
 	STATS_COLLECTOR,
 	AUTOVACUUM,
+	PARALLEL_QUERY,
 	CLIENT_CONN,
 	CLIENT_CONN_STATEMENT,
 	CLIENT_CONN_LOCALE,
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index a037f81..ad2a33f 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -707,6 +707,9 @@ FunctionParameterMode
 FunctionScan
 FunctionScanPerFuncState
 FunctionScanState
+Funnel
+FunnelPath
+FunnelState
 FuzzyAttrMatchState
 GBT_NUMKEY
 GBT_NUMKEY_R
@@ -1195,6 +1198,8 @@ OverrideSearchPath
 OverrideStackEntry
 PACE_HEADER
 PACL
+parallel_estimate_ctx
+ParallelStmt
 PATH
 PBOOL
 PCtxtHandle
