From 7f3cc3f322a971cdd23221b45810ceb47e50be84 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <bharath.rupireddy@enterprisedb.com>
Date: Tue, 22 Dec 2020 13:30:22 +0530
Subject: [PATCH v15 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              | 104 ++-
 src/backend/optimizer/path/allpaths.c        |  39 +
 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, 1073 insertions(+), 28 deletions(-)

diff --git a/src/backend/commands/createas.c b/src/backend/commands/createas.c
index 210927d4f4..20d4f805d0 100644
--- a/src/backend/commands/createas.c
+++ b/src/backend/commands/createas.c
@@ -655,6 +655,78 @@ intorel_destroy(DestReceiver *self)
 	pfree(self);
 }
 
+/*
+ * PushDownCTASParallelInsertState --- push the dest receiver down to the
+ * Gather nodes.
+ *
+ * In this function we only care 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;
+
+			/*
+			 * For parallelizing inserts in CTAS i.e. making each parallel
+			 * worker insert the tuples, we must send information such as into
+			 * clause (for each worker to build separate dest receiver), object
+			 * id (for each worker to open the created table).
+			 */
+			((DR_intorel *) dest)->is_parallel = true;
+			gstate->dest = dest;
+
+			/*
+			 * Since there are no rows that are transferred from workers to
+			 * Gather node, so we set it to 0 to be visible in estimated row
+			 * count of explain plans.
+			 */
+			ps->plan->plan_rows = 0;
+		}
+	}
+
+	return parallel;
+}
+
 /*
  * ChooseParallelInsertsInCTAS --- determine whether or not parallel
  * insertion is possible, if yes set the parallel insert state i.e. push down
@@ -664,6 +736,7 @@ void ChooseParallelInsertsInCTAS(IntoClause *into, QueryDesc *queryDesc,
 								 uint8 *tuple_cost_flags)
 {
 	bool allow = false;
+	bool gather_exists = false;
 
 	if (!IS_CTAS(into))
 		return;
@@ -678,30 +751,12 @@ void ChooseParallelInsertsInCTAS(IntoClause *into, QueryDesc *queryDesc,
 	if (!into->rel || into->rel->relpersistence == RELPERSISTENCE_TEMP)
 		return;
 
-	if (queryDesc && IsA(queryDesc->planstate, GatherState) &&
-		!queryDesc->planstate->ps_ProjInfo)
-	{
-		GatherState *gstate = (GatherState *) queryDesc->planstate;
-		DestReceiver *dest	= queryDesc->dest;
-
-		/*
-		 * For parallelizing inserts in CTAS i.e. making each parallel worker
-		 * insert the tuples, we must send information such as into clause (for
-		 * each worker to build separate dest receiver), object id (for each
-		 * worker to open the created table).
-		 */
-		((DR_intorel *) dest)->is_parallel = true;
-		gstate->dest = dest;
-
-		/*
-		 * Since there are no rows that are transferred from workers to Gather
-		 * node, so we set it to 0 to be visible in estimated row count of
-		 * explain plans.
-		 */
-		queryDesc->planstate->plan->plan_rows = 0;
+	if (!queryDesc)
+		return;
 
-		allow = true;
-	}
+	allow = PushDownCTASParallelInsertState(queryDesc->dest,
+											queryDesc->planstate,
+											&gather_exists);
 
 	/*
 	 * It should not happen that in cost_gather we have ignored the parallel
@@ -712,8 +767,7 @@ void ChooseParallelInsertsInCTAS(IntoClause *into, QueryDesc *queryDesc,
 	 * which the parallel plan may have been chosen, but we do not allow the
 	 * parallel inserts now.
 	 */
-	if (!allow && tuple_cost_flags && queryDesc &&
-		IsA(queryDesc->planstate, GatherState))
+	if (!allow && tuple_cost_flags && gather_exists)
 	{
 		/*
 		 * If we have correctly ignored parallel tuple cost in planner while
diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index 627d08b78a..00819eea3f 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,6 +1104,44 @@ 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 set the flag to ignore parallel
+		 * tuple cost by the Gather path in cost_gather if the SELECT is for
+		 * CTAS.
+		 */
+		if (childrel->rtekind == RTE_SUBQUERY)
+		{
+			/*
+			 * We set the flag for two cases when there is no parent path will
+			 * be created(such as : limit,sort,distinct...):
+			 * i) query_level is 1
+			 * ii) query_level > 1 then set 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;
+			}
+		}
+
+		if (root->parse->CTASParallelInsInfo &
+			CTAS_PARALLEL_INS_IGN_TUP_COST_APPEND)
+		{
+			root->parse->CTASParallelInsInfo &=
+										~CTAS_PARALLEL_INS_IGN_TUP_COST_APPEND;
+		}
+
 		/*
 		 * Compute the child's size.
 		 */
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index f1134711b0..7555cde61a 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 0ae2f49e0c..4103ac65f0 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,
 	/* Set to this after the cost is ignored. */
-	CTAS_PARALLEL_INS_TUP_COST_IGNORED = 1 << 2
+	CTAS_PARALLEL_INS_TUP_COST_IGNORED = 1 << 2,
+	/* Set to 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 357fbbbe8d..e0296e88a3 100644
--- a/src/test/regress/expected/write_parallel.out
+++ b/src/test/regress/expected/write_parallel.out
@@ -649,6 +649,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 46b958014a..4e03d0ab6b 100644
--- a/src/test/regress/sql/write_parallel.sql
+++ b/src/test/regress/sql/write_parallel.sql
@@ -251,6 +251,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

