From 999cec40f6725ccb7a5ceea1700d25f6e3d48b80 Mon Sep 17 00:00:00 2001
From: bucoo <bucoo@sohu.com>
Date: Wed, 15 Sep 2021 15:13:34 +0800
Subject: [PATCH 2/2] parallel union/aggregate/distinct support using
 redistribute plan

---
 src/backend/commands/explain.c                |  59 ++
 src/backend/executor/Makefile                 |   1 +
 src/backend/executor/execParallel.c           |  43 +
 src/backend/executor/execProcnode.c           |  10 +
 src/backend/executor/nodeGather.c             |   1 +
 src/backend/executor/nodeGatherMerge.c        |   1 +
 src/backend/executor/nodeRedistribute.c       | 941 ++++++++++++++++++
 src/backend/nodes/copyfuncs.c                 |  25 +
 src/backend/nodes/outfuncs.c                  |  28 +
 src/backend/nodes/readfuncs.c                 |  18 +
 src/backend/optimizer/path/costsize.c         |  15 +
 src/backend/optimizer/plan/createplan.c       |  31 +-
 src/backend/optimizer/plan/planner.c          | 260 +++++
 src/backend/optimizer/plan/setrefs.c          |   1 +
 src/backend/optimizer/plan/subselect.c        |   1 +
 src/backend/optimizer/prep/prepunion.c        |  79 +-
 src/backend/optimizer/util/pathnode.c         |  26 +
 src/backend/optimizer/util/tlist.c            |  21 +
 src/backend/storage/ipc/shm_mq.c              | 100 ++
 src/backend/utils/activity/wait_event.c       |   6 +
 src/backend/utils/misc/guc.c                  |  12 +
 src/backend/utils/sort/sharedtuplestore.c     |  53 +-
 src/include/executor/execParallel.h           |   1 +
 src/include/executor/nodeRedistribute.h       |  21 +
 src/include/nodes/execnodes.h                 |  39 +
 src/include/nodes/nodes.h                     |   3 +
 src/include/nodes/pathnodes.h                 |  10 +
 src/include/nodes/plannodes.h                 |  14 +
 src/include/optimizer/cost.h                  |   2 +
 src/include/optimizer/pathnode.h              |   4 +
 src/include/optimizer/tlist.h                 |   1 +
 src/include/storage/shm_mq.h                  |   4 +
 src/include/utils/sharedtuplestore.h          |   7 +
 src/include/utils/wait_event.h                |   4 +-
 .../test_shm_mq/expected/test_shm_mq.out      |   6 +
 .../modules/test_shm_mq/sql/test_shm_mq.sql   |   1 +
 src/test/modules/test_shm_mq/test.c           |   6 +-
 .../modules/test_shm_mq/test_shm_mq--1.0.sql  |   3 +-
 .../regress/expected/partition_aggregate.out  |  75 ++
 src/test/regress/expected/select_distinct.out |  46 +
 src/test/regress/expected/select_parallel.out |  48 +
 src/test/regress/expected/union.out           |  95 ++
 src/test/regress/sql/partition_aggregate.sql  |  26 +
 src/test/regress/sql/select_distinct.sql      |  15 +
 src/test/regress/sql/select_parallel.sql      |  19 +
 src/test/regress/sql/union.sql                |  32 +
 46 files changed, 2187 insertions(+), 27 deletions(-)
 create mode 100644 src/backend/executor/nodeRedistribute.c
 create mode 100644 src/include/executor/nodeRedistribute.h

diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c
index 10644dfac4..68c1468138 100644
--- a/src/backend/commands/explain.c
+++ b/src/backend/commands/explain.c
@@ -117,6 +117,7 @@ static void show_tidbitmap_info(BitmapHeapScanState *planstate,
 static void show_instrumentation_count(const char *qlabel, int which,
 									   PlanState *planstate, ExplainState *es);
 static void show_foreignscan_info(ForeignScanState *fsstate, ExplainState *es);
+static void show_redistribute_info(RedistributeState *node, ExplainState *es);
 static void show_eval_params(Bitmapset *bms_params, ExplainState *es);
 static const char *explain_get_index_name(Oid indexId);
 static void show_buffer_usage(ExplainState *es, const BufferUsage *usage,
@@ -1388,6 +1389,9 @@ ExplainNode(PlanState *planstate, List *ancestors,
 		case T_Hash:
 			pname = sname = "Hash";
 			break;
+		case T_Redistribute:
+			pname = sname = "Redistribute";
+			break;
 		default:
 			pname = sname = "???";
 			break;
@@ -2019,6 +2023,14 @@ ExplainNode(PlanState *planstate, List *ancestors,
 			show_memoize_info(castNode(MemoizeState, planstate), ancestors,
 							  es);
 			break;
+		case T_Redistribute:
+			show_sort_group_keys(planstate, "Hash Key",
+								 ((Redistribute*)plan)->numCols, 0,
+								 ((Redistribute*)plan)->hashColIdx,
+								 NULL, NULL, NULL,
+								 ancestors, es);
+			show_redistribute_info(castNode(RedistributeState, planstate), es);
+			break;
 		default:
 			break;
 	}
@@ -3429,6 +3441,53 @@ show_foreignscan_info(ForeignScanState *fsstate, ExplainState *es)
 	}
 }
 
+/*
+ * Show extra information for a Redistribute node's one worker.
+ */
+static void
+show_redistribute_one_info(RedistributeInstrumentation *instrument, ExplainState *es)
+{
+	if (es->format == EXPLAIN_FORMAT_TEXT)
+	{
+		ExplainIndentText(es);
+		appendStringInfo(es->str, "Parts: %d", instrument->parts_got);
+		appendStringInfo(es->str, "  Disk Usage: " UINT64_FORMAT "kB", instrument->disk_used / 1024);
+		appendStringInfo(es->str, "  Disk Rows: " UINT64_FORMAT, instrument->disk_rows);
+		appendStringInfoChar(es->str, '\n');
+	}else
+	{
+		ExplainPropertyInteger("Parts", NULL, instrument->parts_got, es);
+		ExplainPropertyInteger("Disk Usage", "kB", instrument->disk_used / 1024, es);
+		ExplainPropertyInteger("Disk Rows", NULL, instrument->disk_rows, es);
+	}
+}
+
+/*
+ * Show extra information for a Redistribute node.
+ */
+static void
+show_redistribute_info(RedistributeState *node, ExplainState *es)
+{
+	SharedRedistributeInfo *info = node->shared_instrument;
+	int						i;
+	if (info == NULL)
+		return;
+
+	if (info->sinstrument[0].parts_got != 0)
+		show_redistribute_one_info(&info->sinstrument[0], es);
+
+	for (i=1;i<info->num_workers;++i)
+	{
+		if (info->sinstrument[i].parts_got == 0)
+			continue;
+		if (es->workers_state)
+			ExplainOpenWorker(i-1, es);
+		show_redistribute_one_info(&info->sinstrument[i], es);
+		if (es->workers_state)
+			ExplainCloseWorker(i-1, es);
+	}
+}
+
 /*
  * Show initplan params evaluated at Gather or Gather Merge node.
  */
diff --git a/src/backend/executor/Makefile b/src/backend/executor/Makefile
index 11118d0ce0..4adea66848 100644
--- a/src/backend/executor/Makefile
+++ b/src/backend/executor/Makefile
@@ -61,6 +61,7 @@ OBJS = \
 	nodeNestloop.o \
 	nodeProjectSet.o \
 	nodeRecursiveunion.o \
+	nodeRedistribute.o \
 	nodeResult.o \
 	nodeSamplescan.o \
 	nodeSeqscan.o \
diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c
index f8a4a40e7b..40248669bf 100644
--- a/src/backend/executor/execParallel.c
+++ b/src/backend/executor/execParallel.c
@@ -36,6 +36,7 @@
 #include "executor/nodeIndexonlyscan.h"
 #include "executor/nodeIndexscan.h"
 #include "executor/nodeMemoize.h"
+#include "executor/nodeRedistribute.h"
 #include "executor/nodeSeqscan.h"
 #include "executor/nodeSort.h"
 #include "executor/nodeSubplan.h"
@@ -297,6 +298,9 @@ ExecParallelEstimate(PlanState *planstate, ExecParallelEstimateContext *e)
 			/* even when not parallel-aware, for EXPLAIN ANALYZE */
 			ExecMemoizeEstimate((MemoizeState *) planstate, e->pcxt);
 			break;
+		case T_RedistributeState:
+			ExecRedistributeEstimate((RedistributeState *) planstate, e->pcxt);
+			break;
 		default:
 			break;
 	}
@@ -521,6 +525,9 @@ ExecParallelInitializeDSM(PlanState *planstate,
 			/* even when not parallel-aware, for EXPLAIN ANALYZE */
 			ExecMemoizeInitializeDSM((MemoizeState *) planstate, d->pcxt);
 			break;
+		case T_RedistributeState:
+			ExecRedistributeInitializeDSM((RedistributeState *) planstate, d->pcxt);
+			break;;
 		default:
 			break;
 	}
@@ -994,6 +1001,10 @@ ExecParallelReInitializeDSM(PlanState *planstate,
 				ExecHashJoinReInitializeDSM((HashJoinState *) planstate,
 											pcxt);
 			break;
+		case T_RedistributeState:
+			ExecRedistributeReInitializeDSM((RedistributeState *) planstate,
+											pcxt);
+			break;
 		case T_HashState:
 		case T_SortState:
 		case T_IncrementalSortState:
@@ -1008,6 +1019,32 @@ ExecParallelReInitializeDSM(PlanState *planstate,
 	return planstate_tree_walker(planstate, ExecParallelReInitializeDSM, pcxt);
 }
 
+static bool
+ExecParallelLaunchedWalker(PlanState *planstate, ParallelContext *pcxt)
+{
+	if (planstate == NULL)
+		return false;
+
+	switch (nodeTag(planstate))
+	{
+		case T_RedistributeState:
+			ExecRedistributeParallelLaunched((RedistributeState *)planstate,
+											 pcxt);
+			break;
+
+		default:
+			break;
+	}
+
+	return planstate_tree_walker(planstate, ExecParallelLaunchedWalker, pcxt);
+}
+
+void
+ExecParallelLaunched(PlanState *planstate, ParallelContext *pcxt)
+{
+	(void)planstate_tree_walker(planstate, ExecParallelLaunchedWalker, pcxt);
+}
+
 /*
  * Copy instrumentation information about this node and its descendants from
  * dynamic shared memory.
@@ -1070,6 +1107,9 @@ ExecParallelRetrieveInstrumentation(PlanState *planstate,
 		case T_MemoizeState:
 			ExecMemoizeRetrieveInstrumentation((MemoizeState *) planstate);
 			break;
+		case T_RedistributeState:
+			ExecRedistributeRetrieveInstrumentation((RedistributeState *) planstate);
+			break;
 		default:
 			break;
 	}
@@ -1366,6 +1406,9 @@ ExecParallelInitializeWorker(PlanState *planstate, ParallelWorkerContext *pwcxt)
 			/* even when not parallel-aware, for EXPLAIN ANALYZE */
 			ExecMemoizeInitializeWorker((MemoizeState *) planstate, pwcxt);
 			break;
+		case T_RedistributeState:
+			ExecRedistributeInitializeWorker((RedistributeState *) planstate, pwcxt);
+			break;
 		default:
 			break;
 	}
diff --git a/src/backend/executor/execProcnode.c b/src/backend/executor/execProcnode.c
index 1752b9bfd8..22ba0e4290 100644
--- a/src/backend/executor/execProcnode.c
+++ b/src/backend/executor/execProcnode.c
@@ -102,6 +102,7 @@
 #include "executor/nodeNestloop.h"
 #include "executor/nodeProjectSet.h"
 #include "executor/nodeRecursiveunion.h"
+#include "executor/nodeRedistribute.h"
 #include "executor/nodeResult.h"
 #include "executor/nodeSamplescan.h"
 #include "executor/nodeSeqscan.h"
@@ -381,6 +382,11 @@ ExecInitNode(Plan *node, EState *estate, int eflags)
 												 estate, eflags);
 			break;
 
+		case T_Redistribute:
+			result = (PlanState *) ExecInitRedistribute((Redistribute *)node,
+														estate, eflags);
+			break;
+
 		default:
 			elog(ERROR, "unrecognized node type: %d", (int) nodeTag(node));
 			result = NULL;		/* keep compiler quiet */
@@ -756,6 +762,10 @@ ExecEndNode(PlanState *node)
 			ExecEndLimit((LimitState *) node);
 			break;
 
+		case T_RedistributeState:
+			ExecEndRedistribute((RedistributeState *) node);
+			break;
+
 		default:
 			elog(ERROR, "unrecognized node type: %d", (int) nodeTag(node));
 			break;
diff --git a/src/backend/executor/nodeGather.c b/src/backend/executor/nodeGather.c
index 734142b7b1..ed5f0068d8 100644
--- a/src/backend/executor/nodeGather.c
+++ b/src/backend/executor/nodeGather.c
@@ -184,6 +184,7 @@ ExecGather(PlanState *pstate)
 			 */
 			pcxt = node->pei->pcxt;
 			LaunchParallelWorkers(pcxt);
+			ExecParallelLaunched(outerPlanState(node), pcxt);
 			/* We save # workers launched for the benefit of EXPLAIN */
 			node->nworkers_launched = pcxt->nworkers_launched;
 
diff --git a/src/backend/executor/nodeGatherMerge.c b/src/backend/executor/nodeGatherMerge.c
index 03f02a19aa..10f1341347 100644
--- a/src/backend/executor/nodeGatherMerge.c
+++ b/src/backend/executor/nodeGatherMerge.c
@@ -225,6 +225,7 @@ ExecGatherMerge(PlanState *pstate)
 			/* Try to launch workers. */
 			pcxt = node->pei->pcxt;
 			LaunchParallelWorkers(pcxt);
+			ExecParallelLaunched(outerPlanState(node), pcxt);
 			/* We save # workers launched for the benefit of EXPLAIN */
 			node->nworkers_launched = pcxt->nworkers_launched;
 
diff --git a/src/backend/executor/nodeRedistribute.c b/src/backend/executor/nodeRedistribute.c
new file mode 100644
index 0000000000..9ca199d902
--- /dev/null
+++ b/src/backend/executor/nodeRedistribute.c
@@ -0,0 +1,941 @@
+
+#include "postgres.h"
+
+#include "common/hashfn.h"
+#include "executor/executor.h"
+#include "executor/nodeRedistribute.h"
+#include "miscadmin.h"
+#include "storage/barrier.h"
+#include "storage/shm_mq.h"
+#include "utils/sharedtuplestore.h"
+#include "utils/typcache.h"
+#include "utils/wait_event.h"
+
+#define INVALID_PART_NUMBER			0xffffffff
+
+/*
+ * local scan status
+ */
+#define	STATUS_LOCALLY_END			0x01	/* outer returned NULL slot */
+#define STATUS_QUERY_STS_END		0x02	/* all tuples on STS scanned */
+#define STATUS_READY_STS_READ		0x04	/* STS ready to scan */
+
+/*
+ * all workers working status
+ */
+#define PRW_WAIT_WORKER_NUMBER		0	/* waiting leader write worker number to shared memory */
+#define PRW_WRITING					1	/* writing tuple(s) to STS and MQ */
+#define PRW_END_STS					2	/* all attached worker called sts_end_write(ready for read) */
+
+#if 1
+#define REDISTRIBUTE_DEBUG(...)	ereport(LOG, errbacktrace(), errmsg(__VA_ARGS__))
+#else
+#define REDISTRIBUTE_DEBUG(...)	((void)0)
+#endif
+
+#define SharedRedistributeOffset(shmem_, offset_)	\
+	(AssertMacro((shmem_)->offset_ >= offsetof(RedistributeShmem, padding)), \
+	 ((char*)(shmem_)) + (shmem_)->offset_)
+
+typedef struct RedistributeWriteInfo
+{
+	SharedTuplestoreAccessor   *sta;
+	shm_mq_handle			   *mqwrite;
+}RedistributeWriteInfo;
+
+typedef struct RedistributeShmem
+{
+	Barrier				part_barrier;				/* see PRW_XXX */
+	pg_atomic_uint32	final_worker_num;			/* how many workers final start successed */
+	pg_atomic_uint32	next_idle_part;				/* next idle part index */
+	uint32				plan_worker_num;			/* how many workers plan to start */
+	Size				sts_shmem_offset;			/* all SharedTuplestore offset (start with this struct) */
+	Size				sts_shmem_size;				/* each SharedTuplestore size in bytes */
+	Size				mq_shmem_offset;			/* all shm_mq offset (start with this struct) */
+	Size				mq_shmem_size;				/* each shm_mq size in bytes */
+	Size				mq_shmem_part_size;			/* equal plan_worker_num * mq_shmem_size */
+	Size				instrument_shmem_offset;	/* RedistributeInstrumentations offset (start with this struct), 0 for not alloc */
+	SharedFileSet		fileset;					/* using by SharedTuplestore */
+
+	char				padding[FLEXIBLE_ARRAY_MEMBER];
+}RedistributeShmem;
+
+typedef struct RedistributeHashInfo
+{
+	FmgrInfo	flinfo;
+	union
+	{
+		FunctionCallInfoBaseData	fcinfo;
+		char	fcinfo_data[SizeForFunctionCallInfo(1)];
+	};
+	AttrNumber	attindex;
+}RedistributeHashInfo;
+
+int	redistribute_query_size = 0;	/* GUC variable */
+
+#define WaitMyLatch(wait_event_info)	\
+	WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0, wait_event_info)
+
+/*
+ * This function should not be call.
+ * When resource not ready, just report an error,
+ * else some function will call ExecSetExecProcNode using different execute function
+ */
+static TupleTableSlot *
+ExecRedistributeError(PlanState *pstate)
+{
+	ereport(ERROR,
+			errcode(ERRCODE_INTERNAL_ERROR),
+			errmsg("DSM not initialized for Redistribute plan"));
+
+	return NULL;	/* keep compiler quiet */
+}
+
+static TupleTableSlot *
+ExecCopyOuterSlot(TupleTableSlot *result, TupleTableSlot *outer)
+{
+	int		natts = result->tts_tupleDescriptor->natts;
+	Assert(natts == outer->tts_tupleDescriptor->natts);
+
+	if (result->tts_ops == outer->tts_ops)
+		return outer;
+
+	/* copy outer slot to result slot and return */
+	ExecClearTuple(result);
+	slot_getsomeattrs(outer, natts);
+	memcpy(result->tts_values, outer->tts_values, sizeof(outer->tts_values[0]) * natts);
+	memcpy(result->tts_isnull, outer->tts_isnull, sizeof(outer->tts_isnull[0]) * natts);
+	return ExecStoreVirtualTuple(result);
+}
+
+/*
+ * ExecRedistributeDirect
+ * this function for no any one parallel work start successed
+ */
+static TupleTableSlot *
+ExecRedistributeDirect(PlanState *pstate)
+{
+	TupleTableSlot *outer_slot = ExecProcNode(outerPlanState(pstate));
+	if (unlikely(TupIsNull(outer_slot)))
+		return ExecClearTuple(pstate->ps_ResultTupleSlot);
+
+	return ExecCopyOuterSlot(pstate->ps_ResultTupleSlot, outer_slot);
+}
+
+static shm_mq*
+GetRedistributeSharedMemoryQueue(RedistributeShmem *shmem, uint32 part, uint32 worknum)
+{
+	/* get shm_mq start address */
+	char *addr = SharedRedistributeOffset(shmem, mq_shmem_offset);
+
+	/* get part start address */
+	addr += shmem->mq_shmem_part_size * part;
+
+	/* get offset of part */
+	addr += shmem->mq_shmem_size * worknum;
+
+	return (shm_mq*)addr;
+}
+
+static void
+ExecInitRedistributeWriter(RedistributeState *node, uint32 my_work_num)
+{
+	MemoryContext		oldcontext = MemoryContextSwitchTo(GetMemoryChunkContext(node));
+	RedistributeShmem  *shmem = node->shmem;
+	uint32				nworkers = pg_atomic_read_u32(&shmem->final_worker_num);
+	uint32				mq_offset = shmem->sts_shmem_size + shmem->mq_shmem_part_size;
+	uint32				i;
+
+	for (i=0;i<nworkers;++i)
+	{
+		RedistributeWriteInfo  *writer;
+		char				   *addr;
+		shm_mq				   *mq;
+		if (unlikely(i == my_work_num))
+			continue;
+
+		addr = SharedRedistributeOffset(shmem, sts_shmem_offset);
+		addr += shmem->sts_shmem_size * i;
+		writer = &node->writer[i];
+		writer->sta = sts_attach((SharedTuplestore*)addr, my_work_num, &shmem->fileset);
+
+		REDISTRIBUTE_DEBUG("writer sts_attach(%p)=%p mq_attach %p part %u index %u",
+						   addr, writer->sta, addr + mq_offset, my_work_num, i);
+
+		mq = GetRedistributeSharedMemoryQueue(shmem, i, my_work_num);
+		shm_mq_set_sender(mq, MyProc);
+		writer->mqwrite = shm_mq_attach(mq, NULL, NULL);
+	}
+
+	MemoryContextSwitchTo(oldcontext);
+}
+
+static void
+ExecInitRedistributeReader(RedistributeState *node)
+{
+	MemoryContext		oldcontext = MemoryContextSwitchTo(GetMemoryChunkContext(node));
+	RedistributeShmem  *shmem = node->shmem;
+	char			   *start;
+	uint32				i;
+	uint32				nworkers = pg_atomic_read_u32(&shmem->final_worker_num);
+	Assert(node->current_part < nworkers);
+
+	/* attach SharedTuplestore */
+	start = SharedRedistributeOffset(shmem, sts_shmem_offset);
+	start += node->current_part * shmem->sts_shmem_size;
+	node->sta = sts_attach((SharedTuplestore*)start,
+						   node->current_part,
+						   &shmem->fileset);
+	REDISTRIBUTE_DEBUG("reader sts_attach(%p)=%p part %u", start, node->sta, node->current_part);
+
+	/* attach shm_mq */
+	node->nextreader = 0;
+	node->nreaders = 0;
+	for (i=0;i<nworkers;++i)
+	{
+		shm_mq *mq;
+		if (unlikely(i == node->current_part))
+			continue;
+
+		mq = GetRedistributeSharedMemoryQueue(shmem, node->current_part, i);
+		REDISTRIBUTE_DEBUG("reader mq_attach %p part %u", mq, node->current_part);
+		shm_mq_set_receiver(mq, MyProc);
+		node->mqreader[node->nreaders] = shm_mq_attach(mq, NULL, NULL);
+		++(node->nreaders);
+	}
+
+	MemoryContextSwitchTo(oldcontext);
+}
+
+static bool
+ExecNextRedistributePart(RedistributeState *node)
+{
+	RedistributeShmem  *shmem = node->shmem;
+	uint32				count_part = pg_atomic_read_u32(&shmem->final_worker_num);
+	uint32				idle_part = pg_atomic_read_u32(&shmem->next_idle_part);
+
+	while (idle_part < count_part)
+	{
+		if (pg_atomic_compare_exchange_u32(&shmem->next_idle_part,
+										   &idle_part,
+										   idle_part + 1))
+		{
+			node->current_part = idle_part;
+			if (node->instrument)
+				++(node->instrument->parts_got);
+			return true;
+		}
+	}
+
+	node->current_part = INVALID_PART_NUMBER;
+	return false;
+}
+
+/*
+ * initialize redistribute hash function info list
+ * and alloc reader and writer array
+ */
+static void
+InitializeRedistributeExecute(RedistributeState *node)
+{
+	AttrNumber				maxattr;
+	int						i;
+	TypeCacheEntry		   *typeCache;
+	RedistributeHashInfo   *info;
+	Form_pg_attribute		attr;
+	TupleDesc				desc = ExecGetResultType(outerPlanState(node));
+	List				   *list = NIL;
+	Redistribute		   *plan = castNode(Redistribute, node->ps.plan);
+	RedistributeShmem	   *shmem = node->shmem;
+	MemoryContext			oldcontext = MemoryContextSwitchTo(GetMemoryChunkContext(node));
+
+	/* initialize hash functions */
+	if (node->hash_funcs == NIL)
+	{
+		maxattr = 0;
+		for (i=0;i<plan->numCols;++i)
+		{
+			info = palloc0(sizeof(*info));
+
+			info->attindex = plan->hashColIdx[i]-1;
+			attr = TupleDescAttr(desc, info->attindex);
+
+			typeCache = lookup_type_cache(attr->atttypid, TYPECACHE_HASH_PROC);
+
+			fmgr_info(typeCache->hash_proc, &info->flinfo);
+			InitFunctionCallInfoData(info->fcinfo, &info->flinfo, 1, attr->attcollation, NULL, NULL);
+			info->fcinfo.args[0].isnull = false;
+			list = lappend(list, info);
+
+			if (plan->hashColIdx[i] > maxattr)
+				maxattr = plan->hashColIdx[i];
+		}
+		node->hash_funcs = list;
+		node->hash_max_attr = maxattr;
+	}
+
+	/* alloc reader and writer array */
+	Assert(pg_atomic_read_u32(&shmem->final_worker_num) > 1);
+	Assert(shmem->plan_worker_num >= pg_atomic_read_u32(&shmem->final_worker_num));
+	if (node->mqreader == NULL)
+		node->mqreader = palloc0(sizeof(node->mqreader[0]) * shmem->plan_worker_num);
+	if (node->writer == NULL)
+	{
+		node->writer = palloc0(sizeof(node->writer[0]) * shmem->plan_worker_num);
+		node->nwriter = shmem->plan_worker_num;
+	}
+
+	MemoryContextSwitchTo(oldcontext);
+}
+
+static uint32
+RedistributeGetHash(RedistributeState *node, TupleTableSlot *input)
+{
+	ListCell			   *lc;
+	RedistributeHashInfo   *info;
+	Datum					datum;
+	uint32					hash = 0;
+
+	/* get all values we need */
+	slot_getsomeattrs(input, node->hash_max_attr);
+
+	foreach (lc, node->hash_funcs)
+	{
+		info = lfirst(lc);
+
+		/* skip NULL */
+		if (input->tts_isnull[info->attindex])
+			continue;
+
+		info->fcinfo.args[0].value = input->tts_values[info->attindex];
+		datum = FunctionCallInvoke(&info->fcinfo);
+		if (unlikely(info->fcinfo.isnull))
+			elog(ERROR, "hash function %u returned NULL", info->flinfo.fn_oid);
+
+		hash = hash_combine(hash, DatumGetUInt32(datum));
+	}
+
+	return hash;
+}
+
+/* like function gather_readnext */
+static MinimalTuple
+GetTupleFromMQ(RedistributeState *node)
+{
+	shm_mq_handle  *mq;
+	void		   *data;
+	Size			nbytes;
+	shm_mq_result	mq_result;
+	int				nvisited = 0;
+
+	for (;;)
+	{
+		/* Check for async events, particularly messages from workers. */
+		CHECK_FOR_INTERRUPTS();
+
+		Assert(node->nextreader < node->nreaders);
+		mq = node->mqreader[node->nextreader];
+
+		mq_result = shm_mq_receive(mq, &nbytes, &data, true);
+retest_:
+		if (unlikely(mq_result == SHM_MQ_DETACHED))
+		{
+			shm_mq_detach(mq);
+			node->mqreader[node->nextreader] = NULL;
+			--(node->nreaders);
+			if (node->nreaders == 0)
+				return NULL;
+
+			memmove(&node->mqreader[node->nextreader],
+					&node->mqreader[node->nextreader + 1],
+					sizeof(node->mqreader[0]) * (node->nreaders - node->nextreader));
+			if (node->nextreader >= node->nreaders)
+				node->nextreader = 0;
+		}
+		else if (mq_result == SHM_MQ_WOULD_BLOCK)
+		{
+			/*
+			 * When SharedTuplestore ready to scan and MQ sender not attached
+			 * then it will no longer attach, and we can assume that the queue has ended
+			 */
+			if (node->status_flags & STATUS_READY_STS_READ &&
+				shm_mq_receiver_get_sender(mq) == NULL)
+			{
+				mq_result = SHM_MQ_DETACHED;
+				goto retest_;
+			}
+
+			/*
+			 * Advance nextreader pointer in round-robin fashion.  Note that we
+			 * only reach this code if we weren't able to get a tuple from the
+			 * current worker.  We used to advance the nextreader pointer after
+			 * every tuple, but it turns out to be much more efficient to keep
+			 * reading from the same queue until that would require blocking.
+			 */
+			node->nextreader++;
+			if (node->nextreader >= node->nreaders)
+				node->nextreader = 0;
+
+			/* Have we visited every (surviving) shm_mq_handle? */
+			nvisited++;
+			if (nvisited >= node->nreaders)
+			{
+				return NULL;
+			}
+		}else
+		{
+			Assert(mq_result == SHM_MQ_SUCCESS);
+			Assert(((MinimalTuple)data)->t_len == nbytes);
+			return (MinimalTuple)data;
+		}
+	}
+}
+
+static void
+PutSlotToQuery(RedistributeWriteInfo *writer, TupleTableSlot *slot,
+			   RedistributeInstrumentation *instrument)
+{
+	MinimalTuple	mtup;
+	shm_mq_result	mq_result;
+	bool			shouldFree;
+
+	mtup = ExecFetchSlotMinimalTuple(slot, &shouldFree);
+	mq_result = shm_mq_send_once(writer->mqwrite, mtup->t_len, mtup);
+	switch (mq_result)
+	{
+	case SHM_MQ_SUCCESS:
+		/* nothing todo */
+		break;
+	case SHM_MQ_WOULD_BLOCK:
+		sts_puttuple(writer->sta, NULL, mtup);
+		if (instrument)
+			++(instrument->disk_rows);
+		break;
+	case SHM_MQ_DETACHED:
+		/* other worker should report an error */
+		break;
+	}
+
+	if (shouldFree)
+		heap_free_minimal_tuple(mtup);
+}
+
+static void
+ExecRedistributeEndWrite(RedistributeState *node)
+{
+	RedistributeWriteInfo *writer;
+	uint32	i = node->nwriter;
+
+	while (i>0)
+	{
+		--i;
+		writer = &node->writer[i];
+		if (writer->sta)
+		{
+			int64	size;
+			REDISTRIBUTE_DEBUG("writer sts_close(%p) part %u", writer->sta, i);
+			size = sts_close(writer->sta);
+			writer->sta = NULL;
+			if (node->instrument)
+				node->instrument->disk_used += size;
+		}
+		if (writer->mqwrite != NULL)
+		{
+			shm_mq_detach(writer->mqwrite);
+			writer->mqwrite = NULL;
+		}
+	}
+}
+
+static TupleTableSlot *
+ExecRedistributeReadOnly(PlanState *pstate)
+{
+	MinimalTuple		mtup;
+	RedistributeState  *node = castNode(RedistributeState, pstate);
+
+loop_:
+	Assert(node->status_flags & STATUS_LOCALLY_END);
+
+	/* try read tuple from MQ and STS */
+	while (node->nreaders > 0)
+	{
+		/* try read tuple from MQ */
+		if ((mtup = GetTupleFromMQ(node)) != NULL)
+			return ExecStoreMinimalTuple(mtup, pstate->ps_ResultTupleSlot, false);
+
+		/*
+		 * GetTupleFromMQ maybe change nreaders,
+		 * so test it again.
+		 */
+		if (unlikely(node->nreaders == 0))
+			break;
+
+		/* test for all attached worker called sts_end_write */
+		if ((node->status_flags & STATUS_READY_STS_READ) == 0)
+		{
+			Barrier *part_barrier = &node->shmem->part_barrier;
+			int part_phase = BarrierAttach(part_barrier);
+			BarrierDetach(part_barrier);
+
+			if (part_phase >= PRW_END_STS)
+			{
+				sts_begin_parallel_scan(node->sta);
+				node->status_flags |= STATUS_READY_STS_READ;
+				SetLatch(MyLatch);
+			}
+		}
+
+		/* try read tuple from STS */
+		if ((node->status_flags & (STATUS_READY_STS_READ|STATUS_QUERY_STS_END)) == STATUS_READY_STS_READ)
+		{
+			mtup = sts_parallel_scan_next(node->sta, NULL);
+			if (mtup != NULL)
+				return ExecStoreMinimalTuple(mtup, pstate->ps_ResultTupleSlot, false);
+			REDISTRIBUTE_DEBUG("reader sts_close(%p)", node->sta);
+			sts_close(node->sta);
+			node->sta = NULL;
+			node->status_flags |= STATUS_QUERY_STS_END;
+		}
+
+		/* wait latch */
+		WaitMyLatch(WAIT_EVENT_MQ_RECEIVE);
+		ResetLatch(MyLatch);
+	}
+
+	if ((node->status_flags & STATUS_READY_STS_READ) == 0)
+	{
+		/* wait for all worker call sts_end_write */
+		Barrier *part_barrier = &node->shmem->part_barrier;
+		BarrierAttach(part_barrier);
+		while (BarrierPhase(part_barrier) < PRW_END_STS)
+			BarrierArriveAndWait(part_barrier, WAIT_EVENT_REDISTRIBUTE_SHARED_TUPLESTORE);
+		BarrierDetach(part_barrier);
+
+		sts_begin_parallel_scan(node->sta);
+		node->status_flags |= STATUS_READY_STS_READ;
+	}
+
+	if ((node->status_flags & STATUS_QUERY_STS_END) == 0)
+	{
+		mtup = sts_parallel_scan_next(node->sta, NULL);
+		if (mtup != NULL)
+			return ExecStoreMinimalTuple(mtup, pstate->ps_ResultTupleSlot, false);
+		REDISTRIBUTE_DEBUG("reader sts_close(%p)", node->sta);
+		sts_close(node->sta);
+		node->sta = NULL;
+		node->status_flags |= STATUS_QUERY_STS_END;
+	}
+
+	/* try read next part */
+	if (ExecNextRedistributePart(node))
+	{
+		ExecInitRedistributeReader(node);
+		sts_begin_parallel_scan(node->sta);
+		node->status_flags &= ~STATUS_QUERY_STS_END;
+		goto loop_;
+	}
+
+	return ExecClearTuple(pstate->ps_ResultTupleSlot);
+}
+
+/*
+ * fetch tuples from outer and other workers
+ */
+static TupleTableSlot *
+ExecRedistributeMixed(PlanState *pstate)
+{
+	uint32				count_part;
+	uint32				hash;
+	uint32				mod;
+	MinimalTuple		mtup;
+	TupleTableSlot	   *slot;
+	RedistributeState  *node = castNode(RedistributeState, pstate);
+
+	/* first try read tuple from MQ */
+	if (node->nreaders > 0)
+	{
+		mtup = GetTupleFromMQ(node);
+		if (mtup != NULL)
+			return ExecStoreMinimalTuple(mtup, pstate->ps_ResultTupleSlot, false);
+	}
+
+	count_part = pg_atomic_read_u32(&node->shmem->final_worker_num);
+	while ((node->status_flags & STATUS_LOCALLY_END) == 0)
+	{
+		slot = ExecProcNode(outerPlanState(pstate));
+		if (TupIsNull(slot))
+		{
+			Assert(BarrierPhase(&node->shmem->part_barrier) == PRW_WRITING);
+			ExecRedistributeEndWrite(node);
+			BarrierArriveAndDetach(&node->shmem->part_barrier);
+			node->status_flags |= STATUS_LOCALLY_END;
+
+			break;
+		}
+
+		hash = RedistributeGetHash(node, slot);
+		Assert(node->current_part != INVALID_PART_NUMBER);
+		Assert(node->current_part < count_part);
+		mod = hash % count_part;
+		if (mod == node->current_part)
+			return ExecCopyOuterSlot(pstate->ps_ResultTupleSlot, slot);
+
+		PutSlotToQuery(&node->writer[mod], slot, node->instrument);
+
+		/* try read tuple from MQ again */
+		if (node->nreaders > 0 &&
+			(mtup = GetTupleFromMQ(node)) != NULL)
+		{
+			return ExecStoreMinimalTuple(mtup, pstate->ps_ResultTupleSlot, false);
+		}
+	}
+
+	/* no more tuples from outer, change to only read tuples from other workers */
+	ExecSetExecProcNode(pstate, ExecRedistributeReadOnly);
+	return ExecRedistributeReadOnly(pstate);
+}
+
+/*
+ * Now, we know how many parallel work start succcessed
+ */
+static TupleTableSlot *
+ExecRedistributeReady(PlanState *pstate)
+{
+	RedistributeState  *node = castNode(RedistributeState, pstate);
+	RedistributeShmem  *shmem = node->shmem;
+	int					part_phase;
+
+	/* attach instrument if exist */
+	if (shmem->instrument_shmem_offset > 0)
+	{
+		int		index = ParallelWorkerNumber + 1;
+		Assert(index < shmem->plan_worker_num);
+		node->instrument = (RedistributeInstrumentation*)SharedRedistributeOffset(shmem, instrument_shmem_offset);
+		node->instrument = &node->instrument[index];
+	}
+
+	/* wait leader change final_worker_num */
+	part_phase = BarrierAttach(&shmem->part_barrier);
+	while (part_phase <= PRW_WAIT_WORKER_NUMBER)
+	{
+		/* only leader can change final_worker_num */
+		BarrierArriveAndWait(&shmem->part_barrier, WAIT_EVENT_REDISTRIBUTE_PARALLEL_START);
+		part_phase = BarrierPhase(&shmem->part_barrier);
+	}
+
+	/* get next part for read and write */
+	if (ExecNextRedistributePart(node) == false)
+	{
+		Assert(part_phase >= PRW_END_STS);
+		BarrierDetach(&shmem->part_barrier);
+		return ExecClearTuple(pstate->ps_ResultTupleSlot);
+	}
+
+	/* intialize execute and read resource */
+	InitializeRedistributeExecute(node);
+	ExecInitRedistributeReader(node);
+
+	if (part_phase >= PRW_END_STS)
+	{
+		/*
+		 * all outer's tuple fetched, don't need fetch anymore
+		 */
+		node->status_flags |= STATUS_LOCALLY_END;
+		BarrierDetach(&shmem->part_barrier);
+		ExecSetExecProcNode(pstate, ExecRedistributeReadOnly);
+	}else
+	{
+		ExecInitRedistributeWriter(node, node->current_part);
+		ExecSetExecProcNode(pstate, ExecRedistributeMixed);
+	}
+
+	Assert(part_phase != PRW_WRITING ||
+		   node->current_part != INVALID_PART_NUMBER);
+
+	return (*pstate->ExecProcNodeReal)(pstate);
+}
+
+RedistributeState *
+ExecInitRedistribute(Redistribute *node, EState *estate, int eflags)
+{
+	RedistributeState *state;
+
+	Assert(outerPlan(node) != NULL);
+	Assert(innerPlan(node) == NULL);
+
+	/*
+	 * create state struct
+	 */
+	state = makeNode(RedistributeState);
+	state->ps.plan = (Plan*)node;
+	state->ps.state = estate;
+	/*
+	 * ExecRedistributeInitializeDSM and ExecRedistributeInitializeWorker
+	 * will change execute function
+	 */
+	state->ps.ExecProcNode = ExecRedistributeError;
+
+	/*
+	 * Miscellaneous initialization
+	 *
+	 * create expression context for node
+	 */
+	ExecAssignExprContext(estate, &state->ps);
+
+	/*
+	 * initialize child nodes
+	 */
+	outerPlanState(state) = ExecInitNode(outerPlan(node), estate, eflags);
+
+	/*
+	 * tuple table and result type initialization
+	 */
+	ExecInitResultTupleSlotTL(&state->ps, &TTSOpsMinimalTuple);
+
+	return state;
+}
+
+/* release parallel work resource */
+static void
+ExecEndRedistributeInternal(RedistributeState *node)
+{
+	uint32	i;
+
+	if (node->writer != NULL &&
+		node->shmem != NULL)
+		ExecRedistributeEndWrite(node);
+
+	if (node->nreaders > 0)
+	{
+		for (i=0;i<node->nreaders;++i)
+			shm_mq_detach(node->mqreader[i]);
+		node->nreaders = node->nextreader = 0;
+	}
+
+	if (node->sta != NULL)
+	{
+		sts_close(node->sta);
+		node->sta = NULL;
+	}
+}
+
+void
+ExecEndRedistribute(RedistributeState *node)
+{
+	ExecEndRedistributeInternal(node);
+	ExecEndNode(outerPlanState(node));
+}
+
+void
+ExecReScanRedistribute(RedistributeState *node)
+{
+	ExecEndRedistributeInternal(node);
+	ExecReScan(outerPlanState(node));
+}
+
+/* parallel scan support */
+static Size
+ComputeRedistributeShmemSize(RedistributeShmem *shmem, int instrument)
+{
+	Size	size;
+	uint32	nworks = shmem->plan_worker_num;
+	Assert(nworks > 1);
+
+	/* RedistributeShmem size */
+	size = MAXALIGN(offsetof(RedistributeShmem, padding));
+
+	/* SharedTuplestore size */
+	Assert(size == MAXALIGN(size));
+	shmem->sts_shmem_offset = size;
+	shmem->sts_shmem_size = MAXALIGN(sts_estimate(nworks));
+	size = add_size(size, mul_size(shmem->sts_shmem_size, nworks));
+
+	/* shm_mq size */
+	Assert(redistribute_query_size > 0);
+	Assert(size == MAXALIGN(size));
+	shmem->mq_shmem_offset = size;
+	shmem->mq_shmem_size = ((Size)(redistribute_query_size)) * 1024;
+	shmem->mq_shmem_part_size = mul_size(shmem->mq_shmem_size, nworks);
+	size = add_size(size, mul_size(shmem->mq_shmem_part_size, nworks));
+
+	/* instrument size if need */
+	if (instrument)
+	{
+		Assert(size == MAXALIGN(size));
+		shmem->instrument_shmem_offset = size;
+		size = add_size(size, sizeof(RedistributeInstrumentation) * nworks);
+	}else
+	{
+		shmem->instrument_shmem_offset = 0;
+	}
+
+	return size;
+}
+
+void
+ExecRedistributeEstimate(RedistributeState *node, ParallelContext *pcxt)
+{
+	RedistributeShmem tmp;
+	tmp.plan_worker_num = pcxt->nworkers + 1;
+
+	shm_toc_estimate_chunk(&pcxt->estimator,
+						   ComputeRedistributeShmemSize(&tmp, node->ps.state->es_instrument));
+	shm_toc_estimate_keys(&pcxt->estimator, 1);
+}
+
+static void
+InitializeRedistributeSharedMemory(RedistributeShmem *shmem,
+								   dsm_segment *seg)
+{
+	uint32	i,j;
+	char	name[MAXPGPATH];
+	char   *addr;
+	Assert(shmem->plan_worker_num > 1);
+
+	/* base init */
+	BarrierInit(&shmem->part_barrier, 0);
+	BarrierAttach(&shmem->part_barrier);	/* let other worker wait */
+	pg_atomic_init_u32(&shmem->final_worker_num, 0);
+	pg_atomic_init_u32(&shmem->next_idle_part, 0);
+	if (seg != NULL)	/* not reinitialize */
+		SharedFileSetInit(&shmem->fileset, seg);
+
+	/* initialize all SharedTuplestore */
+	Assert(shmem->sts_shmem_offset >= offsetof(RedistributeShmem, padding) &&
+		   shmem->sts_shmem_size > 0);
+	addr = SharedRedistributeOffset(shmem, sts_shmem_offset);
+	for (i=0;i<shmem->plan_worker_num;++i)
+	{
+		AssertPointerAlignment(addr, MAXIMUM_ALIGNOF);
+		snprintf(name, sizeof(name), "rd%d", i);
+		sts_create((SharedTuplestore*)addr,
+				   shmem->plan_worker_num,
+				   0,
+				   SHARED_TUPLESTORE_SINGLE_PASS,
+				   name);
+		addr += shmem->sts_shmem_size;
+	}
+
+	/* initialize shm_mq */
+	Assert(shmem->mq_shmem_offset >= offsetof(RedistributeShmem, padding) &&
+		   shmem->mq_shmem_size > 0);
+	addr = SharedRedistributeOffset(shmem, mq_shmem_offset);
+	for (i=0;i<shmem->plan_worker_num;++i)
+	{
+		/* initialize message query for each worker */
+		for (j=0;j<shmem->plan_worker_num;++j)
+		{
+			AssertPointerAlignment(addr, MAXIMUM_ALIGNOF);
+			shm_mq_create(addr, shmem->mq_shmem_size);
+			addr += shmem->mq_shmem_size;
+		}
+	}
+
+	/* initialize instrument if alloced */
+	if (shmem->instrument_shmem_offset > 0)
+	{
+		Assert(shmem->instrument_shmem_offset >= offsetof(RedistributeShmem, padding));
+		addr = SharedRedistributeOffset(shmem, instrument_shmem_offset);
+		AssertPointerAlignment(addr, MAXIMUM_ALIGNOF);
+
+		MemSet(addr, 0, sizeof(RedistributeInstrumentation) * shmem->plan_worker_num);
+	}
+}
+
+void
+ExecRedistributeInitializeDSM(RedistributeState *node, ParallelContext *pcxt)
+{
+	RedistributeShmem tmp;
+
+	tmp.plan_worker_num = pcxt->nworkers + 1;
+	node->shmem = shm_toc_allocate(pcxt->toc,
+								   ComputeRedistributeShmemSize(&tmp, node->ps.state->es_instrument));
+	*node->shmem = tmp;
+	shm_toc_insert(pcxt->toc,
+				   node->ps.plan->plan_node_id,
+				   node->shmem);
+
+	InitializeRedistributeSharedMemory(node->shmem,
+									   pcxt->seg);
+}
+
+void
+ExecRedistributeReInitializeDSM(RedistributeState *node, ParallelContext *pcxt)
+{
+	RedistributeShmem *shmem = node->shmem;
+	if (shmem)
+	{
+		SharedFileSetDeleteAll(&shmem->fileset);
+		InitializeRedistributeSharedMemory(shmem, NULL);
+	}
+	ExecSetExecProcNode(&node->ps, ExecRedistributeError);
+}
+
+/*
+ * find shared memory and change execute proc function
+ */
+void
+ExecRedistributeInitializeWorker(RedistributeState *node, ParallelWorkerContext *pwcxt)
+{
+	node->shmem = shm_toc_lookup(pwcxt->toc,
+								 node->ps.plan->plan_node_id,
+								 false);
+	ExecSetExecProcNode(&node->ps, ExecRedistributeReady);
+}
+
+/*
+ * Call after all parallel workers startup, make sure got parallel
+ * workers startup success count number(ParallelContext::nworkers_launched).
+ */
+void
+ExecRedistributeParallelLaunched(RedistributeState *node, ParallelContext *pcxt)
+{
+	RedistributeShmem *shmem;
+	if (pcxt->nworkers_launched > 0)
+	{
+		/*
+		 * All workers should be waiting leader change final_worker_num,
+		 * Leader already attached part_barrier at initialize DSM
+		 */
+		shmem = node->shmem;
+		Assert(BarrierPhase(&shmem->part_barrier) == PRW_WAIT_WORKER_NUMBER);
+		pg_atomic_write_u32(&shmem->final_worker_num, pcxt->nworkers_launched + 1);
+		BarrierArriveAndDetach(&shmem->part_barrier);
+
+		ExecSetExecProcNode(&node->ps, ExecRedistributeReady);
+	}
+	else
+	{
+		/*
+		 * have no any parallel work start success,
+		 * just return tuples from outer
+		 */
+		ExecSetExecProcNode(&node->ps, ExecRedistributeDirect);
+	}
+}
+
+/* ----------------------------------------------------------------
+ *		ExecRedistributeRetrieveInstrumentation
+ *
+ *		Transfer redistribute statistics from DSM to private memory.
+ * ----------------------------------------------------------------
+ */
+void
+ExecRedistributeRetrieveInstrumentation(RedistributeState *node)
+{
+	Size					size;
+	SharedRedistributeInfo *info;
+	RedistributeShmem	   *shmem = node->shmem;
+
+	if (shmem == NULL ||
+		shmem->instrument_shmem_offset == 0)
+		return;
+
+	size = shmem->plan_worker_num * sizeof(RedistributeInstrumentation);
+	info = palloc(size + offsetof(SharedRedistributeInfo, sinstrument));
+	memcpy(&info->sinstrument[0],
+		   SharedRedistributeOffset(shmem, instrument_shmem_offset),
+		   size);
+	info->num_workers = shmem->plan_worker_num;
+	node->shared_instrument = info;
+}
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 228387eaee..201e2a9a1b 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -1235,6 +1235,28 @@ _copyLimit(const Limit *from)
 	return newnode;
 }
 
+/*
+ * _copyRedistribute
+ */
+static Redistribute *
+_copyRedistribute(const Redistribute *from)
+{
+	Redistribute	   *newnode = makeNode(Redistribute);
+
+	/*
+	 * copy node superclass fields
+	 */
+	CopyPlanFields((const Plan *) from, (Plan *) newnode);
+
+	/*
+	 * copy remainder of node
+	 */
+	COPY_SCALAR_FIELD(numCols);
+	COPY_POINTER_FIELD(hashColIdx, from->numCols * sizeof(from->hashColIdx[0]));
+
+	return newnode;
+}
+
 /*
  * _copyNestLoopParam
  */
@@ -5128,6 +5150,9 @@ copyObjectImpl(const void *from)
 		case T_Limit:
 			retval = _copyLimit(from);
 			break;
+		case T_Redistribute:
+			retval = _copyRedistribute(from);
+			break;
 		case T_NestLoopParam:
 			retval = _copyNestLoopParam(from);
 			break;
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index f2a6a6e7a0..4e60386b1f 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -971,6 +971,17 @@ _outLimit(StringInfo str, const Limit *node)
 	WRITE_OID_ARRAY(uniqCollations, node->uniqNumCols);
 }
 
+static void
+_outRedistribute(StringInfo str, const Redistribute *node)
+{
+	WRITE_NODE_TYPE("REDISTRIBUTE");
+
+	_outPlanInfo(str, (const Plan *) node);
+
+	WRITE_INT_FIELD(numCols);
+	WRITE_ATTRNUMBER_ARRAY(hashColIdx, node->numCols);
+}
+
 static void
 _outNestLoopParam(StringInfo str, const NestLoopParam *node)
 {
@@ -2216,6 +2227,17 @@ _outLimitPath(StringInfo str, const LimitPath *node)
 	WRITE_ENUM_FIELD(limitOption, LimitOption);
 }
 
+static void
+_outRedistributePath(StringInfo str, const RedistributePath *node)
+{
+	WRITE_NODE_TYPE("REDISTRIBUTEPATH");
+
+	_outPathInfo(str, (const Path *) node);
+
+	WRITE_NODE_FIELD(subpath);
+	WRITE_NODE_FIELD(hashClause);
+}
+
 static void
 _outGatherMergePath(StringInfo str, const GatherMergePath *node)
 {
@@ -3987,6 +4009,9 @@ outNode(StringInfo str, const void *obj)
 			case T_Limit:
 				_outLimit(str, obj);
 				break;
+			case T_Redistribute:
+				_outRedistribute(str, obj);
+				break;
 			case T_NestLoopParam:
 				_outNestLoopParam(str, obj);
 				break;
@@ -4257,6 +4282,9 @@ outNode(StringInfo str, const void *obj)
 			case T_LimitPath:
 				_outLimitPath(str, obj);
 				break;
+			case T_RedistributePath:
+				_outRedistributePath(str, obj);
+				break;
 			case T_GatherMergePath:
 				_outGatherMergePath(str, obj);
 				break;
diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c
index 0dd1ad7dfc..e02b8355b9 100644
--- a/src/backend/nodes/readfuncs.c
+++ b/src/backend/nodes/readfuncs.c
@@ -2493,6 +2493,22 @@ _readLimit(void)
 	READ_DONE();
 }
 
+/*
+ * _readRedistribute
+ */
+static Redistribute *
+_readRedistribute(void)
+{
+	READ_LOCALS(Redistribute);
+
+	ReadCommonPlan(&local_node->plan);
+
+	READ_INT_FIELD(numCols);
+	READ_ATTRNUMBER_ARRAY(hashColIdx, local_node->numCols);
+
+	READ_DONE();
+}
+
 /*
  * _readNestLoopParam
  */
@@ -2949,6 +2965,8 @@ parseNodeString(void)
 		return_value = _readLockRows();
 	else if (MATCH("LIMIT", 5))
 		return_value = _readLimit();
+	else if (MATCH("REDISTRIBUTE", 12))
+		return_value = _readRedistribute();
 	else if (MATCH("NESTLOOPPARAM", 13))
 		return_value = _readNestLoopParam();
 	else if (MATCH("PLANROWMARK", 11))
diff --git a/src/backend/optimizer/path/costsize.c b/src/backend/optimizer/path/costsize.c
index 1096a236d5..c81cedc74f 100644
--- a/src/backend/optimizer/path/costsize.c
+++ b/src/backend/optimizer/path/costsize.c
@@ -4113,6 +4113,21 @@ final_cost_hashjoin(PlannerInfo *root, HashPath *path,
 	path->jpath.path.total_cost = startup_cost + run_cost;
 }
 
+void
+cost_redistribute(RedistributePath *path)
+{
+	Cost	run_cost;
+	double	rows = path->path.rows;
+
+	/* hash cost */
+	run_cost = cpu_operator_cost * list_length(path->hashClause) * rows;
+
+	/* communication cost */
+	run_cost += parallel_tuple_cost * (rows - rows / (path->path.parallel_workers+1));
+
+	path->path.startup_cost = path->subpath->startup_cost;
+	path->path.total_cost = path->subpath->total_cost + run_cost;
+}
 
 /*
  * cost_subplan
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index a5f6d678cc..84488c9858 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -311,7 +311,9 @@ static ModifyTable *make_modifytable(PlannerInfo *root, Plan *subplan,
 									 List *rowMarks, OnConflictExpr *onconflict, int epqParam);
 static GatherMerge *create_gather_merge_plan(PlannerInfo *root,
 											 GatherMergePath *best_path);
-
+static Redistribute *create_redistribute_plan(PlannerInfo *root,
+											  RedistributePath *best_path,
+											  int flags);
 
 /*
  * create_plan
@@ -536,6 +538,11 @@ create_plan_recurse(PlannerInfo *root, Path *best_path, int flags)
 			plan = (Plan *) create_gather_merge_plan(root,
 													 (GatherMergePath *) best_path);
 			break;
+		case T_Redistribute:
+			plan = (Plan *) create_redistribute_plan(root,
+													 (RedistributePath *) best_path,
+													 flags);
+			break;
 		default:
 			elog(ERROR, "unrecognized node type: %d",
 				 (int) best_path->pathtype);
@@ -1922,6 +1929,28 @@ create_gather_merge_plan(PlannerInfo *root, GatherMergePath *best_path)
 	return gm_plan;
 }
 
+static Redistribute *
+create_redistribute_plan(PlannerInfo *root, RedistributePath *best_path, int flags)
+{
+	Redistribute   *plan;
+	Plan		   *subplan;
+
+	subplan = create_plan_recurse(root, best_path->subpath,
+								  flags | CP_SMALL_TLIST);
+
+	plan = makeNode(Redistribute);
+	plan->plan.targetlist = subplan->targetlist;
+	plan->plan.qual = NIL;
+	outerPlan(plan) = subplan;
+	innerPlan(plan) = NULL;
+	plan->numCols = list_length(best_path->hashClause);
+	plan->hashColIdx = extract_grouping_cols(best_path->hashClause,
+											 subplan->targetlist);
+
+	copy_generic_path_info(&plan->plan, &best_path->path);
+	return plan;
+}
+
 /*
  * create_projection_plan
  *
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index 1e42d75465..77f9d43797 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -192,6 +192,9 @@ static RelOptInfo *create_distinct_paths(PlannerInfo *root,
 static void create_partial_distinct_paths(PlannerInfo *root,
 										  RelOptInfo *input_rel,
 										  RelOptInfo *final_distinct_rel);
+static void create_parallel_distinct_paths(PlannerInfo *root,
+										   RelOptInfo *input_rel,
+										   RelOptInfo *distinct_rel);
 static RelOptInfo *create_final_distinct_paths(PlannerInfo *root,
 											   RelOptInfo *input_rel,
 											   RelOptInfo *distinct_rel);
@@ -3361,6 +3364,7 @@ create_grouping_paths(PlannerInfo *root,
 		extra.havingQual = parse->havingQual;
 		extra.targetList = parse->targetList;
 		extra.partial_costs_set = false;
+		extra.hashableList = grouping_get_hashable(parse->groupClause);
 
 		/*
 		 * Determine whether partitionwise aggregation is in theory possible.
@@ -4257,6 +4261,7 @@ create_distinct_paths(PlannerInfo *root, RelOptInfo *input_rel)
 	distinct_rel->userid = input_rel->userid;
 	distinct_rel->useridiscurrent = input_rel->useridiscurrent;
 	distinct_rel->fdwroutine = input_rel->fdwroutine;
+	distinct_rel->reltarget = root->upper_targets[UPPERREL_DISTINCT];
 
 	/* build distinct paths based on input_rel's pathlist */
 	create_final_distinct_paths(root, input_rel, distinct_rel);
@@ -4264,6 +4269,11 @@ create_distinct_paths(PlannerInfo *root, RelOptInfo *input_rel)
 	/* now build distinct paths based on input_rel's partial_pathlist */
 	create_partial_distinct_paths(root, input_rel, distinct_rel);
 
+	/* now build parallel distinct paths based on input_rel's partial_pathlist */
+	create_parallel_distinct_paths(root, input_rel, distinct_rel);
+
+	generate_useful_gather_paths(root, distinct_rel, false);
+
 	/* Give a helpful error if we failed to create any paths */
 	if (distinct_rel->pathlist == NIL)
 		ereport(ERROR,
@@ -4565,6 +4575,95 @@ create_final_distinct_paths(PlannerInfo *root, RelOptInfo *input_rel,
 	return distinct_rel;
 }
 
+static void
+create_parallel_distinct_paths(PlannerInfo *root,
+							   RelOptInfo *input_rel,
+							   RelOptInfo *distinct_rel)
+{
+	List	   *hashable_clause;
+	Query	   *parse;
+	List	   *distinctExprs;
+	double		numDistinctRows;
+	Path	   *cheapest_partial_path;
+	Path	   *path;
+
+	if (distinct_rel->consider_parallel == false ||
+		redistribute_query_size <= 0 ||
+		input_rel->partial_pathlist == NIL)
+		return;
+
+	parse = root->parse;
+
+	hashable_clause = grouping_get_hashable(parse->distinctClause);
+	if (hashable_clause == NIL)
+		return;
+
+	cheapest_partial_path = linitial(input_rel->partial_pathlist);
+
+	distinctExprs = get_sortgrouplist_exprs(parse->distinctClause,
+											parse->targetList);
+
+	/* estimate how many distinct rows we'll get from each worker */
+	numDistinctRows = estimate_num_groups(root, distinctExprs,
+										  input_rel->cheapest_total_path->rows,
+										  NULL, NULL);
+	numDistinctRows /= (cheapest_partial_path->parallel_workers + 1);
+	if (numDistinctRows < 1.0)
+		numDistinctRows = 1.0;
+
+	if (grouping_is_sortable(parse->distinctClause))
+	{
+		List	   *needed_pathkeys;
+
+		/* For explicit-sort case, always use the more rigorous clause */
+		if (list_length(root->distinct_pathkeys) <
+			list_length(root->sort_pathkeys))
+		{
+			needed_pathkeys = root->sort_pathkeys;
+			/* Assert checks that parser didn't mess up... */
+			Assert(pathkeys_contained_in(root->distinct_pathkeys,
+										 needed_pathkeys));
+		}
+		else
+			needed_pathkeys = root->distinct_pathkeys;
+
+		path = (Path *) create_redistribute_path(root,
+												 distinct_rel,
+												 cheapest_partial_path,
+												 hashable_clause);
+		path = (Path *) create_sort_path(root,
+										 distinct_rel,
+										 path,
+										 needed_pathkeys,
+										 -1.0);
+		path = (Path *)create_upper_unique_path(root,
+												distinct_rel,
+												path,
+												list_length(root->distinct_pathkeys),
+												numDistinctRows);
+		add_partial_path(distinct_rel, path);
+	}
+
+	if (grouping_is_hashable(parse->distinctClause))
+	{
+		path = (Path *) create_redistribute_path(root,
+												 distinct_rel,
+												 cheapest_partial_path,
+												 hashable_clause);
+		path = (Path *)create_agg_path(root,
+									   distinct_rel,
+									   path,
+									   path->pathtarget,
+									   AGG_HASHED,
+									   AGGSPLIT_SIMPLE,
+									   parse->distinctClause,
+									   NIL,
+									   NULL,
+									   numDistinctRows);
+		add_partial_path(distinct_rel, path);
+	}
+}
+
 /*
  * create_ordered_paths
  *
@@ -6294,6 +6393,57 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
 			}
 		}
 
+		if (parse->groupingSets == NIL && /* not support grouping sets */
+			extra->hashableList != NIL &&
+			grouped_rel->consider_parallel &&
+			input_rel->partial_pathlist != NIL &&
+			redistribute_query_size > 0)
+		{
+			Path   *path = linitial(input_rel->partial_pathlist);
+			double	rows = dNumGroups / (path->parallel_workers + 1);
+			if (rows < 1.0)
+				rows = 1.0;
+
+			/* Redistribute tuples using hashable group columns */
+			path = (Path*) create_redistribute_path(root,
+													grouped_rel,
+													path,
+													extra->hashableList);
+
+			/* Sort the path */
+			path = (Path*) create_sort_path(root,
+											grouped_rel,
+											path,
+											root->group_pathkeys,
+											-1.0);
+
+			Assert(parse->groupClause != NIL);
+			if (parse->hasAggs)
+			{
+				add_partial_path(grouped_rel, (Path *)
+						 create_agg_path(root,
+										 grouped_rel,
+										 path,
+										 grouped_rel->reltarget,
+										 AGG_SORTED,
+										 AGGSPLIT_SIMPLE,
+										 parse->groupClause,
+										 havingQual,
+										 agg_costs,
+										 rows));
+			}
+			else
+			{
+				add_partial_path(grouped_rel, (Path *)
+								 create_group_path(root,
+												   grouped_rel,
+												   path,
+												   parse->groupClause,
+												   havingQual,
+												   dNumGroups));
+			}
+		}
+
 		/*
 		 * Instead of operating directly on the input relation, we can
 		 * consider finalizing a partially aggregated path.
@@ -6397,6 +6547,58 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
 											   havingQual,
 											   dNumGroups));
 			}
+
+			if (parse->groupingSets == NIL && /* not support grouping sets */
+				extra->hashableList != NIL &&
+				grouped_rel->consider_parallel &&
+				partially_grouped_rel->partial_pathlist != NIL &&
+				redistribute_query_size > 0)
+			{
+				Path   *path = linitial(partially_grouped_rel->partial_pathlist);
+				double	rows = dNumGroups / (path->parallel_workers + 1);
+				if (rows < 1.0)
+					rows = 1.0;
+
+				/* Redistribute tuples using hashable group columns */
+				path = (Path*) create_redistribute_path(root,
+														grouped_rel,
+														path,
+														extra->hashableList);
+
+				/* Sort the path */
+				path = (Path*) create_sort_path(root,
+												grouped_rel,
+												path,
+												root->group_pathkeys,
+												-1.0);
+
+				Assert(parse->groupClause != NIL);
+				if (parse->hasAggs)
+				{
+					add_partial_path(grouped_rel, (Path *)
+							 create_agg_path(root,
+											 grouped_rel,
+											 path,
+											 grouped_rel->reltarget,
+											 AGG_SORTED,
+											 AGGSPLIT_FINAL_DESERIAL,
+											 parse->groupClause,
+											 havingQual,
+											 agg_costs,
+											 rows));
+				}
+				else
+				{
+					add_partial_path(grouped_rel, (Path *)
+									 create_group_path(root,
+													   grouped_rel,
+													   path,
+													   parse->groupClause,
+													   havingQual,
+													   dNumGroups));
+				}
+			}
+
 		}
 	}
 
@@ -6427,6 +6629,34 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
 									 havingQual,
 									 agg_costs,
 									 dNumGroups));
+
+			if (redistribute_query_size > 0 &&
+				extra->hashableList != NIL &&
+				grouped_rel->consider_parallel &&
+				input_rel->partial_pathlist != NIL)
+			{
+				Path   *path = linitial(input_rel->partial_pathlist);
+				double	rows = dNumGroups / (path->parallel_workers + 1);
+				if (rows < 1.0)
+					rows = 1.0;
+
+				/* Redistribute tuples using hashable group columns */
+				path = (Path*) create_redistribute_path(root,
+														grouped_rel,
+														path,
+														extra->hashableList);
+
+				path = (Path *) create_agg_path(root, grouped_rel,
+												path,
+												grouped_rel->reltarget,
+												AGG_HASHED,
+												AGGSPLIT_SIMPLE,
+												parse->groupClause,
+												havingQual,
+												agg_costs,
+												rows);
+				add_partial_path(grouped_rel, path);
+			}
 		}
 
 		/*
@@ -6448,6 +6678,36 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
 									 havingQual,
 									 agg_final_costs,
 									 dNumGroups));
+
+			if (redistribute_query_size > 0 &&
+				extra->hashableList != NIL &&
+				grouped_rel->consider_parallel &&
+				partially_grouped_rel->partial_pathlist != NIL)
+			{
+				double	rows;
+				path = linitial(partially_grouped_rel->partial_pathlist);
+				rows = dNumGroups / (path->parallel_workers + 1);
+				if (rows < 1.0)
+					rows = 1.0;
+
+				/* Redistribute tuples using hashable group columns */
+				path = (Path*) create_redistribute_path(root,
+														grouped_rel,
+														path,
+														extra->hashableList);
+
+				path = (Path *) create_agg_path(root,
+												grouped_rel,
+												path,
+												grouped_rel->reltarget,
+												AGG_HASHED,
+												AGGSPLIT_FINAL_DESERIAL,
+												parse->groupClause,
+												havingQual,
+												agg_final_costs,
+												rows);
+				add_partial_path(grouped_rel, path);
+			}
 		}
 	}
 
diff --git a/src/backend/optimizer/plan/setrefs.c b/src/backend/optimizer/plan/setrefs.c
index 3d23a2e5ac..0f03eb2cff 100644
--- a/src/backend/optimizer/plan/setrefs.c
+++ b/src/backend/optimizer/plan/setrefs.c
@@ -810,6 +810,7 @@ set_plan_refs(PlannerInfo *root, Plan *plan, int rtoffset)
 		case T_IncrementalSort:
 		case T_Unique:
 		case T_SetOp:
+		case T_Redistribute:
 
 			/*
 			 * These plan types don't actually bother to evaluate their
diff --git a/src/backend/optimizer/plan/subselect.c b/src/backend/optimizer/plan/subselect.c
index c9f7a09d10..ef9aaeb4c4 100644
--- a/src/backend/optimizer/plan/subselect.c
+++ b/src/backend/optimizer/plan/subselect.c
@@ -2756,6 +2756,7 @@ finalize_plan(PlannerInfo *root, Plan *plan,
 		case T_Unique:
 		case T_SetOp:
 		case T_Group:
+		case T_Redistribute:
 			/* no node-type-specific fields need fixing */
 			break;
 
diff --git a/src/backend/optimizer/prep/prepunion.c b/src/backend/optimizer/prep/prepunion.c
index e9256a2d4d..5bb5acf5d1 100644
--- a/src/backend/optimizer/prep/prepunion.c
+++ b/src/backend/optimizer/prep/prepunion.c
@@ -67,7 +67,7 @@ static List *plan_union_children(PlannerInfo *root,
 								 List *refnames_tlist,
 								 List **tlist_list);
 static Path *make_union_unique(SetOperationStmt *op, Path *path, List *tlist,
-							   PlannerInfo *root);
+							   PlannerInfo *root, List *groupList, List *sortKeys);
 static void postprocess_setop_rel(PlannerInfo *root, RelOptInfo *rel);
 static bool choose_hashed_setop(PlannerInfo *root, List *groupClauses,
 								Path *input_path,
@@ -355,6 +355,7 @@ recurse_set_operations(Node *setOp, PlannerInfo *root,
 			rel = generate_nonunion_paths(op, root,
 										  refnames_tlist,
 										  pTargetList);
+		generate_useful_gather_paths(root, rel, false);
 		if (pNumGroups)
 			*pNumGroups = rel->rows;
 
@@ -553,6 +554,8 @@ generate_union_paths(SetOperationStmt *op, PlannerInfo *root,
 	List	   *tlist_list;
 	List	   *tlist;
 	Path	   *path;
+	List	   *groupList = NIL;
+	List	   *sortKeys = NIL;
 
 	/*
 	 * If plain UNION, tell children to fetch all tuples.
@@ -588,6 +591,15 @@ generate_union_paths(SetOperationStmt *op, PlannerInfo *root,
 
 	*pTargetList = tlist;
 
+	if (!op->all)
+	{
+		/* Identify the grouping semantics */
+		groupList = generate_setop_grouplist(op, tlist);
+		if (groupList != NIL &&
+			grouping_is_sortable(groupList))
+			sortKeys = make_pathkeys_for_sortclauses(root, groupList, tlist);
+	}
+
 	/* Build path lists and relid set. */
 	foreach(lc, rellist)
 	{
@@ -628,7 +640,7 @@ generate_union_paths(SetOperationStmt *op, PlannerInfo *root,
 	 * node(s) to remove duplicates.
 	 */
 	if (!op->all)
-		path = make_union_unique(op, path, tlist, root);
+		path = make_union_unique(op, path, tlist, root, groupList, sortKeys);
 
 	add_path(result_rel, path);
 
@@ -679,11 +691,62 @@ generate_union_paths(SetOperationStmt *op, PlannerInfo *root,
 							   NIL, NULL,
 							   parallel_workers, enable_parallel_append,
 							   -1);
+
+		if (!op->all &&
+			redistribute_query_size > 0)
+		{
+			Path   *parallel_path;
+			List   *hashable_list;
+
+			/* create parallel sort union */
+			if (sortKeys != NIL &&
+				(hashable_list = grouping_get_hashable(groupList)) != NIL)
+			{
+				Assert(list_length(sortKeys) >= list_length(hashable_list));
+
+				parallel_path = (Path*) create_redistribute_path(root,
+																 result_rel,
+																 ppath,
+																 hashable_list);
+				parallel_path = (Path*) create_sort_path(root,
+														 result_rel,
+														 parallel_path,
+														 sortKeys,
+														 -1.0);
+				parallel_path = (Path*) create_upper_unique_path(root,
+																result_rel,
+																parallel_path,
+																list_length(sortKeys),
+																ppath->rows);
+				add_partial_path(result_rel, parallel_path);
+			}
+
+			/* create parallel hash union */
+			if (grouping_is_hashable(groupList))
+			{
+				parallel_path = (Path*) create_redistribute_path(root,
+																 result_rel,
+																 ppath,
+																 groupList);
+				parallel_path = (Path*) create_agg_path(root,
+														result_rel,
+														parallel_path,
+														create_pathtarget(root, tlist),
+														AGG_HASHED,
+														AGGSPLIT_SIMPLE,
+														groupList,
+														NIL,
+														NULL,
+														ppath->rows);
+				add_partial_path(result_rel, parallel_path);
+			}
+		}
+
 		ppath = (Path *)
 			create_gather_path(root, result_rel, ppath,
 							   result_rel->reltarget, NULL, NULL);
 		if (!op->all)
-			ppath = make_union_unique(op, ppath, tlist, root);
+			ppath = make_union_unique(op, ppath, tlist, root, groupList, sortKeys);
 		add_path(result_rel, ppath);
 	}
 
@@ -934,15 +997,11 @@ plan_union_children(PlannerInfo *root,
  */
 static Path *
 make_union_unique(SetOperationStmt *op, Path *path, List *tlist,
-				  PlannerInfo *root)
+				  PlannerInfo *root, List *groupList, List *sortKeys)
 {
 	RelOptInfo *result_rel = fetch_upper_rel(root, UPPERREL_SETOP, NULL);
-	List	   *groupList;
 	double		dNumGroups;
 
-	/* Identify the grouping semantics */
-	groupList = generate_setop_grouplist(op, tlist);
-
 	/*
 	 * XXX for the moment, take the number of distinct groups as equal to the
 	 * total input size, ie, the worst case.  This is too conservative, but
@@ -977,9 +1036,7 @@ make_union_unique(SetOperationStmt *op, Path *path, List *tlist,
 				create_sort_path(root,
 								 result_rel,
 								 path,
-								 make_pathkeys_for_sortclauses(root,
-															   groupList,
-															   tlist),
+								 sortKeys,
 								 -1.0);
 		path = (Path *) create_upper_unique_path(root,
 												 result_rel,
diff --git a/src/backend/optimizer/util/pathnode.c b/src/backend/optimizer/util/pathnode.c
index cedb3848dd..a515edc827 100644
--- a/src/backend/optimizer/util/pathnode.c
+++ b/src/backend/optimizer/util/pathnode.c
@@ -3822,6 +3822,32 @@ adjust_limit_rows_costs(double *rows,	/* in/out parameter */
 	}
 }
 
+RedistributePath *
+create_redistribute_path(PlannerInfo *root,
+						 RelOptInfo *rel,
+						 Path *subpath,
+						 List *groupClause)
+{
+	RedistributePath   *path;
+	Assert(rel->consider_parallel && subpath->parallel_safe);
+
+	path = makeNode(RedistributePath);
+	path->path.pathtype = T_Redistribute;
+	path->path.parent = rel;
+	/* Redistribute doesn't project, so use source path's pathtarget */
+	path->path.pathtarget = subpath->pathtarget;
+	path->path.parallel_aware = true;
+	path->path.parallel_safe = true;
+	path->path.parallel_workers = subpath->parallel_workers;
+	path->path.rows = subpath->rows;
+	path->path.pathkeys = subpath->pathkeys;
+	path->subpath = subpath;
+	path->hashClause = groupClause;
+
+	cost_redistribute(path);
+
+	return path;
+}
 
 /*
  * reparameterize_path
diff --git a/src/backend/optimizer/util/tlist.c b/src/backend/optimizer/util/tlist.c
index 311579d059..d6b42ce67a 100644
--- a/src/backend/optimizer/util/tlist.c
+++ b/src/backend/optimizer/util/tlist.c
@@ -560,6 +560,27 @@ grouping_is_hashable(List *groupClause)
 	return true;
 }
 
+/*
+ * grouping_get_hashable - extract all hashable clause
+ *
+ * Some time we don't all clauses can be hash
+ */
+List *
+grouping_get_hashable(List *groupClause)
+{
+	ListCell   *lc;
+	List	   *result = NIL;
+
+	foreach (lc, groupClause)
+	{
+		SortGroupClause *groupcl = lfirst_node(SortGroupClause, lc);
+
+		if (groupcl->hashable)
+			result = lappend(result, groupcl);
+	}
+
+	return result;
+}
 
 /*****************************************************************************
  *		PathTarget manipulation functions
diff --git a/src/backend/storage/ipc/shm_mq.c b/src/backend/storage/ipc/shm_mq.c
index 91a7093e03..8b6ffc595e 100644
--- a/src/backend/storage/ipc/shm_mq.c
+++ b/src/backend/storage/ipc/shm_mq.c
@@ -259,6 +259,68 @@ shm_mq_get_sender(shm_mq *mq)
 	return sender;
 }
 
+/*
+ * Receiver get peer
+ */
+PGPROC *
+shm_mq_receiver_get_sender(shm_mq_handle *mqh)
+{
+	PGPROC	   *sender;
+	shm_mq	   *mq = mqh->mqh_queue;
+	Assert(mq->mq_receiver == MyProc);
+
+	/*
+	 * If the counterparty is known to have attached, we can read mq_receiver
+	 * without acquiring the spinlock and assume it isn't NULL.  Otherwise,
+	 * more caution is needed.
+	 */
+	if (mqh->mqh_counterparty_attached)
+	{
+		sender = mq->mq_sender;
+	}
+	else
+	{
+		SpinLockAcquire(&mq->mq_mutex);
+		sender = mq->mq_sender;
+		SpinLockRelease(&mq->mq_mutex);
+		if (sender != NULL)
+			mqh->mqh_counterparty_attached = true;
+	}
+
+	return sender;
+}
+
+/*
+ * Sender get peer
+ */
+PGPROC *
+shm_mq_sender_get_reciver(shm_mq_handle *mqh)
+{
+	PGPROC	   *receiver;
+	shm_mq	   *mq = mqh->mqh_queue;
+	Assert(mq->mq_sender == MyProc);
+
+	/*
+	 * If the counterparty is known to have attached, we can read mq_receiver
+	 * without acquiring the spinlock and assume it isn't NULL.  Otherwise,
+	 * more caution is needed.
+	 */
+	if (mqh->mqh_counterparty_attached)
+	{
+		receiver = mq->mq_receiver;
+	}
+	else
+	{
+		SpinLockAcquire(&mq->mq_mutex);
+		receiver = mq->mq_receiver;
+		SpinLockRelease(&mq->mq_mutex);
+		if (receiver != NULL)
+			mqh->mqh_counterparty_attached = true;
+	}
+
+	return receiver;
+}
+
 /*
  * Attach to a shared message queue so we can send or receive messages.
  *
@@ -523,6 +585,44 @@ shm_mq_sendv(shm_mq_handle *mqh, shm_mq_iovec *iov, int iovcnt, bool nowait)
 	return SHM_MQ_SUCCESS;
 }
 
+/*
+ * like shm_mq_send(nowait), but return SHM_MQ_WOULD_BLOCK immediately when
+ * no space for write data, not write any data to queue
+ */
+shm_mq_result
+shm_mq_send_once(shm_mq_handle *mqh, Size nbytes, const void *data)
+{
+	PGPROC	   *receiver;
+	shm_mq	   *mq = mqh->mqh_queue;
+	uint64		rb = pg_atomic_read_u64(&mq->mq_bytes_read);
+	uint64		wb = pg_atomic_read_u64(&mq->mq_bytes_written);
+	Size		written;
+	shm_mq_result result;
+
+	Assert(mq->mq_sender == MyProc);
+	Assert(mqh->mqh_length_word_complete == false);
+	Assert(mqh->mqh_partial_bytes == 0);
+
+	/* test space */
+	if (mq->mq_ring_size - (wb - rb) < nbytes + sizeof(Size))
+		return SHM_MQ_WOULD_BLOCK;
+
+	/* write date */
+	if ((result = shm_mq_send_bytes(mqh, sizeof(Size), &nbytes, true, &written)) != SHM_MQ_SUCCESS ||
+		(result = shm_mq_send_bytes(mqh, nbytes, data, true, &written)) != SHM_MQ_SUCCESS)
+	{
+		/* should not get SHM_MQ_WOULD_BLOCK */
+		Assert(result == SHM_MQ_DETACHED);
+		return result;
+	}
+
+	/* Notify receiver of the newly-written data(if attached), and return. */
+	receiver = shm_mq_sender_get_reciver(mqh);
+	if (receiver != NULL)
+		SetLatch(&receiver->procLatch);
+	return SHM_MQ_SUCCESS;
+}
+
 /*
  * Receive a message from a shared message queue.
  *
diff --git a/src/backend/utils/activity/wait_event.c b/src/backend/utils/activity/wait_event.c
index ef7e6bfb77..f86c2f36ae 100644
--- a/src/backend/utils/activity/wait_event.c
+++ b/src/backend/utils/activity/wait_event.c
@@ -451,6 +451,12 @@ pgstat_get_wait_ipc(WaitEventIPC w)
 		case WAIT_EVENT_XACT_GROUP_UPDATE:
 			event_name = "XactGroupUpdate";
 			break;
+		case WAIT_EVENT_REDISTRIBUTE_PARALLEL_START:
+			event_name = "RedistributeParallelStartup";
+			break;
+		case WAIT_EVENT_REDISTRIBUTE_SHARED_TUPLESTORE:
+			event_name = "RedistributeSharedTuplestore";
+			break;
 			/* no default case, so that compiler will warn */
 	}
 
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 23236fa4c3..6642fa84bc 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -137,6 +137,7 @@ extern char *temp_tablespaces;
 extern bool ignore_checksum_failure;
 extern bool ignore_invalid_pages;
 extern bool synchronize_seqscans;
+extern int	redistribute_query_size;	/* in nodeRedistribute.c */
 
 #ifdef TRACE_SYNCSCAN
 extern bool trace_syncscan;
@@ -3558,6 +3559,17 @@ static struct config_int ConfigureNamesInt[] =
 		check_client_connection_check_interval, NULL, NULL
 	},
 
+	{
+		{"redistribute_query_size", PGC_USERSET, 0,
+			gettext_noop("Sets the maximum shared memory to be used by redistribute plan"),
+			NULL,
+			GUC_UNIT_KB,
+		},
+		&redistribute_query_size,
+		0, 0, MAX_KILOBYTES,
+		NULL, NULL, NULL
+	},
+
 	/* End-of-list marker */
 	{
 		{NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL
diff --git a/src/backend/utils/sort/sharedtuplestore.c b/src/backend/utils/sort/sharedtuplestore.c
index 033088f9bc..13071d43a7 100644
--- a/src/backend/utils/sort/sharedtuplestore.c
+++ b/src/backend/utils/sort/sharedtuplestore.c
@@ -131,11 +131,22 @@ sts_initialize(SharedTuplestore *sts, int participants,
 			   SharedFileSet *fileset,
 			   const char *name)
 {
-	SharedTuplestoreAccessor *accessor;
-	int			i;
-
-	Assert(my_participant_number < participants);
+	sts_create(sts, participants, meta_data_size, flags, name);
+	return sts_attach(sts, my_participant_number, fileset);
+}
 
+/*
+ * The caller must supply a SharedFileSet, which is essentially a directory
+ * that will be cleaned up automatically, and a name which must be unique
+ * across all SharedTuplestores created in the same SharedFileSet.
+ */
+void
+sts_create(SharedTuplestore *sts, int participants,
+		   size_t meta_data_size,
+		   int flags,
+		   const char *name)
+{
+	int		i;
 	sts->nparticipants = participants;
 	sts->meta_data_size = meta_data_size;
 	sts->flags = flags;
@@ -160,14 +171,6 @@ sts_initialize(SharedTuplestore *sts, int participants,
 		sts->participants[i].read_page = 0;
 		sts->participants[i].writing = false;
 	}
-
-	accessor = palloc0(sizeof(SharedTuplestoreAccessor));
-	accessor->participant = my_participant_number;
-	accessor->sts = sts;
-	accessor->fileset = fileset;
-	accessor->context = CurrentMemoryContext;
-
-	return accessor;
 }
 
 /*
@@ -205,6 +208,32 @@ sts_flush_chunk(SharedTuplestoreAccessor *accessor)
 		STS_CHUNK_PAGES;
 }
 
+/*
+ * close opend files and free memory resource
+ * return written size
+ */
+int64
+sts_close(SharedTuplestoreAccessor *accessor)
+{
+	int64	size = 0;
+	if (accessor->write_file != NULL)
+	{
+		sts_flush_chunk(accessor);
+		size = BufFileSize(accessor->write_file);
+		BufFileClose(accessor->write_file);
+		pfree(accessor->write_chunk);
+		accessor->write_chunk = NULL;
+		accessor->write_file = NULL;
+		accessor->sts->participants[accessor->participant].writing = false;
+	}
+
+	if (accessor->read_file)
+		BufFileClose(accessor->read_file);
+	pfree(accessor);
+
+	return size;
+}
+
 /*
  * Finish writing tuples.  This must be called by all backends that have
  * written data before any backend begins reading it.
diff --git a/src/include/executor/execParallel.h b/src/include/executor/execParallel.h
index 3888175a2f..be67633358 100644
--- a/src/include/executor/execParallel.h
+++ b/src/include/executor/execParallel.h
@@ -45,6 +45,7 @@ extern void ExecParallelFinish(ParallelExecutorInfo *pei);
 extern void ExecParallelCleanup(ParallelExecutorInfo *pei);
 extern void ExecParallelReinitialize(PlanState *planstate,
 									 ParallelExecutorInfo *pei, Bitmapset *sendParam);
+extern void ExecParallelLaunched(PlanState *planstate, ParallelContext *pcxt);
 
 extern void ParallelQueryMain(dsm_segment *seg, shm_toc *toc);
 
diff --git a/src/include/executor/nodeRedistribute.h b/src/include/executor/nodeRedistribute.h
new file mode 100644
index 0000000000..960d434eb2
--- /dev/null
+++ b/src/include/executor/nodeRedistribute.h
@@ -0,0 +1,21 @@
+#ifndef NODEREDISTRIBUTE_H
+#define NODEREDISTRIBUTE_H
+
+#include "access/parallel.h"
+#include "nodes/execnodes.h"
+
+extern RedistributeState *ExecInitRedistribute(Redistribute *node, EState *estate, int eflags);
+extern void ExecEndRedistribute(RedistributeState *node);
+extern void ExecReScanRedistribute(RedistributeState *node);
+
+/* parallel scan support */
+extern void ExecRedistributeEstimate(RedistributeState *node, ParallelContext *pcxt);
+extern void ExecRedistributeInitializeDSM(RedistributeState *node, ParallelContext *pcxt);
+extern void ExecRedistributeReInitializeDSM(RedistributeState *node, ParallelContext *pcxt);
+extern void ExecRedistributeInitializeWorker(RedistributeState *node,
+											 ParallelWorkerContext *pwcxt);
+extern void ExecRedistributeParallelLaunched(RedistributeState *node,
+											 ParallelContext *pcxt);
+extern void ExecRedistributeRetrieveInstrumentation(RedistributeState *node);
+
+#endif							/* NODEREDISTRIBUTE_H */
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 37cb4f3d59..0c506faeb9 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -2640,4 +2640,43 @@ typedef struct LimitState
 	TupleTableSlot *last_slot;	/* slot for evaluation of ties */
 } LimitState;
 
+/* ---------------------
+ *	per-worker redistribute information
+ * ---------------------
+ */
+typedef struct RedistributeInstrumentation
+{
+	Size	disk_used;		/* bytes of disk space used */
+	Size	disk_rows;		/* rows of disk writed */
+	uint32	parts_got;		/* parts fetched */
+}RedistributeInstrumentation;
+
+/* ----------------
+ *	 Shared memory container for per-worker aggregate information
+ * ----------------
+ */
+typedef struct SharedRedistributeInfo
+{
+	int			num_workers;
+	RedistributeInstrumentation sinstrument[FLEXIBLE_ARRAY_MEMBER];
+} SharedRedistributeInfo;
+
+typedef struct RedistributeState
+{
+	PlanState		ps;
+	List		   *hash_funcs;				/* for compute hash with redistribute */
+	int				hash_max_attr;			/* max attribute from subplan's output */
+	struct RedistributeShmem *shmem;		/* private in nodeRedistribute.c */
+	struct RedistributeWriteInfo *writer;	/* private in nodeRedistribute.c */
+	struct SharedTuplestoreAccessor *sta;	/* in disk tuples from other workers */
+	struct shm_mq_handle  **mqreader;		/* array with nreaders active entries */
+	uint32			nreaders;				/* number of working workers for writing */
+	uint32			nextreader;				/* next one to try to read from */
+	uint32			nwriter;				/* number of array writer */
+	uint32			status_flags;			/* execute status flags */
+	uint32			current_part;			/* current partition number */
+	RedistributeInstrumentation *instrument;
+	SharedRedistributeInfo *shared_instrument;
+}RedistributeState;
+
 #endif							/* EXECNODES_H */
diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index b3ee4194d3..b935c56474 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -87,6 +87,7 @@ typedef enum NodeTag
 	T_SetOp,
 	T_LockRows,
 	T_Limit,
+	T_Redistribute,
 	/* these aren't subclasses of Plan: */
 	T_NestLoopParam,
 	T_PlanRowMark,
@@ -146,6 +147,7 @@ typedef enum NodeTag
 	T_SetOpState,
 	T_LockRowsState,
 	T_LimitState,
+	T_RedistributeState,
 
 	/*
 	 * TAGS FOR PRIMITIVE NODES (primnodes.h)
@@ -262,6 +264,7 @@ typedef enum NodeTag
 	T_LockRowsPath,
 	T_ModifyTablePath,
 	T_LimitPath,
+	T_RedistributePath,
 	/* these aren't subclasses of Path: */
 	T_EquivalenceClass,
 	T_EquivalenceMember,
diff --git a/src/include/nodes/pathnodes.h b/src/include/nodes/pathnodes.h
index c8a8eec6e2..0c73682ae5 100644
--- a/src/include/nodes/pathnodes.h
+++ b/src/include/nodes/pathnodes.h
@@ -1908,6 +1908,15 @@ typedef struct LimitPath
 	LimitOption limitOption;	/* FETCH FIRST with ties or exact number */
 } LimitPath;
 
+/*
+ * Redistribute represents redistribute tuples
+ */
+typedef struct RedistributePath
+{
+	Path		path;
+	Path	   *subpath;		/* path representing input source */
+	List	   *hashClause;		/* a list of SortGroupClause's */
+}RedistributePath;
 
 /*
  * Restriction clause info.
@@ -2596,6 +2605,7 @@ typedef struct
 	bool		target_parallel_safe;
 	Node	   *havingQual;
 	List	   *targetList;
+	List	   *hashableList;	/* hashable group list */
 	PartitionwiseAggregateType patype;
 } GroupPathExtraData;
 
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index ec9a8b0c81..48f08149f1 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -1034,6 +1034,20 @@ typedef struct Limit
 } Limit;
 
 
+/* ----------------
+ *		limit node
+ *
+ * Note: as of Postgres 8.2, the offset and count expressions are expected
+ * to yield int8, rather than int4 as before.
+ * ----------------
+ */
+typedef struct Redistribute
+{
+	Plan		plan;
+	int			numCols;		/* number of hash columns */
+	AttrNumber *hashColIdx;		/* their indexes in the target list */
+}Redistribute;
+
 /*
  * RowMarkType -
  *	  enums for types of row-marking operations
diff --git a/src/include/optimizer/cost.h b/src/include/optimizer/cost.h
index 2113bc82de..17180c01f8 100644
--- a/src/include/optimizer/cost.h
+++ b/src/include/optimizer/cost.h
@@ -68,6 +68,7 @@ extern PGDLLIMPORT bool enable_parallel_hash;
 extern PGDLLIMPORT bool enable_partition_pruning;
 extern PGDLLIMPORT bool enable_async_append;
 extern PGDLLIMPORT int constraint_exclusion;
+extern PGDLLIMPORT int redistribute_query_size; /* in nodeRedistribute.c */
 
 extern double index_pages_fetched(double tuples_fetched, BlockNumber pages,
 								  double index_pages, PlannerInfo *root);
@@ -169,6 +170,7 @@ extern void cost_gather_merge(GatherMergePath *path, PlannerInfo *root,
 							  RelOptInfo *rel, ParamPathInfo *param_info,
 							  Cost input_startup_cost, Cost input_total_cost,
 							  double *rows);
+extern void cost_redistribute(RedistributePath *path);
 extern void cost_subplan(PlannerInfo *root, SubPlan *subplan, Plan *plan);
 extern void cost_qual_eval(QualCost *cost, List *quals, PlannerInfo *root);
 extern void cost_qual_eval_node(QualCost *cost, Node *qual, PlannerInfo *root);
diff --git a/src/include/optimizer/pathnode.h b/src/include/optimizer/pathnode.h
index f704d39980..3ceecc2bf2 100644
--- a/src/include/optimizer/pathnode.h
+++ b/src/include/optimizer/pathnode.h
@@ -290,6 +290,10 @@ extern Path *reparameterize_path(PlannerInfo *root, Path *path,
 								 double loop_count);
 extern Path *reparameterize_path_by_child(PlannerInfo *root, Path *path,
 										  RelOptInfo *child_rel);
+extern RedistributePath *create_redistribute_path(PlannerInfo *root,
+												  RelOptInfo *rel,
+												  Path *subpath,
+												  List *groupClause);
 
 /*
  * prototypes for relnode.c
diff --git a/src/include/optimizer/tlist.h b/src/include/optimizer/tlist.h
index d62a09665a..afa06de584 100644
--- a/src/include/optimizer/tlist.h
+++ b/src/include/optimizer/tlist.h
@@ -35,6 +35,7 @@ extern Oid *extract_grouping_collations(List *groupClause, List *tlist);
 extern AttrNumber *extract_grouping_cols(List *groupClause, List *tlist);
 extern bool grouping_is_sortable(List *groupClause);
 extern bool grouping_is_hashable(List *groupClause);
+extern List *grouping_get_hashable(List *groupClause);
 
 extern PathTarget *make_pathtarget_from_tlist(List *tlist);
 extern List *make_tlist_from_pathtarget(PathTarget *target);
diff --git a/src/include/storage/shm_mq.h b/src/include/storage/shm_mq.h
index e693f3f760..2bd6f70161 100644
--- a/src/include/storage/shm_mq.h
+++ b/src/include/storage/shm_mq.h
@@ -54,6 +54,8 @@ extern void shm_mq_set_sender(shm_mq *mq, PGPROC *);
 /* Accessor methods for sender and receiver. */
 extern PGPROC *shm_mq_get_receiver(shm_mq *);
 extern PGPROC *shm_mq_get_sender(shm_mq *);
+extern PGPROC *shm_mq_receiver_get_sender(shm_mq_handle *mqh);
+extern PGPROC *shm_mq_sender_get_reciver(shm_mq_handle *mqh);
 
 /* Set up backend-local queue state. */
 extern shm_mq_handle *shm_mq_attach(shm_mq *mq, dsm_segment *seg,
@@ -73,6 +75,8 @@ extern shm_mq_result shm_mq_send(shm_mq_handle *mqh,
 								 Size nbytes, const void *data, bool nowait);
 extern shm_mq_result shm_mq_sendv(shm_mq_handle *mqh,
 								  shm_mq_iovec *iov, int iovcnt, bool nowait);
+extern shm_mq_result shm_mq_send_once(shm_mq_handle *mqh,
+									  Size nbytes, const void *data);
 extern shm_mq_result shm_mq_receive(shm_mq_handle *mqh,
 									Size *nbytesp, void **datap, bool nowait);
 
diff --git a/src/include/utils/sharedtuplestore.h b/src/include/utils/sharedtuplestore.h
index 01ad6efe51..ff6df228bc 100644
--- a/src/include/utils/sharedtuplestore.h
+++ b/src/include/utils/sharedtuplestore.h
@@ -39,10 +39,17 @@ extern SharedTuplestoreAccessor *sts_initialize(SharedTuplestore *sts,
 												SharedFileSet *fileset,
 												const char *name);
 
+extern void sts_create(SharedTuplestore *sts, int participants,
+					   size_t meta_data_size,
+					   int flags,
+					   const char *name);
+
 extern SharedTuplestoreAccessor *sts_attach(SharedTuplestore *sts,
 											int my_participant_number,
 											SharedFileSet *fileset);
 
+extern int64 sts_close(SharedTuplestoreAccessor *accessor);
+
 extern void sts_end_write(SharedTuplestoreAccessor *accessor);
 
 extern void sts_reinitialize(SharedTuplestoreAccessor *accessor);
diff --git a/src/include/utils/wait_event.h b/src/include/utils/wait_event.h
index 6007827b44..2487737e42 100644
--- a/src/include/utils/wait_event.h
+++ b/src/include/utils/wait_event.h
@@ -125,7 +125,9 @@ typedef enum
 	WAIT_EVENT_SYNC_REP,
 	WAIT_EVENT_WAL_RECEIVER_EXIT,
 	WAIT_EVENT_WAL_RECEIVER_WAIT_START,
-	WAIT_EVENT_XACT_GROUP_UPDATE
+	WAIT_EVENT_XACT_GROUP_UPDATE,
+	WAIT_EVENT_REDISTRIBUTE_PARALLEL_START,
+	WAIT_EVENT_REDISTRIBUTE_SHARED_TUPLESTORE
 } WaitEventIPC;
 
 /* ----------
diff --git a/src/test/modules/test_shm_mq/expected/test_shm_mq.out b/src/test/modules/test_shm_mq/expected/test_shm_mq.out
index c4858b0c20..ed5d526d89 100644
--- a/src/test/modules/test_shm_mq/expected/test_shm_mq.out
+++ b/src/test/modules/test_shm_mq/expected/test_shm_mq.out
@@ -34,3 +34,9 @@ SELECT test_shm_mq_pipelined(16384, (select string_agg(chr(32+(random()*95)::int
  
 (1 row)
 
+SELECT test_shm_mq_pipelined(16384, (select string_agg(chr(32+(random()*95)::int), '') from generate_series(1,4000)), 600, 3, true, true);
+ test_shm_mq_pipelined 
+-----------------------
+ 
+(1 row)
+
diff --git a/src/test/modules/test_shm_mq/sql/test_shm_mq.sql b/src/test/modules/test_shm_mq/sql/test_shm_mq.sql
index 9de19d304a..247417b616 100644
--- a/src/test/modules/test_shm_mq/sql/test_shm_mq.sql
+++ b/src/test/modules/test_shm_mq/sql/test_shm_mq.sql
@@ -10,3 +10,4 @@ SELECT test_shm_mq(1024, 'a', 2001, 1);
 SELECT test_shm_mq(32768, (select string_agg(chr(32+(random()*95)::int), '') from generate_series(1,(100+900*random())::int)), 10000, 1);
 SELECT test_shm_mq(100, (select string_agg(chr(32+(random()*95)::int), '') from generate_series(1,(100+200*random())::int)), 10000, 1);
 SELECT test_shm_mq_pipelined(16384, (select string_agg(chr(32+(random()*95)::int), '') from generate_series(1,270000)), 200, 3);
+SELECT test_shm_mq_pipelined(16384, (select string_agg(chr(32+(random()*95)::int), '') from generate_series(1,4000)), 600, 3, true, true);
diff --git a/src/test/modules/test_shm_mq/test.c b/src/test/modules/test_shm_mq/test.c
index 2d8d695f97..adef9e106b 100644
--- a/src/test/modules/test_shm_mq/test.c
+++ b/src/test/modules/test_shm_mq/test.c
@@ -136,6 +136,7 @@ test_shm_mq_pipelined(PG_FUNCTION_ARGS)
 	int32		loop_count = PG_GETARG_INT32(2);
 	int32		nworkers = PG_GETARG_INT32(3);
 	bool		verify = PG_GETARG_BOOL(4);
+	bool		use_once = PG_GETARG_BOOL(5);
 	int32		send_count = 0;
 	int32		receive_count = 0;
 	dsm_segment *seg;
@@ -177,7 +178,10 @@ test_shm_mq_pipelined(PG_FUNCTION_ARGS)
 		 */
 		if (send_count < loop_count)
 		{
-			res = shm_mq_send(outqh, message_size, message_contents, true);
+			if (use_once)
+				res = shm_mq_send_once(outqh, message_size, message_contents);
+			else
+				res = shm_mq_send(outqh, message_size, message_contents, true);
 			if (res == SHM_MQ_SUCCESS)
 			{
 				++send_count;
diff --git a/src/test/modules/test_shm_mq/test_shm_mq--1.0.sql b/src/test/modules/test_shm_mq/test_shm_mq--1.0.sql
index 56db05d93d..e8fc9e9348 100644
--- a/src/test/modules/test_shm_mq/test_shm_mq--1.0.sql
+++ b/src/test/modules/test_shm_mq/test_shm_mq--1.0.sql
@@ -14,6 +14,7 @@ CREATE FUNCTION test_shm_mq_pipelined(queue_size pg_catalog.int8,
 					   message pg_catalog.text,
 					   repeat_count pg_catalog.int4 default 1,
 					   num_workers pg_catalog.int4 default 1,
-					   verify pg_catalog.bool default true)
+					   verify pg_catalog.bool default true,
+					   use_once pg_catalog.bool default false)
     RETURNS pg_catalog.void STRICT
 	AS 'MODULE_PATHNAME' LANGUAGE C;
diff --git a/src/test/regress/expected/partition_aggregate.out b/src/test/regress/expected/partition_aggregate.out
index dfa4b036b5..698c761bda 100644
--- a/src/test/regress/expected/partition_aggregate.out
+++ b/src/test/regress/expected/partition_aggregate.out
@@ -1517,3 +1517,78 @@ SELECT x, sum(y), avg(y), count(*) FROM pagg_tab_para GROUP BY x HAVING avg(y) <
  21 | 6000 | 6.0000000000000000 |  1000
 (6 rows)
 
+-- simple agg in parallel
+BEGIN;
+SET redistribute_query_size = '128kB';
+SET min_parallel_table_scan_size = 0;
+SET parallel_tuple_cost = 0;
+SET parallel_setup_cost = 0;
+SET enable_indexonlyscan = OFF;
+EXPLAIN (COSTS OFF)
+SELECT unique2,count(*) FROM tenk1 GROUP BY 1;
+                  QUERY PLAN                  
+----------------------------------------------
+ Gather
+   Workers Planned: 2
+   ->  HashAggregate
+         Group Key: unique2
+         ->  Parallel Redistribute
+               Hash Key: unique2
+               ->  Parallel Seq Scan on tenk1
+(7 rows)
+
+EXPLAIN (COSTS OFF)
+SELECT count(*) FROM (SELECT unique2,count(*) FROM tenk1 GROUP BY 1) foo;
+                        QUERY PLAN                        
+----------------------------------------------------------
+ Finalize Aggregate
+   ->  Gather
+         Workers Planned: 2
+         ->  Partial Aggregate
+               ->  HashAggregate
+                     Group Key: tenk1.unique2
+                     ->  Parallel Redistribute
+                           Hash Key: tenk1.unique2
+                           ->  Parallel Seq Scan on tenk1
+(9 rows)
+
+SELECT count(*) FROM (SELECT unique2,count(*) FROM tenk1 GROUP BY 1) foo;
+ count 
+-------
+ 10000
+(1 row)
+
+EXPLAIN (COSTS OFF)
+SELECT count(*) FROM
+  (SELECT unique2,count(*) id FROM tenk1 GROUP BY 1) t1
+    INNER JOIN
+  (SELECT unique2  id FROM tenk1) t2
+  USING(id);
+                               QUERY PLAN                               
+------------------------------------------------------------------------
+ Finalize Aggregate
+   ->  Gather
+         Workers Planned: 2
+         ->  Partial Aggregate
+               ->  Parallel Hash Join
+                     Hash Cond: ((count(*)) = tenk1.unique2)
+                     ->  HashAggregate
+                           Group Key: tenk1_1.unique2
+                           ->  Parallel Redistribute
+                                 Hash Key: tenk1_1.unique2
+                                 ->  Parallel Seq Scan on tenk1 tenk1_1
+                     ->  Parallel Hash
+                           ->  Parallel Seq Scan on tenk1
+(13 rows)
+
+SELECT count(*) FROM
+  (SELECT unique2,count(*) id FROM tenk1 GROUP BY 1) t1
+    INNER JOIN
+  (SELECT unique2  id FROM tenk1) t2
+  USING(id);
+ count 
+-------
+ 10000
+(1 row)
+
+ABORT;
diff --git a/src/test/regress/expected/select_distinct.out b/src/test/regress/expected/select_distinct.out
index 58122c6f88..2c0b865677 100644
--- a/src/test/regress/expected/select_distinct.out
+++ b/src/test/regress/expected/select_distinct.out
@@ -375,3 +375,49 @@ SELECT null IS NOT DISTINCT FROM null as "yes";
  t
 (1 row)
 
+-- parallel distinct
+BEGIN;
+SET redistribute_query_size = '128kB';
+SET min_parallel_table_scan_size =0;
+SET parallel_tuple_cost = 0;
+SET parallel_setup_cost = 0;
+SET enable_indexonlyscan = OFF;
+EXPLAIN (costs off)
+SELECT COUNT(*) FROM (SELECT DISTINCT unique2 FROM tenk1) foo;
+                     QUERY PLAN                     
+----------------------------------------------------
+ Aggregate
+   ->  Gather
+         Workers Planned: 2
+         ->  HashAggregate
+               Group Key: tenk1.unique2
+               ->  Parallel Redistribute
+                     Hash Key: tenk1.unique2
+                     ->  Parallel Seq Scan on tenk1
+(8 rows)
+
+SELECT COUNT(*) FROM (SELECT DISTINCT unique2 FROM tenk1) foo;
+ count 
+-------
+ 10000
+(1 row)
+
+explain (costs off)
+SELECT DISTINCT * FROM (SELECT DISTINCT unique2 FROM tenk1) foo;
+                           QUERY PLAN                           
+----------------------------------------------------------------
+ Gather Merge
+   Workers Planned: 2
+   ->  Unique
+         ->  Sort
+               Sort Key: tenk1.unique2
+               ->  Parallel Redistribute
+                     Hash Key: tenk1.unique2
+                     ->  HashAggregate
+                           Group Key: tenk1.unique2
+                           ->  Parallel Redistribute
+                                 Hash Key: tenk1.unique2
+                                 ->  Parallel Seq Scan on tenk1
+(12 rows)
+
+ABORT;
diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
index 4ea1aa7dfd..34a61b62e0 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -845,6 +845,54 @@ select * from
 
 reset enable_material;
 reset enable_hashagg;
+--test rescan for redistribute
+SET redistribute_query_size = '128kB';
+SET enable_sort = off;
+explain (costs off)
+select * from
+  (select string4, count(unique2)
+   from tenk1 group by string4) ss
+  right join (values (1),(2),(3)) v(x) on true order by 3,1;
+                              QUERY PLAN                              
+----------------------------------------------------------------------
+ Sort
+   Sort Key: "*VALUES*".column1, tenk1.string4
+   ->  Nested Loop Left Join
+         ->  Values Scan on "*VALUES*"
+         ->  Materialize
+               ->  Gather
+                     Workers Planned: 4
+                     ->  Finalize HashAggregate
+                           Group Key: tenk1.string4
+                           ->  Parallel Redistribute
+                                 Hash Key: tenk1.string4
+                                 ->  Partial HashAggregate
+                                       Group Key: tenk1.string4
+                                       ->  Parallel Seq Scan on tenk1
+(14 rows)
+
+select * from
+  (select string4, count(unique2)
+   from tenk1 group by string4) ss
+  right join (values (1),(2),(3)) v(x) on true order by 3,1;
+ string4 | count | x 
+---------+-------+---
+ AAAAxx  |  2500 | 1
+ HHHHxx  |  2500 | 1
+ OOOOxx  |  2500 | 1
+ VVVVxx  |  2500 | 1
+ AAAAxx  |  2500 | 2
+ HHHHxx  |  2500 | 2
+ OOOOxx  |  2500 | 2
+ VVVVxx  |  2500 | 2
+ AAAAxx  |  2500 | 3
+ HHHHxx  |  2500 | 3
+ OOOOxx  |  2500 | 3
+ VVVVxx  |  2500 | 3
+(12 rows)
+
+reset enable_sort;
+reset redistribute_query_size;
 -- check parallelized int8 aggregate (bug #14897)
 explain (costs off)
 select avg(unique1::int8) from tenk1;
diff --git a/src/test/regress/expected/union.out b/src/test/regress/expected/union.out
index dece7310cf..29fb62bb61 100644
--- a/src/test/regress/expected/union.out
+++ b/src/test/regress/expected/union.out
@@ -1432,3 +1432,98 @@ where (x = 0) or (q1 >= q2 and q1 <= q2);
  4567890123456789 |  4567890123456789 | 1
 (6 rows)
 
+-- parallel union
+BEGIN;
+SET redistribute_query_size = '128kB';
+SET min_parallel_table_scan_size =0;
+SET parallel_tuple_cost = 0;
+SET parallel_setup_cost = 0;
+SET enable_indexonlyscan = OFF;
+-- using redistribute+sort sort
+SET enable_hashagg = OFF;
+EXPLAIN (costs off)
+SELECT count(*) FROM (SELECT unique2 FROM tenk1 UNION SELECT unique2 FROM tenk1) foo;
+                                  QUERY PLAN                                  
+------------------------------------------------------------------------------
+ Finalize Aggregate
+   ->  Gather
+         Workers Planned: 2
+         ->  Partial Aggregate
+               ->  Unique
+                     ->  Sort
+                           Sort Key: tenk1.unique2
+                           ->  Parallel Redistribute
+                                 Hash Key: tenk1.unique2
+                                 ->  Parallel Append
+                                       ->  Parallel Seq Scan on tenk1
+                                       ->  Parallel Seq Scan on tenk1 tenk1_1
+(12 rows)
+
+SELECT count(*) FROM (SELECT unique2 FROM tenk1 UNION SELECT unique2 FROM tenk1) foo;
+ count 
+-------
+ 10000
+(1 row)
+
+EXPLAIN (costs off)
+SELECT count(*) FROM (SELECT unique2 FROM tenk1 UNION SELECT unique2 FROM tenk1) t1 INNER JOIN tenk1 USING(unique2);
+                                        QUERY PLAN                                        
+------------------------------------------------------------------------------------------
+ Finalize Aggregate
+   ->  Gather
+         Workers Planned: 2
+         ->  Partial Aggregate
+               ->  Parallel Hash Join
+                     Hash Cond: (tenk1.unique2 = tenk1_1.unique2)
+                     ->  Parallel Seq Scan on tenk1
+                     ->  Parallel Hash
+                           ->  Unique
+                                 ->  Sort
+                                       Sort Key: tenk1_1.unique2
+                                       ->  Parallel Redistribute
+                                             Hash Key: tenk1_1.unique2
+                                             ->  Parallel Append
+                                                   ->  Parallel Seq Scan on tenk1 tenk1_1
+                                                   ->  Parallel Seq Scan on tenk1 tenk1_2
+(16 rows)
+
+SELECT count(*) FROM (SELECT unique2 FROM tenk1 UNION SELECT unique2 FROM tenk1) t1 INNER JOIN tenk1 USING(unique2);
+ count 
+-------
+ 10000
+(1 row)
+
+reset enable_hashagg;
+-- using batch hash
+SET enable_sort = OFF;
+EXPLAIN (costs off)
+SELECT count(*) from (
+  SELECT hundred from tenk1
+    union
+  SELECT hundred from tenk1) foo;
+                               QUERY PLAN                               
+------------------------------------------------------------------------
+ Finalize Aggregate
+   ->  Gather
+         Workers Planned: 2
+         ->  Partial Aggregate
+               ->  HashAggregate
+                     Group Key: tenk1.hundred
+                     ->  Parallel Redistribute
+                           Hash Key: tenk1.hundred
+                           ->  Parallel Append
+                                 ->  Parallel Seq Scan on tenk1
+                                 ->  Parallel Seq Scan on tenk1 tenk1_1
+(11 rows)
+
+SELECT count(*) from (
+  SELECT hundred from tenk1
+    union
+  SELECT hundred from tenk1) foo;
+ count 
+-------
+   100
+(1 row)
+
+reset enable_sort;
+ABORT;
diff --git a/src/test/regress/sql/partition_aggregate.sql b/src/test/regress/sql/partition_aggregate.sql
index c17294b15b..091f5d3eaa 100644
--- a/src/test/regress/sql/partition_aggregate.sql
+++ b/src/test/regress/sql/partition_aggregate.sql
@@ -331,3 +331,29 @@ RESET parallel_setup_cost;
 EXPLAIN (COSTS OFF)
 SELECT x, sum(y), avg(y), count(*) FROM pagg_tab_para GROUP BY x HAVING avg(y) < 7 ORDER BY 1, 2, 3;
 SELECT x, sum(y), avg(y), count(*) FROM pagg_tab_para GROUP BY x HAVING avg(y) < 7 ORDER BY 1, 2, 3;
+
+-- simple agg in parallel
+BEGIN;
+SET redistribute_query_size = '128kB';
+SET min_parallel_table_scan_size = 0;
+SET parallel_tuple_cost = 0;
+SET parallel_setup_cost = 0;
+SET enable_indexonlyscan = OFF;
+
+EXPLAIN (COSTS OFF)
+SELECT unique2,count(*) FROM tenk1 GROUP BY 1;
+EXPLAIN (COSTS OFF)
+SELECT count(*) FROM (SELECT unique2,count(*) FROM tenk1 GROUP BY 1) foo;
+SELECT count(*) FROM (SELECT unique2,count(*) FROM tenk1 GROUP BY 1) foo;
+EXPLAIN (COSTS OFF)
+SELECT count(*) FROM
+  (SELECT unique2,count(*) id FROM tenk1 GROUP BY 1) t1
+    INNER JOIN
+  (SELECT unique2  id FROM tenk1) t2
+  USING(id);
+SELECT count(*) FROM
+  (SELECT unique2,count(*) id FROM tenk1 GROUP BY 1) t1
+    INNER JOIN
+  (SELECT unique2  id FROM tenk1) t2
+  USING(id);
+ABORT;
diff --git a/src/test/regress/sql/select_distinct.sql b/src/test/regress/sql/select_distinct.sql
index 1bfe59c26f..d90ef7bc27 100644
--- a/src/test/regress/sql/select_distinct.sql
+++ b/src/test/regress/sql/select_distinct.sql
@@ -174,3 +174,18 @@ SELECT 1 IS NOT DISTINCT FROM 2 as "no";
 SELECT 2 IS NOT DISTINCT FROM 2 as "yes";
 SELECT 2 IS NOT DISTINCT FROM null as "no";
 SELECT null IS NOT DISTINCT FROM null as "yes";
+
+-- parallel distinct
+BEGIN;
+SET redistribute_query_size = '128kB';
+SET min_parallel_table_scan_size =0;
+SET parallel_tuple_cost = 0;
+SET parallel_setup_cost = 0;
+SET enable_indexonlyscan = OFF;
+
+EXPLAIN (costs off)
+SELECT COUNT(*) FROM (SELECT DISTINCT unique2 FROM tenk1) foo;
+SELECT COUNT(*) FROM (SELECT DISTINCT unique2 FROM tenk1) foo;
+explain (costs off)
+SELECT DISTINCT * FROM (SELECT DISTINCT unique2 FROM tenk1) foo;
+ABORT;
\ No newline at end of file
diff --git a/src/test/regress/sql/select_parallel.sql b/src/test/regress/sql/select_parallel.sql
index f924731248..c7603ab02a 100644
--- a/src/test/regress/sql/select_parallel.sql
+++ b/src/test/regress/sql/select_parallel.sql
@@ -314,6 +314,25 @@ reset enable_material;
 
 reset enable_hashagg;
 
+--test rescan for redistribute
+SET redistribute_query_size = '128kB';
+SET enable_sort = off;
+
+explain (costs off)
+select * from
+  (select string4, count(unique2)
+   from tenk1 group by string4) ss
+  right join (values (1),(2),(3)) v(x) on true order by 3,1;
+
+select * from
+  (select string4, count(unique2)
+   from tenk1 group by string4) ss
+  right join (values (1),(2),(3)) v(x) on true order by 3,1;
+
+reset enable_sort;
+
+reset redistribute_query_size;
+
 -- check parallelized int8 aggregate (bug #14897)
 explain (costs off)
 select avg(unique1::int8) from tenk1;
diff --git a/src/test/regress/sql/union.sql b/src/test/regress/sql/union.sql
index ca8c9b4d12..acaf88759f 100644
--- a/src/test/regress/sql/union.sql
+++ b/src/test/regress/sql/union.sql
@@ -540,3 +540,35 @@ select * from
    union all
    select *, 1 as x from int8_tbl b) ss
 where (x = 0) or (q1 >= q2 and q1 <= q2);
+
+-- parallel union
+BEGIN;
+SET redistribute_query_size = '128kB';
+SET min_parallel_table_scan_size =0;
+SET parallel_tuple_cost = 0;
+SET parallel_setup_cost = 0;
+SET enable_indexonlyscan = OFF;
+
+-- using redistribute+sort sort
+SET enable_hashagg = OFF;
+EXPLAIN (costs off)
+SELECT count(*) FROM (SELECT unique2 FROM tenk1 UNION SELECT unique2 FROM tenk1) foo;
+SELECT count(*) FROM (SELECT unique2 FROM tenk1 UNION SELECT unique2 FROM tenk1) foo;
+EXPLAIN (costs off)
+SELECT count(*) FROM (SELECT unique2 FROM tenk1 UNION SELECT unique2 FROM tenk1) t1 INNER JOIN tenk1 USING(unique2);
+SELECT count(*) FROM (SELECT unique2 FROM tenk1 UNION SELECT unique2 FROM tenk1) t1 INNER JOIN tenk1 USING(unique2);
+reset enable_hashagg;
+
+-- using batch hash
+SET enable_sort = OFF;
+EXPLAIN (costs off)
+SELECT count(*) from (
+  SELECT hundred from tenk1
+    union
+  SELECT hundred from tenk1) foo;
+SELECT count(*) from (
+  SELECT hundred from tenk1
+    union
+  SELECT hundred from tenk1) foo;
+reset enable_sort;
+ABORT;
-- 
2.25.1

