From 4f71a8b1e0cf50d488af8c925151aed335bb2e8c Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <bharath.rupireddy@enterprisedb.com>
Date: Mon, 28 Dec 2020 17:00:38 +0530
Subject: [PATCH v17 4/4] Enable CTAS Parallel Inserts For Append

This patch allows pushing down the CTAS dest receiver even if there
exists Gather node under Top Append node. It also add the code for
influencing the planner to consider parallel tuple cost zero and
asserts for wrong enforcement if later parallel insertion is not
possible. Test cases are also included in this patch.
---
 src/backend/commands/createas.c              | 118 ++-
 src/backend/optimizer/path/allpaths.c        |  38 +
 src/backend/optimizer/plan/planner.c         |  10 +-
 src/include/commands/createas.h              |   4 +-
 src/test/regress/expected/write_parallel.out | 722 +++++++++++++++++++
 src/test/regress/sql/write_parallel.sql      | 222 ++++++
 6 files changed, 1076 insertions(+), 38 deletions(-)

diff --git a/src/backend/commands/createas.c b/src/backend/commands/createas.c
index 5046e02aac..148e7c7ea2 100644
--- a/src/backend/commands/createas.c
+++ b/src/backend/commands/createas.c
@@ -690,35 +690,99 @@ IsParallelInsertionAllowedInCTAS(IntoClause *into)
 	return true;
 }
 
+/*
+ * PushDownCTASParallelInsertState --- push the dest receiver down to the
+ * Gather nodes.
+ *
+ * In this function we only care about Append and Gather nodes.
+ *
+ * Push the dest receiver to Gather node when it is either at the top of the
+ * plan or under top Append node unless it does not have any projections to do.
+ * Required information from the pushed dest receiver is sent to workers so
+ * that they can perform parallel insertions into the target table.
+ *
+ * If the top node is Append, then this function recursively checks the sub
+ * plans for Gather nodes, when found one(and if it does not have projections),
+ * then sets the dest receiver information.
+ *
+ * In any case, this function returns true if at least one Gather node can allow
+ * parallel insertions by the workers. Otherwise returns false.
+ */
+static bool
+PushDownCTASParallelInsertState(DestReceiver *dest, PlanState *ps,
+								bool *gather_exists)
+{
+	bool parallel = false;
+
+	if (ps == NULL)
+		return parallel;
+
+	if (IsA(ps, AppendState))
+	{
+		AppendState *aps = (AppendState *) ps;
+
+		for (int i = 0; i < aps->as_nplans; i++)
+		{
+			parallel |= PushDownCTASParallelInsertState(dest,
+														aps->appendplans[i],
+														gather_exists);
+		}
+	}
+	else if (IsA(ps, GatherState))
+	{
+		/*
+		 * Set to true if there exists at least one Gather node either at the
+		 * top of the plan or as a direct sub node under Append node.
+		 */
+		 *gather_exists |= true;
+
+		if (!ps->ps_ProjInfo)
+		{
+			GatherState *gstate = (GatherState *) ps;
+
+			parallel = true;
+
+			/* Okay to parallelize inserts, so mark it. */
+			((DR_intorel *) dest)->is_parallel = true;
+
+			/*
+			 * For parallelizing inserts in CTAS we must send information such
+			 * as into clause (to build separate dest receiver), object id (to
+			 * open the created table) to each workers. Since this information
+			 * is available in the CTAS dest receiver, store a reference to it
+			 * in the Gather state so that it will be used in
+			 * ExecInitParallelPlan to pick the required information.
+			 */
+			gstate->dest = dest;
+		}
+	}
+
+	return parallel;
+}
+
 /*
  * SetCTASParallelInsertState --- set the parallel insert state
  *
- * See if the upper node is Gather and it doesn't have any projections, then
- * set the parallel insert state such as a flag in the dest receiver and also
- * store the dest receiver reference in the Gather node so that the required
- * information will be sent to workers.
+ * See if the upper node is either Gather or Append but it does have Gather
+ * nodes under it, then set the parallel insert state in respective Gather
+ * nodes if they do not have any projections. The parallel insert state
+ * includes a flag in the dest receiver and also a dest receiver reference in
+ * the Gather node so that the required information will be sent to workers.
  */
 void
 SetCTASParallelInsertState(QueryDesc *queryDesc, uint8 *tuple_cost_flags)
 {
-	GatherState *gstate;
-	DestReceiver *dest;
+	bool allow = false;
+	bool gather_exists = false;
 
 	Assert(queryDesc);
 
-	gstate = (GatherState *) queryDesc->planstate;
-	dest	= queryDesc->dest;
-
+	allow = PushDownCTASParallelInsertState(queryDesc->dest,
+											queryDesc->planstate,
+											&gather_exists);
 	/*
-	 * Parallel insertions are not possible if the upper node is not Gather.
-	 */
-	if (!IsA(gstate, GatherState))
-		return;
-
-	/*
-	 * If the upper Gather node has some projections to perform, then we can
-	 * not allow parallel insertions. But before returning, ensure that we have
-	 * not done wrong parallel tuple cost enforcement in the planner.
+	 * Ensure that we have not done wrong parallel tuple cost enforcement in
+	 * the planner.
 	 *
 	 * The main reason for this assertion is to check if we enforced planner to
 	 * ignore the parallel tuple cost (with the intention of choosing parallel
@@ -730,25 +794,9 @@ SetCTASParallelInsertState(QueryDesc *queryDesc, uint8 *tuple_cost_flags)
 	 * case it occurs, that means the planner may have chosen this parallel
 	 * plan because of our wrong enforcement. So let's catch that here.
 	 */
-	if (tuple_cost_flags && gstate->ps.ps_ProjInfo)
+	if (!allow && tuple_cost_flags && gather_exists)
 		Assert(!(*tuple_cost_flags & CTAS_PARALLEL_INS_TUP_COST_IGNORED));
 
-	if (!gstate->ps.ps_ProjInfo)
-	{
-		/* Okay to parallelize inserts, so mark it. */
-		((DR_intorel *) dest)->is_parallel = true;
-
-		/*
-		 * For parallelizing inserts in CTAS we must send information such as
-		 * into clause (to build separate dest receiver), object id (to open
-		 * the created table) to each workers. Since this information is
-		 * available in the CTAS dest receiver, store a reference to it in the
-		 * Gather state so that it will be used in ExecInitParallelPlan to pick
-		 * the required information.
-		 */
-		gstate->dest = dest;
-	}
-
 	if (tuple_cost_flags)
 		*tuple_cost_flags = CTAS_PARALLEL_INS_UNDEF;
 }
diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index 627d08b78a..b0835c32bd 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -23,6 +23,7 @@
 #include "catalog/pg_class.h"
 #include "catalog/pg_operator.h"
 #include "catalog/pg_proc.h"
+#include "commands/createas.h"
 #include "foreign/fdwapi.h"
 #include "miscadmin.h"
 #include "nodes/makefuncs.h"
@@ -1103,11 +1104,48 @@ set_append_rel_size(PlannerInfo *root, RelOptInfo *rel,
 		if (root->glob->parallelModeOK && rel->consider_parallel)
 			set_rel_consider_parallel(root, childrel, childRTE);
 
+		/*
+		 * When subplan is subquery, It's possible to do parallel insert if top
+		 * node of subquery is Gather, so we turn on a flag to ignore parallel
+		 * tuple cost in cost_gather if the SELECT is for CTAS.
+		 */
+		if (childrel->rtekind == RTE_SUBQUERY)
+		{
+			/*
+			 * When there is no parent path generating clause(such as limit,
+			 * sort, distinct...), we can turn on the flag for two cases:
+			 * i) query_level is 1
+			 * ii) query_level > 1 then turn on the flag in the parent_root.
+			 * The case ii) is to check append under append:
+			 * Append
+			 *	->Append
+			 *	->Gather
+			 *	->Other plan
+			 */
+			if (root->parse->CTASParallelInsInfo &
+				CTAS_PARALLEL_INS_SELECT &&
+				(root->query_level == 1 ||
+				 root->parent_root->parse->CTASParallelInsInfo &
+				 CTAS_PARALLEL_INS_IGN_TUP_COST_APPEND) &&
+				!(HAS_PARENT_PATH_GENERATING_CLAUSE(root)))
+			{
+				root->parse->CTASParallelInsInfo |=
+										CTAS_PARALLEL_INS_IGN_TUP_COST_APPEND;
+			}
+		}
+
 		/*
 		 * Compute the child's size.
 		 */
 		set_rel_size(root, childrel, childRTindex, childRTE);
 
+		if (root->parse->CTASParallelInsInfo &
+			CTAS_PARALLEL_INS_IGN_TUP_COST_APPEND)
+		{
+			root->parse->CTASParallelInsInfo &=
+										~CTAS_PARALLEL_INS_IGN_TUP_COST_APPEND;
+		}
+
 		/*
 		 * It is possible that constraint exclusion detected a contradiction
 		 * within a child subquery, even though we didn't prove one above. If
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index 20a18628a7..5e607f598a 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -7350,8 +7350,14 @@ can_partial_agg(PlannerInfo *root)
 static bool
 ignore_parallel_tuple_cost(PlannerInfo *root)
 {
-	if (root->query_level == 1 &&
-		(root->parse->CTASParallelInsInfo & CTAS_PARALLEL_INS_SELECT))
+	if (root->query_level != 1 &&
+		(root->parent_root->parse->CTASParallelInsInfo &
+		 CTAS_PARALLEL_INS_IGN_TUP_COST_APPEND))
+	{
+		root->parse->CTASParallelInsInfo |= CTAS_PARALLEL_INS_SELECT;
+	}
+
+	if (root->parse->CTASParallelInsInfo & CTAS_PARALLEL_INS_SELECT)
 	{
 		/*
 		 * In each of the HAS_PARENT_PATH_GENERATING_CLAUSE cases, a parent
diff --git a/src/include/commands/createas.h b/src/include/commands/createas.h
index 140226ad5a..bf52b05ea0 100644
--- a/src/include/commands/createas.h
+++ b/src/include/commands/createas.h
@@ -54,7 +54,9 @@ typedef enum CTASParallelInsertOpt
 	 */
 	CTAS_PARALLEL_INS_TUP_COST_CAN_IGN = 1 << 1,
 	/* Turn on this after the cost is ignored. */
-	CTAS_PARALLEL_INS_TUP_COST_IGNORED = 1 << 2
+	CTAS_PARALLEL_INS_TUP_COST_IGNORED = 1 << 2,
+	/* Turn on this in case tuple cost needs to be ignored for Append cases. */
+	CTAS_PARALLEL_INS_IGN_TUP_COST_APPEND = 1 << 3
 } CTASParallelInsertOpt;
 
 #define IS_CTAS(intoclause) (intoclause && IsA(intoclause, IntoClause))
diff --git a/src/test/regress/expected/write_parallel.out b/src/test/regress/expected/write_parallel.out
index 38a18c5a9b..356a2d0002 100644
--- a/src/test/regress/expected/write_parallel.out
+++ b/src/test/regress/expected/write_parallel.out
@@ -631,6 +631,728 @@ drop table parallel_write;
 reset enable_nestloop;
 reset enable_mergejoin;
 reset enable_hashjoin;
+-- test cases for performing parallel inserts when Append node is at the top
+-- and Gather node is in one of its direct sub plans.
+-- case 1: parallel inserts must occur at each Gather node as we can push the
+-- CTAS dest receiver.
+-- Append
+-- 	->Gather
+-- 		->Parallel Seq Scan
+-- 	->Gather
+-- 		->Parallel Seq Scan
+-- 	->Gather
+-- 		->Parallel Seq Scan
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 where col1 = 5 union all
+    select * from temp2 where col2 = 5 union all
+    select * from temp2 where col2 = 5;');
+                            explain_pictas                            
+----------------------------------------------------------------------
+ Append (actual rows=N loops=N)
+   ->  Gather (actual rows=N loops=N)
+         Workers Planned: 3
+         Workers Launched: N
+       ->  Create parallel_write
+       ->  Parallel Seq Scan on temp1 (actual rows=N loops=N)
+               Filter: (col1 = 5)
+               Rows Removed by Filter: N
+   ->  Gather (actual rows=N loops=N)
+         Workers Planned: 3
+         Workers Launched: N
+       ->  Create parallel_write
+       ->  Parallel Seq Scan on temp2 (actual rows=N loops=N)
+               Filter: (col2 = 5)
+               Rows Removed by Filter: N
+   ->  Gather (actual rows=N loops=N)
+         Workers Planned: 3
+         Workers Launched: N
+       ->  Create parallel_write
+       ->  Parallel Seq Scan on temp2 temp2_1 (actual rows=N loops=N)
+               Filter: (col2 = 5)
+               Rows Removed by Filter: N
+(22 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+     3
+(1 row)
+
+drop table parallel_write;
+-- case 2: parallel inserts must occur at the top Gather node as we can push
+-- the CTAS dest receiver to it.
+-- Gather
+-- 	->Parallel Append
+-- 		->Parallel Seq Scan
+-- 		->Parallel Seq Scan
+-- 		->Parallel Seq Scan
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 union all
+    select * from temp2 union all
+    select * from temp2;');
+                             explain_pictas                             
+------------------------------------------------------------------------
+ Gather (actual rows=N loops=N)
+   Workers Planned: 3
+   Workers Launched: N
+ ->  Create parallel_write
+   ->  Parallel Append (actual rows=N loops=N)
+         ->  Parallel Seq Scan on temp1 (actual rows=N loops=N)
+         ->  Parallel Seq Scan on temp2 (actual rows=N loops=N)
+         ->  Parallel Seq Scan on temp2 temp2_1 (actual rows=N loops=N)
+(8 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+    15
+(1 row)
+
+drop table parallel_write;
+select explain_pictas(
+'create table parallel_write as
+    select (select col2 from temp2 limit 1) col2 from temp1 union all
+    select (select col2 from temp2 limit 1) col2 from temp1 union all
+    select * from temp2;');
+                                 explain_pictas                                 
+--------------------------------------------------------------------------------
+ Gather (actual rows=N loops=N)
+   Workers Planned: 3
+   Params Evaluated: $1, $3
+   Workers Launched: N
+ ->  Create parallel_write
+   InitPlan 1 (returns $1)
+     ->  Limit (actual rows=N loops=N)
+           ->  Gather (actual rows=N loops=N)
+                 Workers Planned: 3
+                 Workers Launched: N
+                 ->  Parallel Seq Scan on temp2 temp2_1 (actual rows=N loops=N)
+   InitPlan 2 (returns $3)
+     ->  Limit (actual rows=N loops=N)
+           ->  Gather (actual rows=N loops=N)
+                 Workers Planned: 3
+                 Workers Launched: N
+                 ->  Parallel Seq Scan on temp2 temp2_2 (actual rows=N loops=N)
+   ->  Parallel Append (actual rows=N loops=N)
+         ->  Parallel Seq Scan on temp1 (actual rows=N loops=N)
+         ->  Parallel Seq Scan on temp1 temp1_1 (actual rows=N loops=N)
+         ->  Parallel Seq Scan on temp2 (actual rows=N loops=N)
+(21 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+    15
+(1 row)
+
+drop table parallel_write;
+-- case 3: parallel inserts must occur at each Gather node as we can push the
+-- CTAS dest receiver. Non-Gather nodes will do inserts by sending tuples to
+-- Append and from there to CTAS dest receiver.
+-- Append
+-- 	->Gather
+-- 		->Parallel Seq Scan
+-- 	->Seq Scan / Join / any other non-Gather node
+-- 	->Gather
+-- 		->Parallel Seq Scan
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 where col1 = 5 union all
+    select (select temp1.col1 from temp2 limit 1) col2 from temp1 union all
+    select * from temp1 where col1 = 5;');
+                                explain_pictas                                
+------------------------------------------------------------------------------
+ Append (actual rows=N loops=N)
+   ->  Gather (actual rows=N loops=N)
+         Workers Planned: 3
+         Workers Launched: N
+       ->  Create parallel_write
+       ->  Parallel Seq Scan on temp1 temp1_1 (actual rows=N loops=N)
+               Filter: (col1 = 5)
+               Rows Removed by Filter: N
+   ->  Seq Scan on temp1 (actual rows=N loops=N)
+         SubPlan 1
+           ->  Limit (actual rows=N loops=N)
+                 ->  Gather (actual rows=N loops=N)
+                       Workers Planned: 3
+                       Workers Launched: N
+                       ->  Parallel Seq Scan on temp2 (actual rows=N loops=N)
+   ->  Gather (actual rows=N loops=N)
+         Workers Planned: 3
+         Workers Launched: N
+       ->  Create parallel_write
+       ->  Parallel Seq Scan on temp1 temp1_2 (actual rows=N loops=N)
+               Filter: (col1 = 5)
+               Rows Removed by Filter: N
+(22 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+     7
+(1 row)
+
+drop table parallel_write;
+alter table temp2 set (parallel_workers = 0);
+select explain_pictas(
+'create table parallel_write as select * from temp1 where col1 = (select 1) union all
+    select * from temp1 where col1 = (select 1) union all
+ select * from temp2 where col2 = (select 2);');
+                             explain_pictas                             
+------------------------------------------------------------------------
+ Append (actual rows=N loops=N)
+   ->  Gather (actual rows=N loops=N)
+         Workers Planned: 3
+         Params Evaluated: $0
+         Workers Launched: N
+       ->  Create parallel_write
+       InitPlan 1 (returns $0)
+           ->  Result (actual rows=N loops=N)
+         ->  Parallel Seq Scan on temp1 (actual rows=N loops=N)
+               Filter: (col1 = $0)
+               Rows Removed by Filter: N
+   ->  Gather (actual rows=N loops=N)
+         Workers Planned: 3
+         Params Evaluated: $1
+         Workers Launched: N
+       ->  Create parallel_write
+       InitPlan 2 (returns $1)
+           ->  Result (actual rows=N loops=N)
+         ->  Parallel Seq Scan on temp1 temp1_1 (actual rows=N loops=N)
+               Filter: (col1 = $1)
+               Rows Removed by Filter: N
+   ->  Seq Scan on temp2 (actual rows=N loops=N)
+         Filter: (col2 = $2)
+         Rows Removed by Filter: N
+         InitPlan 3 (returns $2)
+           ->  Result (actual rows=N loops=N)
+(26 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+     3
+(1 row)
+
+alter table temp2 reset (parallel_workers);
+drop table parallel_write;
+-- case 4: parallel inserts must not occur as there will be no direct Gather
+-- node under Append node. Non-Gather nodes will do inserts by sending tuples
+-- to Append and from there to CTAS dest receiver.
+-- Append
+-- 	->Seq Scan / Join / any other non-Gather node
+-- 	->Seq Scan / Join / any other non-Gather node
+-- 	->Seq Scan / Join / any other non-Gather node
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 union all
+    select * from temp2 union all
+    select (select temp1.col1 from temp2 limit 1) col2 from temp1;');
+                                    explain_pictas                                    
+--------------------------------------------------------------------------------------
+ Append (actual rows=N loops=N)
+   ->  Seq Scan on temp1 (actual rows=N loops=N)
+   ->  Seq Scan on temp2 (actual rows=N loops=N)
+   ->  Seq Scan on temp1 temp1_1 (actual rows=N loops=N)
+         SubPlan 1
+           ->  Limit (actual rows=N loops=N)
+                 ->  Gather (actual rows=N loops=N)
+                       Workers Planned: 3
+                       Workers Launched: N
+                       ->  Parallel Seq Scan on temp2 temp2_1 (actual rows=N loops=N)
+(10 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+    15
+(1 row)
+
+drop table parallel_write;
+-- case 5: parallel inserts must occur at the top Gather node as we can push
+-- the CTAS dest receiver to it.
+-- Gather
+-- 	->Parallel Append
+-- 		->Seq Scan / Join / any other non-Gather node
+-- 		->Parallel Seq Scan
+-- 		->Parallel Seq Scan
+alter table temp2 set (parallel_workers = 0);
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 union all
+    select * from temp2 union all
+    select * from temp1;');
+                             explain_pictas                             
+------------------------------------------------------------------------
+ Gather (actual rows=N loops=N)
+   Workers Planned: 3
+   Workers Launched: N
+ ->  Create parallel_write
+   ->  Parallel Append (actual rows=N loops=N)
+         ->  Seq Scan on temp2 (actual rows=N loops=N)
+         ->  Parallel Seq Scan on temp1 (actual rows=N loops=N)
+         ->  Parallel Seq Scan on temp1 temp1_1 (actual rows=N loops=N)
+(8 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+    15
+(1 row)
+
+drop table parallel_write;
+alter table temp2 reset (parallel_workers);
+-- case 6: parallel inserts must occur at each Gather node as we can push the
+-- CTAS dest receiver.
+-- Append
+--  ->Append
+--     ->Gather
+-- 		->Parallel Seq Scan
+--  ->Append
+--     ->Gather
+-- 		->Parallel Seq Scan
+--  ->Gather
+--  ->Gather
+select explain_pictas(
+'create table parallel_write as
+    select * from (select * from temp1 where col1 = (select 1) union all
+            select * from temp2 where col2 = (select 2)) as tt
+        where col1 = (select 1) union all
+    select * from temp2 where col2 = (select 2) union all
+    select * from temp2 where col2 = (select 2);');
+                               explain_pictas                               
+----------------------------------------------------------------------------
+ Append (actual rows=N loops=N)
+   ->  Append (actual rows=N loops=N)
+         InitPlan 1 (returns $0)
+           ->  Result (actual rows=N loops=N)
+         ->  Gather (actual rows=N loops=N)
+               Workers Planned: 3
+               Params Evaluated: $0, $1
+               Workers Launched: N
+             ->  Create parallel_write
+             InitPlan 2 (returns $1)
+                 ->  Result (actual rows=N loops=N)
+               ->  Result (actual rows=N loops=N)
+                     One-Time Filter: ($1 = $0)
+                     ->  Parallel Seq Scan on temp1 (actual rows=N loops=N)
+                           Filter: (col1 = $0)
+                           Rows Removed by Filter: N
+         ->  Gather (actual rows=N loops=N)
+               Workers Planned: 3
+               Params Evaluated: $0, $2
+               Workers Launched: N
+             ->  Create parallel_write
+             InitPlan 3 (returns $2)
+                 ->  Result (actual rows=N loops=N)
+               ->  Result (actual rows=N loops=N)
+                     One-Time Filter: ($2 = $0)
+                     ->  Parallel Seq Scan on temp2 (never executed)
+                           Filter: (col2 = $0)
+   ->  Gather (actual rows=N loops=N)
+         Workers Planned: 3
+         Params Evaluated: $3
+         Workers Launched: N
+       ->  Create parallel_write
+       InitPlan 4 (returns $3)
+           ->  Result (actual rows=N loops=N)
+         ->  Parallel Seq Scan on temp2 temp2_1 (actual rows=N loops=N)
+               Filter: (col2 = $3)
+               Rows Removed by Filter: N
+   ->  Gather (actual rows=N loops=N)
+         Workers Planned: 3
+         Params Evaluated: $4
+         Workers Launched: N
+       ->  Create parallel_write
+       InitPlan 5 (returns $4)
+           ->  Result (actual rows=N loops=N)
+         ->  Parallel Seq Scan on temp2 temp2_2 (actual rows=N loops=N)
+               Filter: (col2 = $4)
+               Rows Removed by Filter: N
+(47 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+     3
+(1 row)
+
+drop table parallel_write;
+select explain_pictas(
+'create table parallel_write as
+    select * from (select * from temp1 where col1 = (select 1) union all
+            select * from temp2 where col2 = (select 2)) as tt
+        where col1 = (select 1) union all
+    select * from (select * from temp1 where col1 = (select 1) union all
+            select * from temp2 where col2 = (select 2)) as tt
+        where col1 = (select 1) union all
+    select * from temp1 where col1 = 5 union all
+    select * from temp2 where col2 = 5;');
+                                   explain_pictas                                   
+------------------------------------------------------------------------------------
+ Append (actual rows=N loops=N)
+   ->  Append (actual rows=N loops=N)
+         InitPlan 1 (returns $0)
+           ->  Result (actual rows=N loops=N)
+         ->  Gather (actual rows=N loops=N)
+               Workers Planned: 3
+               Params Evaluated: $0, $1
+               Workers Launched: N
+             ->  Create parallel_write
+             InitPlan 2 (returns $1)
+                 ->  Result (actual rows=N loops=N)
+               ->  Result (actual rows=N loops=N)
+                     One-Time Filter: ($1 = $0)
+                     ->  Parallel Seq Scan on temp1 (actual rows=N loops=N)
+                           Filter: (col1 = $0)
+                           Rows Removed by Filter: N
+         ->  Gather (actual rows=N loops=N)
+               Workers Planned: 3
+               Params Evaluated: $0, $2
+               Workers Launched: N
+             ->  Create parallel_write
+             InitPlan 3 (returns $2)
+                 ->  Result (actual rows=N loops=N)
+               ->  Result (actual rows=N loops=N)
+                     One-Time Filter: ($2 = $0)
+                     ->  Parallel Seq Scan on temp2 (never executed)
+                           Filter: (col2 = $0)
+   ->  Append (actual rows=N loops=N)
+         InitPlan 4 (returns $3)
+           ->  Result (actual rows=N loops=N)
+         ->  Gather (actual rows=N loops=N)
+               Workers Planned: 3
+               Params Evaluated: $3, $4
+               Workers Launched: N
+             ->  Create parallel_write
+             InitPlan 5 (returns $4)
+                 ->  Result (actual rows=N loops=N)
+               ->  Result (actual rows=N loops=N)
+                     One-Time Filter: ($4 = $3)
+                     ->  Parallel Seq Scan on temp1 temp1_1 (actual rows=N loops=N)
+                           Filter: (col1 = $3)
+                           Rows Removed by Filter: N
+         ->  Gather (actual rows=N loops=N)
+               Workers Planned: 3
+               Params Evaluated: $3, $5
+               Workers Launched: N
+             ->  Create parallel_write
+             InitPlan 6 (returns $5)
+                 ->  Result (actual rows=N loops=N)
+               ->  Result (actual rows=N loops=N)
+                     One-Time Filter: ($5 = $3)
+                     ->  Parallel Seq Scan on temp2 temp2_1 (never executed)
+                           Filter: (col2 = $3)
+   ->  Gather (actual rows=N loops=N)
+         Workers Planned: 3
+         Workers Launched: N
+       ->  Create parallel_write
+       ->  Parallel Seq Scan on temp1 temp1_2 (actual rows=N loops=N)
+               Filter: (col1 = 5)
+               Rows Removed by Filter: N
+   ->  Gather (actual rows=N loops=N)
+         Workers Planned: 3
+         Workers Launched: N
+       ->  Create parallel_write
+       ->  Parallel Seq Scan on temp2 temp2_2 (actual rows=N loops=N)
+               Filter: (col2 = 5)
+               Rows Removed by Filter: N
+(67 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+     4
+(1 row)
+
+drop table parallel_write;
+-- case 7: parallel inserts must occur at each Gather node as we can push the
+-- CTAS dest receiver. Non-Gather nodes will do inserts by sending tuples
+-- to Append and from there to CTAS dest receiver.
+-- Append
+--  ->Append
+--     ->Gather
+-- 		->Parallel Seq Scan
+--  ->Append
+--     ->Gather
+-- 		->Parallel Seq Scan
+--  ->Seq Scan / Join / any other non-Gather node
+--  ->Gather
+alter table temp2 set (parallel_workers = 0);
+select explain_pictas(
+'create table parallel_write as
+    select * from (select * from temp1 where col1 = (select 1) union all
+            select * from temp2 where col2 = (select 2)) as tt
+        where col1 = (select 1) union all
+    select * from temp2 where col2 = (select 2) union all
+    select * from temp1 where col1 = (select 2);');
+                               explain_pictas                               
+----------------------------------------------------------------------------
+ Append (actual rows=N loops=N)
+   ->  Append (actual rows=N loops=N)
+         InitPlan 1 (returns $0)
+           ->  Result (actual rows=N loops=N)
+         ->  Gather (actual rows=N loops=N)
+               Workers Planned: 3
+               Params Evaluated: $0, $1
+               Workers Launched: N
+             ->  Create parallel_write
+             InitPlan 2 (returns $1)
+                 ->  Result (actual rows=N loops=N)
+               ->  Result (actual rows=N loops=N)
+                     One-Time Filter: ($1 = $0)
+                     ->  Parallel Seq Scan on temp1 (actual rows=N loops=N)
+                           Filter: (col1 = $0)
+                           Rows Removed by Filter: N
+         ->  Result (actual rows=N loops=N)
+               One-Time Filter: ($2 = $0)
+               InitPlan 3 (returns $2)
+                 ->  Result (actual rows=N loops=N)
+               ->  Seq Scan on temp2 (never executed)
+                     Filter: (col2 = $0)
+   ->  Seq Scan on temp2 temp2_1 (actual rows=N loops=N)
+         Filter: (col2 = $3)
+         Rows Removed by Filter: N
+         InitPlan 4 (returns $3)
+           ->  Result (actual rows=N loops=N)
+   ->  Gather (actual rows=N loops=N)
+         Workers Planned: 3
+         Params Evaluated: $4
+         Workers Launched: N
+       ->  Create parallel_write
+       InitPlan 5 (returns $4)
+           ->  Result (actual rows=N loops=N)
+         ->  Parallel Seq Scan on temp1 temp1_1 (actual rows=N loops=N)
+               Filter: (col1 = $4)
+               Rows Removed by Filter: N
+(37 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+     3
+(1 row)
+
+drop table parallel_write;
+select explain_pictas(
+'create table parallel_write as
+    select * from (select * from temp1 where col1 = (select 1) union all
+            select * from temp2 where col2 = (select 2)) as tt
+        where col1 = (select 1) union all
+    select * from (select * from temp1 where col1 = (select 1) union all
+            select * from temp2 where col2 = (select 2)) as tt
+        where col1 = (select 1) union all
+    select * from temp1 where col1 = 5 union all
+    select * from temp2 where col2 = 5;');
+                                   explain_pictas                                   
+------------------------------------------------------------------------------------
+ Append (actual rows=N loops=N)
+   ->  Append (actual rows=N loops=N)
+         InitPlan 1 (returns $0)
+           ->  Result (actual rows=N loops=N)
+         ->  Gather (actual rows=N loops=N)
+               Workers Planned: 3
+               Params Evaluated: $0, $1
+               Workers Launched: N
+             ->  Create parallel_write
+             InitPlan 2 (returns $1)
+                 ->  Result (actual rows=N loops=N)
+               ->  Result (actual rows=N loops=N)
+                     One-Time Filter: ($1 = $0)
+                     ->  Parallel Seq Scan on temp1 (actual rows=N loops=N)
+                           Filter: (col1 = $0)
+                           Rows Removed by Filter: N
+         ->  Result (actual rows=N loops=N)
+               One-Time Filter: ($2 = $0)
+               InitPlan 3 (returns $2)
+                 ->  Result (actual rows=N loops=N)
+               ->  Seq Scan on temp2 (never executed)
+                     Filter: (col2 = $0)
+   ->  Append (actual rows=N loops=N)
+         InitPlan 4 (returns $3)
+           ->  Result (actual rows=N loops=N)
+         ->  Gather (actual rows=N loops=N)
+               Workers Planned: 3
+               Params Evaluated: $3, $4
+               Workers Launched: N
+             ->  Create parallel_write
+             InitPlan 5 (returns $4)
+                 ->  Result (actual rows=N loops=N)
+               ->  Result (actual rows=N loops=N)
+                     One-Time Filter: ($4 = $3)
+                     ->  Parallel Seq Scan on temp1 temp1_1 (actual rows=N loops=N)
+                           Filter: (col1 = $3)
+                           Rows Removed by Filter: N
+         ->  Result (actual rows=N loops=N)
+               One-Time Filter: ($5 = $3)
+               InitPlan 6 (returns $5)
+                 ->  Result (actual rows=N loops=N)
+               ->  Seq Scan on temp2 temp2_1 (never executed)
+                     Filter: (col2 = $3)
+   ->  Gather (actual rows=N loops=N)
+         Workers Planned: 3
+         Workers Launched: N
+       ->  Create parallel_write
+       ->  Parallel Seq Scan on temp1 temp1_2 (actual rows=N loops=N)
+               Filter: (col1 = 5)
+               Rows Removed by Filter: N
+   ->  Seq Scan on temp2 temp2_2 (actual rows=N loops=N)
+         Filter: (col2 = 5)
+         Rows Removed by Filter: N
+(53 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+     4
+(1 row)
+
+drop table parallel_write;
+alter table temp2 reset (parallel_workers);
+-- case 8: parallel inserts must not occur because there is no Gather or Append
+-- node at the top for union, except/except all, intersect/intersect all
+-- cases.
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 union
+    select * from temp2;');
+                            explain_pictas                            
+----------------------------------------------------------------------
+ HashAggregate (actual rows=N loops=N)
+   Group Key: temp1.col1
+   Batches: 1  Memory Usage: 217kB
+   ->  Append (actual rows=N loops=N)
+         ->  Gather (actual rows=N loops=N)
+               Workers Planned: 3
+               Workers Launched: N
+               ->  Parallel Seq Scan on temp1 (actual rows=N loops=N)
+         ->  Gather (actual rows=N loops=N)
+               Workers Planned: 3
+               Workers Launched: N
+               ->  Parallel Seq Scan on temp2 (actual rows=N loops=N)
+(12 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+     5
+(1 row)
+
+drop table parallel_write;
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 except
+    select * from temp2 where col2 < 3;');
+                               explain_pictas                               
+----------------------------------------------------------------------------
+ HashSetOp Except (actual rows=N loops=N)
+   ->  Append (actual rows=N loops=N)
+         ->  Subquery Scan on "*SELECT* 1" (actual rows=N loops=N)
+               ->  Gather (actual rows=N loops=N)
+                     Workers Planned: 3
+                     Workers Launched: N
+                     ->  Parallel Seq Scan on temp1 (actual rows=N loops=N)
+         ->  Subquery Scan on "*SELECT* 2" (actual rows=N loops=N)
+               ->  Gather (actual rows=N loops=N)
+                     Workers Planned: 3
+                     Workers Launched: N
+                     ->  Parallel Seq Scan on temp2 (actual rows=N loops=N)
+                           Filter: (col2 < 3)
+                           Rows Removed by Filter: N
+(14 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+     3
+(1 row)
+
+drop table parallel_write;
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 except all
+    select * from temp2 where col2 < 3;');
+                               explain_pictas                               
+----------------------------------------------------------------------------
+ HashSetOp Except All (actual rows=N loops=N)
+   ->  Append (actual rows=N loops=N)
+         ->  Subquery Scan on "*SELECT* 1" (actual rows=N loops=N)
+               ->  Gather (actual rows=N loops=N)
+                     Workers Planned: 3
+                     Workers Launched: N
+                     ->  Parallel Seq Scan on temp1 (actual rows=N loops=N)
+         ->  Subquery Scan on "*SELECT* 2" (actual rows=N loops=N)
+               ->  Gather (actual rows=N loops=N)
+                     Workers Planned: 3
+                     Workers Launched: N
+                     ->  Parallel Seq Scan on temp2 (actual rows=N loops=N)
+                           Filter: (col2 < 3)
+                           Rows Removed by Filter: N
+(14 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+     3
+(1 row)
+
+drop table parallel_write;
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 intersect
+    select * from temp2;');
+                               explain_pictas                               
+----------------------------------------------------------------------------
+ HashSetOp Intersect (actual rows=N loops=N)
+   ->  Append (actual rows=N loops=N)
+         ->  Subquery Scan on "*SELECT* 1" (actual rows=N loops=N)
+               ->  Gather (actual rows=N loops=N)
+                     Workers Planned: 3
+                     Workers Launched: N
+                     ->  Parallel Seq Scan on temp1 (actual rows=N loops=N)
+         ->  Subquery Scan on "*SELECT* 2" (actual rows=N loops=N)
+               ->  Gather (actual rows=N loops=N)
+                     Workers Planned: 3
+                     Workers Launched: N
+                     ->  Parallel Seq Scan on temp2 (actual rows=N loops=N)
+(12 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+     5
+(1 row)
+
+drop table parallel_write;
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 intersect all
+    select * from temp2;');
+                               explain_pictas                               
+----------------------------------------------------------------------------
+ HashSetOp Intersect All (actual rows=N loops=N)
+   ->  Append (actual rows=N loops=N)
+         ->  Subquery Scan on "*SELECT* 1" (actual rows=N loops=N)
+               ->  Gather (actual rows=N loops=N)
+                     Workers Planned: 3
+                     Workers Launched: N
+                     ->  Parallel Seq Scan on temp1 (actual rows=N loops=N)
+         ->  Subquery Scan on "*SELECT* 2" (actual rows=N loops=N)
+               ->  Gather (actual rows=N loops=N)
+                     Workers Planned: 3
+                     Workers Launched: N
+                     ->  Parallel Seq Scan on temp2 (actual rows=N loops=N)
+(12 rows)
+
+select count(*) from parallel_write;
+ count 
+-------
+     5
+(1 row)
+
+drop table parallel_write;
 drop table temp1;
 drop table temp2;
 drop table temp3;
diff --git a/src/test/regress/sql/write_parallel.sql b/src/test/regress/sql/write_parallel.sql
index 40aadafc2a..32e6ad8636 100644
--- a/src/test/regress/sql/write_parallel.sql
+++ b/src/test/regress/sql/write_parallel.sql
@@ -246,6 +246,228 @@ reset enable_nestloop;
 reset enable_mergejoin;
 reset enable_hashjoin;
 
+-- test cases for performing parallel inserts when Append node is at the top
+-- and Gather node is in one of its direct sub plans.
+
+-- case 1: parallel inserts must occur at each Gather node as we can push the
+-- CTAS dest receiver.
+-- Append
+-- 	->Gather
+-- 		->Parallel Seq Scan
+-- 	->Gather
+-- 		->Parallel Seq Scan
+-- 	->Gather
+-- 		->Parallel Seq Scan
+
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 where col1 = 5 union all
+    select * from temp2 where col2 = 5 union all
+    select * from temp2 where col2 = 5;');
+select count(*) from parallel_write;
+drop table parallel_write;
+
+-- case 2: parallel inserts must occur at the top Gather node as we can push
+-- the CTAS dest receiver to it.
+-- Gather
+-- 	->Parallel Append
+-- 		->Parallel Seq Scan
+-- 		->Parallel Seq Scan
+-- 		->Parallel Seq Scan
+
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 union all
+    select * from temp2 union all
+    select * from temp2;');
+select count(*) from parallel_write;
+drop table parallel_write;
+
+select explain_pictas(
+'create table parallel_write as
+    select (select col2 from temp2 limit 1) col2 from temp1 union all
+    select (select col2 from temp2 limit 1) col2 from temp1 union all
+    select * from temp2;');
+select count(*) from parallel_write;
+drop table parallel_write;
+
+-- case 3: parallel inserts must occur at each Gather node as we can push the
+-- CTAS dest receiver. Non-Gather nodes will do inserts by sending tuples to
+-- Append and from there to CTAS dest receiver.
+-- Append
+-- 	->Gather
+-- 		->Parallel Seq Scan
+-- 	->Seq Scan / Join / any other non-Gather node
+-- 	->Gather
+-- 		->Parallel Seq Scan
+
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 where col1 = 5 union all
+    select (select temp1.col1 from temp2 limit 1) col2 from temp1 union all
+    select * from temp1 where col1 = 5;');
+select count(*) from parallel_write;
+drop table parallel_write;
+
+alter table temp2 set (parallel_workers = 0);
+select explain_pictas(
+'create table parallel_write as select * from temp1 where col1 = (select 1) union all
+    select * from temp1 where col1 = (select 1) union all
+ select * from temp2 where col2 = (select 2);');
+select count(*) from parallel_write;
+alter table temp2 reset (parallel_workers);
+drop table parallel_write;
+
+-- case 4: parallel inserts must not occur as there will be no direct Gather
+-- node under Append node. Non-Gather nodes will do inserts by sending tuples
+-- to Append and from there to CTAS dest receiver.
+-- Append
+-- 	->Seq Scan / Join / any other non-Gather node
+-- 	->Seq Scan / Join / any other non-Gather node
+-- 	->Seq Scan / Join / any other non-Gather node
+
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 union all
+    select * from temp2 union all
+    select (select temp1.col1 from temp2 limit 1) col2 from temp1;');
+select count(*) from parallel_write;
+drop table parallel_write;
+
+-- case 5: parallel inserts must occur at the top Gather node as we can push
+-- the CTAS dest receiver to it.
+-- Gather
+-- 	->Parallel Append
+-- 		->Seq Scan / Join / any other non-Gather node
+-- 		->Parallel Seq Scan
+-- 		->Parallel Seq Scan
+
+alter table temp2 set (parallel_workers = 0);
+
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 union all
+    select * from temp2 union all
+    select * from temp1;');
+select count(*) from parallel_write;
+drop table parallel_write;
+
+alter table temp2 reset (parallel_workers);
+
+-- case 6: parallel inserts must occur at each Gather node as we can push the
+-- CTAS dest receiver.
+-- Append
+--  ->Append
+--     ->Gather
+-- 		->Parallel Seq Scan
+--  ->Append
+--     ->Gather
+-- 		->Parallel Seq Scan
+--  ->Gather
+--  ->Gather
+
+select explain_pictas(
+'create table parallel_write as
+    select * from (select * from temp1 where col1 = (select 1) union all
+            select * from temp2 where col2 = (select 2)) as tt
+        where col1 = (select 1) union all
+    select * from temp2 where col2 = (select 2) union all
+    select * from temp2 where col2 = (select 2);');
+select count(*) from parallel_write;
+drop table parallel_write;
+
+select explain_pictas(
+'create table parallel_write as
+    select * from (select * from temp1 where col1 = (select 1) union all
+            select * from temp2 where col2 = (select 2)) as tt
+        where col1 = (select 1) union all
+    select * from (select * from temp1 where col1 = (select 1) union all
+            select * from temp2 where col2 = (select 2)) as tt
+        where col1 = (select 1) union all
+    select * from temp1 where col1 = 5 union all
+    select * from temp2 where col2 = 5;');
+select count(*) from parallel_write;
+drop table parallel_write;
+
+-- case 7: parallel inserts must occur at each Gather node as we can push the
+-- CTAS dest receiver. Non-Gather nodes will do inserts by sending tuples
+-- to Append and from there to CTAS dest receiver.
+-- Append
+--  ->Append
+--     ->Gather
+-- 		->Parallel Seq Scan
+--  ->Append
+--     ->Gather
+-- 		->Parallel Seq Scan
+--  ->Seq Scan / Join / any other non-Gather node
+--  ->Gather
+
+alter table temp2 set (parallel_workers = 0);
+
+select explain_pictas(
+'create table parallel_write as
+    select * from (select * from temp1 where col1 = (select 1) union all
+            select * from temp2 where col2 = (select 2)) as tt
+        where col1 = (select 1) union all
+    select * from temp2 where col2 = (select 2) union all
+    select * from temp1 where col1 = (select 2);');
+select count(*) from parallel_write;
+drop table parallel_write;
+
+select explain_pictas(
+'create table parallel_write as
+    select * from (select * from temp1 where col1 = (select 1) union all
+            select * from temp2 where col2 = (select 2)) as tt
+        where col1 = (select 1) union all
+    select * from (select * from temp1 where col1 = (select 1) union all
+            select * from temp2 where col2 = (select 2)) as tt
+        where col1 = (select 1) union all
+    select * from temp1 where col1 = 5 union all
+    select * from temp2 where col2 = 5;');
+select count(*) from parallel_write;
+drop table parallel_write;
+
+alter table temp2 reset (parallel_workers);
+
+-- case 8: parallel inserts must not occur because there is no Gather or Append
+-- node at the top for union, except/except all, intersect/intersect all
+-- cases.
+
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 union
+    select * from temp2;');
+select count(*) from parallel_write;
+drop table parallel_write;
+
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 except
+    select * from temp2 where col2 < 3;');
+select count(*) from parallel_write;
+drop table parallel_write;
+
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 except all
+    select * from temp2 where col2 < 3;');
+select count(*) from parallel_write;
+drop table parallel_write;
+
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 intersect
+    select * from temp2;');
+select count(*) from parallel_write;
+drop table parallel_write;
+
+select explain_pictas(
+'create table parallel_write as
+    select * from temp1 intersect all
+    select * from temp2;');
+select count(*) from parallel_write;
+drop table parallel_write;
+
 drop table temp1;
 drop table temp2;
 drop table temp3;
-- 
2.25.1

