Fix BUG #17335: Duplicate result rows in Gather node

diff --git a/src/backend/optimizer/plan/setrefs.c b/src/backend/optimizer/plan/setrefs.c
index e44ae971b4..a7b11b7f03 100644
--- a/src/backend/optimizer/plan/setrefs.c
+++ b/src/backend/optimizer/plan/setrefs.c
@@ -1512,8 +1512,16 @@ set_append_references(PlannerInfo *root,
 		lfirst(l) = set_plan_refs(root, (Plan *) lfirst(l), rtoffset);
 	}
 
-	/* Now, if there's just one, forget the Append and return that child */
-	if (list_length(aplan->appendplans) == 1)
+	/*
+	 * See if it's safe to get rid of the Append entirely.  For this to be
+	 * safe, there must be only one child plan and that child plan's parallel
+	 * awareness must match that of the Append's.  The reason for the latter
+	 * is that the if the Append is parallel aware and the child is not then
+	 * the calling plan may execute the non-parallel aware child multiple
+	 * times.
+	 */
+	if (list_length(aplan->appendplans) == 1 &&
+		((Plan *) linitial(aplan->appendplans))->parallel_aware == aplan->plan.parallel_aware)
 		return clean_up_removed_plan_level((Plan *) aplan,
 										   (Plan *) linitial(aplan->appendplans));
 
@@ -1576,8 +1584,16 @@ set_mergeappend_references(PlannerInfo *root,
 		lfirst(l) = set_plan_refs(root, (Plan *) lfirst(l), rtoffset);
 	}
 
-	/* Now, if there's just one, forget the MergeAppend and return that child */
-	if (list_length(mplan->mergeplans) == 1)
+	/*
+	 * See if it's safe to get rid of the MergeAppend entirely.  For this to
+	 * be safe, there must be only one child plan and that child plan's
+	 * parallel awareness must match that of the MergeAppend's.  The reason
+	 * for the latter is that the if the MergeAppend is parallel aware and the
+	 * child is not then the calling plan may execute the non-parallel aware
+	 * child multiple times.
+	 */
+	if (list_length(mplan->mergeplans) == 1 &&
+		((Plan *) linitial(mplan->mergeplans))->parallel_aware == mplan->plan.parallel_aware)
 		return clean_up_removed_plan_level((Plan *) mplan,
 										   (Plan *) linitial(mplan->mergeplans));
 

From 359df37ae76170a4621cafd3ad8b318473c94a46 Mon Sep 17 00:00:00 2001
From: Yura Sokolov <y.sokolov@postgrespro.ru>
Date: Sun, 23 Jan 2022 14:53:21 +0300
Subject: [PATCH v2] Fix duplicate result rows after Append path removal.

It could happen Append path is created with "parallel_aware" flag,
but its single child is not. Append path parent (Gather or Gather Merge)
thinks its child is parallel_aware, but after Append path removal Gather's
child become not parallel_aware. Then when Gather/Gather Merge decides
to run child in several workers or worker + leader participation, it
gathers duplicate result rows from several child path invocations.

To fix it don't remove Append/MergeAppend node if it's parallel_aware !=
single child parallel_aware.

Authors: David Rowley, Sokolov Yura.
---
 src/backend/optimizer/plan/setrefs.c          |  24 +++-
 .../expected/gather_removed_append.out        | 135 ++++++++++++++++++
 src/test/regress/parallel_schedule            |   1 +
 .../regress/sql/gather_removed_append.sql     |  82 +++++++++++
 4 files changed, 238 insertions(+), 4 deletions(-)
 create mode 100644 src/test/regress/expected/gather_removed_append.out
 create mode 100644 src/test/regress/sql/gather_removed_append.sql

diff --git a/src/backend/optimizer/plan/setrefs.c b/src/backend/optimizer/plan/setrefs.c
index e44ae971b4b..a7b11b7f03a 100644
--- a/src/backend/optimizer/plan/setrefs.c
+++ b/src/backend/optimizer/plan/setrefs.c
@@ -1512,8 +1512,16 @@ set_append_references(PlannerInfo *root,
 		lfirst(l) = set_plan_refs(root, (Plan *) lfirst(l), rtoffset);
 	}
 
-	/* Now, if there's just one, forget the Append and return that child */
-	if (list_length(aplan->appendplans) == 1)
+	/*
+	 * See if it's safe to get rid of the Append entirely.  For this to be
+	 * safe, there must be only one child plan and that child plan's parallel
+	 * awareness must match that of the Append's.  The reason for the latter
+	 * is that the if the Append is parallel aware and the child is not then
+	 * the calling plan may execute the non-parallel aware child multiple
+	 * times.
+	 */
+	if (list_length(aplan->appendplans) == 1 &&
+		((Plan *) linitial(aplan->appendplans))->parallel_aware == aplan->plan.parallel_aware)
 		return clean_up_removed_plan_level((Plan *) aplan,
 										   (Plan *) linitial(aplan->appendplans));
 
@@ -1576,8 +1584,16 @@ set_mergeappend_references(PlannerInfo *root,
 		lfirst(l) = set_plan_refs(root, (Plan *) lfirst(l), rtoffset);
 	}
 
-	/* Now, if there's just one, forget the MergeAppend and return that child */
-	if (list_length(mplan->mergeplans) == 1)
+	/*
+	 * See if it's safe to get rid of the MergeAppend entirely.  For this to
+	 * be safe, there must be only one child plan and that child plan's
+	 * parallel awareness must match that of the MergeAppend's.  The reason
+	 * for the latter is that the if the MergeAppend is parallel aware and the
+	 * child is not then the calling plan may execute the non-parallel aware
+	 * child multiple times.
+	 */
+	if (list_length(mplan->mergeplans) == 1 &&
+		((Plan *) linitial(mplan->mergeplans))->parallel_aware == mplan->plan.parallel_aware)
 		return clean_up_removed_plan_level((Plan *) mplan,
 										   (Plan *) linitial(mplan->mergeplans));
 
diff --git a/src/test/regress/expected/gather_removed_append.out b/src/test/regress/expected/gather_removed_append.out
new file mode 100644
index 00000000000..1c2d40d7c76
--- /dev/null
+++ b/src/test/regress/expected/gather_removed_append.out
@@ -0,0 +1,135 @@
+-- Test correctness of parallel query execution after removal
+-- of Append path due to single non-trivial child.
+DROP TABLE IF EXISTS gather_append_1, gather_append_2;
+NOTICE:  table "gather_append_1" does not exist, skipping
+NOTICE:  table "gather_append_2" does not exist, skipping
+CREATE TABLE gather_append_1 (
+    fk int,
+    f bool
+);
+INSERT INTO gather_append_1 (fk, f) SELECT i, i%50=0 from generate_series(1, 2000) as i;
+CREATE INDEX gather_append_1_ix on gather_append_1 (f);
+CREATE TABLE gather_append_2 (
+    fk int,
+    val serial
+);
+INSERT INTO gather_append_2 (fk) SELECT fk from gather_append_1, generate_series(1, 5) as i;
+ANALYZE gather_append_1, gather_append_2;
+SET max_parallel_workers_per_gather = 0;
+-- Find correct rows count
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);
+ count 
+-------
+   200
+(1 row)
+
+SET parallel_setup_cost = 0;
+SET parallel_tuple_cost = 0.1;
+SET min_parallel_table_scan_size = 0;
+SET max_parallel_workers_per_gather = 2;
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);
+ count 
+-------
+   200
+(1 row)
+
+EXPLAIN (ANALYZE true, COSTS false, TIMING false, SUMMARY false, VERBOSE false, BUFFERS false)
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);
+                                                  QUERY PLAN                                                   
+---------------------------------------------------------------------------------------------------------------
+ Finalize Aggregate (actual rows=1 loops=1)
+   ->  Gather (actual rows=2 loops=1)
+         Workers Planned: 1
+         Workers Launched: 1
+         ->  Partial Aggregate (actual rows=1 loops=2)
+               ->  Parallel Hash Left Join (actual rows=100 loops=2)
+                     Hash Cond: (gather_append_1.fk = gather_append_2.fk)
+                     ->  Parallel Append (actual rows=20 loops=2)
+                           ->  Index Scan using gather_append_1_ix on gather_append_1 (actual rows=40 loops=1)
+                                 Index Cond: (f = true)
+                     ->  Parallel Hash (actual rows=5000 loops=2)
+                           Buckets: 16384  Batches: 1  Memory Usage: 544kB
+                           ->  Parallel Seq Scan on gather_append_2 (actual rows=5000 loops=2)
+(13 rows)
+
+-- Result rows in root node should be equal to non-parallel count
+EXPLAIN (ANALYZE true, COSTS false, TIMING false, SUMMARY false, VERBOSE false, BUFFERS false)
+SELECT val
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk)
+ORDER BY val;
+                                               QUERY PLAN                                                
+---------------------------------------------------------------------------------------------------------
+ Gather Merge (actual rows=200 loops=1)
+   Workers Planned: 1
+   Workers Launched: 1
+   ->  Sort (actual rows=100 loops=2)
+         Sort Key: gather_append_2.val
+         Sort Method: quicksort  Memory: 25kB
+         Worker 0:  Sort Method: quicksort  Memory: 25kB
+         ->  Parallel Hash Left Join (actual rows=100 loops=2)
+               Hash Cond: (gather_append_1.fk = gather_append_2.fk)
+               ->  Parallel Append (actual rows=20 loops=2)
+                     ->  Index Scan using gather_append_1_ix on gather_append_1 (actual rows=40 loops=1)
+                           Index Cond: (f = true)
+               ->  Parallel Hash (actual rows=5000 loops=2)
+                     Buckets: 16384  Batches: 1  Memory Usage: 576kB
+                     ->  Parallel Seq Scan on gather_append_2 (actual rows=5000 loops=2)
+(15 rows)
+
+-- Result rows in root node should be equal to non-parallel count
+EXPLAIN (ANALYZE true, COSTS false, TIMING false, SUMMARY false, VERBOSE false, BUFFERS false)
+SELECT val
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk)
+ORDER BY val;
+                                               QUERY PLAN                                                
+---------------------------------------------------------------------------------------------------------
+ Gather Merge (actual rows=200 loops=1)
+   Workers Planned: 1
+   Workers Launched: 1
+   ->  Sort (actual rows=100 loops=2)
+         Sort Key: gather_append_2.val
+         Sort Method: quicksort  Memory: 25kB
+         Worker 0:  Sort Method: quicksort  Memory: 25kB
+         ->  Parallel Hash Left Join (actual rows=100 loops=2)
+               Hash Cond: (gather_append_1.fk = gather_append_2.fk)
+               ->  Parallel Append (actual rows=20 loops=2)
+                     ->  Index Scan using gather_append_1_ix on gather_append_1 (actual rows=40 loops=1)
+                           Index Cond: (f = true)
+               ->  Parallel Hash (actual rows=5000 loops=2)
+                     Buckets: 16384  Batches: 1  Memory Usage: 544kB
+                     ->  Parallel Seq Scan on gather_append_2 (actual rows=5000 loops=2)
+(15 rows)
+
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index 5b0c73d7e37..84f2f81255d 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -100,6 +100,7 @@ test: rules psql psql_crosstab amutils stats_ext collate.linux.utf8
 test: select_parallel
 test: write_parallel
 test: vacuum_parallel
+test: gather_removed_append
 
 # no relation related tests can be put in this group
 test: publication subscription
diff --git a/src/test/regress/sql/gather_removed_append.sql b/src/test/regress/sql/gather_removed_append.sql
new file mode 100644
index 00000000000..df1b796a4f6
--- /dev/null
+++ b/src/test/regress/sql/gather_removed_append.sql
@@ -0,0 +1,82 @@
+-- Test correctness of parallel query execution after removal
+-- of Append path due to single non-trivial child.
+
+DROP TABLE IF EXISTS gather_append_1, gather_append_2;
+
+CREATE TABLE gather_append_1 (
+    fk int,
+    f bool
+);
+
+INSERT INTO gather_append_1 (fk, f) SELECT i, i%50=0 from generate_series(1, 2000) as i;
+
+CREATE INDEX gather_append_1_ix on gather_append_1 (f);
+
+CREATE TABLE gather_append_2 (
+    fk int,
+    val serial
+);
+
+INSERT INTO gather_append_2 (fk) SELECT fk from gather_append_1, generate_series(1, 5) as i;
+
+ANALYZE gather_append_1, gather_append_2;
+
+SET max_parallel_workers_per_gather = 0;
+
+-- Find correct rows count
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);
+
+SET parallel_setup_cost = 0;
+SET parallel_tuple_cost = 0.1;
+SET min_parallel_table_scan_size = 0;
+SET max_parallel_workers_per_gather = 2;
+
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);
+
+EXPLAIN (ANALYZE true, COSTS false, TIMING false, SUMMARY false, VERBOSE false, BUFFERS false)
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);
+
+-- Result rows in root node should be equal to non-parallel count
+EXPLAIN (ANALYZE true, COSTS false, TIMING false, SUMMARY false, VERBOSE false, BUFFERS false)
+SELECT val
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk)
+ORDER BY val;
+
+-- Result rows in root node should be equal to non-parallel count
+EXPLAIN (ANALYZE true, COSTS false, TIMING false, SUMMARY false, VERBOSE false, BUFFERS false)
+SELECT val
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk)
+ORDER BY val;
-- 
2.34.1

From 9ed2139495b2026433ff5e7c4092fcfa8f10e4d1 Mon Sep 17 00:00:00 2001
From: Yura Sokolov <y.sokolov@postgrespro.ru>
Date: Sun, 23 Jan 2022 14:53:21 +0300
Subject: [PATCH v3] Fix duplicate result rows after Append path removal.

It could happen Append path is created with "parallel_aware" flag,
but its single child is not. Append path parent (Gather or Gather Merge)
thinks its child is parallel_aware, but after Append path removal Gather's
child become not parallel_aware. Then when Gather/Gather Merge decides
to run child in several workers or worker + leader participation, it
gathers duplicate result rows from several child path invocations.

To fix it don't remove Append/MergeAppend node if it's parallel_aware !=
single child parallel_aware.

Authors: David Rowley, Sokolov Yura.
---
 src/backend/optimizer/plan/setrefs.c          |  24 ++-
 .../expected/gather_removed_append.out        | 154 ++++++++++++++++++
 src/test/regress/parallel_schedule            |   1 +
 .../regress/sql/gather_removed_append.sql     | 102 ++++++++++++
 4 files changed, 277 insertions(+), 4 deletions(-)
 create mode 100644 src/test/regress/expected/gather_removed_append.out
 create mode 100644 src/test/regress/sql/gather_removed_append.sql

diff --git a/src/backend/optimizer/plan/setrefs.c b/src/backend/optimizer/plan/setrefs.c
index e44ae971b4b..a7b11b7f03a 100644
--- a/src/backend/optimizer/plan/setrefs.c
+++ b/src/backend/optimizer/plan/setrefs.c
@@ -1512,8 +1512,16 @@ set_append_references(PlannerInfo *root,
 		lfirst(l) = set_plan_refs(root, (Plan *) lfirst(l), rtoffset);
 	}
 
-	/* Now, if there's just one, forget the Append and return that child */
-	if (list_length(aplan->appendplans) == 1)
+	/*
+	 * See if it's safe to get rid of the Append entirely.  For this to be
+	 * safe, there must be only one child plan and that child plan's parallel
+	 * awareness must match that of the Append's.  The reason for the latter
+	 * is that the if the Append is parallel aware and the child is not then
+	 * the calling plan may execute the non-parallel aware child multiple
+	 * times.
+	 */
+	if (list_length(aplan->appendplans) == 1 &&
+		((Plan *) linitial(aplan->appendplans))->parallel_aware == aplan->plan.parallel_aware)
 		return clean_up_removed_plan_level((Plan *) aplan,
 										   (Plan *) linitial(aplan->appendplans));
 
@@ -1576,8 +1584,16 @@ set_mergeappend_references(PlannerInfo *root,
 		lfirst(l) = set_plan_refs(root, (Plan *) lfirst(l), rtoffset);
 	}
 
-	/* Now, if there's just one, forget the MergeAppend and return that child */
-	if (list_length(mplan->mergeplans) == 1)
+	/*
+	 * See if it's safe to get rid of the MergeAppend entirely.  For this to
+	 * be safe, there must be only one child plan and that child plan's
+	 * parallel awareness must match that of the MergeAppend's.  The reason
+	 * for the latter is that the if the MergeAppend is parallel aware and the
+	 * child is not then the calling plan may execute the non-parallel aware
+	 * child multiple times.
+	 */
+	if (list_length(mplan->mergeplans) == 1 &&
+		((Plan *) linitial(mplan->mergeplans))->parallel_aware == mplan->plan.parallel_aware)
 		return clean_up_removed_plan_level((Plan *) mplan,
 										   (Plan *) linitial(mplan->mergeplans));
 
diff --git a/src/test/regress/expected/gather_removed_append.out b/src/test/regress/expected/gather_removed_append.out
new file mode 100644
index 00000000000..1676b00d9dc
--- /dev/null
+++ b/src/test/regress/expected/gather_removed_append.out
@@ -0,0 +1,154 @@
+-- Test correctness of parallel query execution after removal
+-- of Append path due to single non-trivial child.
+DROP TABLE IF EXISTS gather_append_1, gather_append_2;
+NOTICE:  table "gather_append_1" does not exist, skipping
+NOTICE:  table "gather_append_2" does not exist, skipping
+CREATE TABLE gather_append_1 (
+    fk int,
+    f bool
+);
+INSERT INTO gather_append_1 (fk, f) SELECT i, i%50=0 from generate_series(1, 2000) as i;
+CREATE INDEX gather_append_1_ix on gather_append_1 (f);
+CREATE TABLE gather_append_2 (
+    fk int,
+    val serial
+);
+INSERT INTO gather_append_2 (fk) SELECT fk from gather_append_1, generate_series(1, 5) as i;
+ANALYZE gather_append_1, gather_append_2;
+SET max_parallel_workers_per_gather = 0;
+-- Find correct rows count
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);
+ count 
+-------
+   200
+(1 row)
+
+SET parallel_setup_cost = 0;
+SET parallel_tuple_cost = 0.1;
+SET min_parallel_table_scan_size = 0;
+SET max_parallel_workers_per_gather = 2;
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);
+ count 
+-------
+   200
+(1 row)
+
+-- The buckets/batches/memory values from the Parallel Hash node can vary between
+-- machines.  Let's just replace the number with an 'N'.
+create function explain_gather(query text) returns setof text
+language plpgsql as
+$$
+declare
+    ln text;
+begin
+    for ln in
+        execute format('explain (analyze, costs off, summary off, timing off, verbose off, buffers off) %s',
+            query)
+    loop
+        ln := regexp_replace(ln, 'Buckets: \d+', 'Buckets: N');
+        ln := regexp_replace(ln, 'Batches: \d+', 'Batches: N');
+        ln := regexp_replace(ln, 'Memory Usage: \d+', 'Memory Usage: N');
+        return next ln;
+    end loop;
+end;
+$$;
+SELECT explain_gather('
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);');
+                                                explain_gather                                                 
+---------------------------------------------------------------------------------------------------------------
+ Finalize Aggregate (actual rows=1 loops=1)
+   ->  Gather (actual rows=2 loops=1)
+         Workers Planned: 1
+         Workers Launched: 1
+         ->  Partial Aggregate (actual rows=1 loops=2)
+               ->  Parallel Hash Left Join (actual rows=100 loops=2)
+                     Hash Cond: (gather_append_1.fk = gather_append_2.fk)
+                     ->  Parallel Append (actual rows=20 loops=2)
+                           ->  Index Scan using gather_append_1_ix on gather_append_1 (actual rows=40 loops=1)
+                                 Index Cond: (f = true)
+                     ->  Parallel Hash (actual rows=5000 loops=2)
+                           Buckets: N  Batches: N  Memory Usage: NkB
+                           ->  Parallel Seq Scan on gather_append_2 (actual rows=5000 loops=2)
+(13 rows)
+
+-- Result rows in root node should be equal to non-parallel count
+SELECT explain_gather('
+SELECT val
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk)
+ORDER BY val;');
+                                             explain_gather                                              
+---------------------------------------------------------------------------------------------------------
+ Gather Merge (actual rows=200 loops=1)
+   Workers Planned: 1
+   Workers Launched: 1
+   ->  Sort (actual rows=100 loops=2)
+         Sort Key: gather_append_2.val
+         Sort Method: quicksort  Memory: 25kB
+         Worker 0:  Sort Method: quicksort  Memory: 25kB
+         ->  Parallel Hash Left Join (actual rows=100 loops=2)
+               Hash Cond: (gather_append_1.fk = gather_append_2.fk)
+               ->  Parallel Append (actual rows=20 loops=2)
+                     ->  Index Scan using gather_append_1_ix on gather_append_1 (actual rows=40 loops=1)
+                           Index Cond: (f = true)
+               ->  Parallel Hash (actual rows=5000 loops=2)
+                     Buckets: N  Batches: N  Memory Usage: NkB
+                     ->  Parallel Seq Scan on gather_append_2 (actual rows=5000 loops=2)
+(15 rows)
+
+-- Result rows in root node should be equal to non-parallel count
+SELECT explain_gather('
+SELECT val
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk)
+ORDER BY val;');
+                                             explain_gather                                              
+---------------------------------------------------------------------------------------------------------
+ Gather Merge (actual rows=200 loops=1)
+   Workers Planned: 1
+   Workers Launched: 1
+   ->  Sort (actual rows=100 loops=2)
+         Sort Key: gather_append_2.val
+         Sort Method: quicksort  Memory: 25kB
+         Worker 0:  Sort Method: quicksort  Memory: 25kB
+         ->  Parallel Hash Left Join (actual rows=100 loops=2)
+               Hash Cond: (gather_append_1.fk = gather_append_2.fk)
+               ->  Parallel Append (actual rows=20 loops=2)
+                     ->  Index Scan using gather_append_1_ix on gather_append_1 (actual rows=40 loops=1)
+                           Index Cond: (f = true)
+               ->  Parallel Hash (actual rows=5000 loops=2)
+                     Buckets: N  Batches: N  Memory Usage: NkB
+                     ->  Parallel Seq Scan on gather_append_2 (actual rows=5000 loops=2)
+(15 rows)
+
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index 5b0c73d7e37..84f2f81255d 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -100,6 +100,7 @@ test: rules psql psql_crosstab amutils stats_ext collate.linux.utf8
 test: select_parallel
 test: write_parallel
 test: vacuum_parallel
+test: gather_removed_append
 
 # no relation related tests can be put in this group
 test: publication subscription
diff --git a/src/test/regress/sql/gather_removed_append.sql b/src/test/regress/sql/gather_removed_append.sql
new file mode 100644
index 00000000000..8c0374a8f8b
--- /dev/null
+++ b/src/test/regress/sql/gather_removed_append.sql
@@ -0,0 +1,102 @@
+-- Test correctness of parallel query execution after removal
+-- of Append path due to single non-trivial child.
+
+DROP TABLE IF EXISTS gather_append_1, gather_append_2;
+
+CREATE TABLE gather_append_1 (
+    fk int,
+    f bool
+);
+
+INSERT INTO gather_append_1 (fk, f) SELECT i, i%50=0 from generate_series(1, 2000) as i;
+
+CREATE INDEX gather_append_1_ix on gather_append_1 (f);
+
+CREATE TABLE gather_append_2 (
+    fk int,
+    val serial
+);
+
+INSERT INTO gather_append_2 (fk) SELECT fk from gather_append_1, generate_series(1, 5) as i;
+
+ANALYZE gather_append_1, gather_append_2;
+
+SET max_parallel_workers_per_gather = 0;
+
+-- Find correct rows count
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);
+
+SET parallel_setup_cost = 0;
+SET parallel_tuple_cost = 0.1;
+SET min_parallel_table_scan_size = 0;
+SET max_parallel_workers_per_gather = 2;
+
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);
+
+-- The buckets/batches/memory values from the Parallel Hash node can vary between
+-- machines.  Let's just replace the number with an 'N'.
+create function explain_gather(query text) returns setof text
+language plpgsql as
+$$
+declare
+    ln text;
+begin
+    for ln in
+        execute format('explain (analyze, costs off, summary off, timing off, verbose off, buffers off) %s',
+            query)
+    loop
+        ln := regexp_replace(ln, 'Buckets: \d+', 'Buckets: N');
+        ln := regexp_replace(ln, 'Batches: \d+', 'Batches: N');
+        ln := regexp_replace(ln, 'Memory Usage: \d+', 'Memory Usage: N');
+        return next ln;
+    end loop;
+end;
+$$;
+
+SELECT explain_gather('
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);');
+
+-- Result rows in root node should be equal to non-parallel count
+SELECT explain_gather('
+SELECT val
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk)
+ORDER BY val;');
+
+-- Result rows in root node should be equal to non-parallel count
+SELECT explain_gather('
+SELECT val
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk)
+ORDER BY val;');
-- 
2.34.1

From fb09491a401f0df828faf6088158f431b2a69381 Mon Sep 17 00:00:00 2001
From: Yura Sokolov <y.sokolov@postgrespro.ru>
Date: Sun, 23 Jan 2022 14:53:21 +0300
Subject: [PATCH v4] Fix duplicate result rows after Append path removal.

It could happen Append path is created with "parallel_aware" flag,
but its single child is not. Append path parent (Gather or Gather Merge)
thinks its child is parallel_aware, but after Append path removal Gather's
child become not parallel_aware. Then when Gather/Gather Merge decides
to run child in several workers or worker + leader participation, it
gathers duplicate result rows from several child path invocations.

To fix it don't remove Append/MergeAppend node if it's parallel_aware !=
single child parallel_aware.

Authors: David Rowley, Sokolov Yura.
---
 src/backend/optimizer/plan/setrefs.c          |  24 +++-
 .../expected/gather_removed_append.out        | 126 ++++++++++++++++++
 src/test/regress/parallel_schedule            |   1 +
 .../regress/sql/gather_removed_append.sql     |  94 +++++++++++++
 4 files changed, 241 insertions(+), 4 deletions(-)
 create mode 100644 src/test/regress/expected/gather_removed_append.out
 create mode 100644 src/test/regress/sql/gather_removed_append.sql

diff --git a/src/backend/optimizer/plan/setrefs.c b/src/backend/optimizer/plan/setrefs.c
index e44ae971b4b..a7b11b7f03a 100644
--- a/src/backend/optimizer/plan/setrefs.c
+++ b/src/backend/optimizer/plan/setrefs.c
@@ -1512,8 +1512,16 @@ set_append_references(PlannerInfo *root,
 		lfirst(l) = set_plan_refs(root, (Plan *) lfirst(l), rtoffset);
 	}
 
-	/* Now, if there's just one, forget the Append and return that child */
-	if (list_length(aplan->appendplans) == 1)
+	/*
+	 * See if it's safe to get rid of the Append entirely.  For this to be
+	 * safe, there must be only one child plan and that child plan's parallel
+	 * awareness must match that of the Append's.  The reason for the latter
+	 * is that the if the Append is parallel aware and the child is not then
+	 * the calling plan may execute the non-parallel aware child multiple
+	 * times.
+	 */
+	if (list_length(aplan->appendplans) == 1 &&
+		((Plan *) linitial(aplan->appendplans))->parallel_aware == aplan->plan.parallel_aware)
 		return clean_up_removed_plan_level((Plan *) aplan,
 										   (Plan *) linitial(aplan->appendplans));
 
@@ -1576,8 +1584,16 @@ set_mergeappend_references(PlannerInfo *root,
 		lfirst(l) = set_plan_refs(root, (Plan *) lfirst(l), rtoffset);
 	}
 
-	/* Now, if there's just one, forget the MergeAppend and return that child */
-	if (list_length(mplan->mergeplans) == 1)
+	/*
+	 * See if it's safe to get rid of the MergeAppend entirely.  For this to
+	 * be safe, there must be only one child plan and that child plan's
+	 * parallel awareness must match that of the MergeAppend's.  The reason
+	 * for the latter is that the if the MergeAppend is parallel aware and the
+	 * child is not then the calling plan may execute the non-parallel aware
+	 * child multiple times.
+	 */
+	if (list_length(mplan->mergeplans) == 1 &&
+		((Plan *) linitial(mplan->mergeplans))->parallel_aware == mplan->plan.parallel_aware)
 		return clean_up_removed_plan_level((Plan *) mplan,
 										   (Plan *) linitial(mplan->mergeplans));
 
diff --git a/src/test/regress/expected/gather_removed_append.out b/src/test/regress/expected/gather_removed_append.out
new file mode 100644
index 00000000000..849cf0ae97e
--- /dev/null
+++ b/src/test/regress/expected/gather_removed_append.out
@@ -0,0 +1,126 @@
+-- Test correctness of parallel query execution after removal
+-- of Append path due to single non-trivial child.
+DROP TABLE IF EXISTS gather_append_1, gather_append_2;
+NOTICE:  table "gather_append_1" does not exist, skipping
+NOTICE:  table "gather_append_2" does not exist, skipping
+CREATE TABLE gather_append_1 (
+    fk int,
+    f bool
+);
+INSERT INTO gather_append_1 (fk, f) SELECT i, i%50=0 from generate_series(1, 2000) as i;
+CREATE INDEX gather_append_1_ix on gather_append_1 (f);
+CREATE TABLE gather_append_2 (
+    fk int,
+    val serial
+);
+INSERT INTO gather_append_2 (fk) SELECT fk from gather_append_1, generate_series(1, 5) as i;
+ANALYZE gather_append_1, gather_append_2;
+SET max_parallel_workers_per_gather = 0;
+-- Find correct rows count
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);
+ count 
+-------
+   200
+(1 row)
+
+SET parallel_setup_cost = 0;
+SET parallel_tuple_cost = 0.1;
+SET min_parallel_table_scan_size = 0;
+SET max_parallel_workers_per_gather = 2;
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);
+ count 
+-------
+   200
+(1 row)
+
+-- The buckets/batches/memory values from the Parallel Hash node can vary between
+-- machines.  Let's just replace the number with an 'N'.
+create function explain_gather(query text) returns setof text
+language plpgsql as
+$$
+declare
+    ln text;
+begin
+    for ln in
+        execute format('explain (analyze, costs off, summary off, timing off, verbose off, buffers off) %s',
+            query)
+    loop
+        if not ln like '%Gather%' then
+            ln := regexp_replace(ln, 'actual rows=\d+ loops=\d+', 'actual rows=R loops=L');
+        end if;
+        ln := regexp_replace(ln, 'Buckets: \d+', 'Buckets: N');
+        ln := regexp_replace(ln, 'Batches: \d+', 'Batches: N');
+        ln := regexp_replace(ln, 'Memory Usage: \d+', 'Memory Usage: N');
+        return next ln;
+    end loop;
+end;
+$$;
+-- Result rows in root node should be equal to non-parallel count
+SELECT explain_gather('
+SELECT val
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);');
+                                          explain_gather                                          
+--------------------------------------------------------------------------------------------------
+ Gather (actual rows=200 loops=1)
+   Workers Planned: 1
+   Workers Launched: 1
+   ->  Parallel Hash Left Join (actual rows=R loops=L)
+         Hash Cond: (gather_append_1.fk = gather_append_2.fk)
+         ->  Parallel Append (actual rows=R loops=L)
+               ->  Index Scan using gather_append_1_ix on gather_append_1 (actual rows=R loops=L)
+                     Index Cond: (f = true)
+         ->  Parallel Hash (actual rows=R loops=L)
+               Buckets: N  Batches: N  Memory Usage: NkB
+               ->  Parallel Seq Scan on gather_append_2 (actual rows=R loops=L)
+(11 rows)
+
+-- Result rows in root node should be equal to non-parallel count
+SELECT explain_gather('
+SELECT val
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk)
+ORDER BY val;');
+                                             explain_gather                                             
+--------------------------------------------------------------------------------------------------------
+ Gather Merge (actual rows=200 loops=1)
+   Workers Planned: 1
+   Workers Launched: 1
+   ->  Sort (actual rows=R loops=L)
+         Sort Key: gather_append_2.val
+         Sort Method: quicksort  Memory: 25kB
+         Worker 0:  Sort Method: quicksort  Memory: 25kB
+         ->  Parallel Hash Left Join (actual rows=R loops=L)
+               Hash Cond: (gather_append_1.fk = gather_append_2.fk)
+               ->  Parallel Append (actual rows=R loops=L)
+                     ->  Index Scan using gather_append_1_ix on gather_append_1 (actual rows=R loops=L)
+                           Index Cond: (f = true)
+               ->  Parallel Hash (actual rows=R loops=L)
+                     Buckets: N  Batches: N  Memory Usage: NkB
+                     ->  Parallel Seq Scan on gather_append_2 (actual rows=R loops=L)
+(15 rows)
+
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index 5b0c73d7e37..84f2f81255d 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -100,6 +100,7 @@ test: rules psql psql_crosstab amutils stats_ext collate.linux.utf8
 test: select_parallel
 test: write_parallel
 test: vacuum_parallel
+test: gather_removed_append
 
 # no relation related tests can be put in this group
 test: publication subscription
diff --git a/src/test/regress/sql/gather_removed_append.sql b/src/test/regress/sql/gather_removed_append.sql
new file mode 100644
index 00000000000..3af88b29f0a
--- /dev/null
+++ b/src/test/regress/sql/gather_removed_append.sql
@@ -0,0 +1,94 @@
+-- Test correctness of parallel query execution after removal
+-- of Append path due to single non-trivial child.
+
+DROP TABLE IF EXISTS gather_append_1, gather_append_2;
+
+CREATE TABLE gather_append_1 (
+    fk int,
+    f bool
+);
+
+INSERT INTO gather_append_1 (fk, f) SELECT i, i%50=0 from generate_series(1, 2000) as i;
+
+CREATE INDEX gather_append_1_ix on gather_append_1 (f);
+
+CREATE TABLE gather_append_2 (
+    fk int,
+    val serial
+);
+
+INSERT INTO gather_append_2 (fk) SELECT fk from gather_append_1, generate_series(1, 5) as i;
+
+ANALYZE gather_append_1, gather_append_2;
+
+SET max_parallel_workers_per_gather = 0;
+
+-- Find correct rows count
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);
+
+SET parallel_setup_cost = 0;
+SET parallel_tuple_cost = 0.1;
+SET min_parallel_table_scan_size = 0;
+SET max_parallel_workers_per_gather = 2;
+
+SELECT count(1)
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);
+
+-- The buckets/batches/memory values from the Parallel Hash node can vary between
+-- machines.  Let's just replace the number with an 'N'.
+create function explain_gather(query text) returns setof text
+language plpgsql as
+$$
+declare
+    ln text;
+begin
+    for ln in
+        execute format('explain (analyze, costs off, summary off, timing off, verbose off, buffers off) %s',
+            query)
+    loop
+        if not ln like '%Gather%' then
+            ln := regexp_replace(ln, 'actual rows=\d+ loops=\d+', 'actual rows=R loops=L');
+        end if;
+        ln := regexp_replace(ln, 'Buckets: \d+', 'Buckets: N');
+        ln := regexp_replace(ln, 'Batches: \d+', 'Batches: N');
+        ln := regexp_replace(ln, 'Memory Usage: \d+', 'Memory Usage: N');
+        return next ln;
+    end loop;
+end;
+$$;
+
+-- Result rows in root node should be equal to non-parallel count
+SELECT explain_gather('
+SELECT val
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk);');
+
+-- Result rows in root node should be equal to non-parallel count
+SELECT explain_gather('
+SELECT val
+FROM (
+  SELECT fk FROM gather_append_1 WHERE f
+  UNION ALL
+  SELECT fk FROM gather_append_1 WHERE false
+) as t
+LEFT OUTER JOIN gather_append_2
+USING (fk)
+ORDER BY val;');
-- 
2.34.1

diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index cd6d72c763..56e28ba739 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -313,6 +313,21 @@ 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 bool contains_a_parallel_path(Path *path);
+static bool contains_only_parallel_aware_and_safe_paths(Path *path);
+
+/*
+ * PathTypeCount
+ *		Used for various checks to assert plans are sane in assert enabled
+ *		builds.
+ */
+typedef struct PathTypeCount
+{
+	uint64 count;
+	uint64 parallel_safe_count;
+	uint64 parallel_aware_count;
+
+} PathTypeCount;
 
 
 /*
@@ -389,6 +404,10 @@ create_plan_recurse(PlannerInfo *root, Path *best_path, int flags)
 	/* Guard against stack overflow due to overly complex plans */
 	check_stack_depth();
 
+	/* Parallel aware paths should contain only parallel aware subpaths. */
+	Assert(!best_path->parallel_aware ||
+		   contains_only_parallel_aware_and_safe_paths(best_path));
+
 	switch (best_path->pathtype)
 	{
 		case T_SeqScan:
@@ -481,6 +500,13 @@ create_plan_recurse(PlannerInfo *root, Path *best_path, int flags)
 		case T_Gather:
 			plan = (Plan *) create_gather_plan(root,
 											   (GatherPath *) best_path);
+
+			/*
+			 * We expect a Gather to contain at least one parallel path unless
+			 * running in single_copy mode.
+			 */
+			Assert(((GatherPath *) best_path)->single_copy ||
+				   contains_a_parallel_path(((GatherPath *) best_path)->subpath));
 			break;
 		case T_Sort:
 			plan = (Plan *) create_sort_plan(root,
@@ -537,6 +563,8 @@ create_plan_recurse(PlannerInfo *root, Path *best_path, int flags)
 		case T_GatherMerge:
 			plan = (Plan *) create_gather_merge_plan(root,
 													 (GatherMergePath *) best_path);
+			/* We expect a GatherMerge to contain at least one parallel path */
+			Assert(contains_a_parallel_path(((GatherMergePath *) best_path)->subpath));
 			break;
 		default:
 			elog(ERROR, "unrecognized node type: %d",
@@ -7052,6 +7080,210 @@ make_modifytable(PlannerInfo *root, Plan *subplan,
 	return node;
 }
 
+/*
+ * path_tree_walker
+ *		Walk a path tree beginning with 'path' and call the 'walker' function
+ *		for that path and each of its subpaths recursively.
+ */
+static void
+path_tree_walker(Path *path, void (*walker) (), void *context)
+
+{
+	if (path == NULL)
+		return;
+
+	/* Guard against stack overflow due to overly complex path trees */
+	check_stack_depth();
+
+	walker(path, context);
+
+	switch (path->pathtype)
+	{
+		case T_SeqScan:
+		case T_SampleScan:
+		case T_IndexScan:
+		case T_IndexOnlyScan:
+		case T_BitmapHeapScan:
+		case T_TidScan:
+		case T_TidRangeScan:
+		case T_SubqueryScan:
+		case T_FunctionScan:
+		case T_TableFuncScan:
+		case T_ValuesScan:
+		case T_CteScan:
+		case T_WorkTableScan:
+		case T_NamedTuplestoreScan:
+		case T_ForeignScan:
+		case T_CustomScan:
+			/* Scan paths have no subpaths */
+			break;
+		case T_HashJoin:
+		case T_MergeJoin:
+		case T_NestLoop:
+			path_tree_walker(((JoinPath *) path)->outerjoinpath, walker, context);
+			path_tree_walker(((JoinPath *) path)->innerjoinpath, walker, context);
+			break;
+		case T_Append:
+			{
+				AppendPath *apath = (AppendPath *) path;
+				ListCell	   *lc;
+
+				foreach(lc, apath->subpaths)
+				{
+					Path *subpath = lfirst(lc);
+
+					path_tree_walker(subpath, walker, context);
+				}
+			}
+			break;
+			case T_MergeAppend:
+			{
+				MergeAppendPath *mpath = (MergeAppendPath *) path;
+				ListCell	   *lc;
+
+				foreach(lc, mpath->subpaths)
+				{
+					Path *subpath = lfirst(lc);
+
+					path_tree_walker(subpath, walker, context);
+				}
+			}
+			break;
+		case T_Result:
+			if (IsA(path, ProjectionPath))
+			{
+				path_tree_walker(((ProjectionPath *) path)->subpath, walker, context);
+			}
+			else if (IsA(path, MinMaxAggPath))
+			{
+				/* MinMaxAggPath has no subpaths */
+			}
+			else if (IsA(path, GroupResultPath))
+			{
+				/* GroupResultPath has no subpaths */
+			}
+			else
+			{
+				/* No subpaths for any other Result type path */
+			}
+			break;
+		case T_ProjectSet:
+			path_tree_walker(((ProjectSetPath *) path)->subpath, walker, context);
+			break;
+		case T_Material:
+			path_tree_walker(((MaterialPath *) path)->subpath, walker, context);
+			break;
+		case T_Memoize:
+			path_tree_walker(((MemoizePath *) path)->subpath, walker, context);
+			break;
+		case T_Unique:
+			if (IsA(path, UpperUniquePath))
+			{
+				path_tree_walker(((UpperUniquePath *) path)->subpath, walker, context);
+			}
+			else
+			{
+				Assert(IsA(path, UniquePath));
+				path_tree_walker(((UniquePath *) path)->subpath, walker, context);
+			}
+			break;
+		case T_Gather:
+			path_tree_walker(((GatherPath *) path)->subpath, walker, context);
+			break;
+		case T_Sort:
+			path_tree_walker(((SortPath *) path)->subpath, walker, context);
+			break;
+		case T_IncrementalSort:
+			path_tree_walker(((IncrementalSortPath *) path)->spath.subpath, walker, context);
+			break;
+		case T_Group:
+			path_tree_walker(((GroupPath *) path)->subpath, walker, context);
+			break;
+		case T_Agg:
+			if (IsA(path, GroupingSetsPath))
+				path_tree_walker(((GroupingSetsPath *) path)->subpath, walker, context);
+			else
+			{
+				Assert(IsA(path, AggPath));
+				path_tree_walker(((AggPath *) path)->subpath, walker, context);
+			}
+			break;
+		case T_WindowAgg:
+			path_tree_walker(((WindowAggPath *) path)->subpath, walker, context);
+			break;
+		case T_SetOp:
+			path_tree_walker(((SetOpPath *) path)->subpath, walker, context);
+			break;
+		case T_RecursiveUnion:
+			path_tree_walker(((RecursiveUnionPath *) path)->leftpath, walker, context);
+			path_tree_walker(((RecursiveUnionPath *) path)->rightpath, walker, context);
+			break;
+		case T_LockRows:
+			path_tree_walker(((LockRowsPath *) path)->subpath, walker, context);
+			break;
+		case T_ModifyTable:
+			path_tree_walker(((ModifyTablePath *) path)->subpath, walker, context);
+			break;
+		case T_Limit:
+			path_tree_walker(((LimitPath *) path)->subpath, walker, context);
+			break;
+		case T_GatherMerge:
+			path_tree_walker(((GatherMergePath *) path)->subpath, walker, context);
+			break;
+		default:
+			elog(ERROR, "unrecognized node type: %d", (int) path->pathtype);
+			break;
+	}
+}
+
+/*
+ * path_type_counter
+ *		Determine the total number of paths and the number of paths that are
+ *		parallel_aware and the number that are parallel safe.
+ */
+static void
+path_type_counter(Path *path, PathTypeCount *pathcount)
+{
+	pathcount->count++;
+	if (path->parallel_aware)
+		pathcount->parallel_aware_count++;
+	else if (path->parallel_safe)
+		pathcount->parallel_safe_count++;
+}
+
+/*
+ * contains_a_parallel_path
+ *		Determine if 'path' or any of its subpaths are parallel aware
+ */
+static bool
+contains_a_parallel_path(Path *path)
+{
+	PathTypeCount pathcount;
+
+	memset(&pathcount, 0, sizeof(pathcount));
+
+	path_tree_walker(path, path_type_counter, (void *) &pathcount);
+
+	return (pathcount.parallel_aware_count > 0);
+}
+
+/*
+ * contains_only_parallel_aware_and_safe_paths
+ *		Returns true if 'path' and all of its subpaths are either parallel
+ *		aware or parallel safe.
+ */
+static bool
+contains_only_parallel_aware_and_safe_paths(Path *path)
+{
+	PathTypeCount pathcount;
+
+	memset(&pathcount, 0, sizeof(pathcount));
+
+	path_tree_walker(path, path_type_counter, (void *) &pathcount);
+
+	return (pathcount.parallel_aware_count + pathcount.parallel_safe_count == pathcount.count);
+}
+
 /*
  * is_projection_capable_path
  *		Check whether a given Path node is able to do projection.

diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index cd6d72c763..898046ca07 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -313,6 +313,20 @@ 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 bool contains_a_parallel_aware_path(Path *path);
+static bool contains_only_parallel_safe_paths(Path *path);
+
+/*
+ * PathTypeCount
+ *		Used for various checks to assert plans are sane in assert enabled
+ *		builds.
+ */
+typedef struct PathTypeCount
+{
+	uint64 count;
+	uint64 parallel_safe_count;
+	uint64 parallel_aware_count;
+} PathTypeCount;
 
 
 /*
@@ -389,6 +403,10 @@ create_plan_recurse(PlannerInfo *root, Path *best_path, int flags)
 	/* Guard against stack overflow due to overly complex plans */
 	check_stack_depth();
 
+	/* Parallel aware paths should contain only parallel safe subpaths. */
+	Assert(!best_path->parallel_aware ||
+		   contains_only_parallel_safe_paths(best_path));
+
 	switch (best_path->pathtype)
 	{
 		case T_SeqScan:
@@ -481,6 +499,14 @@ create_plan_recurse(PlannerInfo *root, Path *best_path, int flags)
 		case T_Gather:
 			plan = (Plan *) create_gather_plan(root,
 											   (GatherPath *) best_path);
+
+			/*
+			 * We expect a Gather to contain at least one parallel aware path
+			 * unless running in single_copy mode.
+			 */
+			Assert(((GatherPath *) best_path)->single_copy ||
+				   contains_a_parallel_aware_path(((GatherPath *)
+												  best_path)->subpath));
 			break;
 		case T_Sort:
 			plan = (Plan *) create_sort_plan(root,
@@ -537,6 +563,9 @@ create_plan_recurse(PlannerInfo *root, Path *best_path, int flags)
 		case T_GatherMerge:
 			plan = (Plan *) create_gather_merge_plan(root,
 													 (GatherMergePath *) best_path);
+			/* GatherMerge must contain at least one parallel aware path */
+			Assert(contains_a_parallel_aware_path(((GatherMergePath *)
+												  best_path)->subpath));
 			break;
 		default:
 			elog(ERROR, "unrecognized node type: %d",
@@ -7052,6 +7081,207 @@ make_modifytable(PlannerInfo *root, Plan *subplan,
 	return node;
 }
 
+/*
+ * path_tree_walker
+ *		Walk a path tree beginning with 'path' and call the 'walker' function
+ *		for that path and each of its subpaths recursively.
+ */
+static void
+path_tree_walker(Path *path, void (*walker) (), void *context)
+
+{
+	if (path == NULL)
+		return;
+
+	/* Guard against stack overflow due to overly complex path trees */
+	check_stack_depth();
+
+	walker(path, context);
+
+	switch (path->pathtype)
+	{
+		case T_SeqScan:
+		case T_SampleScan:
+		case T_IndexScan:
+		case T_IndexOnlyScan:
+		case T_BitmapHeapScan:
+		case T_TidScan:
+		case T_TidRangeScan:
+		case T_SubqueryScan:
+		case T_FunctionScan:
+		case T_TableFuncScan:
+		case T_ValuesScan:
+		case T_CteScan:
+		case T_WorkTableScan:
+		case T_NamedTuplestoreScan:
+		case T_ForeignScan:
+		case T_CustomScan:
+			/* Scan paths have no subpaths */
+			break;
+		case T_HashJoin:
+		case T_MergeJoin:
+		case T_NestLoop:
+			path_tree_walker(((JoinPath *) path)->outerjoinpath, walker, context);
+			path_tree_walker(((JoinPath *) path)->innerjoinpath, walker, context);
+			break;
+		case T_Append:
+			{
+				AppendPath *apath = (AppendPath *) path;
+				ListCell	   *lc;
+
+				foreach(lc, apath->subpaths)
+				{
+					Path *subpath = lfirst(lc);
+
+					path_tree_walker(subpath, walker, context);
+				}
+			}
+			break;
+			case T_MergeAppend:
+			{
+				MergeAppendPath *mpath = (MergeAppendPath *) path;
+				ListCell	   *lc;
+
+				foreach(lc, mpath->subpaths)
+				{
+					Path *subpath = lfirst(lc);
+
+					path_tree_walker(subpath, walker, context);
+				}
+			}
+			break;
+		case T_Result:
+			if (IsA(path, ProjectionPath))
+			{
+				path_tree_walker(((ProjectionPath *) path)->subpath, walker, context);
+			}
+			else if (IsA(path, MinMaxAggPath))
+			{
+				/* MinMaxAggPath has no subpaths */
+			}
+			else if (IsA(path, GroupResultPath))
+			{
+				/* GroupResultPath has no subpaths */
+			}
+			else
+			{
+				/* No subpaths for any other Result type path */
+			}
+			break;
+		case T_ProjectSet:
+			path_tree_walker(((ProjectSetPath *) path)->subpath, walker, context);
+			break;
+		case T_Material:
+			path_tree_walker(((MaterialPath *) path)->subpath, walker, context);
+			break;
+		case T_Memoize:
+			path_tree_walker(((MemoizePath *) path)->subpath, walker, context);
+			break;
+		case T_Unique:
+			if (IsA(path, UpperUniquePath))
+				path_tree_walker(((UpperUniquePath *) path)->subpath, walker, context);
+			else
+			{
+				Assert(IsA(path, UniquePath));
+				path_tree_walker(((UniquePath *) path)->subpath, walker, context);
+			}
+			break;
+		case T_Gather:
+			path_tree_walker(((GatherPath *) path)->subpath, walker, context);
+			break;
+		case T_Sort:
+			path_tree_walker(((SortPath *) path)->subpath, walker, context);
+			break;
+		case T_IncrementalSort:
+			path_tree_walker(((IncrementalSortPath *) path)->spath.subpath, walker, context);
+			break;
+		case T_Group:
+			path_tree_walker(((GroupPath *) path)->subpath, walker, context);
+			break;
+		case T_Agg:
+			if (IsA(path, GroupingSetsPath))
+				path_tree_walker(((GroupingSetsPath *) path)->subpath, walker, context);
+			else
+			{
+				Assert(IsA(path, AggPath));
+				path_tree_walker(((AggPath *) path)->subpath, walker, context);
+			}
+			break;
+		case T_WindowAgg:
+			path_tree_walker(((WindowAggPath *) path)->subpath, walker, context);
+			break;
+		case T_SetOp:
+			path_tree_walker(((SetOpPath *) path)->subpath, walker, context);
+			break;
+		case T_RecursiveUnion:
+			path_tree_walker(((RecursiveUnionPath *) path)->leftpath, walker, context);
+			path_tree_walker(((RecursiveUnionPath *) path)->rightpath, walker, context);
+			break;
+		case T_LockRows:
+			path_tree_walker(((LockRowsPath *) path)->subpath, walker, context);
+			break;
+		case T_ModifyTable:
+			path_tree_walker(((ModifyTablePath *) path)->subpath, walker, context);
+			break;
+		case T_Limit:
+			path_tree_walker(((LimitPath *) path)->subpath, walker, context);
+			break;
+		case T_GatherMerge:
+			path_tree_walker(((GatherMergePath *) path)->subpath, walker, context);
+			break;
+		default:
+			elog(ERROR, "unrecognized node type: %d", (int) path->pathtype);
+			break;
+	}
+}
+
+/*
+ * path_type_counter
+ *		Determine the total number of paths and the number of paths that are
+ *		parallel_aware and the number that are parallel safe.
+ */
+static void
+path_type_counter(Path *path, PathTypeCount *pathcount)
+{
+	pathcount->count++;
+	if (path->parallel_aware)
+		pathcount->parallel_aware_count++;
+	if (path->parallel_safe)
+		pathcount->parallel_safe_count++;
+}
+
+/*
+ * contains_a_parallel_aware_path
+ *		Determine if 'path' or any of its subpaths are parallel aware
+ */
+static bool
+contains_a_parallel_aware_path(Path *path)
+{
+	PathTypeCount pathcount;
+
+	memset(&pathcount, 0, sizeof(pathcount));
+
+	path_tree_walker(path, path_type_counter, (void *) &pathcount);
+
+	return (pathcount.parallel_aware_count > 0);
+}
+
+/*
+ * contains_only_parallel_safe_paths
+ *		Returns true if 'path' and all of its subpaths are parallel safe
+ */
+static bool
+contains_only_parallel_safe_paths(Path *path)
+{
+	PathTypeCount pathcount;
+
+	memset(&pathcount, 0, sizeof(pathcount));
+
+	path_tree_walker(path, path_type_counter, (void *) &pathcount);
+
+	return (pathcount.parallel_safe_count == pathcount.count);
+}
+
 /*
  * is_projection_capable_path
  *		Check whether a given Path node is able to do projection.