public inbox for [email protected]
help / color / mirror / Atom feedFrom: Yura Sokolov <[email protected]>
To: David Rowley <[email protected]>
Cc: PostgreSQL Developers <[email protected]>
Subject: Re: Fix BUG #17335: Duplicate result rows in Gather node
Date: Sun, 23 Jan 2022 14:56:46 +0300
Message-ID: <[email protected]> (raw)
In-Reply-To: <CAApHDvqMUshyci-F2CZ-jkSbmuhzCudhF5t5_z0shFxd9D_hVg@mail.gmail.com>
References: <[email protected]>
<CAApHDvqMUshyci-F2CZ-jkSbmuhzCudhF5t5_z0shFxd9D_hVg@mail.gmail.com>
В Чт, 20/01/2022 в 09:32 +1300, David Rowley пишет:
> On Fri, 31 Dec 2021 at 00:14, Yura Sokolov <[email protected]> wrote:
> > Suggested quick (and valid) fix in the patch attached:
> > - If Append has single child, then copy its parallel awareness.
>
> I've been looking at this and I've gone through changing my mind about
> what's the right fix quite a number of times.
>
> My current thoughts are that I don't really like the fact that we can
> have plans in the following shape:
>
> Finalize Aggregate
> -> Gather
> Workers Planned: 1
> -> Partial Aggregate
> -> Parallel Hash Left Join
> Hash Cond: (gather_append_1.fk = gather_append_2.fk)
> -> Index Scan using gather_append_1_ix on gather_append_1
> Index Cond: (f = true)
> -> Parallel Hash
> -> Parallel Seq Scan on gather_append_2
>
> It's only made safe by the fact that Gather will only use 1 worker.
> To me, it just seems too fragile to assume that's always going to be
> the case. I feel like this fix just relies on the fact that
> create_gather_path() and create_gather_merge_path() do
> "pathnode->num_workers = subpath->parallel_workers;". If someone
> decided that was to work a different way, then we risk this breaking
> again. Additionally, today we have Gather and GatherMerge, but we may
> one day end up with more node types that gather results from parallel
> workers, or even a completely different way of executing plans.
It seems strange parallel_aware and parallel_safe flags neither affect
execution nor are properly checked.
Except parallel_safe is checked in ExecSerializePlan which is called from
ExecInitParallelPlan, which is called from ExecGather and ExecGatherMerge.
But looks like this check doesn't affect execution as well.
>
> I think a safer way to fix this is to just not remove the
> Append/MergeAppend node if the parallel_aware flag of the only-child
> and the Append/MergeAppend don't match. I've done that in the
> attached.
>
> I believe the code at the end of add_paths_to_append_rel() can remain as is.
I found clean_up_removed_plan_level also called from set_subqueryscan_references.
Is there a need to patch there as well?
And there is strange state:
- in the loop by subpaths, pathnode->node.parallel_safe is set to AND of
all its subpath's parallel_safe
(therefore there were need to copy it in my patch version),
- that means, our AppendPath is parallel_aware but not parallel_safe.
It is ridiculous a bit.
And it is strange AppendPath could have more parallel_workers than sum of
its children parallel_workers.
So it looks like whole machinery around parallel_aware/parallel_safe has
no enough consistency.
Either way, I attach you version of fix with my tests as new patch version.
regards,
Yura Sokolov
Attachments:
[text/x-patch] v2-0001-Fix-duplicate-result-rows-after-Append-path-remov.patch (11.7K, ../[email protected]/2-v2-0001-Fix-duplicate-result-rows-after-Append-path-remov.patch)
download | inline diff:
From 359df37ae76170a4621cafd3ad8b318473c94a46 Mon Sep 17 00:00:00 2001
From: Yura Sokolov <[email protected]>
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
view thread (15+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected]
Subject: Re: Fix BUG #17335: Duplicate result rows in Gather node
In-Reply-To: <[email protected]>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox