public inbox for [email protected]
help / color / mirror / Atom feedRe: Parallelize correlated subqueries that execute within each worker
22+ messages / 8 participants
[nested] [flat]
* Re: Parallelize correlated subqueries that execute within each worker
@ 2023-01-23 14:00 James Coleman <[email protected]>
0 siblings, 2 replies; 22+ messages in thread
From: James Coleman @ 2023-01-23 14:00 UTC (permalink / raw)
To: Tomas Vondra <[email protected]>; +Cc: vignesh C <[email protected]>; Andres Freund <[email protected]>; Robert Haas <[email protected]>; pgsql-hackers; Amit Kapila <[email protected]>; Tom Lane <[email protected]>
On Sat, Jan 21, 2023 at 10:07 PM James Coleman <[email protected]> wrote:
> ...
> While working through Tomas's comment about a conditional in the
> max_parallel_hazard_waker being guaranteed true I realized that in the
> current version of the patch the safe_param_ids tracking in
> is_parallel_safe isn't actually needed any longer. That seemed
> suspicious, and so I started digging, and I found out that in the
> current approach all of the tests pass with only the changes in
> clauses.c. I don't believe that the other changes aren't needed;
> rather I believe there isn't yet a test case exercising them, but I
> realize that means I can't prove they're needed. I spent some time
> poking at this, but at least with my current level of imagination I
> haven't stumbled across a query that would exercise these checks.
I played with this a good bit more yesterday, I'm now a good bit more
confident this is correct. I've cleaned up the patch; see attached for
v7.
Here's some of my thought process:
The comments in src/include/nodes/pathnodes.h:2953 tell us that
PARAM_EXEC params are used to pass values around from one plan node to
another in the following ways:
1. Values down into subqueries (for outer references in subqueries)
2. Up out of subqueries (for the results of a subplan)
3. From a NestLoop plan node into its inner relation (when the inner
scan is parameterized with values from the outer relation)
Case (2) is already known to be safe (we currently add these params to
safe_param_ids in max_parallel_hazard_walker when we encounter a
SubPlan node).
I also believe case (3) is already handled. We don't build partial
paths for joins when joinrel->lateral_relids is non-empty, and join
order calculations already require that parameterization here go the
correct way (i.e., inner depends on outer rather than the other way
around).
That leaves us with only case (1) to consider in this patch. Another
way of saying this is that this is really the only thing the
safe_param_ids tracking is guarding against. For params passed down
into subqueries we can further distinguish between init plans and
"regular" subplans. We already know that params from init plans are
safe (at the right level). So we're concerned here with a way to know
if the params passed to subplans are safe. We already track required
rels in ParamPathInfo, so it's fairly simple to do this test.
Which this patch we do in fact now see (as expected) rels with
non-empty lateral_relids showing up in generate_[useful_]gather_paths.
And the partial paths can now have non-empty required outer rels. I'm
not able to come up with a plan that would actually be caught by those
checks; I theorize that because of the few places we actually call
generate_[useful_]gather_paths we are in practice already excluding
those, but for now I've left these as a conditional rather than an
assertion because it seems like the kind of guard we'd want to ensure
those methods are safe.
The other other place that we actually create gather[_merge] paths is
gather_grouping_paths(), and there I've chosen to use assertions,
because the point at which grouping happens in planning suggests to me
that we shouldn't have lateral dependencies at that point. If someone
is concerned about that, I'd be happy to change those to conditionals
also.
James Coleman
Attachments:
[application/octet-stream] v8-0001-Add-tests-before-change.patch (7.2K, ../../CAAaqYe9V+2TTwGDUu12NLu1tDf0SnwvYnTOrKz=uTh6cLx3McA@mail.gmail.com/2-v8-0001-Add-tests-before-change.patch)
download | inline diff:
From d7a74b2f26edaf36d5098b4b45e19c437a4ebbdf Mon Sep 17 00:00:00 2001
From: jcoleman <[email protected]>
Date: Mon, 26 Sep 2022 20:30:23 -0400
Subject: [PATCH v8 1/2] Add tests before change
---
src/test/regress/expected/select_parallel.out | 108 ++++++++++++++++++
src/test/regress/sql/select_parallel.sql | 25 ++++
2 files changed, 133 insertions(+)
diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
index 91f74fe47a..9b4d7dd44a 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -311,6 +311,114 @@ select count(*) from tenk1 where (two, four) not in
10000
(1 row)
+-- test parallel plans for queries containing correlated subplans
+-- where the subplan only needs params available from the current
+-- worker's scan.
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t, generate_series(1, 10);
+ QUERY PLAN
+----------------------------------------------------------------------------
+ Gather
+ Output: (SubPlan 1)
+ Workers Planned: 4
+ -> Nested Loop
+ Output: t.unique1
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ -> Function Scan on pg_catalog.generate_series
+ Output: generate_series.generate_series
+ Function Call: generate_series(1, 10)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(14 rows)
+
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t;
+ QUERY PLAN
+----------------------------------------------------------------------
+ Gather
+ Output: (SubPlan 1)
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(9 rows)
+
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t
+ limit 1;
+ QUERY PLAN
+-------------------------------------------------------------------
+ Limit
+ Output: ((SubPlan 1))
+ -> Seq Scan on public.tenk1 t
+ Output: (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(8 rows)
+
+explain (costs off, verbose) select t.unique1
+ from tenk1 t
+ where t.unique1 = (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+ QUERY PLAN
+-------------------------------------------------------------
+ Seq Scan on public.tenk1 t
+ Output: t.unique1
+ Filter: (t.unique1 = (SubPlan 1))
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(7 rows)
+
+explain (costs off, verbose) select *
+ from tenk1 t
+ order by (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+ QUERY PLAN
+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Sort
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, ((SubPlan 1))
+ Sort Key: ((SubPlan 1))
+ -> Gather
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, (SubPlan 1)
+ Workers Planned: 4
+ -> Parallel Seq Scan on public.tenk1 t
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(12 rows)
+
+-- test subplan in join/lateral join
+explain (costs off, verbose, timing off) select t.unique1, l.*
+ from tenk1 t
+ join lateral (
+ select (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1 offset 0)
+ ) l on true;
+ QUERY PLAN
+----------------------------------------------------------------------
+ Gather
+ Output: t.unique1, (SubPlan 1)
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(9 rows)
+
-- this is not parallel-safe due to use of random() within SubLink's testexpr:
explain (costs off)
select * from tenk1 where (unique1 + random())::integer not in
diff --git a/src/test/regress/sql/select_parallel.sql b/src/test/regress/sql/select_parallel.sql
index 62fb68c7a0..21c2f1c742 100644
--- a/src/test/regress/sql/select_parallel.sql
+++ b/src/test/regress/sql/select_parallel.sql
@@ -111,6 +111,31 @@ explain (costs off)
(select hundred, thousand from tenk2 where thousand > 100);
select count(*) from tenk1 where (two, four) not in
(select hundred, thousand from tenk2 where thousand > 100);
+-- test parallel plans for queries containing correlated subplans
+-- where the subplan only needs params available from the current
+-- worker's scan.
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t, generate_series(1, 10);
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t;
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t
+ limit 1;
+explain (costs off, verbose) select t.unique1
+ from tenk1 t
+ where t.unique1 = (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+explain (costs off, verbose) select *
+ from tenk1 t
+ order by (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+-- test subplan in join/lateral join
+explain (costs off, verbose, timing off) select t.unique1, l.*
+ from tenk1 t
+ join lateral (
+ select (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1 offset 0)
+ ) l on true;
-- this is not parallel-safe due to use of random() within SubLink's testexpr:
explain (costs off)
select * from tenk1 where (unique1 + random())::integer not in
--
2.32.1 (Apple Git-133)
[application/octet-stream] v8-0002-Parallelize-correlated-subqueries.patch (30.4K, ../../CAAaqYe9V+2TTwGDUu12NLu1tDf0SnwvYnTOrKz=uTh6cLx3McA@mail.gmail.com/3-v8-0002-Parallelize-correlated-subqueries.patch)
download | inline diff:
From 3ddb8ea8ba949965b01cf8fd741f84d60f891945 Mon Sep 17 00:00:00 2001
From: jcoleman <[email protected]>
Date: Wed, 18 Jan 2023 20:43:26 -0500
Subject: [PATCH v8 2/2] Parallelize correlated subqueries
When params are provided at the current query level (i.e., are generated
within a single worker and not shared across workers) we can safely
execute these in parallel.
Alternative approach using just relids subset check
---
doc/src/sgml/parallel.sgml | 3 +-
src/backend/optimizer/path/allpaths.c | 45 ++++--
src/backend/optimizer/path/joinpath.c | 10 +-
src/backend/optimizer/plan/planner.c | 8 ++
src/backend/optimizer/util/clauses.c | 74 ++++------
.../regress/expected/incremental_sort.out | 41 +++---
src/test/regress/expected/partition_prune.out | 104 +++++++-------
src/test/regress/expected/select_parallel.out | 128 ++++++++++--------
8 files changed, 221 insertions(+), 192 deletions(-)
diff --git a/doc/src/sgml/parallel.sgml b/doc/src/sgml/parallel.sgml
index 5acc9537d6..fd32572ec8 100644
--- a/doc/src/sgml/parallel.sgml
+++ b/doc/src/sgml/parallel.sgml
@@ -518,7 +518,8 @@ EXPLAIN SELECT * FROM pgbench_accounts WHERE filler LIKE '%x%';
<listitem>
<para>
- Plan nodes that reference a correlated <literal>SubPlan</literal>.
+ Plan nodes that reference a correlated <literal>SubPlan</literal> where
+ the result is shared between workers.
</para>
</listitem>
</itemizedlist>
diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index c2fc568dc8..0ff94690c3 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -3018,11 +3018,19 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
ListCell *lc;
double rows;
double *rowsp = NULL;
+ Relids required_outer = rel->lateral_relids;
/* If there are no partial paths, there's nothing to do here. */
if (rel->partial_pathlist == NIL)
return;
+ /*
+ * Delay gather path creation until the level in the join tree where all
+ * params used in a worker are generated within that worker.
+ */
+ if (!bms_is_subset(required_outer, rel->relids))
+ return;
+
/* Should we override the rel's rowcount estimate? */
if (override_rows)
rowsp = &rows;
@@ -3033,12 +3041,17 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
* of partial_pathlist because of the way add_partial_path works.
*/
cheapest_partial_path = linitial(rel->partial_pathlist);
- rows =
- cheapest_partial_path->rows * cheapest_partial_path->parallel_workers;
- simple_gather_path = (Path *)
- create_gather_path(root, rel, cheapest_partial_path, rel->reltarget,
- NULL, rowsp);
- add_path(rel, simple_gather_path);
+
+ /* We can't pass params to workers. */
+ if (bms_is_subset(PATH_REQ_OUTER(cheapest_partial_path), rel->relids))
+ {
+ rows =
+ cheapest_partial_path->rows * cheapest_partial_path->parallel_workers;
+ simple_gather_path = (Path *)
+ create_gather_path(root, rel, cheapest_partial_path, rel->reltarget,
+ required_outer, rowsp);
+ add_path(rel, simple_gather_path);
+ }
/*
* For each useful ordering, we can consider an order-preserving Gather
@@ -3052,9 +3065,13 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
if (subpath->pathkeys == NIL)
continue;
+ /* We can't pass params to workers. */
+ if (!bms_is_subset(PATH_REQ_OUTER(subpath), rel->relids))
+ continue;
+
rows = subpath->rows * subpath->parallel_workers;
path = create_gather_merge_path(root, rel, subpath, rel->reltarget,
- subpath->pathkeys, NULL, rowsp);
+ subpath->pathkeys, required_outer, rowsp);
add_path(rel, &path->path);
}
}
@@ -3156,11 +3173,19 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
double *rowsp = NULL;
List *useful_pathkeys_list = NIL;
Path *cheapest_partial_path = NULL;
+ Relids required_outer = rel->lateral_relids;
/* If there are no partial paths, there's nothing to do here. */
if (rel->partial_pathlist == NIL)
return;
+ /*
+ * Delay gather path creation until the level in the join tree where all
+ * params used in a worker are generated within that worker.
+ */
+ if (!bms_is_subset(required_outer, rel->relids))
+ return;
+
/* Should we override the rel's rowcount estimate? */
if (override_rows)
rowsp = &rows;
@@ -3217,6 +3242,10 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
(presorted_keys == 0 || !enable_incremental_sort))
continue;
+ /* We can't pass params to workers. */
+ if (!bms_is_subset(PATH_REQ_OUTER(subpath), rel->relids))
+ continue;
+
/*
* Consider regular sort for any path that's not presorted or if
* incremental sort is disabled. We've no need to consider both
@@ -3249,7 +3278,7 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
subpath,
rel->reltarget,
subpath->pathkeys,
- NULL,
+ required_outer,
rowsp);
add_path(rel, &path->path);
diff --git a/src/backend/optimizer/path/joinpath.c b/src/backend/optimizer/path/joinpath.c
index d345c0437a..644b90ad1e 100644
--- a/src/backend/optimizer/path/joinpath.c
+++ b/src/backend/optimizer/path/joinpath.c
@@ -1791,10 +1791,12 @@ match_unsorted_outer(PlannerInfo *root,
* Consider partial nestloop and mergejoin plan if outerrel has any
* partial path and the joinrel is parallel-safe. However, we can't
* handle JOIN_UNIQUE_OUTER, because the outer path will be partial, and
- * therefore we won't be able to properly guarantee uniqueness. Nor can
- * we handle joins needing lateral rels, since partial paths must not be
- * parameterized. Similarly, we can't handle JOIN_FULL and JOIN_RIGHT,
- * because they can produce false null extended rows.
+ * therefore we won't be able to properly guarantee uniqueness. Similarly,
+ * we can't handle JOIN_FULL and JOIN_RIGHT, because they can produce false
+ * null extended rows.
+ *
+ * While partial paths may now be parameterized so long as all of the params
+ * can be generated wholly within a worker we punt on supporting that here.
*/
if (joinrel->consider_parallel &&
save_jointype != JOIN_UNIQUE_OUTER &&
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index 05f44faf6e..fcde41d9f3 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -7344,11 +7344,16 @@ gather_grouping_paths(PlannerInfo *root, RelOptInfo *rel)
ListCell *lc;
Path *cheapest_partial_path;
+ /* By grouping time we shouldn't have any lateral dependencies. */
+ Assert(rel->lateral_relids == NULL);
+
/* Try Gather for unordered paths and Gather Merge for ordered ones. */
generate_useful_gather_paths(root, rel, true);
/* Try cheapest partial path + explicit Sort + Gather Merge. */
cheapest_partial_path = linitial(rel->partial_pathlist);
+ /* By grouping time we shouldn't have any lateral dependencies. */
+ Assert(PATH_REQ_OUTER(cheapest_partial_path) == NULL);
if (!pathkeys_contained_in(root->group_pathkeys,
cheapest_partial_path->pathkeys))
{
@@ -7400,6 +7405,9 @@ gather_grouping_paths(PlannerInfo *root, RelOptInfo *rel)
if (presorted_keys == 0)
continue;
+ /* By grouping time we shouldn't have any lateral dependencies. */
+ Assert(PATH_REQ_OUTER(path) == NULL);
+
path = (Path *) create_incremental_sort_path(root,
rel,
path,
diff --git a/src/backend/optimizer/util/clauses.c b/src/backend/optimizer/util/clauses.c
index aa584848cf..4a3a0489a8 100644
--- a/src/backend/optimizer/util/clauses.c
+++ b/src/backend/optimizer/util/clauses.c
@@ -89,7 +89,6 @@ typedef struct
{
char max_hazard; /* worst proparallel hazard found so far */
char max_interesting; /* worst proparallel hazard of interest */
- List *safe_param_ids; /* PARAM_EXEC Param IDs to treat as safe */
} max_parallel_hazard_context;
static bool contain_agg_clause_walker(Node *node, void *context);
@@ -618,7 +617,6 @@ max_parallel_hazard(Query *parse)
context.max_hazard = PROPARALLEL_SAFE;
context.max_interesting = PROPARALLEL_UNSAFE;
- context.safe_param_ids = NIL;
(void) max_parallel_hazard_walker((Node *) parse, &context);
return context.max_hazard;
}
@@ -629,43 +627,24 @@ max_parallel_hazard(Query *parse)
*
* root->glob->maxParallelHazard must previously have been set to the
* result of max_parallel_hazard() on the whole query.
+ *
+ * The caller is responsible for verifying that PARAM_EXEC Params are generated
+ * at the current plan level.
*/
bool
is_parallel_safe(PlannerInfo *root, Node *node)
{
max_parallel_hazard_context context;
- PlannerInfo *proot;
- ListCell *l;
/*
- * Even if the original querytree contained nothing unsafe, we need to
- * search the expression if we have generated any PARAM_EXEC Params while
- * planning, because those are parallel-restricted and there might be one
- * in this expression. But otherwise we don't need to look.
+ * If we've already checked the querytree don't burn cycles doing it again.
*/
- if (root->glob->maxParallelHazard == PROPARALLEL_SAFE &&
- root->glob->paramExecTypes == NIL)
+ if (root->glob->maxParallelHazard == PROPARALLEL_SAFE)
return true;
+
/* Else use max_parallel_hazard's search logic, but stop on RESTRICTED */
context.max_hazard = PROPARALLEL_SAFE;
context.max_interesting = PROPARALLEL_RESTRICTED;
- context.safe_param_ids = NIL;
-
- /*
- * The params that refer to the same or parent query level are considered
- * parallel-safe. The idea is that we compute such params at Gather or
- * Gather Merge node and pass their value to workers.
- */
- for (proot = root; proot != NULL; proot = proot->parent_root)
- {
- foreach(l, proot->init_plans)
- {
- SubPlan *initsubplan = (SubPlan *) lfirst(l);
-
- context.safe_param_ids = list_concat(context.safe_param_ids,
- initsubplan->setParam);
- }
- }
return !max_parallel_hazard_walker(node, &context);
}
@@ -775,39 +754,34 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context)
}
/*
- * Only parallel-safe SubPlans can be sent to workers. Within the
- * testexpr of the SubPlan, Params representing the output columns of the
- * subplan can be treated as parallel-safe, so temporarily add their IDs
- * to the safe_param_ids list while examining the testexpr.
+ * Only parallel-safe SubPlans can be sent to workers.
*/
else if (IsA(node, SubPlan))
{
SubPlan *subplan = (SubPlan *) node;
- List *save_safe_param_ids;
if (!subplan->parallel_safe &&
max_parallel_hazard_test(PROPARALLEL_RESTRICTED, context))
return true;
- save_safe_param_ids = context->safe_param_ids;
- context->safe_param_ids = list_concat_copy(context->safe_param_ids,
- subplan->paramIds);
+
if (max_parallel_hazard_walker(subplan->testexpr, context))
- return true; /* no need to restore safe_param_ids */
- list_free(context->safe_param_ids);
- context->safe_param_ids = save_safe_param_ids;
- /* we must also check args, but no special Param treatment there */
+ return true;
+
if (max_parallel_hazard_walker((Node *) subplan->args, context))
return true;
+
/* don't want to recurse normally, so we're done */
return false;
}
/*
- * We can't pass Params to workers at the moment either, so they are also
- * parallel-restricted, unless they are PARAM_EXTERN Params or are
- * PARAM_EXEC Params listed in safe_param_ids, meaning they could be
- * either generated within workers or can be computed by the leader and
- * then their value can be passed to workers.
+ * We can't pass all types of Params to workers at the moment either.
+ * PARAM_EXTERN Params are always allowed. PARAM_EXEC Params are parallel-
+ * safe when they can be computed by the leader and their value passed to
+ * workers or are generated within a worker. However we don't always know
+ * whether a param will be generated within a worker when we are parsing a
+ * querytree. In that case we leave it to the consumer to verify that the
+ * current plan level provides these params.
*/
else if (IsA(node, Param))
{
@@ -816,12 +790,12 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context)
if (param->paramkind == PARAM_EXTERN)
return false;
- if (param->paramkind != PARAM_EXEC ||
- !list_member_int(context->safe_param_ids, param->paramid))
- {
- if (max_parallel_hazard_test(PROPARALLEL_RESTRICTED, context))
- return true;
- }
+ if (param->paramkind == PARAM_EXEC)
+ return false;
+
+ if (max_parallel_hazard_test(PROPARALLEL_RESTRICTED, context))
+ return true;
+
return false; /* nothing to recurse to */
}
diff --git a/src/test/regress/expected/incremental_sort.out b/src/test/regress/expected/incremental_sort.out
index 0c3433f8e5..2b52b0ca3c 100644
--- a/src/test/regress/expected/incremental_sort.out
+++ b/src/test/regress/expected/incremental_sort.out
@@ -1597,20 +1597,21 @@ explain (costs off) select distinct
unique1,
(select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
from tenk1 t, generate_series(1, 1000);
- QUERY PLAN
----------------------------------------------------------------------------------
+ QUERY PLAN
+---------------------------------------------------------------------------------------
Unique
- -> Sort
- Sort Key: t.unique1, ((SubPlan 1))
- -> Gather
- Workers Planned: 2
- -> Nested Loop
- -> Parallel Index Only Scan using tenk1_unique1 on tenk1 t
- -> Function Scan on generate_series
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on tenk1
- Index Cond: (unique1 = t.unique1)
-(11 rows)
+ -> Gather Merge
+ Workers Planned: 2
+ -> Unique
+ -> Sort
+ Sort Key: t.unique1, ((SubPlan 1))
+ -> Nested Loop
+ -> Parallel Index Only Scan using tenk1_unique1 on tenk1 t
+ -> Function Scan on generate_series
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on tenk1
+ Index Cond: (unique1 = t.unique1)
+(12 rows)
explain (costs off) select
unique1,
@@ -1619,16 +1620,16 @@ from tenk1 t, generate_series(1, 1000)
order by 1, 2;
QUERY PLAN
---------------------------------------------------------------------------
- Sort
- Sort Key: t.unique1, ((SubPlan 1))
- -> Gather
- Workers Planned: 2
+ Gather Merge
+ Workers Planned: 2
+ -> Sort
+ Sort Key: t.unique1, ((SubPlan 1))
-> Nested Loop
-> Parallel Index Only Scan using tenk1_unique1 on tenk1 t
-> Function Scan on generate_series
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on tenk1
- Index Cond: (unique1 = t.unique1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on tenk1
+ Index Cond: (unique1 = t.unique1)
(10 rows)
-- Parallel sort but with expression not available until the upper rel.
diff --git a/src/test/regress/expected/partition_prune.out b/src/test/regress/expected/partition_prune.out
index 7555764c77..5c45f9c0a5 100644
--- a/src/test/regress/expected/partition_prune.out
+++ b/src/test/regress/expected/partition_prune.out
@@ -1284,60 +1284,64 @@ EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM part p(x) ORDER BY x;
--
-- pruning won't work for mc3p, because some keys are Params
explain (costs off) select * from mc2p t1, lateral (select count(*) from mc3p t2 where t2.a = t1.b and abs(t2.b) = 1 and t2.c = 1) s where t1.a = 1;
- QUERY PLAN
------------------------------------------------------------------------
- Nested Loop
- -> Append
- -> Seq Scan on mc2p1 t1_1
- Filter: (a = 1)
- -> Seq Scan on mc2p2 t1_2
- Filter: (a = 1)
- -> Seq Scan on mc2p_default t1_3
- Filter: (a = 1)
- -> Aggregate
- -> Append
- -> Seq Scan on mc3p0 t2_1
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p1 t2_2
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p2 t2_3
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p3 t2_4
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p4 t2_5
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p5 t2_6
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p6 t2_7
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p7 t2_8
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p_default t2_9
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
-(28 rows)
+ QUERY PLAN
+-----------------------------------------------------------------------------
+ Gather
+ Workers Planned: 2
+ -> Nested Loop
+ -> Parallel Append
+ -> Parallel Seq Scan on mc2p1 t1_1
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p2 t1_2
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p_default t1_3
+ Filter: (a = 1)
+ -> Aggregate
+ -> Append
+ -> Seq Scan on mc3p0 t2_1
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p1 t2_2
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p2 t2_3
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p3 t2_4
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p4 t2_5
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p5 t2_6
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p6 t2_7
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p7 t2_8
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p_default t2_9
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+(30 rows)
-- pruning should work fine, because values for a prefix of keys (a, b) are
-- available
explain (costs off) select * from mc2p t1, lateral (select count(*) from mc3p t2 where t2.c = t1.b and abs(t2.b) = 1 and t2.a = 1) s where t1.a = 1;
- QUERY PLAN
------------------------------------------------------------------------
- Nested Loop
- -> Append
- -> Seq Scan on mc2p1 t1_1
- Filter: (a = 1)
- -> Seq Scan on mc2p2 t1_2
- Filter: (a = 1)
- -> Seq Scan on mc2p_default t1_3
- Filter: (a = 1)
- -> Aggregate
- -> Append
- -> Seq Scan on mc3p0 t2_1
- Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p1 t2_2
- Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p_default t2_3
- Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
-(16 rows)
+ QUERY PLAN
+-----------------------------------------------------------------------------
+ Gather
+ Workers Planned: 2
+ -> Nested Loop
+ -> Parallel Append
+ -> Parallel Seq Scan on mc2p1 t1_1
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p2 t1_2
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p_default t1_3
+ Filter: (a = 1)
+ -> Aggregate
+ -> Append
+ -> Seq Scan on mc3p0 t2_1
+ Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p1 t2_2
+ Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p_default t2_3
+ Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
+(18 rows)
-- also here, because values for all keys are provided
explain (costs off) select * from mc2p t1, lateral (select count(*) from mc3p t2 where t2.a = 1 and abs(t2.b) = 1 and t2.c = 1) s where t1.a = 1;
diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
index 9b4d7dd44a..01443e2ffb 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -137,8 +137,8 @@ create table part_pa_test_p2 partition of part_pa_test for values from (0) to (m
explain (costs off)
select (select max((select pa1.b from part_pa_test pa1 where pa1.a = pa2.a)))
from part_pa_test pa2;
- QUERY PLAN
---------------------------------------------------------------
+ QUERY PLAN
+----------------------------------------------------------------
Aggregate
-> Gather
Workers Planned: 3
@@ -148,12 +148,14 @@ explain (costs off)
SubPlan 2
-> Result
SubPlan 1
- -> Append
- -> Seq Scan on part_pa_test_p1 pa1_1
- Filter: (a = pa2.a)
- -> Seq Scan on part_pa_test_p2 pa1_2
- Filter: (a = pa2.a)
-(14 rows)
+ -> Gather
+ Workers Planned: 3
+ -> Parallel Append
+ -> Parallel Seq Scan on part_pa_test_p1 pa1_1
+ Filter: (a = pa2.a)
+ -> Parallel Seq Scan on part_pa_test_p2 pa1_2
+ Filter: (a = pa2.a)
+(16 rows)
drop table part_pa_test;
-- test with leader participation disabled
@@ -320,19 +322,19 @@ explain (costs off, verbose) select
QUERY PLAN
----------------------------------------------------------------------------
Gather
- Output: (SubPlan 1)
+ Output: ((SubPlan 1))
Workers Planned: 4
-> Nested Loop
- Output: t.unique1
+ Output: (SubPlan 1)
-> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
Output: t.unique1
-> Function Scan on pg_catalog.generate_series
Output: generate_series.generate_series
Function Call: generate_series(1, 10)
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(14 rows)
explain (costs off, verbose) select
@@ -341,63 +343,69 @@ explain (costs off, verbose) select
QUERY PLAN
----------------------------------------------------------------------
Gather
- Output: (SubPlan 1)
+ Output: ((SubPlan 1))
Workers Planned: 4
-> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
- Output: t.unique1
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ Output: (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(9 rows)
explain (costs off, verbose) select
(select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
from tenk1 t
limit 1;
- QUERY PLAN
--------------------------------------------------------------------
+ QUERY PLAN
+----------------------------------------------------------------------------
Limit
Output: ((SubPlan 1))
- -> Seq Scan on public.tenk1 t
- Output: (SubPlan 1)
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
-(8 rows)
+ -> Gather
+ Output: ((SubPlan 1))
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(11 rows)
explain (costs off, verbose) select t.unique1
from tenk1 t
where t.unique1 = (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
- QUERY PLAN
--------------------------------------------------------------
- Seq Scan on public.tenk1 t
+ QUERY PLAN
+----------------------------------------------------------------------
+ Gather
Output: t.unique1
- Filter: (t.unique1 = (SubPlan 1))
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
-(7 rows)
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ Filter: (t.unique1 = (SubPlan 1))
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(10 rows)
explain (costs off, verbose) select *
from tenk1 t
order by (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
- QUERY PLAN
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
- Sort
+ QUERY PLAN
+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Gather Merge
Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, ((SubPlan 1))
- Sort Key: ((SubPlan 1))
- -> Gather
- Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, (SubPlan 1)
- Workers Planned: 4
+ Workers Planned: 4
+ -> Sort
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, ((SubPlan 1))
+ Sort Key: ((SubPlan 1))
-> Parallel Seq Scan on public.tenk1 t
- Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(12 rows)
-- test subplan in join/lateral join
@@ -409,14 +417,14 @@ explain (costs off, verbose, timing off) select t.unique1, l.*
QUERY PLAN
----------------------------------------------------------------------
Gather
- Output: t.unique1, (SubPlan 1)
+ Output: t.unique1, ((SubPlan 1))
Workers Planned: 4
-> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
- Output: t.unique1
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ Output: t.unique1, (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(9 rows)
-- this is not parallel-safe due to use of random() within SubLink's testexpr:
@@ -1322,8 +1330,10 @@ SELECT 1 FROM tenk1_vw_sec
-> Parallel Index Only Scan using tenk1_unique1 on tenk1
SubPlan 1
-> Aggregate
- -> Seq Scan on int4_tbl
- Filter: (f1 < tenk1_vw_sec.unique1)
-(9 rows)
+ -> Gather
+ Workers Planned: 1
+ -> Parallel Seq Scan on int4_tbl
+ Filter: (f1 < tenk1_vw_sec.unique1)
+(11 rows)
rollback;
--
2.32.1 (Apple Git-133)
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2023-02-06 16:40 Antonin Houska <[email protected]>
parent: James Coleman <[email protected]>
1 sibling, 1 reply; 22+ messages in thread
From: Antonin Houska @ 2023-02-06 16:40 UTC (permalink / raw)
To: James Coleman <[email protected]>; +Cc: Tomas Vondra <[email protected]>; vignesh C <[email protected]>; Andres Freund <[email protected]>; Robert Haas <[email protected]>; pgsql-hackers; Amit Kapila <[email protected]>; Tom Lane <[email protected]>
James Coleman <[email protected]> wrote:
> On Sat, Jan 21, 2023 at 10:07 PM James Coleman <[email protected]> wrote:
> > ...
> > While working through Tomas's comment about a conditional in the
> > max_parallel_hazard_waker being guaranteed true I realized that in the
> > current version of the patch the safe_param_ids tracking in
> > is_parallel_safe isn't actually needed any longer. That seemed
> > suspicious, and so I started digging, and I found out that in the
> > current approach all of the tests pass with only the changes in
> > clauses.c. I don't believe that the other changes aren't needed;
> > rather I believe there isn't yet a test case exercising them, but I
> > realize that means I can't prove they're needed. I spent some time
> > poking at this, but at least with my current level of imagination I
> > haven't stumbled across a query that would exercise these checks.
>
> I played with this a good bit more yesterday, I'm now a good bit more
> confident this is correct. I've cleaned up the patch; see attached for
> v7.
>
> Here's some of my thought process:
> The comments in src/include/nodes/pathnodes.h:2953 tell us that
> PARAM_EXEC params are used to pass values around from one plan node to
> another in the following ways:
> 1. Values down into subqueries (for outer references in subqueries)
> 2. Up out of subqueries (for the results of a subplan)
> 3. From a NestLoop plan node into its inner relation (when the inner
> scan is parameterized with values from the outer relation)
>
> Case (2) is already known to be safe (we currently add these params to
> safe_param_ids in max_parallel_hazard_walker when we encounter a
> SubPlan node).
>
> I also believe case (3) is already handled. We don't build partial
> paths for joins when joinrel->lateral_relids is non-empty, and join
> order calculations already require that parameterization here go the
> correct way (i.e., inner depends on outer rather than the other way
> around).
>
> That leaves us with only case (1) to consider in this patch. Another
> way of saying this is that this is really the only thing the
> safe_param_ids tracking is guarding against. For params passed down
> into subqueries we can further distinguish between init plans and
> "regular" subplans. We already know that params from init plans are
> safe (at the right level). So we're concerned here with a way to know
> if the params passed to subplans are safe. We already track required
> rels in ParamPathInfo, so it's fairly simple to do this test.
>
> Which this patch we do in fact now see (as expected) rels with
> non-empty lateral_relids showing up in generate_[useful_]gather_paths.
> And the partial paths can now have non-empty required outer rels. I'm
> not able to come up with a plan that would actually be caught by those
> checks; I theorize that because of the few places we actually call
> generate_[useful_]gather_paths we are in practice already excluding
> those, but for now I've left these as a conditional rather than an
> assertion because it seems like the kind of guard we'd want to ensure
> those methods are safe.
Maybe we can later (in separate patches) relax the restrictions imposed on
partial path creation a little bit, so that more parameterized partial paths
are created.
One particular case that should be rejected by your checks is a partial index
path, which can be parameterized, but I couldn't construct a query that makes
your checks fire. Maybe the reason is that a parameterized index path is
mostly used on the inner side of a NL join, however no partial path can be
used there. (The problem is that each worker evaluating the NL join would only
see a subset of the inner relation, which whould lead to incorrect results.)
So I'd also choose conditions rather than assert statements.
Following are my (minor) findings:
In generate_gather_paths() you added this test
/*
* Delay gather path creation until the level in the join tree where all
* params used in a worker are generated within that worker.
*/
if (!bms_is_subset(required_outer, rel->relids))
return;
but I'm not sure if required_outer can contain anything of rel->relids. How
about using bms_is_empty(required) outer, or even this?
if (required_outer)
return;
Similarly,
/* We can't pass params to workers. */
if (!bms_is_subset(PATH_REQ_OUTER(cheapest_partial_path), rel->relids))
might look like
if (!bms_is_empty(PATH_REQ_OUTER(cheapest_partial_path)))
or
if (PATH_REQ_OUTER(cheapest_partial_path))
In particular, build_index_paths() does the following when setting
outer_relids (which eventually becomes (path->param_info->ppi_req_outer):
/* Enforce convention that outer_relids is exactly NULL if empty */
if (bms_is_empty(outer_relids))
outer_relids = NULL;
Another question is whether in this call
simple_gather_path = (Path *)
create_gather_path(root, rel, cheapest_partial_path, rel->reltarget,
required_outer, rowsp);
required_outer should be passed to create_gather_path(). Shouldn't it rather
be PATH_REQ_OUTER(cheapest_partial_path) that you test just above? Again,
build_index_paths() initializes outer_relids this way
outer_relids = bms_copy(rel->lateral_relids);
but then it may add some more relations:
/* OK to include this clause */
index_clauses = lappend(index_clauses, iclause);
outer_relids = bms_add_members(outer_relids,
rinfo->clause_relids);
So I think that PATH_REQ_OUTER(cheapest_partial_path) in
generate_gather_paths() can eventually contain more relations than
required_outer, and therefore it's safer to check the first.
Similar comments might apply to generate_useful_gather_paths(). Here I also
suggest to move this test
/* We can't pass params to workers. */
if (!bms_is_subset(PATH_REQ_OUTER(subpath), rel->relids))
continue;
to the top of the loop because it's relatively cheap.
--
Antonin Houska
Web: https://www.cybertec-postgresql.com
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2023-02-08 13:38 James Coleman <[email protected]>
parent: Antonin Houska <[email protected]>
0 siblings, 1 reply; 22+ messages in thread
From: James Coleman @ 2023-02-08 13:38 UTC (permalink / raw)
To: Antonin Houska <[email protected]>; +Cc: Tomas Vondra <[email protected]>; vignesh C <[email protected]>; Andres Freund <[email protected]>; Robert Haas <[email protected]>; pgsql-hackers; Amit Kapila <[email protected]>; Tom Lane <[email protected]>
On Mon, Feb 6, 2023 at 11:39 AM Antonin Houska <[email protected]> wrote:
>
> James Coleman <[email protected]> wrote:
> > Which this patch we do in fact now see (as expected) rels with
> > non-empty lateral_relids showing up in generate_[useful_]gather_paths.
> > And the partial paths can now have non-empty required outer rels. I'm
> > not able to come up with a plan that would actually be caught by those
> > checks; I theorize that because of the few places we actually call
> > generate_[useful_]gather_paths we are in practice already excluding
> > those, but for now I've left these as a conditional rather than an
> > assertion because it seems like the kind of guard we'd want to ensure
> > those methods are safe.
>
> Maybe we can later (in separate patches) relax the restrictions imposed on
> partial path creation a little bit, so that more parameterized partial paths
> are created.
>
> One particular case that should be rejected by your checks is a partial index
> path, which can be parameterized, but I couldn't construct a query that makes
> your checks fire. Maybe the reason is that a parameterized index path is
> mostly used on the inner side of a NL join, however no partial path can be
> used there. (The problem is that each worker evaluating the NL join would only
> see a subset of the inner relation, which whould lead to incorrect results.)
>
> So I'd also choose conditions rather than assert statements.
Thanks for confirming.
>
> Following are my (minor) findings:
>
> In generate_gather_paths() you added this test
>
> /*
> * Delay gather path creation until the level in the join tree where all
> * params used in a worker are generated within that worker.
> */
> if (!bms_is_subset(required_outer, rel->relids))
> return;
>
> but I'm not sure if required_outer can contain anything of rel->relids. How
> about using bms_is_empty(required) outer, or even this?
>
> if (required_outer)
> return;
>
> Similarly,
>
> /* We can't pass params to workers. */
> if (!bms_is_subset(PATH_REQ_OUTER(cheapest_partial_path), rel->relids))
>
> might look like
>
> if (!bms_is_empty(PATH_REQ_OUTER(cheapest_partial_path)))
>
> or
>
> if (PATH_REQ_OUTER(cheapest_partial_path))
I'm not sure about this change. Deciding is difficult given the fact
that we don't seem to currently generate these paths, but I don't see
a reason why lateral_relids can't be present on the rel, and if so,
then we need to check to see if they're a subset of relids we can
satisfy rather than checking that they don't exist.
> In particular, build_index_paths() does the following when setting
> outer_relids (which eventually becomes (path->param_info->ppi_req_outer):
>
> /* Enforce convention that outer_relids is exactly NULL if empty */
> if (bms_is_empty(outer_relids))
> outer_relids = NULL;
>
>
> Another question is whether in this call
>
> simple_gather_path = (Path *)
> create_gather_path(root, rel, cheapest_partial_path, rel->reltarget,
> required_outer, rowsp);
>
> required_outer should be passed to create_gather_path(). Shouldn't it rather
> be PATH_REQ_OUTER(cheapest_partial_path) that you test just above? Again,
> build_index_paths() initializes outer_relids this way
>
> outer_relids = bms_copy(rel->lateral_relids);
>
> but then it may add some more relations:
>
> /* OK to include this clause */
> index_clauses = lappend(index_clauses, iclause);
> outer_relids = bms_add_members(outer_relids,
> rinfo->clause_relids);
>
> So I think that PATH_REQ_OUTER(cheapest_partial_path) in
> generate_gather_paths() can eventually contain more relations than
> required_outer, and therefore it's safer to check the first.
Yes, this is a good catch. Originally I didn't know about
PATH_REQ_OUTER, and I'd missed using it in these places.
>
> Similar comments might apply to generate_useful_gather_paths(). Here I also
> suggest to move this test
>
> /* We can't pass params to workers. */
> if (!bms_is_subset(PATH_REQ_OUTER(subpath), rel->relids))
> continue;
>
> to the top of the loop because it's relatively cheap.
Moved.
Attached is v9.
James Coleman
Attachments:
[application/octet-stream] v9-0001-Add-tests-before-change.patch (7.2K, ../../CAAaqYe8fF=qc2Z+ttmrBkCMhV976Sy83uS1w9CyXMV7=X7NSMg@mail.gmail.com/2-v9-0001-Add-tests-before-change.patch)
download | inline diff:
From d7a74b2f26edaf36d5098b4b45e19c437a4ebbdf Mon Sep 17 00:00:00 2001
From: jcoleman <[email protected]>
Date: Mon, 26 Sep 2022 20:30:23 -0400
Subject: [PATCH v9 1/2] Add tests before change
---
src/test/regress/expected/select_parallel.out | 108 ++++++++++++++++++
src/test/regress/sql/select_parallel.sql | 25 ++++
2 files changed, 133 insertions(+)
diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
index 91f74fe47a..9b4d7dd44a 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -311,6 +311,114 @@ select count(*) from tenk1 where (two, four) not in
10000
(1 row)
+-- test parallel plans for queries containing correlated subplans
+-- where the subplan only needs params available from the current
+-- worker's scan.
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t, generate_series(1, 10);
+ QUERY PLAN
+----------------------------------------------------------------------------
+ Gather
+ Output: (SubPlan 1)
+ Workers Planned: 4
+ -> Nested Loop
+ Output: t.unique1
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ -> Function Scan on pg_catalog.generate_series
+ Output: generate_series.generate_series
+ Function Call: generate_series(1, 10)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(14 rows)
+
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t;
+ QUERY PLAN
+----------------------------------------------------------------------
+ Gather
+ Output: (SubPlan 1)
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(9 rows)
+
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t
+ limit 1;
+ QUERY PLAN
+-------------------------------------------------------------------
+ Limit
+ Output: ((SubPlan 1))
+ -> Seq Scan on public.tenk1 t
+ Output: (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(8 rows)
+
+explain (costs off, verbose) select t.unique1
+ from tenk1 t
+ where t.unique1 = (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+ QUERY PLAN
+-------------------------------------------------------------
+ Seq Scan on public.tenk1 t
+ Output: t.unique1
+ Filter: (t.unique1 = (SubPlan 1))
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(7 rows)
+
+explain (costs off, verbose) select *
+ from tenk1 t
+ order by (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+ QUERY PLAN
+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Sort
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, ((SubPlan 1))
+ Sort Key: ((SubPlan 1))
+ -> Gather
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, (SubPlan 1)
+ Workers Planned: 4
+ -> Parallel Seq Scan on public.tenk1 t
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(12 rows)
+
+-- test subplan in join/lateral join
+explain (costs off, verbose, timing off) select t.unique1, l.*
+ from tenk1 t
+ join lateral (
+ select (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1 offset 0)
+ ) l on true;
+ QUERY PLAN
+----------------------------------------------------------------------
+ Gather
+ Output: t.unique1, (SubPlan 1)
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(9 rows)
+
-- this is not parallel-safe due to use of random() within SubLink's testexpr:
explain (costs off)
select * from tenk1 where (unique1 + random())::integer not in
diff --git a/src/test/regress/sql/select_parallel.sql b/src/test/regress/sql/select_parallel.sql
index 62fb68c7a0..21c2f1c742 100644
--- a/src/test/regress/sql/select_parallel.sql
+++ b/src/test/regress/sql/select_parallel.sql
@@ -111,6 +111,31 @@ explain (costs off)
(select hundred, thousand from tenk2 where thousand > 100);
select count(*) from tenk1 where (two, four) not in
(select hundred, thousand from tenk2 where thousand > 100);
+-- test parallel plans for queries containing correlated subplans
+-- where the subplan only needs params available from the current
+-- worker's scan.
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t, generate_series(1, 10);
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t;
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t
+ limit 1;
+explain (costs off, verbose) select t.unique1
+ from tenk1 t
+ where t.unique1 = (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+explain (costs off, verbose) select *
+ from tenk1 t
+ order by (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+-- test subplan in join/lateral join
+explain (costs off, verbose, timing off) select t.unique1, l.*
+ from tenk1 t
+ join lateral (
+ select (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1 offset 0)
+ ) l on true;
-- this is not parallel-safe due to use of random() within SubLink's testexpr:
explain (costs off)
select * from tenk1 where (unique1 + random())::integer not in
--
2.32.1 (Apple Git-133)
[application/octet-stream] v9-0002-Parallelize-correlated-subqueries.patch (30.5K, ../../CAAaqYe8fF=qc2Z+ttmrBkCMhV976Sy83uS1w9CyXMV7=X7NSMg@mail.gmail.com/3-v9-0002-Parallelize-correlated-subqueries.patch)
download | inline diff:
From 8b36aa8fa43bfe6a32a1dbce0623ad9243f0b303 Mon Sep 17 00:00:00 2001
From: jcoleman <[email protected]>
Date: Wed, 18 Jan 2023 20:43:26 -0500
Subject: [PATCH v9 2/2] Parallelize correlated subqueries
When params are provided at the current query level (i.e., are generated
within a single worker and not shared across workers) we can safely
execute these in parallel.
Alternative approach using just relids subset check
---
doc/src/sgml/parallel.sgml | 3 +-
src/backend/optimizer/path/allpaths.c | 46 +++++--
src/backend/optimizer/path/joinpath.c | 10 +-
src/backend/optimizer/plan/planner.c | 8 ++
src/backend/optimizer/util/clauses.c | 74 ++++------
.../regress/expected/incremental_sort.out | 41 +++---
src/test/regress/expected/partition_prune.out | 104 +++++++-------
src/test/regress/expected/select_parallel.out | 128 ++++++++++--------
8 files changed, 222 insertions(+), 192 deletions(-)
diff --git a/doc/src/sgml/parallel.sgml b/doc/src/sgml/parallel.sgml
index 5acc9537d6..fd32572ec8 100644
--- a/doc/src/sgml/parallel.sgml
+++ b/doc/src/sgml/parallel.sgml
@@ -518,7 +518,8 @@ EXPLAIN SELECT * FROM pgbench_accounts WHERE filler LIKE '%x%';
<listitem>
<para>
- Plan nodes that reference a correlated <literal>SubPlan</literal>.
+ Plan nodes that reference a correlated <literal>SubPlan</literal> where
+ the result is shared between workers.
</para>
</listitem>
</itemizedlist>
diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index c2fc568dc8..a579d4c092 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -3018,11 +3018,19 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
ListCell *lc;
double rows;
double *rowsp = NULL;
+ Relids required_outer = rel->lateral_relids;
/* If there are no partial paths, there's nothing to do here. */
if (rel->partial_pathlist == NIL)
return;
+ /*
+ * Delay gather path creation until the level in the join tree where all
+ * params used in a worker are generated within that worker.
+ */
+ if (!bms_is_subset(required_outer, rel->relids))
+ return;
+
/* Should we override the rel's rowcount estimate? */
if (override_rows)
rowsp = &rows;
@@ -3033,12 +3041,17 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
* of partial_pathlist because of the way add_partial_path works.
*/
cheapest_partial_path = linitial(rel->partial_pathlist);
- rows =
- cheapest_partial_path->rows * cheapest_partial_path->parallel_workers;
- simple_gather_path = (Path *)
- create_gather_path(root, rel, cheapest_partial_path, rel->reltarget,
- NULL, rowsp);
- add_path(rel, simple_gather_path);
+
+ /* We can't pass params to workers. */
+ if (bms_is_subset(PATH_REQ_OUTER(cheapest_partial_path), rel->relids))
+ {
+ rows =
+ cheapest_partial_path->rows * cheapest_partial_path->parallel_workers;
+ simple_gather_path = (Path *)
+ create_gather_path(root, rel, cheapest_partial_path, rel->reltarget,
+ PATH_REQ_OUTER(cheapest_partial_path), rowsp);
+ add_path(rel, simple_gather_path);
+ }
/*
* For each useful ordering, we can consider an order-preserving Gather
@@ -3052,9 +3065,14 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
if (subpath->pathkeys == NIL)
continue;
+ /* We can't pass params to workers. */
+ if (!bms_is_subset(PATH_REQ_OUTER(subpath), rel->relids))
+ continue;
+
rows = subpath->rows * subpath->parallel_workers;
path = create_gather_merge_path(root, rel, subpath, rel->reltarget,
- subpath->pathkeys, NULL, rowsp);
+ subpath->pathkeys,
+ PATH_REQ_OUTER(subpath), rowsp);
add_path(rel, &path->path);
}
}
@@ -3156,11 +3174,19 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
double *rowsp = NULL;
List *useful_pathkeys_list = NIL;
Path *cheapest_partial_path = NULL;
+ Relids required_outer = rel->lateral_relids;
/* If there are no partial paths, there's nothing to do here. */
if (rel->partial_pathlist == NIL)
return;
+ /*
+ * Delay gather path creation until the level in the join tree where all
+ * params used in a worker are generated within that worker.
+ */
+ if (!bms_is_subset(required_outer, rel->relids))
+ return;
+
/* Should we override the rel's rowcount estimate? */
if (override_rows)
rowsp = &rows;
@@ -3190,6 +3216,10 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
Path *subpath = (Path *) lfirst(lc2);
GatherMergePath *path;
+ /* We can't pass params to workers. */
+ if (!bms_is_subset(PATH_REQ_OUTER(subpath), rel->relids))
+ continue;
+
is_sorted = pathkeys_count_contained_in(useful_pathkeys,
subpath->pathkeys,
&presorted_keys);
@@ -3249,7 +3279,7 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
subpath,
rel->reltarget,
subpath->pathkeys,
- NULL,
+ PATH_REQ_OUTER(subpath),
rowsp);
add_path(rel, &path->path);
diff --git a/src/backend/optimizer/path/joinpath.c b/src/backend/optimizer/path/joinpath.c
index d345c0437a..644b90ad1e 100644
--- a/src/backend/optimizer/path/joinpath.c
+++ b/src/backend/optimizer/path/joinpath.c
@@ -1791,10 +1791,12 @@ match_unsorted_outer(PlannerInfo *root,
* Consider partial nestloop and mergejoin plan if outerrel has any
* partial path and the joinrel is parallel-safe. However, we can't
* handle JOIN_UNIQUE_OUTER, because the outer path will be partial, and
- * therefore we won't be able to properly guarantee uniqueness. Nor can
- * we handle joins needing lateral rels, since partial paths must not be
- * parameterized. Similarly, we can't handle JOIN_FULL and JOIN_RIGHT,
- * because they can produce false null extended rows.
+ * therefore we won't be able to properly guarantee uniqueness. Similarly,
+ * we can't handle JOIN_FULL and JOIN_RIGHT, because they can produce false
+ * null extended rows.
+ *
+ * While partial paths may now be parameterized so long as all of the params
+ * can be generated wholly within a worker we punt on supporting that here.
*/
if (joinrel->consider_parallel &&
save_jointype != JOIN_UNIQUE_OUTER &&
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index 05f44faf6e..fcde41d9f3 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -7344,11 +7344,16 @@ gather_grouping_paths(PlannerInfo *root, RelOptInfo *rel)
ListCell *lc;
Path *cheapest_partial_path;
+ /* By grouping time we shouldn't have any lateral dependencies. */
+ Assert(rel->lateral_relids == NULL);
+
/* Try Gather for unordered paths and Gather Merge for ordered ones. */
generate_useful_gather_paths(root, rel, true);
/* Try cheapest partial path + explicit Sort + Gather Merge. */
cheapest_partial_path = linitial(rel->partial_pathlist);
+ /* By grouping time we shouldn't have any lateral dependencies. */
+ Assert(PATH_REQ_OUTER(cheapest_partial_path) == NULL);
if (!pathkeys_contained_in(root->group_pathkeys,
cheapest_partial_path->pathkeys))
{
@@ -7400,6 +7405,9 @@ gather_grouping_paths(PlannerInfo *root, RelOptInfo *rel)
if (presorted_keys == 0)
continue;
+ /* By grouping time we shouldn't have any lateral dependencies. */
+ Assert(PATH_REQ_OUTER(path) == NULL);
+
path = (Path *) create_incremental_sort_path(root,
rel,
path,
diff --git a/src/backend/optimizer/util/clauses.c b/src/backend/optimizer/util/clauses.c
index aa584848cf..4a3a0489a8 100644
--- a/src/backend/optimizer/util/clauses.c
+++ b/src/backend/optimizer/util/clauses.c
@@ -89,7 +89,6 @@ typedef struct
{
char max_hazard; /* worst proparallel hazard found so far */
char max_interesting; /* worst proparallel hazard of interest */
- List *safe_param_ids; /* PARAM_EXEC Param IDs to treat as safe */
} max_parallel_hazard_context;
static bool contain_agg_clause_walker(Node *node, void *context);
@@ -618,7 +617,6 @@ max_parallel_hazard(Query *parse)
context.max_hazard = PROPARALLEL_SAFE;
context.max_interesting = PROPARALLEL_UNSAFE;
- context.safe_param_ids = NIL;
(void) max_parallel_hazard_walker((Node *) parse, &context);
return context.max_hazard;
}
@@ -629,43 +627,24 @@ max_parallel_hazard(Query *parse)
*
* root->glob->maxParallelHazard must previously have been set to the
* result of max_parallel_hazard() on the whole query.
+ *
+ * The caller is responsible for verifying that PARAM_EXEC Params are generated
+ * at the current plan level.
*/
bool
is_parallel_safe(PlannerInfo *root, Node *node)
{
max_parallel_hazard_context context;
- PlannerInfo *proot;
- ListCell *l;
/*
- * Even if the original querytree contained nothing unsafe, we need to
- * search the expression if we have generated any PARAM_EXEC Params while
- * planning, because those are parallel-restricted and there might be one
- * in this expression. But otherwise we don't need to look.
+ * If we've already checked the querytree don't burn cycles doing it again.
*/
- if (root->glob->maxParallelHazard == PROPARALLEL_SAFE &&
- root->glob->paramExecTypes == NIL)
+ if (root->glob->maxParallelHazard == PROPARALLEL_SAFE)
return true;
+
/* Else use max_parallel_hazard's search logic, but stop on RESTRICTED */
context.max_hazard = PROPARALLEL_SAFE;
context.max_interesting = PROPARALLEL_RESTRICTED;
- context.safe_param_ids = NIL;
-
- /*
- * The params that refer to the same or parent query level are considered
- * parallel-safe. The idea is that we compute such params at Gather or
- * Gather Merge node and pass their value to workers.
- */
- for (proot = root; proot != NULL; proot = proot->parent_root)
- {
- foreach(l, proot->init_plans)
- {
- SubPlan *initsubplan = (SubPlan *) lfirst(l);
-
- context.safe_param_ids = list_concat(context.safe_param_ids,
- initsubplan->setParam);
- }
- }
return !max_parallel_hazard_walker(node, &context);
}
@@ -775,39 +754,34 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context)
}
/*
- * Only parallel-safe SubPlans can be sent to workers. Within the
- * testexpr of the SubPlan, Params representing the output columns of the
- * subplan can be treated as parallel-safe, so temporarily add their IDs
- * to the safe_param_ids list while examining the testexpr.
+ * Only parallel-safe SubPlans can be sent to workers.
*/
else if (IsA(node, SubPlan))
{
SubPlan *subplan = (SubPlan *) node;
- List *save_safe_param_ids;
if (!subplan->parallel_safe &&
max_parallel_hazard_test(PROPARALLEL_RESTRICTED, context))
return true;
- save_safe_param_ids = context->safe_param_ids;
- context->safe_param_ids = list_concat_copy(context->safe_param_ids,
- subplan->paramIds);
+
if (max_parallel_hazard_walker(subplan->testexpr, context))
- return true; /* no need to restore safe_param_ids */
- list_free(context->safe_param_ids);
- context->safe_param_ids = save_safe_param_ids;
- /* we must also check args, but no special Param treatment there */
+ return true;
+
if (max_parallel_hazard_walker((Node *) subplan->args, context))
return true;
+
/* don't want to recurse normally, so we're done */
return false;
}
/*
- * We can't pass Params to workers at the moment either, so they are also
- * parallel-restricted, unless they are PARAM_EXTERN Params or are
- * PARAM_EXEC Params listed in safe_param_ids, meaning they could be
- * either generated within workers or can be computed by the leader and
- * then their value can be passed to workers.
+ * We can't pass all types of Params to workers at the moment either.
+ * PARAM_EXTERN Params are always allowed. PARAM_EXEC Params are parallel-
+ * safe when they can be computed by the leader and their value passed to
+ * workers or are generated within a worker. However we don't always know
+ * whether a param will be generated within a worker when we are parsing a
+ * querytree. In that case we leave it to the consumer to verify that the
+ * current plan level provides these params.
*/
else if (IsA(node, Param))
{
@@ -816,12 +790,12 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context)
if (param->paramkind == PARAM_EXTERN)
return false;
- if (param->paramkind != PARAM_EXEC ||
- !list_member_int(context->safe_param_ids, param->paramid))
- {
- if (max_parallel_hazard_test(PROPARALLEL_RESTRICTED, context))
- return true;
- }
+ if (param->paramkind == PARAM_EXEC)
+ return false;
+
+ if (max_parallel_hazard_test(PROPARALLEL_RESTRICTED, context))
+ return true;
+
return false; /* nothing to recurse to */
}
diff --git a/src/test/regress/expected/incremental_sort.out b/src/test/regress/expected/incremental_sort.out
index 0c3433f8e5..2b52b0ca3c 100644
--- a/src/test/regress/expected/incremental_sort.out
+++ b/src/test/regress/expected/incremental_sort.out
@@ -1597,20 +1597,21 @@ explain (costs off) select distinct
unique1,
(select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
from tenk1 t, generate_series(1, 1000);
- QUERY PLAN
----------------------------------------------------------------------------------
+ QUERY PLAN
+---------------------------------------------------------------------------------------
Unique
- -> Sort
- Sort Key: t.unique1, ((SubPlan 1))
- -> Gather
- Workers Planned: 2
- -> Nested Loop
- -> Parallel Index Only Scan using tenk1_unique1 on tenk1 t
- -> Function Scan on generate_series
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on tenk1
- Index Cond: (unique1 = t.unique1)
-(11 rows)
+ -> Gather Merge
+ Workers Planned: 2
+ -> Unique
+ -> Sort
+ Sort Key: t.unique1, ((SubPlan 1))
+ -> Nested Loop
+ -> Parallel Index Only Scan using tenk1_unique1 on tenk1 t
+ -> Function Scan on generate_series
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on tenk1
+ Index Cond: (unique1 = t.unique1)
+(12 rows)
explain (costs off) select
unique1,
@@ -1619,16 +1620,16 @@ from tenk1 t, generate_series(1, 1000)
order by 1, 2;
QUERY PLAN
---------------------------------------------------------------------------
- Sort
- Sort Key: t.unique1, ((SubPlan 1))
- -> Gather
- Workers Planned: 2
+ Gather Merge
+ Workers Planned: 2
+ -> Sort
+ Sort Key: t.unique1, ((SubPlan 1))
-> Nested Loop
-> Parallel Index Only Scan using tenk1_unique1 on tenk1 t
-> Function Scan on generate_series
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on tenk1
- Index Cond: (unique1 = t.unique1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on tenk1
+ Index Cond: (unique1 = t.unique1)
(10 rows)
-- Parallel sort but with expression not available until the upper rel.
diff --git a/src/test/regress/expected/partition_prune.out b/src/test/regress/expected/partition_prune.out
index 7555764c77..5c45f9c0a5 100644
--- a/src/test/regress/expected/partition_prune.out
+++ b/src/test/regress/expected/partition_prune.out
@@ -1284,60 +1284,64 @@ EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM part p(x) ORDER BY x;
--
-- pruning won't work for mc3p, because some keys are Params
explain (costs off) select * from mc2p t1, lateral (select count(*) from mc3p t2 where t2.a = t1.b and abs(t2.b) = 1 and t2.c = 1) s where t1.a = 1;
- QUERY PLAN
------------------------------------------------------------------------
- Nested Loop
- -> Append
- -> Seq Scan on mc2p1 t1_1
- Filter: (a = 1)
- -> Seq Scan on mc2p2 t1_2
- Filter: (a = 1)
- -> Seq Scan on mc2p_default t1_3
- Filter: (a = 1)
- -> Aggregate
- -> Append
- -> Seq Scan on mc3p0 t2_1
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p1 t2_2
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p2 t2_3
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p3 t2_4
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p4 t2_5
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p5 t2_6
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p6 t2_7
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p7 t2_8
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p_default t2_9
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
-(28 rows)
+ QUERY PLAN
+-----------------------------------------------------------------------------
+ Gather
+ Workers Planned: 2
+ -> Nested Loop
+ -> Parallel Append
+ -> Parallel Seq Scan on mc2p1 t1_1
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p2 t1_2
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p_default t1_3
+ Filter: (a = 1)
+ -> Aggregate
+ -> Append
+ -> Seq Scan on mc3p0 t2_1
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p1 t2_2
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p2 t2_3
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p3 t2_4
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p4 t2_5
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p5 t2_6
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p6 t2_7
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p7 t2_8
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p_default t2_9
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+(30 rows)
-- pruning should work fine, because values for a prefix of keys (a, b) are
-- available
explain (costs off) select * from mc2p t1, lateral (select count(*) from mc3p t2 where t2.c = t1.b and abs(t2.b) = 1 and t2.a = 1) s where t1.a = 1;
- QUERY PLAN
------------------------------------------------------------------------
- Nested Loop
- -> Append
- -> Seq Scan on mc2p1 t1_1
- Filter: (a = 1)
- -> Seq Scan on mc2p2 t1_2
- Filter: (a = 1)
- -> Seq Scan on mc2p_default t1_3
- Filter: (a = 1)
- -> Aggregate
- -> Append
- -> Seq Scan on mc3p0 t2_1
- Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p1 t2_2
- Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p_default t2_3
- Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
-(16 rows)
+ QUERY PLAN
+-----------------------------------------------------------------------------
+ Gather
+ Workers Planned: 2
+ -> Nested Loop
+ -> Parallel Append
+ -> Parallel Seq Scan on mc2p1 t1_1
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p2 t1_2
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p_default t1_3
+ Filter: (a = 1)
+ -> Aggregate
+ -> Append
+ -> Seq Scan on mc3p0 t2_1
+ Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p1 t2_2
+ Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p_default t2_3
+ Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
+(18 rows)
-- also here, because values for all keys are provided
explain (costs off) select * from mc2p t1, lateral (select count(*) from mc3p t2 where t2.a = 1 and abs(t2.b) = 1 and t2.c = 1) s where t1.a = 1;
diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
index 9b4d7dd44a..01443e2ffb 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -137,8 +137,8 @@ create table part_pa_test_p2 partition of part_pa_test for values from (0) to (m
explain (costs off)
select (select max((select pa1.b from part_pa_test pa1 where pa1.a = pa2.a)))
from part_pa_test pa2;
- QUERY PLAN
---------------------------------------------------------------
+ QUERY PLAN
+----------------------------------------------------------------
Aggregate
-> Gather
Workers Planned: 3
@@ -148,12 +148,14 @@ explain (costs off)
SubPlan 2
-> Result
SubPlan 1
- -> Append
- -> Seq Scan on part_pa_test_p1 pa1_1
- Filter: (a = pa2.a)
- -> Seq Scan on part_pa_test_p2 pa1_2
- Filter: (a = pa2.a)
-(14 rows)
+ -> Gather
+ Workers Planned: 3
+ -> Parallel Append
+ -> Parallel Seq Scan on part_pa_test_p1 pa1_1
+ Filter: (a = pa2.a)
+ -> Parallel Seq Scan on part_pa_test_p2 pa1_2
+ Filter: (a = pa2.a)
+(16 rows)
drop table part_pa_test;
-- test with leader participation disabled
@@ -320,19 +322,19 @@ explain (costs off, verbose) select
QUERY PLAN
----------------------------------------------------------------------------
Gather
- Output: (SubPlan 1)
+ Output: ((SubPlan 1))
Workers Planned: 4
-> Nested Loop
- Output: t.unique1
+ Output: (SubPlan 1)
-> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
Output: t.unique1
-> Function Scan on pg_catalog.generate_series
Output: generate_series.generate_series
Function Call: generate_series(1, 10)
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(14 rows)
explain (costs off, verbose) select
@@ -341,63 +343,69 @@ explain (costs off, verbose) select
QUERY PLAN
----------------------------------------------------------------------
Gather
- Output: (SubPlan 1)
+ Output: ((SubPlan 1))
Workers Planned: 4
-> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
- Output: t.unique1
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ Output: (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(9 rows)
explain (costs off, verbose) select
(select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
from tenk1 t
limit 1;
- QUERY PLAN
--------------------------------------------------------------------
+ QUERY PLAN
+----------------------------------------------------------------------------
Limit
Output: ((SubPlan 1))
- -> Seq Scan on public.tenk1 t
- Output: (SubPlan 1)
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
-(8 rows)
+ -> Gather
+ Output: ((SubPlan 1))
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(11 rows)
explain (costs off, verbose) select t.unique1
from tenk1 t
where t.unique1 = (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
- QUERY PLAN
--------------------------------------------------------------
- Seq Scan on public.tenk1 t
+ QUERY PLAN
+----------------------------------------------------------------------
+ Gather
Output: t.unique1
- Filter: (t.unique1 = (SubPlan 1))
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
-(7 rows)
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ Filter: (t.unique1 = (SubPlan 1))
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(10 rows)
explain (costs off, verbose) select *
from tenk1 t
order by (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
- QUERY PLAN
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
- Sort
+ QUERY PLAN
+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Gather Merge
Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, ((SubPlan 1))
- Sort Key: ((SubPlan 1))
- -> Gather
- Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, (SubPlan 1)
- Workers Planned: 4
+ Workers Planned: 4
+ -> Sort
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, ((SubPlan 1))
+ Sort Key: ((SubPlan 1))
-> Parallel Seq Scan on public.tenk1 t
- Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(12 rows)
-- test subplan in join/lateral join
@@ -409,14 +417,14 @@ explain (costs off, verbose, timing off) select t.unique1, l.*
QUERY PLAN
----------------------------------------------------------------------
Gather
- Output: t.unique1, (SubPlan 1)
+ Output: t.unique1, ((SubPlan 1))
Workers Planned: 4
-> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
- Output: t.unique1
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ Output: t.unique1, (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(9 rows)
-- this is not parallel-safe due to use of random() within SubLink's testexpr:
@@ -1322,8 +1330,10 @@ SELECT 1 FROM tenk1_vw_sec
-> Parallel Index Only Scan using tenk1_unique1 on tenk1
SubPlan 1
-> Aggregate
- -> Seq Scan on int4_tbl
- Filter: (f1 < tenk1_vw_sec.unique1)
-(9 rows)
+ -> Gather
+ Workers Planned: 1
+ -> Parallel Seq Scan on int4_tbl
+ Filter: (f1 < tenk1_vw_sec.unique1)
+(11 rows)
rollback;
--
2.32.1 (Apple Git-133)
^ permalink raw reply [nested|flat] 22+ messages in thread
* [PATCH] f!fixes for LZ4
@ 2023-02-24 20:07 Justin Pryzby <[email protected]>
0 siblings, 0 replies; 22+ messages in thread
From: Justin Pryzby @ 2023-02-24 20:07 UTC (permalink / raw)
---
doc/src/sgml/ref/pg_dump.sgml | 4 ++--
src/bin/pg_dump/compress_gzip.c | 7 -------
src/bin/pg_dump/compress_io.c | 2 +-
src/bin/pg_dump/compress_io.h | 4 ++--
src/bin/pg_dump/compress_lz4.c | 4 ++--
src/bin/pg_dump/pg_backup_archiver.c | 4 ++--
src/bin/pg_dump/pg_dump.c | 2 --
src/bin/pg_dump/t/002_pg_dump.pl | 6 +++---
8 files changed, 12 insertions(+), 21 deletions(-)
diff --git a/doc/src/sgml/ref/pg_dump.sgml b/doc/src/sgml/ref/pg_dump.sgml
index 49d218905fb..6fbe49f7ede 100644
--- a/doc/src/sgml/ref/pg_dump.sgml
+++ b/doc/src/sgml/ref/pg_dump.sgml
@@ -331,7 +331,7 @@ PostgreSQL documentation
can read. A directory format archive can be manipulated with
standard Unix tools; for example, files in an uncompressed archive
can be compressed with the <application>gzip</application> or
- <application>lz4</application>tool.
+ <application>lz4</application> tools.
This format is compressed by default using <literal>gzip</literal>
and also supports parallel dumps.
</para>
@@ -655,7 +655,7 @@ PostgreSQL documentation
<para>
Specify the compression method and/or the compression level to use.
The compression method can be set to <literal>gzip</literal> or
- <literal>lz4</literal> or <literal>none</literal> for no compression.
+ <literal>lz4</literal>, or <literal>none</literal> for no compression.
A compression detail string can optionally be specified. If the
detail string is an integer, it specifies the compression level.
Otherwise, it should be a comma-separated list of items, each of the
diff --git a/src/bin/pg_dump/compress_gzip.c b/src/bin/pg_dump/compress_gzip.c
index 0af65afeb4e..af68fd27a12 100644
--- a/src/bin/pg_dump/compress_gzip.c
+++ b/src/bin/pg_dump/compress_gzip.c
@@ -123,13 +123,6 @@ WriteDataToArchiveGzip(ArchiveHandle *AH, CompressorState *cs,
gzipcs->outbuf = pg_malloc(ZLIB_OUT_SIZE + 1);
gzipcs->outsize = ZLIB_OUT_SIZE;
- /*
- * A level of zero simply copies the input one block at the time. This
- * is probably not what the user wanted when calling this interface.
- */
- if (cs->compression_spec.level == 0)
- pg_fatal("requested to compress the archive yet no level was specified");
-
if (deflateInit(zp, cs->compression_spec.level) != Z_OK)
pg_fatal("could not initialize compression library: %s", zp->msg);
diff --git a/src/bin/pg_dump/compress_io.c b/src/bin/pg_dump/compress_io.c
index ce06f1eac9c..9239dbb2755 100644
--- a/src/bin/pg_dump/compress_io.c
+++ b/src/bin/pg_dump/compress_io.c
@@ -83,7 +83,7 @@
* used by the caller in an error message.
*/
char *
-supports_compression(const pg_compress_specification compression_spec)
+pgdump_supports_compression(const pg_compress_specification compression_spec)
{
const pg_compress_algorithm algorithm = compression_spec.algorithm;
bool supported = false;
diff --git a/src/bin/pg_dump/compress_io.h b/src/bin/pg_dump/compress_io.h
index bbde2693915..46815fa2ebe 100644
--- a/src/bin/pg_dump/compress_io.h
+++ b/src/bin/pg_dump/compress_io.h
@@ -21,7 +21,7 @@
#define ZLIB_OUT_SIZE 4096
#define ZLIB_IN_SIZE 4096
-extern char *supports_compression(const pg_compress_specification compression_spec);
+extern char *pgdump_supports_compression(const pg_compress_specification compression_spec);
/*
* Prototype for callback function used in writeData()
@@ -172,7 +172,7 @@ struct CompressFileHandle
extern CompressFileHandle *InitCompressFileHandle(const pg_compress_specification compression_spec);
/*
- * Initialize a compress file stream. Deffer the compression algorithm
+ * Initialize a compress file stream. Infer the compression algorithm
* from 'path', either by examining its suffix or by appending the supported
* suffixes in 'path'.
*/
diff --git a/src/bin/pg_dump/compress_lz4.c b/src/bin/pg_dump/compress_lz4.c
index fe1014e6e77..63e794cdc68 100644
--- a/src/bin/pg_dump/compress_lz4.c
+++ b/src/bin/pg_dump/compress_lz4.c
@@ -161,8 +161,8 @@ typedef struct LZ4File
} LZ4File;
/*
- * LZ4 equivalent to feof() or gzeof(). The end of file is reached if there
- * is no decompressed output in the overflow buffer and the end of the file
+ * LZ4 equivalent to feof() or gzeof(). Return true iff there is no
+ * decompressed output in the overflow buffer and the end of the backing file
* is reached.
*/
static int
diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c
index 61ebb8fe85d..2063d6f239d 100644
--- a/src/bin/pg_dump/pg_backup_archiver.c
+++ b/src/bin/pg_dump/pg_backup_archiver.c
@@ -388,7 +388,7 @@ RestoreArchive(Archive *AHX)
{
if (te->hadDumper && (te->reqs & REQ_DATA) != 0)
{
- char *errmsg = supports_compression(AH->compression_spec);
+ char *errmsg = pgdump_supports_compression(AH->compression_spec);
if (errmsg)
pg_fatal("cannot restore from compressed archive (%s)",
errmsg);
@@ -3745,7 +3745,7 @@ ReadHead(ArchiveHandle *AH)
else
AH->compression_spec.algorithm = PG_COMPRESSION_GZIP;
- errmsg = supports_compression(AH->compression_spec);
+ errmsg = pgdump_supports_compression(AH->compression_spec);
if (errmsg)
{
pg_log_warning("archive is compressed, but this installation does not support compression (%s) -- no data will be available",
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 24ba936332d..ce2242195f3 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -733,8 +733,6 @@ main(int argc, char **argv)
#ifdef HAVE_LIBZ
parse_compress_specification(PG_COMPRESSION_GZIP, NULL,
&compression_spec);
-#else
- /* Nothing to do in the default case */
#endif
}
diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl
index 72b19ee6cde..ad7bc5c194b 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -4248,10 +4248,10 @@ foreach my $run (sort keys %pgdump_runs)
my $test_key = $run;
my $run_db = 'postgres';
- # Skip command-level tests for gzip if there is no support for it.
+ # Skip command-level tests for gzip/lz4 if they're not supported.
if ($pgdump_runs{$run}->{compile_option} &&
- ($pgdump_runs{$run}->{compile_option} eq 'gzip' && !$supports_gzip) ||
- ($pgdump_runs{$run}->{compile_option} eq 'lz4' && !$supports_lz4))
+ (($pgdump_runs{$run}->{compile_option} eq 'gzip' && !$supports_gzip) ||
+ ($pgdump_runs{$run}->{compile_option} eq 'lz4' && !$supports_lz4)))
{
note "$run: skipped due to no $pgdump_runs{$run}->{compile_option} support";
next;
--
2.34.1
--lqtP4FxvQo6hb6bg--
^ permalink raw reply [nested|flat] 22+ messages in thread
* [PATCH 1/4] f!fixes for LZ4
@ 2023-02-24 20:07 Justin Pryzby <[email protected]>
0 siblings, 0 replies; 22+ messages in thread
From: Justin Pryzby @ 2023-02-24 20:07 UTC (permalink / raw)
---
src/bin/pg_dump/compress_gzip.c | 8 --------
src/bin/pg_dump/compress_io.h | 2 +-
src/bin/pg_dump/compress_lz4.c | 12 ++++--------
src/bin/pg_dump/pg_dump.c | 2 --
4 files changed, 5 insertions(+), 19 deletions(-)
diff --git a/src/bin/pg_dump/compress_gzip.c b/src/bin/pg_dump/compress_gzip.c
index 0af65afeb4e..52f41c2e58c 100644
--- a/src/bin/pg_dump/compress_gzip.c
+++ b/src/bin/pg_dump/compress_gzip.c
@@ -123,17 +123,9 @@ WriteDataToArchiveGzip(ArchiveHandle *AH, CompressorState *cs,
gzipcs->outbuf = pg_malloc(ZLIB_OUT_SIZE + 1);
gzipcs->outsize = ZLIB_OUT_SIZE;
- /*
- * A level of zero simply copies the input one block at the time. This
- * is probably not what the user wanted when calling this interface.
- */
- if (cs->compression_spec.level == 0)
- pg_fatal("requested to compress the archive yet no level was specified");
-
if (deflateInit(zp, cs->compression_spec.level) != Z_OK)
pg_fatal("could not initialize compression library: %s", zp->msg);
- /* Just be paranoid - maybe End is called after Start, with no Write */
zp->next_out = gzipcs->outbuf;
zp->avail_out = gzipcs->outsize;
}
diff --git a/src/bin/pg_dump/compress_io.h b/src/bin/pg_dump/compress_io.h
index bbde2693915..cdb15951ea9 100644
--- a/src/bin/pg_dump/compress_io.h
+++ b/src/bin/pg_dump/compress_io.h
@@ -172,7 +172,7 @@ struct CompressFileHandle
extern CompressFileHandle *InitCompressFileHandle(const pg_compress_specification compression_spec);
/*
- * Initialize a compress file stream. Deffer the compression algorithm
+ * Initialize a compress file stream. Infer the compression algorithm
* from 'path', either by examining its suffix or by appending the supported
* suffixes in 'path'.
*/
diff --git a/src/bin/pg_dump/compress_lz4.c b/src/bin/pg_dump/compress_lz4.c
index fe1014e6e77..7dacfeae469 100644
--- a/src/bin/pg_dump/compress_lz4.c
+++ b/src/bin/pg_dump/compress_lz4.c
@@ -105,12 +105,8 @@ EndCompressorLZ4(ArchiveHandle *AH, CompressorState *cs)
LZ4CompressorState *LZ4cs;
LZ4cs = (LZ4CompressorState *) cs->private_data;
- if (LZ4cs)
- {
- pg_free(LZ4cs->outbuf);
- pg_free(LZ4cs);
- cs->private_data = NULL;
- }
+ pg_free(LZ4cs->outbuf);
+ pg_free(LZ4cs);
}
@@ -161,8 +157,8 @@ typedef struct LZ4File
} LZ4File;
/*
- * LZ4 equivalent to feof() or gzeof(). The end of file is reached if there
- * is no decompressed output in the overflow buffer and the end of the file
+ * LZ4 equivalent to feof() or gzeof(). Return true iff there is no
+ * decompressed output in the overflow buffer and the end of the backing file
* is reached.
*/
static int
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 24ba936332d..ce2242195f3 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -733,8 +733,6 @@ main(int argc, char **argv)
#ifdef HAVE_LIBZ
parse_compress_specification(PG_COMPRESSION_GZIP, NULL,
&compression_spec);
-#else
- /* Nothing to do in the default case */
#endif
}
--
2.34.1
--Fw50BJRL4GwAnGn1
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="0002-f-fixes-for-LZ4-which-also-conflict-with-the-ZSTD-pa.patch"
^ permalink raw reply [nested|flat] 22+ messages in thread
* [PATCH 1/4] f!fixes for LZ4
@ 2023-02-24 20:07 Justin Pryzby <[email protected]>
0 siblings, 0 replies; 22+ messages in thread
From: Justin Pryzby @ 2023-02-24 20:07 UTC (permalink / raw)
---
doc/src/sgml/ref/pg_dump.sgml | 4 ++--
src/bin/pg_dump/compress_gzip.c | 8 --------
src/bin/pg_dump/compress_io.c | 2 +-
src/bin/pg_dump/compress_io.h | 4 ++--
src/bin/pg_dump/compress_lz4.c | 4 ++--
src/bin/pg_dump/pg_backup_archiver.c | 4 ++--
src/bin/pg_dump/pg_dump.c | 2 --
src/bin/pg_dump/t/002_pg_dump.pl | 6 +++---
8 files changed, 12 insertions(+), 22 deletions(-)
diff --git a/doc/src/sgml/ref/pg_dump.sgml b/doc/src/sgml/ref/pg_dump.sgml
index 49d218905fb..6fbe49f7ede 100644
--- a/doc/src/sgml/ref/pg_dump.sgml
+++ b/doc/src/sgml/ref/pg_dump.sgml
@@ -331,7 +331,7 @@ PostgreSQL documentation
can read. A directory format archive can be manipulated with
standard Unix tools; for example, files in an uncompressed archive
can be compressed with the <application>gzip</application> or
- <application>lz4</application>tool.
+ <application>lz4</application> tools.
This format is compressed by default using <literal>gzip</literal>
and also supports parallel dumps.
</para>
@@ -655,7 +655,7 @@ PostgreSQL documentation
<para>
Specify the compression method and/or the compression level to use.
The compression method can be set to <literal>gzip</literal> or
- <literal>lz4</literal> or <literal>none</literal> for no compression.
+ <literal>lz4</literal>, or <literal>none</literal> for no compression.
A compression detail string can optionally be specified. If the
detail string is an integer, it specifies the compression level.
Otherwise, it should be a comma-separated list of items, each of the
diff --git a/src/bin/pg_dump/compress_gzip.c b/src/bin/pg_dump/compress_gzip.c
index 0af65afeb4e..52f41c2e58c 100644
--- a/src/bin/pg_dump/compress_gzip.c
+++ b/src/bin/pg_dump/compress_gzip.c
@@ -123,17 +123,9 @@ WriteDataToArchiveGzip(ArchiveHandle *AH, CompressorState *cs,
gzipcs->outbuf = pg_malloc(ZLIB_OUT_SIZE + 1);
gzipcs->outsize = ZLIB_OUT_SIZE;
- /*
- * A level of zero simply copies the input one block at the time. This
- * is probably not what the user wanted when calling this interface.
- */
- if (cs->compression_spec.level == 0)
- pg_fatal("requested to compress the archive yet no level was specified");
-
if (deflateInit(zp, cs->compression_spec.level) != Z_OK)
pg_fatal("could not initialize compression library: %s", zp->msg);
- /* Just be paranoid - maybe End is called after Start, with no Write */
zp->next_out = gzipcs->outbuf;
zp->avail_out = gzipcs->outsize;
}
diff --git a/src/bin/pg_dump/compress_io.c b/src/bin/pg_dump/compress_io.c
index ce06f1eac9c..9239dbb2755 100644
--- a/src/bin/pg_dump/compress_io.c
+++ b/src/bin/pg_dump/compress_io.c
@@ -83,7 +83,7 @@
* used by the caller in an error message.
*/
char *
-supports_compression(const pg_compress_specification compression_spec)
+pgdump_supports_compression(const pg_compress_specification compression_spec)
{
const pg_compress_algorithm algorithm = compression_spec.algorithm;
bool supported = false;
diff --git a/src/bin/pg_dump/compress_io.h b/src/bin/pg_dump/compress_io.h
index bbde2693915..46815fa2ebe 100644
--- a/src/bin/pg_dump/compress_io.h
+++ b/src/bin/pg_dump/compress_io.h
@@ -21,7 +21,7 @@
#define ZLIB_OUT_SIZE 4096
#define ZLIB_IN_SIZE 4096
-extern char *supports_compression(const pg_compress_specification compression_spec);
+extern char *pgdump_supports_compression(const pg_compress_specification compression_spec);
/*
* Prototype for callback function used in writeData()
@@ -172,7 +172,7 @@ struct CompressFileHandle
extern CompressFileHandle *InitCompressFileHandle(const pg_compress_specification compression_spec);
/*
- * Initialize a compress file stream. Deffer the compression algorithm
+ * Initialize a compress file stream. Infer the compression algorithm
* from 'path', either by examining its suffix or by appending the supported
* suffixes in 'path'.
*/
diff --git a/src/bin/pg_dump/compress_lz4.c b/src/bin/pg_dump/compress_lz4.c
index fe1014e6e77..63e794cdc68 100644
--- a/src/bin/pg_dump/compress_lz4.c
+++ b/src/bin/pg_dump/compress_lz4.c
@@ -161,8 +161,8 @@ typedef struct LZ4File
} LZ4File;
/*
- * LZ4 equivalent to feof() or gzeof(). The end of file is reached if there
- * is no decompressed output in the overflow buffer and the end of the file
+ * LZ4 equivalent to feof() or gzeof(). Return true iff there is no
+ * decompressed output in the overflow buffer and the end of the backing file
* is reached.
*/
static int
diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c
index 61ebb8fe85d..2063d6f239d 100644
--- a/src/bin/pg_dump/pg_backup_archiver.c
+++ b/src/bin/pg_dump/pg_backup_archiver.c
@@ -388,7 +388,7 @@ RestoreArchive(Archive *AHX)
{
if (te->hadDumper && (te->reqs & REQ_DATA) != 0)
{
- char *errmsg = supports_compression(AH->compression_spec);
+ char *errmsg = pgdump_supports_compression(AH->compression_spec);
if (errmsg)
pg_fatal("cannot restore from compressed archive (%s)",
errmsg);
@@ -3745,7 +3745,7 @@ ReadHead(ArchiveHandle *AH)
else
AH->compression_spec.algorithm = PG_COMPRESSION_GZIP;
- errmsg = supports_compression(AH->compression_spec);
+ errmsg = pgdump_supports_compression(AH->compression_spec);
if (errmsg)
{
pg_log_warning("archive is compressed, but this installation does not support compression (%s) -- no data will be available",
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 24ba936332d..ce2242195f3 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -733,8 +733,6 @@ main(int argc, char **argv)
#ifdef HAVE_LIBZ
parse_compress_specification(PG_COMPRESSION_GZIP, NULL,
&compression_spec);
-#else
- /* Nothing to do in the default case */
#endif
}
diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl
index 72b19ee6cde..ad7bc5c194b 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -4248,10 +4248,10 @@ foreach my $run (sort keys %pgdump_runs)
my $test_key = $run;
my $run_db = 'postgres';
- # Skip command-level tests for gzip if there is no support for it.
+ # Skip command-level tests for gzip/lz4 if they're not supported.
if ($pgdump_runs{$run}->{compile_option} &&
- ($pgdump_runs{$run}->{compile_option} eq 'gzip' && !$supports_gzip) ||
- ($pgdump_runs{$run}->{compile_option} eq 'lz4' && !$supports_lz4))
+ (($pgdump_runs{$run}->{compile_option} eq 'gzip' && !$supports_gzip) ||
+ ($pgdump_runs{$run}->{compile_option} eq 'lz4' && !$supports_lz4)))
{
note "$run: skipped due to no $pgdump_runs{$run}->{compile_option} support";
next;
--
2.34.1
--VbXIs2ae+j0NhctA
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="0002-pg_dump-zstd-compression.patch"
^ permalink raw reply [nested|flat] 22+ messages in thread
* [PATCH] f!fixes for LZ4
@ 2023-02-24 20:07 Justin Pryzby <[email protected]>
0 siblings, 0 replies; 22+ messages in thread
From: Justin Pryzby @ 2023-02-24 20:07 UTC (permalink / raw)
---
doc/src/sgml/ref/pg_dump.sgml | 4 ++--
src/bin/pg_dump/compress_io.c | 2 +-
src/bin/pg_dump/compress_io.h | 4 ++--
src/bin/pg_dump/compress_lz4.c | 4 ++--
src/bin/pg_dump/pg_backup_archiver.c | 4 ++--
src/bin/pg_dump/pg_dump.c | 2 --
src/bin/pg_dump/t/002_pg_dump.pl | 6 +++---
7 files changed, 12 insertions(+), 14 deletions(-)
diff --git a/doc/src/sgml/ref/pg_dump.sgml b/doc/src/sgml/ref/pg_dump.sgml
index 49d218905fb..6fbe49f7ede 100644
--- a/doc/src/sgml/ref/pg_dump.sgml
+++ b/doc/src/sgml/ref/pg_dump.sgml
@@ -331,7 +331,7 @@ PostgreSQL documentation
can read. A directory format archive can be manipulated with
standard Unix tools; for example, files in an uncompressed archive
can be compressed with the <application>gzip</application> or
- <application>lz4</application>tool.
+ <application>lz4</application> tools.
This format is compressed by default using <literal>gzip</literal>
and also supports parallel dumps.
</para>
@@ -655,7 +655,7 @@ PostgreSQL documentation
<para>
Specify the compression method and/or the compression level to use.
The compression method can be set to <literal>gzip</literal> or
- <literal>lz4</literal> or <literal>none</literal> for no compression.
+ <literal>lz4</literal>, or <literal>none</literal> for no compression.
A compression detail string can optionally be specified. If the
detail string is an integer, it specifies the compression level.
Otherwise, it should be a comma-separated list of items, each of the
diff --git a/src/bin/pg_dump/compress_io.c b/src/bin/pg_dump/compress_io.c
index ce06f1eac9c..9239dbb2755 100644
--- a/src/bin/pg_dump/compress_io.c
+++ b/src/bin/pg_dump/compress_io.c
@@ -83,7 +83,7 @@
* used by the caller in an error message.
*/
char *
-supports_compression(const pg_compress_specification compression_spec)
+pgdump_supports_compression(const pg_compress_specification compression_spec)
{
const pg_compress_algorithm algorithm = compression_spec.algorithm;
bool supported = false;
diff --git a/src/bin/pg_dump/compress_io.h b/src/bin/pg_dump/compress_io.h
index bbde2693915..46815fa2ebe 100644
--- a/src/bin/pg_dump/compress_io.h
+++ b/src/bin/pg_dump/compress_io.h
@@ -21,7 +21,7 @@
#define ZLIB_OUT_SIZE 4096
#define ZLIB_IN_SIZE 4096
-extern char *supports_compression(const pg_compress_specification compression_spec);
+extern char *pgdump_supports_compression(const pg_compress_specification compression_spec);
/*
* Prototype for callback function used in writeData()
@@ -172,7 +172,7 @@ struct CompressFileHandle
extern CompressFileHandle *InitCompressFileHandle(const pg_compress_specification compression_spec);
/*
- * Initialize a compress file stream. Deffer the compression algorithm
+ * Initialize a compress file stream. Infer the compression algorithm
* from 'path', either by examining its suffix or by appending the supported
* suffixes in 'path'.
*/
diff --git a/src/bin/pg_dump/compress_lz4.c b/src/bin/pg_dump/compress_lz4.c
index fe1014e6e77..63e794cdc68 100644
--- a/src/bin/pg_dump/compress_lz4.c
+++ b/src/bin/pg_dump/compress_lz4.c
@@ -161,8 +161,8 @@ typedef struct LZ4File
} LZ4File;
/*
- * LZ4 equivalent to feof() or gzeof(). The end of file is reached if there
- * is no decompressed output in the overflow buffer and the end of the file
+ * LZ4 equivalent to feof() or gzeof(). Return true iff there is no
+ * decompressed output in the overflow buffer and the end of the backing file
* is reached.
*/
static int
diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c
index 61ebb8fe85d..2063d6f239d 100644
--- a/src/bin/pg_dump/pg_backup_archiver.c
+++ b/src/bin/pg_dump/pg_backup_archiver.c
@@ -388,7 +388,7 @@ RestoreArchive(Archive *AHX)
{
if (te->hadDumper && (te->reqs & REQ_DATA) != 0)
{
- char *errmsg = supports_compression(AH->compression_spec);
+ char *errmsg = pgdump_supports_compression(AH->compression_spec);
if (errmsg)
pg_fatal("cannot restore from compressed archive (%s)",
errmsg);
@@ -3745,7 +3745,7 @@ ReadHead(ArchiveHandle *AH)
else
AH->compression_spec.algorithm = PG_COMPRESSION_GZIP;
- errmsg = supports_compression(AH->compression_spec);
+ errmsg = pgdump_supports_compression(AH->compression_spec);
if (errmsg)
{
pg_log_warning("archive is compressed, but this installation does not support compression (%s) -- no data will be available",
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 24ba936332d..ce2242195f3 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -733,8 +733,6 @@ main(int argc, char **argv)
#ifdef HAVE_LIBZ
parse_compress_specification(PG_COMPRESSION_GZIP, NULL,
&compression_spec);
-#else
- /* Nothing to do in the default case */
#endif
}
diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl
index 72b19ee6cde..ad7bc5c194b 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -4248,10 +4248,10 @@ foreach my $run (sort keys %pgdump_runs)
my $test_key = $run;
my $run_db = 'postgres';
- # Skip command-level tests for gzip if there is no support for it.
+ # Skip command-level tests for gzip/lz4 if they're not supported.
if ($pgdump_runs{$run}->{compile_option} &&
- ($pgdump_runs{$run}->{compile_option} eq 'gzip' && !$supports_gzip) ||
- ($pgdump_runs{$run}->{compile_option} eq 'lz4' && !$supports_lz4))
+ (($pgdump_runs{$run}->{compile_option} eq 'gzip' && !$supports_gzip) ||
+ ($pgdump_runs{$run}->{compile_option} eq 'lz4' && !$supports_lz4)))
{
note "$run: skipped due to no $pgdump_runs{$run}->{compile_option} support";
next;
--
2.34.1
--p8BcnzLwh3ipgLRM--
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2023-03-08 18:07 Antonin Houska <[email protected]>
parent: James Coleman <[email protected]>
0 siblings, 0 replies; 22+ messages in thread
From: Antonin Houska @ 2023-03-08 18:07 UTC (permalink / raw)
To: James Coleman <[email protected]>; +Cc: Tomas Vondra <[email protected]>; vignesh C <[email protected]>; Andres Freund <[email protected]>; Robert Haas <[email protected]>; pgsql-hackers; Amit Kapila <[email protected]>; Tom Lane <[email protected]>
James Coleman <[email protected]> wrote:
> On Mon, Feb 6, 2023 at 11:39 AM Antonin Houska <[email protected]> wrote:
> Attached is v9.
ok, I've changed the status to RfC
--
Antonin Houska
Web: https://www.cybertec-postgresql.com
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2023-06-06 09:35 Richard Guo <[email protected]>
parent: James Coleman <[email protected]>
1 sibling, 1 reply; 22+ messages in thread
From: Richard Guo @ 2023-06-06 09:35 UTC (permalink / raw)
To: James Coleman <[email protected]>; +Cc: Tomas Vondra <[email protected]>; vignesh C <[email protected]>; Andres Freund <[email protected]>; Robert Haas <[email protected]>; pgsql-hackers; Amit Kapila <[email protected]>; Tom Lane <[email protected]>
On Mon, Jan 23, 2023 at 10:00 PM James Coleman <[email protected]> wrote:
> Which this patch we do in fact now see (as expected) rels with
> non-empty lateral_relids showing up in generate_[useful_]gather_paths.
> And the partial paths can now have non-empty required outer rels. I'm
> not able to come up with a plan that would actually be caught by those
> checks; I theorize that because of the few places we actually call
> generate_[useful_]gather_paths we are in practice already excluding
> those, but for now I've left these as a conditional rather than an
> assertion because it seems like the kind of guard we'd want to ensure
> those methods are safe.
I'm trying to understand this part. AFAICS we will not create partial
paths for a rel, base or join, if it has lateral references. So it
seems to me that in generate_[useful_]gather_paths after we've checked
that there are partial paths, the checks for lateral_relids are not
necessary because lateral_relids should always be empty in this case.
Maybe I'm missing something.
And while trying the v9 patch I came across a crash with the query
below.
set min_parallel_table_scan_size to 0;
set parallel_setup_cost to 0;
set parallel_tuple_cost to 0;
explain (costs off)
select * from pg_description t1 where objoid in
(select objoid from pg_description t2 where t2.description =
t1.description);
QUERY PLAN
--------------------------------------------------------
Seq Scan on pg_description t1
Filter: (SubPlan 1)
SubPlan 1
-> Gather
Workers Planned: 2
-> Parallel Seq Scan on pg_description t2
Filter: (description = t1.description)
(7 rows)
select * from pg_description t1 where objoid in
(select objoid from pg_description t2 where t2.description =
t1.description);
WARNING: terminating connection because of crash of another server process
Seems something is wrong when extracting the argument from the Param in
parallel worker.
BTW another rebase is needed as it no longer applies to HEAD.
Thanks
Richard
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2023-06-12 02:23 James Coleman <[email protected]>
parent: Richard Guo <[email protected]>
0 siblings, 2 replies; 22+ messages in thread
From: James Coleman @ 2023-06-12 02:23 UTC (permalink / raw)
To: Richard Guo <[email protected]>; +Cc: Tomas Vondra <[email protected]>; vignesh C <[email protected]>; Andres Freund <[email protected]>; Robert Haas <[email protected]>; pgsql-hackers; Amit Kapila <[email protected]>; Tom Lane <[email protected]>
On Tue, Jun 6, 2023 at 4:36 AM Richard Guo <[email protected]> wrote:
>
>
> On Mon, Jan 23, 2023 at 10:00 PM James Coleman <[email protected]> wrote:
>>
>> Which this patch we do in fact now see (as expected) rels with
>> non-empty lateral_relids showing up in generate_[useful_]gather_paths.
>> And the partial paths can now have non-empty required outer rels. I'm
>> not able to come up with a plan that would actually be caught by those
>> checks; I theorize that because of the few places we actually call
>> generate_[useful_]gather_paths we are in practice already excluding
>> those, but for now I've left these as a conditional rather than an
>> assertion because it seems like the kind of guard we'd want to ensure
>> those methods are safe.
>
>
> I'm trying to understand this part. AFAICS we will not create partial
> paths for a rel, base or join, if it has lateral references. So it
> seems to me that in generate_[useful_]gather_paths after we've checked
> that there are partial paths, the checks for lateral_relids are not
> necessary because lateral_relids should always be empty in this case.
> Maybe I'm missing something.
At first I was thinking "isn't the point of the patch to generate
partial paths for rels with lateral references" given what I'd written
back in January, but I added "Assert(bms_is_empty(required_outer));"
to both of those functions and the assertion never fails running the
tests (including my newly parallelizable queries). I'm almost positive
I'd checked this back in January (not only had I'd explicitly written
that I'd confirmed we had non-empty lateral_relids there, but also it
was the entire based of the alternate approach to the patch), but...I
can't go back to 5 months ago and remember what I'd done.
Ah! Your comment about "after we've checked that there are partial
paths" triggered a thought. I think originally I'd had the
"bms_is_subset(required_outer, rel->relids)" check first in these
functions. And indeed if I run the tests with that the assertion moved
to above the partial paths check, I get failures in
generate_useful_gather_paths specifically. Mystery solved!
> And while trying the v9 patch I came across a crash with the query
> below.
>
> set min_parallel_table_scan_size to 0;
> set parallel_setup_cost to 0;
> set parallel_tuple_cost to 0;
>
> explain (costs off)
> select * from pg_description t1 where objoid in
> (select objoid from pg_description t2 where t2.description = t1.description);
> QUERY PLAN
> --------------------------------------------------------
> Seq Scan on pg_description t1
> Filter: (SubPlan 1)
> SubPlan 1
> -> Gather
> Workers Planned: 2
> -> Parallel Seq Scan on pg_description t2
> Filter: (description = t1.description)
> (7 rows)
>
> select * from pg_description t1 where objoid in
> (select objoid from pg_description t2 where t2.description = t1.description);
> WARNING: terminating connection because of crash of another server process
>
> Seems something is wrong when extracting the argument from the Param in
> parallel worker.
With what I'm trying to change I don't think this plan should ever be
generated since it means we'd have to pass a param from the outer seq
scan into the parallel subplan, which we can't do (currently).
I've attached the full backtrace to the email, but as you hinted at
the parallel worker is trying to get the param (in this case
detoasting it), but the param doesn't exist on the worker, so it seg
faults. Looking at this further I think there's an existing test case
that exposes the misplanning here (the one right under the comment
"Parallel Append is not to be used when the subpath depends on the
outer param" in select_parallel.sql), but it doesn't seg fault because
the param is an integer, doesn't need to be detoasted, and therefore
(I think) we skate by (but probably with wrong results in depending on
the dataset).
Interestingly this is one of the existing test queries my original
patch approach didn't change, so this gives me something specific to
work with improving the path. Thanks for testing this and bringing
this to my attention!
BTW are you by any chance testing on ARM macOS? I reproduced the issue
there, but for some reason I did not reproduce the error (and the plan
wasn't parallelized) when I tested this on linux. Perhaps I missed
setting something up; it seems odd.
> BTW another rebase is needed as it no longer applies to HEAD.
Apologies; I'd rebased, but hadn't updated the thread. See attached
for an updated series (albeit still broken on your test query).
Thanks,
James
* thread #1, queue = 'com.apple.main-thread', stop reason = EXC_BAD_ACCESS (code=1, address=0x0)
* frame #0: 0x000000010449cab0 postgres`toast_raw_datum_size(value=0) at detoast.c:550:6
frame #1: 0x0000000104cf9438 postgres`texteq(fcinfo=0x0000000112809c80) at varlena.c:1645:10
frame #2: 0x00000001047ca7ac postgres`ExecInterpExpr(state=0x00000001128097e0, econtext=0x0000000112808c48, isnull=0x000000016b97e39f) at execExprInterp.c:758:8
frame #3: 0x00000001047c98bc postgres`ExecInterpExprStillValid(state=0x00000001128097e0, econtext=0x0000000112808c48, isNull=0x000000016b97e39f) at execExprInterp.c:1870:9
frame #4: 0x00000001047ee1b0 postgres`ExecEvalExprSwitchContext(state=0x00000001128097e0, econtext=0x0000000112808c48, isNull=0x000000016b97e39f) at executor.h:355:13
frame #5: 0x00000001047edddc postgres`ExecQual(state=0x00000001128097e0, econtext=0x0000000112808c48) at executor.h:424:8
frame #6: 0x00000001047ed964 postgres`ExecScan(node=0x0000000112808d68, accessMtd=(postgres`SeqNext at nodeSeqscan.c:51), recheckMtd=(postgres`SeqRecheck at nodeSeqscan.c:90)) at exec
Scan.c:226:23
frame #7: 0x0000000104837ea8 postgres`ExecSeqScan(pstate=0x0000000112808d68) at nodeSeqscan.c:112:9
frame #8: 0x00000001047e8ca0 postgres`ExecProcNodeFirst(node=0x0000000112808d68) at execProcnode.c:464:9
frame #9: 0x00000001047dfd90 postgres`ExecProcNode(node=0x0000000112808d68) at executor.h:273:9
frame #10: 0x00000001047daa28 postgres`ExecutePlan(estate=0x0000000112808318, planstate=0x0000000112808d68, use_parallel_mode=false, operation=CMD_SELECT, sendTuples=true, numberTuple
s=0, direction=ForwardScanDirection, dest=0x00000001130389d8, execute_once=true) at execMain.c:1661:10
frame #11: 0x00000001047da8d4 postgres`standard_ExecutorRun(queryDesc=0x0000000113039738, direction=ForwardScanDirection, count=0, execute_once=true) at execMain.c:365:3
frame #12: 0x00000001047da644 postgres`ExecutorRun(queryDesc=0x0000000113039738, direction=ForwardScanDirection, count=0, execute_once=true) at execMain.c:309:3
frame #13: 0x00000001047e3230 postgres`ParallelQueryMain(seg=0x000000011300a5b8, toc=0x0000000105434000) at execParallel.c:1464:2
frame #14: 0x000000010458a8d0 postgres`ParallelWorkerMain(main_arg=4065634722) at parallel.c:1521:2
frame #15: 0x00000001049e70e4 postgres`StartBackgroundWorker at bgworker.c:861:2
frame #16: 0x00000001049f64b8 postgres`do_start_bgworker(rw=0x0000000113804080) at postmaster.c:5762:4
frame #17: 0x00000001049eed90 postgres`maybe_start_bgworkers at postmaster.c:5986:9
frame #18: 0x00000001049f1764 postgres`process_pm_pmsignal at postmaster.c:5149:3
frame #19: 0x00000001049eef7c postgres`ServerLoop at postmaster.c:1770:5
frame #20: 0x00000001049edb48 postgres`PostmasterMain(argc=3, argv=0x00006000016453c0) at postmaster.c:1463:11
frame #21: 0x00000001048782f4 postgres`main(argc=3, argv=0x00006000016453c0) at main.c:198:3
frame #22: 0x00000001a08cff28 dyld`start + 2236
Attachments:
[text/plain] backtrace.txt (2.9K, ../../CAAaqYe-_TObm5KwmZLYXBJ3BJGh4cUZWM0v1mY1gWTMkRNQXDQ@mail.gmail.com/2-backtrace.txt)
download | inline:
* thread #1, queue = 'com.apple.main-thread', stop reason = EXC_BAD_ACCESS (code=1, address=0x0)
* frame #0: 0x000000010449cab0 postgres`toast_raw_datum_size(value=0) at detoast.c:550:6
frame #1: 0x0000000104cf9438 postgres`texteq(fcinfo=0x0000000112809c80) at varlena.c:1645:10
frame #2: 0x00000001047ca7ac postgres`ExecInterpExpr(state=0x00000001128097e0, econtext=0x0000000112808c48, isnull=0x000000016b97e39f) at execExprInterp.c:758:8
frame #3: 0x00000001047c98bc postgres`ExecInterpExprStillValid(state=0x00000001128097e0, econtext=0x0000000112808c48, isNull=0x000000016b97e39f) at execExprInterp.c:1870:9
frame #4: 0x00000001047ee1b0 postgres`ExecEvalExprSwitchContext(state=0x00000001128097e0, econtext=0x0000000112808c48, isNull=0x000000016b97e39f) at executor.h:355:13
frame #5: 0x00000001047edddc postgres`ExecQual(state=0x00000001128097e0, econtext=0x0000000112808c48) at executor.h:424:8
frame #6: 0x00000001047ed964 postgres`ExecScan(node=0x0000000112808d68, accessMtd=(postgres`SeqNext at nodeSeqscan.c:51), recheckMtd=(postgres`SeqRecheck at nodeSeqscan.c:90)) at exec
Scan.c:226:23
frame #7: 0x0000000104837ea8 postgres`ExecSeqScan(pstate=0x0000000112808d68) at nodeSeqscan.c:112:9
frame #8: 0x00000001047e8ca0 postgres`ExecProcNodeFirst(node=0x0000000112808d68) at execProcnode.c:464:9
frame #9: 0x00000001047dfd90 postgres`ExecProcNode(node=0x0000000112808d68) at executor.h:273:9
frame #10: 0x00000001047daa28 postgres`ExecutePlan(estate=0x0000000112808318, planstate=0x0000000112808d68, use_parallel_mode=false, operation=CMD_SELECT, sendTuples=true, numberTuple
s=0, direction=ForwardScanDirection, dest=0x00000001130389d8, execute_once=true) at execMain.c:1661:10
frame #11: 0x00000001047da8d4 postgres`standard_ExecutorRun(queryDesc=0x0000000113039738, direction=ForwardScanDirection, count=0, execute_once=true) at execMain.c:365:3
frame #12: 0x00000001047da644 postgres`ExecutorRun(queryDesc=0x0000000113039738, direction=ForwardScanDirection, count=0, execute_once=true) at execMain.c:309:3
frame #13: 0x00000001047e3230 postgres`ParallelQueryMain(seg=0x000000011300a5b8, toc=0x0000000105434000) at execParallel.c:1464:2
frame #14: 0x000000010458a8d0 postgres`ParallelWorkerMain(main_arg=4065634722) at parallel.c:1521:2
frame #15: 0x00000001049e70e4 postgres`StartBackgroundWorker at bgworker.c:861:2
frame #16: 0x00000001049f64b8 postgres`do_start_bgworker(rw=0x0000000113804080) at postmaster.c:5762:4
frame #17: 0x00000001049eed90 postgres`maybe_start_bgworkers at postmaster.c:5986:9
frame #18: 0x00000001049f1764 postgres`process_pm_pmsignal at postmaster.c:5149:3
frame #19: 0x00000001049eef7c postgres`ServerLoop at postmaster.c:1770:5
frame #20: 0x00000001049edb48 postgres`PostmasterMain(argc=3, argv=0x00006000016453c0) at postmaster.c:1463:11
frame #21: 0x00000001048782f4 postgres`main(argc=3, argv=0x00006000016453c0) at main.c:198:3
frame #22: 0x00000001a08cff28 dyld`start + 2236
[application/octet-stream] v10-0001-Add-tests-before-change.patch (7.2K, ../../CAAaqYe-_TObm5KwmZLYXBJ3BJGh4cUZWM0v1mY1gWTMkRNQXDQ@mail.gmail.com/3-v10-0001-Add-tests-before-change.patch)
download | inline diff:
From 59c67d670e024173e8417867cd15d7edc257f012 Mon Sep 17 00:00:00 2001
From: jcoleman <[email protected]>
Date: Mon, 26 Sep 2022 20:30:23 -0400
Subject: [PATCH v10 1/2] Add tests before change
---
src/test/regress/expected/select_parallel.out | 108 ++++++++++++++++++
src/test/regress/sql/select_parallel.sql | 25 ++++
2 files changed, 133 insertions(+)
diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
index d88353d496..04810d2756 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -311,6 +311,114 @@ select count(*) from tenk1 where (two, four) not in
10000
(1 row)
+-- test parallel plans for queries containing correlated subplans
+-- where the subplan only needs params available from the current
+-- worker's scan.
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t, generate_series(1, 10);
+ QUERY PLAN
+----------------------------------------------------------------------------
+ Gather
+ Output: (SubPlan 1)
+ Workers Planned: 4
+ -> Nested Loop
+ Output: t.unique1
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ -> Function Scan on pg_catalog.generate_series
+ Output: generate_series.generate_series
+ Function Call: generate_series(1, 10)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(14 rows)
+
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t;
+ QUERY PLAN
+----------------------------------------------------------------------
+ Gather
+ Output: (SubPlan 1)
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(9 rows)
+
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t
+ limit 1;
+ QUERY PLAN
+-------------------------------------------------------------------
+ Limit
+ Output: ((SubPlan 1))
+ -> Seq Scan on public.tenk1 t
+ Output: (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(8 rows)
+
+explain (costs off, verbose) select t.unique1
+ from tenk1 t
+ where t.unique1 = (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+ QUERY PLAN
+-------------------------------------------------------------
+ Seq Scan on public.tenk1 t
+ Output: t.unique1
+ Filter: (t.unique1 = (SubPlan 1))
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(7 rows)
+
+explain (costs off, verbose) select *
+ from tenk1 t
+ order by (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+ QUERY PLAN
+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Sort
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, ((SubPlan 1))
+ Sort Key: ((SubPlan 1))
+ -> Gather
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, (SubPlan 1)
+ Workers Planned: 4
+ -> Parallel Seq Scan on public.tenk1 t
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(12 rows)
+
+-- test subplan in join/lateral join
+explain (costs off, verbose, timing off) select t.unique1, l.*
+ from tenk1 t
+ join lateral (
+ select (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1 offset 0)
+ ) l on true;
+ QUERY PLAN
+----------------------------------------------------------------------
+ Gather
+ Output: t.unique1, (SubPlan 1)
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(9 rows)
+
-- this is not parallel-safe due to use of random() within SubLink's testexpr:
explain (costs off)
select * from tenk1 where (unique1 + random())::integer not in
diff --git a/src/test/regress/sql/select_parallel.sql b/src/test/regress/sql/select_parallel.sql
index 80c914dc02..fd968285c1 100644
--- a/src/test/regress/sql/select_parallel.sql
+++ b/src/test/regress/sql/select_parallel.sql
@@ -111,6 +111,31 @@ explain (costs off)
(select hundred, thousand from tenk2 where thousand > 100);
select count(*) from tenk1 where (two, four) not in
(select hundred, thousand from tenk2 where thousand > 100);
+-- test parallel plans for queries containing correlated subplans
+-- where the subplan only needs params available from the current
+-- worker's scan.
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t, generate_series(1, 10);
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t;
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t
+ limit 1;
+explain (costs off, verbose) select t.unique1
+ from tenk1 t
+ where t.unique1 = (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+explain (costs off, verbose) select *
+ from tenk1 t
+ order by (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+-- test subplan in join/lateral join
+explain (costs off, verbose, timing off) select t.unique1, l.*
+ from tenk1 t
+ join lateral (
+ select (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1 offset 0)
+ ) l on true;
-- this is not parallel-safe due to use of random() within SubLink's testexpr:
explain (costs off)
select * from tenk1 where (unique1 + random())::integer not in
--
2.39.2 (Apple Git-143)
[application/octet-stream] v10-0002-Parallelize-correlated-subqueries.patch (30.5K, ../../CAAaqYe-_TObm5KwmZLYXBJ3BJGh4cUZWM0v1mY1gWTMkRNQXDQ@mail.gmail.com/4-v10-0002-Parallelize-correlated-subqueries.patch)
download | inline diff:
From 8cd41d370c96e58f097a651b525fd585408f618a Mon Sep 17 00:00:00 2001
From: jcoleman <[email protected]>
Date: Wed, 18 Jan 2023 20:43:26 -0500
Subject: [PATCH v10 2/2] Parallelize correlated subqueries
When params are provided at the current query level (i.e., are generated
within a single worker and not shared across workers) we can safely
execute these in parallel.
Alternative approach using just relids subset check
---
doc/src/sgml/parallel.sgml | 3 +-
src/backend/optimizer/path/allpaths.c | 46 +++++--
src/backend/optimizer/path/joinpath.c | 10 +-
src/backend/optimizer/plan/planner.c | 8 ++
src/backend/optimizer/util/clauses.c | 74 ++++------
.../regress/expected/incremental_sort.out | 41 +++---
src/test/regress/expected/partition_prune.out | 104 +++++++-------
src/test/regress/expected/select_parallel.out | 128 ++++++++++--------
8 files changed, 222 insertions(+), 192 deletions(-)
diff --git a/doc/src/sgml/parallel.sgml b/doc/src/sgml/parallel.sgml
index 5acc9537d6..fd32572ec8 100644
--- a/doc/src/sgml/parallel.sgml
+++ b/doc/src/sgml/parallel.sgml
@@ -518,7 +518,8 @@ EXPLAIN SELECT * FROM pgbench_accounts WHERE filler LIKE '%x%';
<listitem>
<para>
- Plan nodes that reference a correlated <literal>SubPlan</literal>.
+ Plan nodes that reference a correlated <literal>SubPlan</literal> where
+ the result is shared between workers.
</para>
</listitem>
</itemizedlist>
diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index 9bdc70c702..c9ccb508b4 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -3064,11 +3064,19 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
ListCell *lc;
double rows;
double *rowsp = NULL;
+ Relids required_outer = rel->lateral_relids;
/* If there are no partial paths, there's nothing to do here. */
if (rel->partial_pathlist == NIL)
return;
+ /*
+ * Delay gather path creation until the level in the join tree where all
+ * params used in a worker are generated within that worker.
+ */
+ if (!bms_is_subset(required_outer, rel->relids))
+ return;
+
/* Should we override the rel's rowcount estimate? */
if (override_rows)
rowsp = &rows;
@@ -3079,12 +3087,17 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
* of partial_pathlist because of the way add_partial_path works.
*/
cheapest_partial_path = linitial(rel->partial_pathlist);
- rows =
- cheapest_partial_path->rows * cheapest_partial_path->parallel_workers;
- simple_gather_path = (Path *)
- create_gather_path(root, rel, cheapest_partial_path, rel->reltarget,
- NULL, rowsp);
- add_path(rel, simple_gather_path);
+
+ /* We can't pass params to workers. */
+ if (bms_is_subset(PATH_REQ_OUTER(cheapest_partial_path), rel->relids))
+ {
+ rows =
+ cheapest_partial_path->rows * cheapest_partial_path->parallel_workers;
+ simple_gather_path = (Path *)
+ create_gather_path(root, rel, cheapest_partial_path, rel->reltarget,
+ PATH_REQ_OUTER(cheapest_partial_path), rowsp);
+ add_path(rel, simple_gather_path);
+ }
/*
* For each useful ordering, we can consider an order-preserving Gather
@@ -3098,9 +3111,14 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
if (subpath->pathkeys == NIL)
continue;
+ /* We can't pass params to workers. */
+ if (!bms_is_subset(PATH_REQ_OUTER(subpath), rel->relids))
+ continue;
+
rows = subpath->rows * subpath->parallel_workers;
path = create_gather_merge_path(root, rel, subpath, rel->reltarget,
- subpath->pathkeys, NULL, rowsp);
+ subpath->pathkeys,
+ PATH_REQ_OUTER(subpath), rowsp);
add_path(rel, &path->path);
}
}
@@ -3202,11 +3220,19 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
double *rowsp = NULL;
List *useful_pathkeys_list = NIL;
Path *cheapest_partial_path = NULL;
+ Relids required_outer = rel->lateral_relids;
/* If there are no partial paths, there's nothing to do here. */
if (rel->partial_pathlist == NIL)
return;
+ /*
+ * Delay gather path creation until the level in the join tree where all
+ * params used in a worker are generated within that worker.
+ */
+ if (!bms_is_subset(required_outer, rel->relids))
+ return;
+
/* Should we override the rel's rowcount estimate? */
if (override_rows)
rowsp = &rows;
@@ -3236,6 +3262,10 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
Path *subpath = (Path *) lfirst(lc2);
GatherMergePath *path;
+ /* We can't pass params to workers. */
+ if (!bms_is_subset(PATH_REQ_OUTER(subpath), rel->relids))
+ continue;
+
is_sorted = pathkeys_count_contained_in(useful_pathkeys,
subpath->pathkeys,
&presorted_keys);
@@ -3295,7 +3325,7 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
subpath,
rel->reltarget,
subpath->pathkeys,
- NULL,
+ PATH_REQ_OUTER(subpath),
rowsp);
add_path(rel, &path->path);
diff --git a/src/backend/optimizer/path/joinpath.c b/src/backend/optimizer/path/joinpath.c
index cd80e61fd7..66d133d0ea 100644
--- a/src/backend/optimizer/path/joinpath.c
+++ b/src/backend/optimizer/path/joinpath.c
@@ -1850,10 +1850,12 @@ match_unsorted_outer(PlannerInfo *root,
* Consider partial nestloop and mergejoin plan if outerrel has any
* partial path and the joinrel is parallel-safe. However, we can't
* handle JOIN_UNIQUE_OUTER, because the outer path will be partial, and
- * therefore we won't be able to properly guarantee uniqueness. Nor can
- * we handle joins needing lateral rels, since partial paths must not be
- * parameterized. Similarly, we can't handle JOIN_FULL, JOIN_RIGHT and
- * JOIN_RIGHT_ANTI, because they can produce false null extended rows.
+ * therefore we won't be able to properly guarantee uniqueness. Similarly,
+ * we can't handle JOIN_FULL JOIN_RIGHT, and JOIN_RIGHT_ANTI, because they
+ * can produce false null extended rows.
+ *
+ * While partial paths may now be parameterized so long as all of the params
+ * can be generated wholly within a worker we punt on supporting that here.
*/
if (joinrel->consider_parallel &&
save_jointype != JOIN_UNIQUE_OUTER &&
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index 1e4dd27dba..2ca363df61 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -7353,11 +7353,16 @@ gather_grouping_paths(PlannerInfo *root, RelOptInfo *rel)
ListCell *lc;
Path *cheapest_partial_path;
+ /* By grouping time we shouldn't have any lateral dependencies. */
+ Assert(rel->lateral_relids == NULL);
+
/* Try Gather for unordered paths and Gather Merge for ordered ones. */
generate_useful_gather_paths(root, rel, true);
/* Try cheapest partial path + explicit Sort + Gather Merge. */
cheapest_partial_path = linitial(rel->partial_pathlist);
+ /* By grouping time we shouldn't have any lateral dependencies. */
+ Assert(PATH_REQ_OUTER(cheapest_partial_path) == NULL);
if (!pathkeys_contained_in(root->group_pathkeys,
cheapest_partial_path->pathkeys))
{
@@ -7409,6 +7414,9 @@ gather_grouping_paths(PlannerInfo *root, RelOptInfo *rel)
if (presorted_keys == 0)
continue;
+ /* By grouping time we shouldn't have any lateral dependencies. */
+ Assert(PATH_REQ_OUTER(path) == NULL);
+
path = (Path *) create_incremental_sort_path(root,
rel,
path,
diff --git a/src/backend/optimizer/util/clauses.c b/src/backend/optimizer/util/clauses.c
index 7f453b04f8..894b93b8c2 100644
--- a/src/backend/optimizer/util/clauses.c
+++ b/src/backend/optimizer/util/clauses.c
@@ -91,7 +91,6 @@ typedef struct
{
char max_hazard; /* worst proparallel hazard found so far */
char max_interesting; /* worst proparallel hazard of interest */
- List *safe_param_ids; /* PARAM_EXEC Param IDs to treat as safe */
} max_parallel_hazard_context;
static bool contain_agg_clause_walker(Node *node, void *context);
@@ -654,7 +653,6 @@ max_parallel_hazard(Query *parse)
context.max_hazard = PROPARALLEL_SAFE;
context.max_interesting = PROPARALLEL_UNSAFE;
- context.safe_param_ids = NIL;
(void) max_parallel_hazard_walker((Node *) parse, &context);
return context.max_hazard;
}
@@ -665,43 +663,24 @@ max_parallel_hazard(Query *parse)
*
* root->glob->maxParallelHazard must previously have been set to the
* result of max_parallel_hazard() on the whole query.
+ *
+ * The caller is responsible for verifying that PARAM_EXEC Params are generated
+ * at the current plan level.
*/
bool
is_parallel_safe(PlannerInfo *root, Node *node)
{
max_parallel_hazard_context context;
- PlannerInfo *proot;
- ListCell *l;
/*
- * Even if the original querytree contained nothing unsafe, we need to
- * search the expression if we have generated any PARAM_EXEC Params while
- * planning, because those are parallel-restricted and there might be one
- * in this expression. But otherwise we don't need to look.
+ * If we've already checked the querytree don't burn cycles doing it again.
*/
- if (root->glob->maxParallelHazard == PROPARALLEL_SAFE &&
- root->glob->paramExecTypes == NIL)
+ if (root->glob->maxParallelHazard == PROPARALLEL_SAFE)
return true;
+
/* Else use max_parallel_hazard's search logic, but stop on RESTRICTED */
context.max_hazard = PROPARALLEL_SAFE;
context.max_interesting = PROPARALLEL_RESTRICTED;
- context.safe_param_ids = NIL;
-
- /*
- * The params that refer to the same or parent query level are considered
- * parallel-safe. The idea is that we compute such params at Gather or
- * Gather Merge node and pass their value to workers.
- */
- for (proot = root; proot != NULL; proot = proot->parent_root)
- {
- foreach(l, proot->init_plans)
- {
- SubPlan *initsubplan = (SubPlan *) lfirst(l);
-
- context.safe_param_ids = list_concat(context.safe_param_ids,
- initsubplan->setParam);
- }
- }
return !max_parallel_hazard_walker(node, &context);
}
@@ -811,39 +790,34 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context)
}
/*
- * Only parallel-safe SubPlans can be sent to workers. Within the
- * testexpr of the SubPlan, Params representing the output columns of the
- * subplan can be treated as parallel-safe, so temporarily add their IDs
- * to the safe_param_ids list while examining the testexpr.
+ * Only parallel-safe SubPlans can be sent to workers.
*/
else if (IsA(node, SubPlan))
{
SubPlan *subplan = (SubPlan *) node;
- List *save_safe_param_ids;
if (!subplan->parallel_safe &&
max_parallel_hazard_test(PROPARALLEL_RESTRICTED, context))
return true;
- save_safe_param_ids = context->safe_param_ids;
- context->safe_param_ids = list_concat_copy(context->safe_param_ids,
- subplan->paramIds);
+
if (max_parallel_hazard_walker(subplan->testexpr, context))
- return true; /* no need to restore safe_param_ids */
- list_free(context->safe_param_ids);
- context->safe_param_ids = save_safe_param_ids;
- /* we must also check args, but no special Param treatment there */
+ return true;
+
if (max_parallel_hazard_walker((Node *) subplan->args, context))
return true;
+
/* don't want to recurse normally, so we're done */
return false;
}
/*
- * We can't pass Params to workers at the moment either, so they are also
- * parallel-restricted, unless they are PARAM_EXTERN Params or are
- * PARAM_EXEC Params listed in safe_param_ids, meaning they could be
- * either generated within workers or can be computed by the leader and
- * then their value can be passed to workers.
+ * We can't pass all types of Params to workers at the moment either.
+ * PARAM_EXTERN Params are always allowed. PARAM_EXEC Params are parallel-
+ * safe when they can be computed by the leader and their value passed to
+ * workers or are generated within a worker. However we don't always know
+ * whether a param will be generated within a worker when we are parsing a
+ * querytree. In that case we leave it to the consumer to verify that the
+ * current plan level provides these params.
*/
else if (IsA(node, Param))
{
@@ -852,12 +826,12 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context)
if (param->paramkind == PARAM_EXTERN)
return false;
- if (param->paramkind != PARAM_EXEC ||
- !list_member_int(context->safe_param_ids, param->paramid))
- {
- if (max_parallel_hazard_test(PROPARALLEL_RESTRICTED, context))
- return true;
- }
+ if (param->paramkind == PARAM_EXEC)
+ return false;
+
+ if (max_parallel_hazard_test(PROPARALLEL_RESTRICTED, context))
+ return true;
+
return false; /* nothing to recurse to */
}
diff --git a/src/test/regress/expected/incremental_sort.out b/src/test/regress/expected/incremental_sort.out
index 0c3433f8e5..2b52b0ca3c 100644
--- a/src/test/regress/expected/incremental_sort.out
+++ b/src/test/regress/expected/incremental_sort.out
@@ -1597,20 +1597,21 @@ explain (costs off) select distinct
unique1,
(select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
from tenk1 t, generate_series(1, 1000);
- QUERY PLAN
----------------------------------------------------------------------------------
+ QUERY PLAN
+---------------------------------------------------------------------------------------
Unique
- -> Sort
- Sort Key: t.unique1, ((SubPlan 1))
- -> Gather
- Workers Planned: 2
- -> Nested Loop
- -> Parallel Index Only Scan using tenk1_unique1 on tenk1 t
- -> Function Scan on generate_series
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on tenk1
- Index Cond: (unique1 = t.unique1)
-(11 rows)
+ -> Gather Merge
+ Workers Planned: 2
+ -> Unique
+ -> Sort
+ Sort Key: t.unique1, ((SubPlan 1))
+ -> Nested Loop
+ -> Parallel Index Only Scan using tenk1_unique1 on tenk1 t
+ -> Function Scan on generate_series
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on tenk1
+ Index Cond: (unique1 = t.unique1)
+(12 rows)
explain (costs off) select
unique1,
@@ -1619,16 +1620,16 @@ from tenk1 t, generate_series(1, 1000)
order by 1, 2;
QUERY PLAN
---------------------------------------------------------------------------
- Sort
- Sort Key: t.unique1, ((SubPlan 1))
- -> Gather
- Workers Planned: 2
+ Gather Merge
+ Workers Planned: 2
+ -> Sort
+ Sort Key: t.unique1, ((SubPlan 1))
-> Nested Loop
-> Parallel Index Only Scan using tenk1_unique1 on tenk1 t
-> Function Scan on generate_series
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on tenk1
- Index Cond: (unique1 = t.unique1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on tenk1
+ Index Cond: (unique1 = t.unique1)
(10 rows)
-- Parallel sort but with expression not available until the upper rel.
diff --git a/src/test/regress/expected/partition_prune.out b/src/test/regress/expected/partition_prune.out
index 2abf759385..4566260fc4 100644
--- a/src/test/regress/expected/partition_prune.out
+++ b/src/test/regress/expected/partition_prune.out
@@ -1489,60 +1489,64 @@ EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM part p(x) ORDER BY x;
--
-- pruning won't work for mc3p, because some keys are Params
explain (costs off) select * from mc2p t1, lateral (select count(*) from mc3p t2 where t2.a = t1.b and abs(t2.b) = 1 and t2.c = 1) s where t1.a = 1;
- QUERY PLAN
------------------------------------------------------------------------
- Nested Loop
- -> Append
- -> Seq Scan on mc2p1 t1_1
- Filter: (a = 1)
- -> Seq Scan on mc2p2 t1_2
- Filter: (a = 1)
- -> Seq Scan on mc2p_default t1_3
- Filter: (a = 1)
- -> Aggregate
- -> Append
- -> Seq Scan on mc3p0 t2_1
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p1 t2_2
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p2 t2_3
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p3 t2_4
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p4 t2_5
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p5 t2_6
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p6 t2_7
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p7 t2_8
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p_default t2_9
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
-(28 rows)
+ QUERY PLAN
+-----------------------------------------------------------------------------
+ Gather
+ Workers Planned: 2
+ -> Nested Loop
+ -> Parallel Append
+ -> Parallel Seq Scan on mc2p1 t1_1
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p2 t1_2
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p_default t1_3
+ Filter: (a = 1)
+ -> Aggregate
+ -> Append
+ -> Seq Scan on mc3p0 t2_1
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p1 t2_2
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p2 t2_3
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p3 t2_4
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p4 t2_5
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p5 t2_6
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p6 t2_7
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p7 t2_8
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p_default t2_9
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+(30 rows)
-- pruning should work fine, because values for a prefix of keys (a, b) are
-- available
explain (costs off) select * from mc2p t1, lateral (select count(*) from mc3p t2 where t2.c = t1.b and abs(t2.b) = 1 and t2.a = 1) s where t1.a = 1;
- QUERY PLAN
------------------------------------------------------------------------
- Nested Loop
- -> Append
- -> Seq Scan on mc2p1 t1_1
- Filter: (a = 1)
- -> Seq Scan on mc2p2 t1_2
- Filter: (a = 1)
- -> Seq Scan on mc2p_default t1_3
- Filter: (a = 1)
- -> Aggregate
- -> Append
- -> Seq Scan on mc3p0 t2_1
- Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p1 t2_2
- Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p_default t2_3
- Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
-(16 rows)
+ QUERY PLAN
+-----------------------------------------------------------------------------
+ Gather
+ Workers Planned: 2
+ -> Nested Loop
+ -> Parallel Append
+ -> Parallel Seq Scan on mc2p1 t1_1
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p2 t1_2
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p_default t1_3
+ Filter: (a = 1)
+ -> Aggregate
+ -> Append
+ -> Seq Scan on mc3p0 t2_1
+ Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p1 t2_2
+ Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p_default t2_3
+ Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
+(18 rows)
-- also here, because values for all keys are provided
explain (costs off) select * from mc2p t1, lateral (select count(*) from mc3p t2 where t2.a = 1 and abs(t2.b) = 1 and t2.c = 1) s where t1.a = 1;
diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
index 04810d2756..eb59b9437c 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -137,8 +137,8 @@ create table part_pa_test_p2 partition of part_pa_test for values from (0) to (m
explain (costs off)
select (select max((select pa1.b from part_pa_test pa1 where pa1.a = pa2.a)))
from part_pa_test pa2;
- QUERY PLAN
---------------------------------------------------------------
+ QUERY PLAN
+----------------------------------------------------------------
Aggregate
-> Gather
Workers Planned: 3
@@ -148,12 +148,14 @@ explain (costs off)
SubPlan 2
-> Result
SubPlan 1
- -> Append
- -> Seq Scan on part_pa_test_p1 pa1_1
- Filter: (a = pa2.a)
- -> Seq Scan on part_pa_test_p2 pa1_2
- Filter: (a = pa2.a)
-(14 rows)
+ -> Gather
+ Workers Planned: 3
+ -> Parallel Append
+ -> Parallel Seq Scan on part_pa_test_p1 pa1_1
+ Filter: (a = pa2.a)
+ -> Parallel Seq Scan on part_pa_test_p2 pa1_2
+ Filter: (a = pa2.a)
+(16 rows)
drop table part_pa_test;
-- test with leader participation disabled
@@ -320,19 +322,19 @@ explain (costs off, verbose) select
QUERY PLAN
----------------------------------------------------------------------------
Gather
- Output: (SubPlan 1)
+ Output: ((SubPlan 1))
Workers Planned: 4
-> Nested Loop
- Output: t.unique1
+ Output: (SubPlan 1)
-> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
Output: t.unique1
-> Function Scan on pg_catalog.generate_series
Output: generate_series.generate_series
Function Call: generate_series(1, 10)
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(14 rows)
explain (costs off, verbose) select
@@ -341,63 +343,69 @@ explain (costs off, verbose) select
QUERY PLAN
----------------------------------------------------------------------
Gather
- Output: (SubPlan 1)
+ Output: ((SubPlan 1))
Workers Planned: 4
-> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
- Output: t.unique1
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ Output: (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(9 rows)
explain (costs off, verbose) select
(select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
from tenk1 t
limit 1;
- QUERY PLAN
--------------------------------------------------------------------
+ QUERY PLAN
+----------------------------------------------------------------------------
Limit
Output: ((SubPlan 1))
- -> Seq Scan on public.tenk1 t
- Output: (SubPlan 1)
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
-(8 rows)
+ -> Gather
+ Output: ((SubPlan 1))
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(11 rows)
explain (costs off, verbose) select t.unique1
from tenk1 t
where t.unique1 = (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
- QUERY PLAN
--------------------------------------------------------------
- Seq Scan on public.tenk1 t
+ QUERY PLAN
+----------------------------------------------------------------------
+ Gather
Output: t.unique1
- Filter: (t.unique1 = (SubPlan 1))
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
-(7 rows)
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ Filter: (t.unique1 = (SubPlan 1))
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(10 rows)
explain (costs off, verbose) select *
from tenk1 t
order by (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
- QUERY PLAN
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
- Sort
+ QUERY PLAN
+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Gather Merge
Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, ((SubPlan 1))
- Sort Key: ((SubPlan 1))
- -> Gather
- Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, (SubPlan 1)
- Workers Planned: 4
+ Workers Planned: 4
+ -> Sort
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, ((SubPlan 1))
+ Sort Key: ((SubPlan 1))
-> Parallel Seq Scan on public.tenk1 t
- Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(12 rows)
-- test subplan in join/lateral join
@@ -409,14 +417,14 @@ explain (costs off, verbose, timing off) select t.unique1, l.*
QUERY PLAN
----------------------------------------------------------------------
Gather
- Output: t.unique1, (SubPlan 1)
+ Output: t.unique1, ((SubPlan 1))
Workers Planned: 4
-> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
- Output: t.unique1
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ Output: t.unique1, (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(9 rows)
-- this is not parallel-safe due to use of random() within SubLink's testexpr:
@@ -1322,8 +1330,10 @@ SELECT 1 FROM tenk1_vw_sec
-> Parallel Index Only Scan using tenk1_unique1 on tenk1
SubPlan 1
-> Aggregate
- -> Seq Scan on int4_tbl
- Filter: (f1 < tenk1_vw_sec.unique1)
-(9 rows)
+ -> Gather
+ Workers Planned: 1
+ -> Parallel Seq Scan on int4_tbl
+ Filter: (f1 < tenk1_vw_sec.unique1)
+(11 rows)
rollback;
--
2.39.2 (Apple Git-143)
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2023-06-26 06:47 Richard Guo <[email protected]>
parent: James Coleman <[email protected]>
1 sibling, 0 replies; 22+ messages in thread
From: Richard Guo @ 2023-06-26 06:47 UTC (permalink / raw)
To: James Coleman <[email protected]>; +Cc: Tomas Vondra <[email protected]>; vignesh C <[email protected]>; Andres Freund <[email protected]>; Robert Haas <[email protected]>; pgsql-hackers; Amit Kapila <[email protected]>; Tom Lane <[email protected]>
On Mon, Jun 12, 2023 at 10:23 AM James Coleman <[email protected]> wrote:
> BTW are you by any chance testing on ARM macOS? I reproduced the issue
> there, but for some reason I did not reproduce the error (and the plan
> wasn't parallelized) when I tested this on linux. Perhaps I missed
> setting something up; it seems odd.
Hmm, that's weird. I was also testing that query on linux. But please
note that several GUC settings are needed to generate parallel plan for
that query.
set min_parallel_table_scan_size to 0;
set parallel_setup_cost to 0;
set parallel_tuple_cost to 0;
Thanks
Richard
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2023-07-04 01:26 James Coleman <[email protected]>
parent: James Coleman <[email protected]>
1 sibling, 1 reply; 22+ messages in thread
From: James Coleman @ 2023-07-04 01:26 UTC (permalink / raw)
To: Richard Guo <[email protected]>; +Cc: Tomas Vondra <[email protected]>; vignesh C <[email protected]>; Andres Freund <[email protected]>; Robert Haas <[email protected]>; pgsql-hackers; Amit Kapila <[email protected]>; Tom Lane <[email protected]>
On Sun, Jun 11, 2023 at 10:23 PM James Coleman <[email protected]> wrote:
>
> ...
> > And while trying the v9 patch I came across a crash with the query
> > below.
> >
> > set min_parallel_table_scan_size to 0;
> > set parallel_setup_cost to 0;
> > set parallel_tuple_cost to 0;
> >
> > explain (costs off)
> > select * from pg_description t1 where objoid in
> > (select objoid from pg_description t2 where t2.description = t1.description);
> > QUERY PLAN
> > --------------------------------------------------------
> > Seq Scan on pg_description t1
> > Filter: (SubPlan 1)
> > SubPlan 1
> > -> Gather
> > Workers Planned: 2
> > -> Parallel Seq Scan on pg_description t2
> > Filter: (description = t1.description)
> > (7 rows)
> >
> > select * from pg_description t1 where objoid in
> > (select objoid from pg_description t2 where t2.description = t1.description);
> > WARNING: terminating connection because of crash of another server process
> >
> > Seems something is wrong when extracting the argument from the Param in
> > parallel worker.
>
> With what I'm trying to change I don't think this plan should ever be
> generated since it means we'd have to pass a param from the outer seq
> scan into the parallel subplan, which we can't do (currently).
>
> I've attached the full backtrace to the email, but as you hinted at
> the parallel worker is trying to get the param (in this case
> detoasting it), but the param doesn't exist on the worker, so it seg
> faults. Looking at this further I think there's an existing test case
> that exposes the misplanning here (the one right under the comment
> "Parallel Append is not to be used when the subpath depends on the
> outer param" in select_parallel.sql), but it doesn't seg fault because
> the param is an integer, doesn't need to be detoasted, and therefore
> (I think) we skate by (but probably with wrong results in depending on
> the dataset).
>
> Interestingly this is one of the existing test queries my original
> patch approach didn't change, so this gives me something specific to
> work with improving the path. Thanks for testing this and bringing
> this to my attention!
Here's what I've found debugging this:
There's only a single gather path ever created when planning this
query, making it easy to know which one is the problem. That gather
path is created with this stacktrace:
frame #0: 0x0000000105291590
postgres`create_gather_path(root=0x000000013081ae78,
rel=0x000000013080c8e8, subpath=0x000000013081c080,
target=0x000000013081c8c0, required_outer=0x0000000000000000,
rows=0x0000000000000000) at pathnode.c:1971:2
frame #1: 0x0000000105208e54
postgres`generate_gather_paths(root=0x000000013081ae78,
rel=0x000000013080c8e8, override_rows=false) at allpaths.c:3097:4
frame #2: 0x00000001052090ec
postgres`generate_useful_gather_paths(root=0x000000013081ae78,
rel=0x000000013080c8e8, override_rows=false) at allpaths.c:3241:2
frame #3: 0x0000000105258754
postgres`apply_scanjoin_target_to_paths(root=0x000000013081ae78,
rel=0x000000013080c8e8, scanjoin_targets=0x000000013081c978,
scanjoin_targets_contain_srfs=0x0000000000000000,
scanjoin_target_parallel_safe=true, tlist_same_exprs=true) at
planner.c:7696:3
frame #4: 0x00000001052533cc
postgres`grouping_planner(root=0x000000013081ae78, tuple_fraction=0.5)
at planner.c:1611:3
frame #5: 0x0000000105251e9c
postgres`subquery_planner(glob=0x00000001308188d8,
parse=0x000000013080caf8, parent_root=0x000000013080cc38,
hasRecursion=false, tuple_fraction=0.5) at planner.c:1062:2
frame #6: 0x000000010526b134
postgres`make_subplan(root=0x000000013080cc38,
orig_subquery=0x000000013080ff58, subLinkType=ANY_SUBLINK,
subLinkId=0, testexpr=0x000000013080d848, isTopQual=true) at
subselect.c:221:12
frame #7: 0x0000000105268b8c
postgres`process_sublinks_mutator(node=0x000000013080d6d8,
context=0x000000016b0998f8) at subselect.c:1950:10
frame #8: 0x0000000105268ad8
postgres`SS_process_sublinks(root=0x000000013080cc38,
expr=0x000000013080d6d8, isQual=true) at subselect.c:1923:9
frame #9: 0x00000001052527b8
postgres`preprocess_expression(root=0x000000013080cc38,
expr=0x000000013080d6d8, kind=0) at planner.c:1169:10
frame #10: 0x0000000105252954
postgres`preprocess_qual_conditions(root=0x000000013080cc38,
jtnode=0x000000013080d108) at planner.c:1214:14
frame #11: 0x0000000105251580
postgres`subquery_planner(glob=0x00000001308188d8,
parse=0x0000000137010d68, parent_root=0x0000000000000000,
hasRecursion=false, tuple_fraction=0) at planner.c:832:2
frame #12: 0x000000010525042c
postgres`standard_planner(parse=0x0000000137010d68,
query_string="explain (costs off)\nselect * from pg_description t1
where objoid in\n (select objoid from pg_description t2 where
t2.description = t1.description);", cursorOptions=2048,
boundParams=0x0000000000000000) at planner.c:411:9
There aren't any lateral markings on the rels. Additionally the
partial path has param_info=null (I found out from Tom in a separate
thread [1] that this is only set for outer relations from the same
query level).
The only param that I could easily find at first was a single param of
type PARAM_EXTERN in root->plan_params in make_subplan().
I spent a lot of time trying to figure out where we could find the
PARAM_EXEC param that's being fed into the subplan, but it doesn't
seem like we have access to any of these things at the point in the
path creation process that it's interesting to us when inserting the
gather nodes.
Given all of that I settled on this approach:
1. Modify is_parallel_safe() to by default ignore PARAM_EXEC params.
2. Add is_parallel_safe_with_params() that checks for the existence of
such params.
3. Store the required params in a bitmapset on each base rel.
4. Union the bitmapset on join rels.
5. Only insert a gather node if that bitmapset is empty.
I have an intuition that there's some spot (e.g. joins) that we should
be removing params from this set (e.g., when we've satisfied them),
but I haven't been able to come up with such a scenario as yet.
The attached v11 fixes the issue you reported.
Thanks,
James Coleman
Attachments:
[application/octet-stream] v11-0002-Parallelize-correlated-subqueries.patch (27.5K, ../../CAAaqYe8F=TNEF7Dk98dgLkksBt9T0QweTZ4kfif9+q5MKtxQ0Q@mail.gmail.com/2-v11-0002-Parallelize-correlated-subqueries.patch)
download | inline diff:
From 7c19d7f498375cd3ee6b676226c1a9b627298c5f Mon Sep 17 00:00:00 2001
From: jcoleman <[email protected]>
Date: Sun, 2 Jul 2023 15:23:49 -0400
Subject: [PATCH v11 2/2] Parallelize correlated subqueries
When params are provided at the current query level (i.e., are generated
within a single worker and not shared across workers) we can safely
execute these in parallel.
We accomplish this by tracking the PARAM_EXEC params that are needed for
a rel to be safely executed in parallel and only inserting a Gather node
if that set is empty.
---
doc/src/sgml/parallel.sgml | 3 +-
src/backend/optimizer/path/allpaths.c | 24 +++-
src/backend/optimizer/util/clauses.c | 73 +++++++++--
src/backend/optimizer/util/relnode.c | 1 +
src/include/nodes/pathnodes.h | 6 +
src/include/optimizer/clauses.h | 1 +
.../regress/expected/incremental_sort.out | 41 +++----
src/test/regress/expected/partition_prune.out | 104 ++++++++--------
src/test/regress/expected/select_parallel.out | 113 ++++++++++--------
9 files changed, 227 insertions(+), 139 deletions(-)
diff --git a/doc/src/sgml/parallel.sgml b/doc/src/sgml/parallel.sgml
index 5acc9537d6..fd32572ec8 100644
--- a/doc/src/sgml/parallel.sgml
+++ b/doc/src/sgml/parallel.sgml
@@ -518,7 +518,8 @@ EXPLAIN SELECT * FROM pgbench_accounts WHERE filler LIKE '%x%';
<listitem>
<para>
- Plan nodes that reference a correlated <literal>SubPlan</literal>.
+ Plan nodes that reference a correlated <literal>SubPlan</literal> where
+ the result is shared between workers.
</para>
</listitem>
</itemizedlist>
diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index 9bdc70c702..cebad18319 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -632,7 +632,7 @@ set_rel_consider_parallel(PlannerInfo *root, RelOptInfo *rel,
if (proparallel != PROPARALLEL_SAFE)
return;
- if (!is_parallel_safe(root, (Node *) rte->tablesample->args))
+ if (!is_parallel_safe_with_params(root, (Node *) rte->tablesample->args, &rel->params_req_for_parallel))
return;
}
@@ -698,7 +698,7 @@ set_rel_consider_parallel(PlannerInfo *root, RelOptInfo *rel,
case RTE_FUNCTION:
/* Check for parallel-restricted functions. */
- if (!is_parallel_safe(root, (Node *) rte->functions))
+ if (!is_parallel_safe_with_params(root, (Node *) rte->functions, &rel->params_req_for_parallel))
return;
break;
@@ -708,7 +708,7 @@ set_rel_consider_parallel(PlannerInfo *root, RelOptInfo *rel,
case RTE_VALUES:
/* Check for parallel-restricted functions. */
- if (!is_parallel_safe(root, (Node *) rte->values_lists))
+ if (!is_parallel_safe_with_params(root, (Node *) rte->values_lists, &rel->params_req_for_parallel))
return;
break;
@@ -745,14 +745,14 @@ set_rel_consider_parallel(PlannerInfo *root, RelOptInfo *rel,
* outer join clauses work correctly. It would likely break equivalence
* classes, too.
*/
- if (!is_parallel_safe(root, (Node *) rel->baserestrictinfo))
+ if (!is_parallel_safe_with_params(root, (Node *) rel->baserestrictinfo, &rel->params_req_for_parallel))
return;
/*
* Likewise, if the relation's outputs are not parallel-safe, give up.
* (Usually, they're just Vars, but sometimes they're not.)
*/
- if (!is_parallel_safe(root, (Node *) rel->reltarget->exprs))
+ if (!is_parallel_safe_with_params(root, (Node *) rel->reltarget->exprs, &rel->params_req_for_parallel))
return;
/* We have a winner. */
@@ -3069,6 +3069,13 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
if (rel->partial_pathlist == NIL)
return;
+ /*
+ * Wait to insert Gather nodes until all PARAM_EXEC params are provided
+ * within the current rel since we can't pass them to workers.
+ */
+ if (!bms_is_empty(rel->params_req_for_parallel))
+ return;
+
/* Should we override the rel's rowcount estimate? */
if (override_rows)
rowsp = &rows;
@@ -3207,6 +3214,13 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
if (rel->partial_pathlist == NIL)
return;
+ /*
+ * Wait to insert Gather nodes until all PARAM_EXEC params are provided
+ * within the current rel since we can't pass them to workers.
+ */
+ if (!bms_is_empty(rel->params_req_for_parallel))
+ return;
+
/* Should we override the rel's rowcount estimate? */
if (override_rows)
rowsp = &rows;
diff --git a/src/backend/optimizer/util/clauses.c b/src/backend/optimizer/util/clauses.c
index 7f453b04f8..a82a7835c4 100644
--- a/src/backend/optimizer/util/clauses.c
+++ b/src/backend/optimizer/util/clauses.c
@@ -91,6 +91,8 @@ typedef struct
{
char max_hazard; /* worst proparallel hazard found so far */
char max_interesting; /* worst proparallel hazard of interest */
+ bool check_params;
+ Bitmapset **required_params;
List *safe_param_ids; /* PARAM_EXEC Param IDs to treat as safe */
} max_parallel_hazard_context;
@@ -654,6 +656,7 @@ max_parallel_hazard(Query *parse)
context.max_hazard = PROPARALLEL_SAFE;
context.max_interesting = PROPARALLEL_UNSAFE;
+ context.check_params = true;
context.safe_param_ids = NIL;
(void) max_parallel_hazard_walker((Node *) parse, &context);
return context.max_hazard;
@@ -670,8 +673,6 @@ bool
is_parallel_safe(PlannerInfo *root, Node *node)
{
max_parallel_hazard_context context;
- PlannerInfo *proot;
- ListCell *l;
/*
* Even if the original querytree contained nothing unsafe, we need to
@@ -685,6 +686,43 @@ is_parallel_safe(PlannerInfo *root, Node *node)
/* Else use max_parallel_hazard's search logic, but stop on RESTRICTED */
context.max_hazard = PROPARALLEL_SAFE;
context.max_interesting = PROPARALLEL_RESTRICTED;
+ context.check_params = false;
+ context.required_params = NULL;
+
+ return !max_parallel_hazard_walker(node, &context);
+}
+
+/*
+ * is_parallel_safe_with_params
+ * As above, but additionally tracking what PARAM_EXEC params required to
+ * be provided within a worker for a gather to be inserted at this level
+ * of the query. Those required params are passed to the caller through
+ * the required_params argument.
+ *
+ * Note: required_params is only valid if node is otherwise parallel safe.
+ */
+bool
+is_parallel_safe_with_params(PlannerInfo *root, Node *node, Bitmapset **required_params)
+{
+ max_parallel_hazard_context context;
+ PlannerInfo *proot;
+ ListCell *l;
+
+ /*
+ * Even if the original querytree contained nothing unsafe, we need to
+ * search the expression if we have generated any PARAM_EXEC Params while
+ * planning, because those will have to be provided for the expression to
+ * remain parallel safe and there might be one in this expression. But
+ * otherwise we don't need to look.
+ */
+ if (root->glob->maxParallelHazard == PROPARALLEL_SAFE &&
+ root->glob->paramExecTypes == NIL)
+ return true;
+ /* Else use max_parallel_hazard's search logic, but stop on RESTRICTED */
+ context.max_hazard = PROPARALLEL_SAFE;
+ context.max_interesting = PROPARALLEL_RESTRICTED;
+ context.check_params = false;
+ context.required_params = required_params;
context.safe_param_ids = NIL;
/*
@@ -824,13 +862,22 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context)
if (!subplan->parallel_safe &&
max_parallel_hazard_test(PROPARALLEL_RESTRICTED, context))
return true;
- save_safe_param_ids = context->safe_param_ids;
- context->safe_param_ids = list_concat_copy(context->safe_param_ids,
- subplan->paramIds);
+
+ if (context->check_params || context->required_params != NULL)
+ {
+ save_safe_param_ids = context->safe_param_ids;
+ context->safe_param_ids = list_concat_copy(context->safe_param_ids,
+ subplan->paramIds);
+ }
if (max_parallel_hazard_walker(subplan->testexpr, context))
return true; /* no need to restore safe_param_ids */
- list_free(context->safe_param_ids);
- context->safe_param_ids = save_safe_param_ids;
+
+ if (context->check_params || context->required_params != NULL)
+ {
+ list_free(context->safe_param_ids);
+ context->safe_param_ids = save_safe_param_ids;
+ }
+
/* we must also check args, but no special Param treatment there */
if (max_parallel_hazard_walker((Node *) subplan->args, context))
return true;
@@ -849,14 +896,18 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context)
{
Param *param = (Param *) node;
- if (param->paramkind == PARAM_EXTERN)
+ if (param->paramkind != PARAM_EXEC || !(context->check_params || context->required_params != NULL))
return false;
- if (param->paramkind != PARAM_EXEC ||
- !list_member_int(context->safe_param_ids, param->paramid))
+ if (!list_member_int(context->safe_param_ids, param->paramid))
{
if (max_parallel_hazard_test(PROPARALLEL_RESTRICTED, context))
- return true;
+ {
+ if (context->required_params != NULL)
+ *context->required_params = bms_add_member(*context->required_params, param->paramid);
+ if (context->check_params)
+ return true;
+ }
}
return false; /* nothing to recurse to */
}
diff --git a/src/backend/optimizer/util/relnode.c b/src/backend/optimizer/util/relnode.c
index 15e3910b79..2e12a61ceb 100644
--- a/src/backend/optimizer/util/relnode.c
+++ b/src/backend/optimizer/util/relnode.c
@@ -687,6 +687,7 @@ build_join_rel(PlannerInfo *root,
joinrel->consider_startup = (root->tuple_fraction > 0);
joinrel->consider_param_startup = false;
joinrel->consider_parallel = false;
+ joinrel->params_req_for_parallel = bms_union(outer_rel->params_req_for_parallel, inner_rel->params_req_for_parallel);
joinrel->reltarget = create_empty_pathtarget();
joinrel->pathlist = NIL;
joinrel->ppilist = NIL;
diff --git a/src/include/nodes/pathnodes.h b/src/include/nodes/pathnodes.h
index c17b53f7ad..4966010fd0 100644
--- a/src/include/nodes/pathnodes.h
+++ b/src/include/nodes/pathnodes.h
@@ -871,6 +871,12 @@ typedef struct RelOptInfo
/* consider parallel paths? */
bool consider_parallel;
+ /*
+ * Params, if any, required to be provided when consider_parallel is true.
+ * Note: if consider_parallel is false then this is not meaningful.
+ */
+ Bitmapset *params_req_for_parallel;
+
/*
* default result targetlist for Paths scanning this relation; list of
* Vars/Exprs, cost, width
diff --git a/src/include/optimizer/clauses.h b/src/include/optimizer/clauses.h
index cbe0607e85..31cf484d0f 100644
--- a/src/include/optimizer/clauses.h
+++ b/src/include/optimizer/clauses.h
@@ -34,6 +34,7 @@ extern bool contain_subplans(Node *clause);
extern char max_parallel_hazard(Query *parse);
extern bool is_parallel_safe(PlannerInfo *root, Node *node);
+extern bool is_parallel_safe_with_params(PlannerInfo *root, Node *node, Bitmapset **required_params);
extern bool contain_nonstrict_functions(Node *clause);
extern bool contain_exec_param(Node *clause, List *param_ids);
extern bool contain_leaked_vars(Node *clause);
diff --git a/src/test/regress/expected/incremental_sort.out b/src/test/regress/expected/incremental_sort.out
index 0c3433f8e5..2b52b0ca3c 100644
--- a/src/test/regress/expected/incremental_sort.out
+++ b/src/test/regress/expected/incremental_sort.out
@@ -1597,20 +1597,21 @@ explain (costs off) select distinct
unique1,
(select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
from tenk1 t, generate_series(1, 1000);
- QUERY PLAN
----------------------------------------------------------------------------------
+ QUERY PLAN
+---------------------------------------------------------------------------------------
Unique
- -> Sort
- Sort Key: t.unique1, ((SubPlan 1))
- -> Gather
- Workers Planned: 2
- -> Nested Loop
- -> Parallel Index Only Scan using tenk1_unique1 on tenk1 t
- -> Function Scan on generate_series
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on tenk1
- Index Cond: (unique1 = t.unique1)
-(11 rows)
+ -> Gather Merge
+ Workers Planned: 2
+ -> Unique
+ -> Sort
+ Sort Key: t.unique1, ((SubPlan 1))
+ -> Nested Loop
+ -> Parallel Index Only Scan using tenk1_unique1 on tenk1 t
+ -> Function Scan on generate_series
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on tenk1
+ Index Cond: (unique1 = t.unique1)
+(12 rows)
explain (costs off) select
unique1,
@@ -1619,16 +1620,16 @@ from tenk1 t, generate_series(1, 1000)
order by 1, 2;
QUERY PLAN
---------------------------------------------------------------------------
- Sort
- Sort Key: t.unique1, ((SubPlan 1))
- -> Gather
- Workers Planned: 2
+ Gather Merge
+ Workers Planned: 2
+ -> Sort
+ Sort Key: t.unique1, ((SubPlan 1))
-> Nested Loop
-> Parallel Index Only Scan using tenk1_unique1 on tenk1 t
-> Function Scan on generate_series
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on tenk1
- Index Cond: (unique1 = t.unique1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on tenk1
+ Index Cond: (unique1 = t.unique1)
(10 rows)
-- Parallel sort but with expression not available until the upper rel.
diff --git a/src/test/regress/expected/partition_prune.out b/src/test/regress/expected/partition_prune.out
index 2abf759385..4566260fc4 100644
--- a/src/test/regress/expected/partition_prune.out
+++ b/src/test/regress/expected/partition_prune.out
@@ -1489,60 +1489,64 @@ EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM part p(x) ORDER BY x;
--
-- pruning won't work for mc3p, because some keys are Params
explain (costs off) select * from mc2p t1, lateral (select count(*) from mc3p t2 where t2.a = t1.b and abs(t2.b) = 1 and t2.c = 1) s where t1.a = 1;
- QUERY PLAN
------------------------------------------------------------------------
- Nested Loop
- -> Append
- -> Seq Scan on mc2p1 t1_1
- Filter: (a = 1)
- -> Seq Scan on mc2p2 t1_2
- Filter: (a = 1)
- -> Seq Scan on mc2p_default t1_3
- Filter: (a = 1)
- -> Aggregate
- -> Append
- -> Seq Scan on mc3p0 t2_1
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p1 t2_2
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p2 t2_3
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p3 t2_4
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p4 t2_5
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p5 t2_6
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p6 t2_7
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p7 t2_8
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p_default t2_9
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
-(28 rows)
+ QUERY PLAN
+-----------------------------------------------------------------------------
+ Gather
+ Workers Planned: 2
+ -> Nested Loop
+ -> Parallel Append
+ -> Parallel Seq Scan on mc2p1 t1_1
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p2 t1_2
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p_default t1_3
+ Filter: (a = 1)
+ -> Aggregate
+ -> Append
+ -> Seq Scan on mc3p0 t2_1
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p1 t2_2
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p2 t2_3
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p3 t2_4
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p4 t2_5
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p5 t2_6
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p6 t2_7
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p7 t2_8
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p_default t2_9
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+(30 rows)
-- pruning should work fine, because values for a prefix of keys (a, b) are
-- available
explain (costs off) select * from mc2p t1, lateral (select count(*) from mc3p t2 where t2.c = t1.b and abs(t2.b) = 1 and t2.a = 1) s where t1.a = 1;
- QUERY PLAN
------------------------------------------------------------------------
- Nested Loop
- -> Append
- -> Seq Scan on mc2p1 t1_1
- Filter: (a = 1)
- -> Seq Scan on mc2p2 t1_2
- Filter: (a = 1)
- -> Seq Scan on mc2p_default t1_3
- Filter: (a = 1)
- -> Aggregate
- -> Append
- -> Seq Scan on mc3p0 t2_1
- Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p1 t2_2
- Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p_default t2_3
- Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
-(16 rows)
+ QUERY PLAN
+-----------------------------------------------------------------------------
+ Gather
+ Workers Planned: 2
+ -> Nested Loop
+ -> Parallel Append
+ -> Parallel Seq Scan on mc2p1 t1_1
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p2 t1_2
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p_default t1_3
+ Filter: (a = 1)
+ -> Aggregate
+ -> Append
+ -> Seq Scan on mc3p0 t2_1
+ Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p1 t2_2
+ Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p_default t2_3
+ Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
+(18 rows)
-- also here, because values for all keys are provided
explain (costs off) select * from mc2p t1, lateral (select count(*) from mc3p t2 where t2.a = 1 and abs(t2.b) = 1 and t2.c = 1) s where t1.a = 1;
diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
index 0ddcf8e0b1..e422854574 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -320,19 +320,19 @@ explain (costs off, verbose) select
QUERY PLAN
----------------------------------------------------------------------------
Gather
- Output: (SubPlan 1)
+ Output: ((SubPlan 1))
Workers Planned: 4
-> Nested Loop
- Output: t.unique1
+ Output: (SubPlan 1)
-> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
Output: t.unique1
-> Function Scan on pg_catalog.generate_series
Output: generate_series.generate_series
Function Call: generate_series(1, 10)
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(14 rows)
explain (costs off, verbose) select
@@ -341,63 +341,69 @@ explain (costs off, verbose) select
QUERY PLAN
----------------------------------------------------------------------
Gather
- Output: (SubPlan 1)
+ Output: ((SubPlan 1))
Workers Planned: 4
-> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
- Output: t.unique1
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ Output: (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(9 rows)
explain (costs off, verbose) select
(select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
from tenk1 t
limit 1;
- QUERY PLAN
--------------------------------------------------------------------
+ QUERY PLAN
+----------------------------------------------------------------------------
Limit
Output: ((SubPlan 1))
- -> Seq Scan on public.tenk1 t
- Output: (SubPlan 1)
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
-(8 rows)
+ -> Gather
+ Output: ((SubPlan 1))
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(11 rows)
explain (costs off, verbose) select t.unique1
from tenk1 t
where t.unique1 = (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
- QUERY PLAN
--------------------------------------------------------------
- Seq Scan on public.tenk1 t
+ QUERY PLAN
+----------------------------------------------------------------------
+ Gather
Output: t.unique1
- Filter: (t.unique1 = (SubPlan 1))
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
-(7 rows)
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ Filter: (t.unique1 = (SubPlan 1))
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(10 rows)
explain (costs off, verbose) select *
from tenk1 t
order by (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
- QUERY PLAN
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
- Sort
+ QUERY PLAN
+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Gather Merge
Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, ((SubPlan 1))
- Sort Key: ((SubPlan 1))
- -> Gather
- Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, (SubPlan 1)
- Workers Planned: 4
+ Workers Planned: 4
+ -> Sort
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, ((SubPlan 1))
+ Sort Key: ((SubPlan 1))
-> Parallel Seq Scan on public.tenk1 t
- Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(12 rows)
-- test subplan in join/lateral join
@@ -409,14 +415,14 @@ explain (costs off, verbose, timing off) select t.unique1, l.*
QUERY PLAN
----------------------------------------------------------------------
Gather
- Output: t.unique1, (SubPlan 1)
+ Output: t.unique1, ((SubPlan 1))
Workers Planned: 4
-> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
- Output: t.unique1
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ Output: t.unique1, (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(9 rows)
-- can't put a gather at the top of a subplan that takes a param
@@ -1349,9 +1355,12 @@ SELECT 1 FROM tenk1_vw_sec
Workers Planned: 4
-> Parallel Index Only Scan using tenk1_unique1 on tenk1
SubPlan 1
- -> Aggregate
- -> Seq Scan on int4_tbl
- Filter: (f1 < tenk1_vw_sec.unique1)
-(9 rows)
+ -> Finalize Aggregate
+ -> Gather
+ Workers Planned: 1
+ -> Partial Aggregate
+ -> Parallel Seq Scan on int4_tbl
+ Filter: (f1 < tenk1_vw_sec.unique1)
+(12 rows)
rollback;
--
2.39.2 (Apple Git-143)
[application/octet-stream] v11-0001-Add-tests-before-change.patch (9.2K, ../../CAAaqYe8F=TNEF7Dk98dgLkksBt9T0QweTZ4kfif9+q5MKtxQ0Q@mail.gmail.com/3-v11-0001-Add-tests-before-change.patch)
download | inline diff:
From 931e64c7e2e2f819f35b5c7878d1c04ae14d9853 Mon Sep 17 00:00:00 2001
From: jcoleman <[email protected]>
Date: Mon, 26 Sep 2022 20:30:23 -0400
Subject: [PATCH v11 1/2] Add tests before change
---
src/test/regress/expected/select_parallel.out | 136 ++++++++++++++++++
src/test/regress/sql/select_parallel.sql | 32 +++++
2 files changed, 168 insertions(+)
diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
index d88353d496..0ddcf8e0b1 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -311,6 +311,142 @@ select count(*) from tenk1 where (two, four) not in
10000
(1 row)
+-- test parallel plans for queries containing correlated subplans
+-- where the subplan only needs params available from the current
+-- worker's scan.
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t, generate_series(1, 10);
+ QUERY PLAN
+----------------------------------------------------------------------------
+ Gather
+ Output: (SubPlan 1)
+ Workers Planned: 4
+ -> Nested Loop
+ Output: t.unique1
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ -> Function Scan on pg_catalog.generate_series
+ Output: generate_series.generate_series
+ Function Call: generate_series(1, 10)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(14 rows)
+
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t;
+ QUERY PLAN
+----------------------------------------------------------------------
+ Gather
+ Output: (SubPlan 1)
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(9 rows)
+
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t
+ limit 1;
+ QUERY PLAN
+-------------------------------------------------------------------
+ Limit
+ Output: ((SubPlan 1))
+ -> Seq Scan on public.tenk1 t
+ Output: (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(8 rows)
+
+explain (costs off, verbose) select t.unique1
+ from tenk1 t
+ where t.unique1 = (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+ QUERY PLAN
+-------------------------------------------------------------
+ Seq Scan on public.tenk1 t
+ Output: t.unique1
+ Filter: (t.unique1 = (SubPlan 1))
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(7 rows)
+
+explain (costs off, verbose) select *
+ from tenk1 t
+ order by (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+ QUERY PLAN
+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Sort
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, ((SubPlan 1))
+ Sort Key: ((SubPlan 1))
+ -> Gather
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, (SubPlan 1)
+ Workers Planned: 4
+ -> Parallel Seq Scan on public.tenk1 t
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(12 rows)
+
+-- test subplan in join/lateral join
+explain (costs off, verbose, timing off) select t.unique1, l.*
+ from tenk1 t
+ join lateral (
+ select (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1 offset 0)
+ ) l on true;
+ QUERY PLAN
+----------------------------------------------------------------------
+ Gather
+ Output: t.unique1, (SubPlan 1)
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(9 rows)
+
+-- can't put a gather at the top of a subplan that takes a param
+explain (costs off, verbose) select * from tenk1 t where t.two in (
+ select t.two
+ from tenk1
+ join tenk1 t3 on t3.stringu1 = tenk1.stringu1
+ where tenk1.four = t.four
+);
+ QUERY PLAN
+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Seq Scan on public.tenk1 t
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4
+ Filter: (SubPlan 1)
+ SubPlan 1
+ -> Hash Join
+ Output: t.two
+ Hash Cond: (tenk1.stringu1 = t3.stringu1)
+ -> Seq Scan on public.tenk1
+ Output: tenk1.unique1, tenk1.unique2, tenk1.two, tenk1.four, tenk1.ten, tenk1.twenty, tenk1.hundred, tenk1.thousand, tenk1.twothousand, tenk1.fivethous, tenk1.tenthous, tenk1.odd, tenk1.even, tenk1.stringu1, tenk1.stringu2, tenk1.string4
+ Filter: (tenk1.four = t.four)
+ -> Hash
+ Output: t3.stringu1
+ -> Gather
+ Output: t3.stringu1
+ Workers Planned: 4
+ -> Parallel Seq Scan on public.tenk1 t3
+ Output: t3.stringu1
+(17 rows)
+
-- this is not parallel-safe due to use of random() within SubLink's testexpr:
explain (costs off)
select * from tenk1 where (unique1 + random())::integer not in
diff --git a/src/test/regress/sql/select_parallel.sql b/src/test/regress/sql/select_parallel.sql
index 80c914dc02..c6908cb919 100644
--- a/src/test/regress/sql/select_parallel.sql
+++ b/src/test/regress/sql/select_parallel.sql
@@ -111,6 +111,38 @@ explain (costs off)
(select hundred, thousand from tenk2 where thousand > 100);
select count(*) from tenk1 where (two, four) not in
(select hundred, thousand from tenk2 where thousand > 100);
+-- test parallel plans for queries containing correlated subplans
+-- where the subplan only needs params available from the current
+-- worker's scan.
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t, generate_series(1, 10);
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t;
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t
+ limit 1;
+explain (costs off, verbose) select t.unique1
+ from tenk1 t
+ where t.unique1 = (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+explain (costs off, verbose) select *
+ from tenk1 t
+ order by (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+-- test subplan in join/lateral join
+explain (costs off, verbose, timing off) select t.unique1, l.*
+ from tenk1 t
+ join lateral (
+ select (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1 offset 0)
+ ) l on true;
+-- can't put a gather at the top of a subplan that takes a param
+explain (costs off, verbose) select * from tenk1 t where t.two in (
+ select t.two
+ from tenk1
+ join tenk1 t3 on t3.stringu1 = tenk1.stringu1
+ where tenk1.four = t.four
+);
-- this is not parallel-safe due to use of random() within SubLink's testexpr:
explain (costs off)
select * from tenk1 where (unique1 + random())::integer not in
--
2.39.2 (Apple Git-143)
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2024-01-09 07:08 vignesh C <[email protected]>
parent: James Coleman <[email protected]>
0 siblings, 2 replies; 22+ messages in thread
From: vignesh C @ 2024-01-09 07:08 UTC (permalink / raw)
To: James Coleman <[email protected]>; +Cc: Richard Guo <[email protected]>; Tomas Vondra <[email protected]>; Andres Freund <[email protected]>; Robert Haas <[email protected]>; pgsql-hackers; Amit Kapila <[email protected]>; Tom Lane <[email protected]>
On Tue, 4 Jul 2023 at 06:56, James Coleman <[email protected]> wrote:
>
> On Sun, Jun 11, 2023 at 10:23 PM James Coleman <[email protected]> wrote:
> >
> > ...
> > > And while trying the v9 patch I came across a crash with the query
> > > below.
> > >
> > > set min_parallel_table_scan_size to 0;
> > > set parallel_setup_cost to 0;
> > > set parallel_tuple_cost to 0;
> > >
> > > explain (costs off)
> > > select * from pg_description t1 where objoid in
> > > (select objoid from pg_description t2 where t2.description = t1.description);
> > > QUERY PLAN
> > > --------------------------------------------------------
> > > Seq Scan on pg_description t1
> > > Filter: (SubPlan 1)
> > > SubPlan 1
> > > -> Gather
> > > Workers Planned: 2
> > > -> Parallel Seq Scan on pg_description t2
> > > Filter: (description = t1.description)
> > > (7 rows)
> > >
> > > select * from pg_description t1 where objoid in
> > > (select objoid from pg_description t2 where t2.description = t1.description);
> > > WARNING: terminating connection because of crash of another server process
> > >
> > > Seems something is wrong when extracting the argument from the Param in
> > > parallel worker.
> >
> > With what I'm trying to change I don't think this plan should ever be
> > generated since it means we'd have to pass a param from the outer seq
> > scan into the parallel subplan, which we can't do (currently).
> >
> > I've attached the full backtrace to the email, but as you hinted at
> > the parallel worker is trying to get the param (in this case
> > detoasting it), but the param doesn't exist on the worker, so it seg
> > faults. Looking at this further I think there's an existing test case
> > that exposes the misplanning here (the one right under the comment
> > "Parallel Append is not to be used when the subpath depends on the
> > outer param" in select_parallel.sql), but it doesn't seg fault because
> > the param is an integer, doesn't need to be detoasted, and therefore
> > (I think) we skate by (but probably with wrong results in depending on
> > the dataset).
> >
> > Interestingly this is one of the existing test queries my original
> > patch approach didn't change, so this gives me something specific to
> > work with improving the path. Thanks for testing this and bringing
> > this to my attention!
>
> Here's what I've found debugging this:
>
> There's only a single gather path ever created when planning this
> query, making it easy to know which one is the problem. That gather
> path is created with this stacktrace:
>
> frame #0: 0x0000000105291590
> postgres`create_gather_path(root=0x000000013081ae78,
> rel=0x000000013080c8e8, subpath=0x000000013081c080,
> target=0x000000013081c8c0, required_outer=0x0000000000000000,
> rows=0x0000000000000000) at pathnode.c:1971:2
> frame #1: 0x0000000105208e54
> postgres`generate_gather_paths(root=0x000000013081ae78,
> rel=0x000000013080c8e8, override_rows=false) at allpaths.c:3097:4
> frame #2: 0x00000001052090ec
> postgres`generate_useful_gather_paths(root=0x000000013081ae78,
> rel=0x000000013080c8e8, override_rows=false) at allpaths.c:3241:2
> frame #3: 0x0000000105258754
> postgres`apply_scanjoin_target_to_paths(root=0x000000013081ae78,
> rel=0x000000013080c8e8, scanjoin_targets=0x000000013081c978,
> scanjoin_targets_contain_srfs=0x0000000000000000,
> scanjoin_target_parallel_safe=true, tlist_same_exprs=true) at
> planner.c:7696:3
> frame #4: 0x00000001052533cc
> postgres`grouping_planner(root=0x000000013081ae78, tuple_fraction=0.5)
> at planner.c:1611:3
> frame #5: 0x0000000105251e9c
> postgres`subquery_planner(glob=0x00000001308188d8,
> parse=0x000000013080caf8, parent_root=0x000000013080cc38,
> hasRecursion=false, tuple_fraction=0.5) at planner.c:1062:2
> frame #6: 0x000000010526b134
> postgres`make_subplan(root=0x000000013080cc38,
> orig_subquery=0x000000013080ff58, subLinkType=ANY_SUBLINK,
> subLinkId=0, testexpr=0x000000013080d848, isTopQual=true) at
> subselect.c:221:12
> frame #7: 0x0000000105268b8c
> postgres`process_sublinks_mutator(node=0x000000013080d6d8,
> context=0x000000016b0998f8) at subselect.c:1950:10
> frame #8: 0x0000000105268ad8
> postgres`SS_process_sublinks(root=0x000000013080cc38,
> expr=0x000000013080d6d8, isQual=true) at subselect.c:1923:9
> frame #9: 0x00000001052527b8
> postgres`preprocess_expression(root=0x000000013080cc38,
> expr=0x000000013080d6d8, kind=0) at planner.c:1169:10
> frame #10: 0x0000000105252954
> postgres`preprocess_qual_conditions(root=0x000000013080cc38,
> jtnode=0x000000013080d108) at planner.c:1214:14
> frame #11: 0x0000000105251580
> postgres`subquery_planner(glob=0x00000001308188d8,
> parse=0x0000000137010d68, parent_root=0x0000000000000000,
> hasRecursion=false, tuple_fraction=0) at planner.c:832:2
> frame #12: 0x000000010525042c
> postgres`standard_planner(parse=0x0000000137010d68,
> query_string="explain (costs off)\nselect * from pg_description t1
> where objoid in\n (select objoid from pg_description t2 where
> t2.description = t1.description);", cursorOptions=2048,
> boundParams=0x0000000000000000) at planner.c:411:9
>
> There aren't any lateral markings on the rels. Additionally the
> partial path has param_info=null (I found out from Tom in a separate
> thread [1] that this is only set for outer relations from the same
> query level).
>
> The only param that I could easily find at first was a single param of
> type PARAM_EXTERN in root->plan_params in make_subplan().
>
> I spent a lot of time trying to figure out where we could find the
> PARAM_EXEC param that's being fed into the subplan, but it doesn't
> seem like we have access to any of these things at the point in the
> path creation process that it's interesting to us when inserting the
> gather nodes.
>
> Given all of that I settled on this approach:
> 1. Modify is_parallel_safe() to by default ignore PARAM_EXEC params.
> 2. Add is_parallel_safe_with_params() that checks for the existence of
> such params.
> 3. Store the required params in a bitmapset on each base rel.
> 4. Union the bitmapset on join rels.
> 5. Only insert a gather node if that bitmapset is empty.
>
> I have an intuition that there's some spot (e.g. joins) that we should
> be removing params from this set (e.g., when we've satisfied them),
> but I haven't been able to come up with such a scenario as yet.
>
> The attached v11 fixes the issue you reported.
One of the tests has failed in CFBot at [1] with:
+++ /tmp/cirrus-ci-build/build/testrun/pg_upgrade/002_pg_upgrade/data/results/select_parallel.out
2023-12-20 20:08:42.480004000 +0000
@@ -137,23 +137,24 @@
explain (costs off)
select (select max((select pa1.b from part_pa_test pa1 where pa1.a = pa2.a)))
from part_pa_test pa2;
- QUERY PLAN
---------------------------------------------------------------
- Aggregate
+ QUERY PLAN
+--------------------------------------------------------------------
+ Finalize Aggregate
-> Gather
Workers Planned: 3
- -> Parallel Append
- -> Parallel Seq Scan on part_pa_test_p1 pa2_1
- -> Parallel Seq Scan on part_pa_test_p2 pa2_2
+ -> Partial Aggregate
+ -> Parallel Append
+ -> Parallel Seq Scan on part_pa_test_p1 pa2_1
+ -> Parallel Seq Scan on part_pa_test_p2 pa2_2
+ SubPlan 1
+ -> Append
+ -> Seq Scan on part_pa_test_p1 pa1_1
+ Filter: (a = pa2.a)
+ -> Seq Scan on part_pa_test_p2 pa1_2
+ Filter: (a = pa2.a)
SubPlan 2
-> Result
- SubPlan 1
- -> Append
- -> Seq Scan on part_pa_test_p1 pa1_1
- Filter: (a = pa2.a)
- -> Seq Scan on part_pa_test_p2 pa1_2
- Filter: (a = pa2.a)
-(14 rows)
+(15 rows)
More details of the failure is available at [2].
[1] - https://cirrus-ci.com/task/5685696451575808
[2] - https://api.cirrus-ci.com/v1/artifact/task/5685696451575808/testrun/build/testrun/pg_upgrade/002_pg_...
Regards,
Vignesh
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2024-01-23 14:33 Akshat Jaimini <[email protected]>
parent: vignesh C <[email protected]>
1 sibling, 1 reply; 22+ messages in thread
From: Akshat Jaimini @ 2024-01-23 14:33 UTC (permalink / raw)
To: [email protected]; +Cc: James Coleman <[email protected]>
Hello!
I was going through the previous conversations for this particular patch and it seems that this patch failed some tests previously?
Imo we should move it to the next CF so that the remaining issues can be resolved accordingly.
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2024-01-23 20:20 Robert Haas <[email protected]>
parent: Akshat Jaimini <[email protected]>
0 siblings, 1 reply; 22+ messages in thread
From: Robert Haas @ 2024-01-23 20:20 UTC (permalink / raw)
To: Akshat Jaimini <[email protected]>; +Cc: [email protected]; James Coleman <[email protected]>
On Tue, Jan 23, 2024 at 9:34 AM Akshat Jaimini <[email protected]> wrote:
> Hello!
> I was going through the previous conversations for this particular patch and it seems that this patch failed some tests previously?
> Imo we should move it to the next CF so that the remaining issues can be resolved accordingly.
So I guess the question here is whether this is thought to be ready
for serious review or whether it is still thought to need work. If the
latter, it should be marked RwF until that happens -- if the former,
then we should try to review it rather than letting it languish
forever.
--
Robert Haas
EDB: http://www.enterprisedb.com
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2024-01-30 16:16 Akshat Jaimini <[email protected]>
parent: Robert Haas <[email protected]>
0 siblings, 1 reply; 22+ messages in thread
From: Akshat Jaimini @ 2024-01-30 16:16 UTC (permalink / raw)
To: [email protected]; +Cc: James Coleman <[email protected]>
I think we should move this patch to the next CF as I believe that work is still going on resolving the last reported bug.
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2024-01-30 16:54 Robert Haas <[email protected]>
parent: Akshat Jaimini <[email protected]>
0 siblings, 1 reply; 22+ messages in thread
From: Robert Haas @ 2024-01-30 16:54 UTC (permalink / raw)
To: Akshat Jaimini <[email protected]>; +Cc: [email protected]; James Coleman <[email protected]>
On Tue, Jan 30, 2024 at 11:17 AM Akshat Jaimini <[email protected]> wrote:
> I think we should move this patch to the next CF as I believe that work is still going on resolving the last reported bug.
We shouldn't just keep pushing this forward to the next CF. It's been
idle since July. If it needs more work, mark it RwF and it can be
reopened when there's something for a reviewer to do.
--
Robert Haas
EDB: http://www.enterprisedb.com
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2024-01-31 02:51 James Coleman <[email protected]>
parent: vignesh C <[email protected]>
1 sibling, 1 reply; 22+ messages in thread
From: James Coleman @ 2024-01-31 02:51 UTC (permalink / raw)
To: vignesh C <[email protected]>; +Cc: Richard Guo <[email protected]>; Tomas Vondra <[email protected]>; Andres Freund <[email protected]>; Robert Haas <[email protected]>; pgsql-hackers; Amit Kapila <[email protected]>; Tom Lane <[email protected]>
On Tue, Jan 9, 2024 at 2:09 AM vignesh C <[email protected]> wrote:
> ...
> > Given all of that I settled on this approach:
> > 1. Modify is_parallel_safe() to by default ignore PARAM_EXEC params.
> > 2. Add is_parallel_safe_with_params() that checks for the existence of
> > such params.
> > 3. Store the required params in a bitmapset on each base rel.
> > 4. Union the bitmapset on join rels.
> > 5. Only insert a gather node if that bitmapset is empty.
> >
> > I have an intuition that there's some spot (e.g. joins) that we should
> > be removing params from this set (e.g., when we've satisfied them),
> > but I haven't been able to come up with such a scenario as yet.
> >
> > The attached v11 fixes the issue you reported.
>
> One of the tests has failed in CFBot at [1] with:
> +++ /tmp/cirrus-ci-build/build/testrun/pg_upgrade/002_pg_upgrade/data/results/select_parallel.out
> 2023-12-20 20:08:42.480004000 +0000
> @@ -137,23 +137,24 @@
> explain (costs off)
> select (select max((select pa1.b from part_pa_test pa1 where pa1.a = pa2.a)))
> from part_pa_test pa2;
> - QUERY PLAN
> ---------------------------------------------------------------
> - Aggregate
> + QUERY PLAN
> +--------------------------------------------------------------------
> + Finalize Aggregate
> -> Gather
> Workers Planned: 3
> - -> Parallel Append
> - -> Parallel Seq Scan on part_pa_test_p1 pa2_1
> - -> Parallel Seq Scan on part_pa_test_p2 pa2_2
> + -> Partial Aggregate
> + -> Parallel Append
> + -> Parallel Seq Scan on part_pa_test_p1 pa2_1
> + -> Parallel Seq Scan on part_pa_test_p2 pa2_2
> + SubPlan 1
> + -> Append
> + -> Seq Scan on part_pa_test_p1 pa1_1
> + Filter: (a = pa2.a)
> + -> Seq Scan on part_pa_test_p2 pa1_2
> + Filter: (a = pa2.a)
> SubPlan 2
> -> Result
> - SubPlan 1
> - -> Append
> - -> Seq Scan on part_pa_test_p1 pa1_1
> - Filter: (a = pa2.a)
> - -> Seq Scan on part_pa_test_p2 pa1_2
> - Filter: (a = pa2.a)
> -(14 rows)
> +(15 rows)
>
> More details of the failure is available at [2].
>
> [1] - https://cirrus-ci.com/task/5685696451575808
> [2] - https://api.cirrus-ci.com/v1/artifact/task/5685696451575808/testrun/build/testrun/pg_upgrade/002_pg_...
Thanks for noting this here.
I've finally had a chance to look at this, and I don't believe there's
any real failure here, merely drift of how the planner works on master
resulting in this query now being eligible for a different plan shape.
I was a bit wary at first because the changing test query is one I'd
previously referenced in [1] as likely exposing the bug I'd fixed
where params where being used across worker boundaries. However
looking at the diff in the patch at that point (v10) that particular
test query formed a different plan shape (there were two gather nodes
being created, and params crossing between them).
But in the current revision of master with the current patch applied
that's no longer true: we have a Gather node, and the Subplan using
the param is properly under that Gather node, and the param should be
being both generated and consumed within the same worker process.
So I've updated the patch to show that plan change as part of the diff.
See attached v12
Regards,
James Coleman
1: https://www.postgresql.org/message-id/CAAaqYe-_TObm5KwmZLYXBJ3BJGh4cUZWM0v1mY1gWTMkRNQXDQ%40mail.gma...
Attachments:
[application/octet-stream] v12-0002-Parallelize-correlated-subqueries.patch (31.3K, ../../CAAaqYe-Ld1q_b9pOf0PRjfQLWCgdkTvnopfeXgmrSvEQu0Ox=A@mail.gmail.com/2-v12-0002-Parallelize-correlated-subqueries.patch)
download | inline diff:
From ac42a4080e002627af2c221591d7c872612a7ce7 Mon Sep 17 00:00:00 2001
From: jcoleman <[email protected]>
Date: Sun, 2 Jul 2023 15:23:49 -0400
Subject: [PATCH v12 2/2] Parallelize correlated subqueries
When params are provided at the current query level (i.e., are generated
within a single worker and not shared across workers) we can safely
execute these in parallel.
We accomplish this by tracking the PARAM_EXEC params that are needed for
a rel to be safely executed in parallel and only inserting a Gather node
if that set is empty.
---
doc/src/sgml/parallel.sgml | 3 +-
src/backend/optimizer/path/allpaths.c | 24 ++-
src/backend/optimizer/util/clauses.c | 73 ++++++--
src/backend/optimizer/util/relnode.c | 1 +
src/include/nodes/pathnodes.h | 6 +
src/include/optimizer/clauses.h | 1 +
.../regress/expected/incremental_sort.out | 41 ++---
src/test/regress/expected/partition_prune.out | 104 +++++------
src/test/regress/expected/select_parallel.out | 161 ++++++++++--------
src/test/regress/sql/select_parallel.sql | 8 +-
10 files changed, 266 insertions(+), 156 deletions(-)
diff --git a/doc/src/sgml/parallel.sgml b/doc/src/sgml/parallel.sgml
index 5acc9537d6..fd32572ec8 100644
--- a/doc/src/sgml/parallel.sgml
+++ b/doc/src/sgml/parallel.sgml
@@ -518,7 +518,8 @@ EXPLAIN SELECT * FROM pgbench_accounts WHERE filler LIKE '%x%';
<listitem>
<para>
- Plan nodes that reference a correlated <literal>SubPlan</literal>.
+ Plan nodes that reference a correlated <literal>SubPlan</literal> where
+ the result is shared between workers.
</para>
</listitem>
</itemizedlist>
diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index 84c4de488a..37c44d393b 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -632,7 +632,7 @@ set_rel_consider_parallel(PlannerInfo *root, RelOptInfo *rel,
if (proparallel != PROPARALLEL_SAFE)
return;
- if (!is_parallel_safe(root, (Node *) rte->tablesample->args))
+ if (!is_parallel_safe_with_params(root, (Node *) rte->tablesample->args, &rel->params_req_for_parallel))
return;
}
@@ -698,7 +698,7 @@ set_rel_consider_parallel(PlannerInfo *root, RelOptInfo *rel,
case RTE_FUNCTION:
/* Check for parallel-restricted functions. */
- if (!is_parallel_safe(root, (Node *) rte->functions))
+ if (!is_parallel_safe_with_params(root, (Node *) rte->functions, &rel->params_req_for_parallel))
return;
break;
@@ -708,7 +708,7 @@ set_rel_consider_parallel(PlannerInfo *root, RelOptInfo *rel,
case RTE_VALUES:
/* Check for parallel-restricted functions. */
- if (!is_parallel_safe(root, (Node *) rte->values_lists))
+ if (!is_parallel_safe_with_params(root, (Node *) rte->values_lists, &rel->params_req_for_parallel))
return;
break;
@@ -745,14 +745,14 @@ set_rel_consider_parallel(PlannerInfo *root, RelOptInfo *rel,
* outer join clauses work correctly. It would likely break equivalence
* classes, too.
*/
- if (!is_parallel_safe(root, (Node *) rel->baserestrictinfo))
+ if (!is_parallel_safe_with_params(root, (Node *) rel->baserestrictinfo, &rel->params_req_for_parallel))
return;
/*
* Likewise, if the relation's outputs are not parallel-safe, give up.
* (Usually, they're just Vars, but sometimes they're not.)
*/
- if (!is_parallel_safe(root, (Node *) rel->reltarget->exprs))
+ if (!is_parallel_safe_with_params(root, (Node *) rel->reltarget->exprs, &rel->params_req_for_parallel))
return;
/* We have a winner. */
@@ -3071,6 +3071,13 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
if (rel->partial_pathlist == NIL)
return;
+ /*
+ * Wait to insert Gather nodes until all PARAM_EXEC params are provided
+ * within the current rel since we can't pass them to workers.
+ */
+ if (!bms_is_empty(rel->params_req_for_parallel))
+ return;
+
/* Should we override the rel's rowcount estimate? */
if (override_rows)
rowsp = &rows;
@@ -3209,6 +3216,13 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
if (rel->partial_pathlist == NIL)
return;
+ /*
+ * Wait to insert Gather nodes until all PARAM_EXEC params are provided
+ * within the current rel since we can't pass them to workers.
+ */
+ if (!bms_is_empty(rel->params_req_for_parallel))
+ return;
+
/* Should we override the rel's rowcount estimate? */
if (override_rows)
rowsp = &rows;
diff --git a/src/backend/optimizer/util/clauses.c b/src/backend/optimizer/util/clauses.c
index 94eb56a1e7..9acad18154 100644
--- a/src/backend/optimizer/util/clauses.c
+++ b/src/backend/optimizer/util/clauses.c
@@ -91,6 +91,8 @@ typedef struct
{
char max_hazard; /* worst proparallel hazard found so far */
char max_interesting; /* worst proparallel hazard of interest */
+ bool check_params;
+ Bitmapset **required_params;
List *safe_param_ids; /* PARAM_EXEC Param IDs to treat as safe */
} max_parallel_hazard_context;
@@ -720,6 +722,7 @@ max_parallel_hazard(Query *parse)
context.max_hazard = PROPARALLEL_SAFE;
context.max_interesting = PROPARALLEL_UNSAFE;
+ context.check_params = true;
context.safe_param_ids = NIL;
(void) max_parallel_hazard_walker((Node *) parse, &context);
return context.max_hazard;
@@ -736,8 +739,6 @@ bool
is_parallel_safe(PlannerInfo *root, Node *node)
{
max_parallel_hazard_context context;
- PlannerInfo *proot;
- ListCell *l;
/*
* Even if the original querytree contained nothing unsafe, we need to
@@ -751,6 +752,43 @@ is_parallel_safe(PlannerInfo *root, Node *node)
/* Else use max_parallel_hazard's search logic, but stop on RESTRICTED */
context.max_hazard = PROPARALLEL_SAFE;
context.max_interesting = PROPARALLEL_RESTRICTED;
+ context.check_params = false;
+ context.required_params = NULL;
+
+ return !max_parallel_hazard_walker(node, &context);
+}
+
+/*
+ * is_parallel_safe_with_params
+ * As above, but additionally tracking what PARAM_EXEC params required to
+ * be provided within a worker for a gather to be inserted at this level
+ * of the query. Those required params are passed to the caller through
+ * the required_params argument.
+ *
+ * Note: required_params is only valid if node is otherwise parallel safe.
+ */
+bool
+is_parallel_safe_with_params(PlannerInfo *root, Node *node, Bitmapset **required_params)
+{
+ max_parallel_hazard_context context;
+ PlannerInfo *proot;
+ ListCell *l;
+
+ /*
+ * Even if the original querytree contained nothing unsafe, we need to
+ * search the expression if we have generated any PARAM_EXEC Params while
+ * planning, because those will have to be provided for the expression to
+ * remain parallel safe and there might be one in this expression. But
+ * otherwise we don't need to look.
+ */
+ if (root->glob->maxParallelHazard == PROPARALLEL_SAFE &&
+ root->glob->paramExecTypes == NIL)
+ return true;
+ /* Else use max_parallel_hazard's search logic, but stop on RESTRICTED */
+ context.max_hazard = PROPARALLEL_SAFE;
+ context.max_interesting = PROPARALLEL_RESTRICTED;
+ context.check_params = false;
+ context.required_params = required_params;
context.safe_param_ids = NIL;
/*
@@ -890,13 +928,22 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context)
if (!subplan->parallel_safe &&
max_parallel_hazard_test(PROPARALLEL_RESTRICTED, context))
return true;
- save_safe_param_ids = context->safe_param_ids;
- context->safe_param_ids = list_concat_copy(context->safe_param_ids,
- subplan->paramIds);
+
+ if (context->check_params || context->required_params != NULL)
+ {
+ save_safe_param_ids = context->safe_param_ids;
+ context->safe_param_ids = list_concat_copy(context->safe_param_ids,
+ subplan->paramIds);
+ }
if (max_parallel_hazard_walker(subplan->testexpr, context))
return true; /* no need to restore safe_param_ids */
- list_free(context->safe_param_ids);
- context->safe_param_ids = save_safe_param_ids;
+
+ if (context->check_params || context->required_params != NULL)
+ {
+ list_free(context->safe_param_ids);
+ context->safe_param_ids = save_safe_param_ids;
+ }
+
/* we must also check args, but no special Param treatment there */
if (max_parallel_hazard_walker((Node *) subplan->args, context))
return true;
@@ -915,14 +962,18 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context)
{
Param *param = (Param *) node;
- if (param->paramkind == PARAM_EXTERN)
+ if (param->paramkind != PARAM_EXEC || !(context->check_params || context->required_params != NULL))
return false;
- if (param->paramkind != PARAM_EXEC ||
- !list_member_int(context->safe_param_ids, param->paramid))
+ if (!list_member_int(context->safe_param_ids, param->paramid))
{
if (max_parallel_hazard_test(PROPARALLEL_RESTRICTED, context))
- return true;
+ {
+ if (context->required_params != NULL)
+ *context->required_params = bms_add_member(*context->required_params, param->paramid);
+ if (context->check_params)
+ return true;
+ }
}
return false; /* nothing to recurse to */
}
diff --git a/src/backend/optimizer/util/relnode.c b/src/backend/optimizer/util/relnode.c
index 22d01cef5b..58228ff269 100644
--- a/src/backend/optimizer/util/relnode.c
+++ b/src/backend/optimizer/util/relnode.c
@@ -699,6 +699,7 @@ build_join_rel(PlannerInfo *root,
joinrel->consider_startup = (root->tuple_fraction > 0);
joinrel->consider_param_startup = false;
joinrel->consider_parallel = false;
+ joinrel->params_req_for_parallel = bms_union(outer_rel->params_req_for_parallel, inner_rel->params_req_for_parallel);
joinrel->reltarget = create_empty_pathtarget();
joinrel->pathlist = NIL;
joinrel->ppilist = NIL;
diff --git a/src/include/nodes/pathnodes.h b/src/include/nodes/pathnodes.h
index 137da178dc..cb5721963d 100644
--- a/src/include/nodes/pathnodes.h
+++ b/src/include/nodes/pathnodes.h
@@ -871,6 +871,12 @@ typedef struct RelOptInfo
/* consider parallel paths? */
bool consider_parallel;
+ /*
+ * Params, if any, required to be provided when consider_parallel is true.
+ * Note: if consider_parallel is false then this is not meaningful.
+ */
+ Bitmapset *params_req_for_parallel;
+
/*
* default result targetlist for Paths scanning this relation; list of
* Vars/Exprs, cost, width
diff --git a/src/include/optimizer/clauses.h b/src/include/optimizer/clauses.h
index 34b301e537..eb4334f565 100644
--- a/src/include/optimizer/clauses.h
+++ b/src/include/optimizer/clauses.h
@@ -34,6 +34,7 @@ extern bool contain_subplans(Node *clause);
extern char max_parallel_hazard(Query *parse);
extern bool is_parallel_safe(PlannerInfo *root, Node *node);
+extern bool is_parallel_safe_with_params(PlannerInfo *root, Node *node, Bitmapset **required_params);
extern bool contain_nonstrict_functions(Node *clause);
extern bool contain_exec_param(Node *clause, List *param_ids);
extern bool contain_leaked_vars(Node *clause);
diff --git a/src/test/regress/expected/incremental_sort.out b/src/test/regress/expected/incremental_sort.out
index 7fdb685313..df1c52414e 100644
--- a/src/test/regress/expected/incremental_sort.out
+++ b/src/test/regress/expected/incremental_sort.out
@@ -1597,20 +1597,21 @@ explain (costs off) select distinct
unique1,
(select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
from tenk1 t, generate_series(1, 1000);
- QUERY PLAN
----------------------------------------------------------------------------------
+ QUERY PLAN
+---------------------------------------------------------------------------------------
Unique
- -> Sort
- Sort Key: t.unique1, ((SubPlan 1))
- -> Gather
- Workers Planned: 2
- -> Nested Loop
- -> Parallel Index Only Scan using tenk1_unique1 on tenk1 t
- -> Function Scan on generate_series
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on tenk1
- Index Cond: (unique1 = t.unique1)
-(11 rows)
+ -> Gather Merge
+ Workers Planned: 2
+ -> Unique
+ -> Sort
+ Sort Key: t.unique1, ((SubPlan 1))
+ -> Nested Loop
+ -> Parallel Index Only Scan using tenk1_unique1 on tenk1 t
+ -> Function Scan on generate_series
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on tenk1
+ Index Cond: (unique1 = t.unique1)
+(12 rows)
explain (costs off) select
unique1,
@@ -1619,16 +1620,16 @@ from tenk1 t, generate_series(1, 1000)
order by 1, 2;
QUERY PLAN
---------------------------------------------------------------------------
- Sort
- Sort Key: t.unique1, ((SubPlan 1))
- -> Gather
- Workers Planned: 2
+ Gather Merge
+ Workers Planned: 2
+ -> Sort
+ Sort Key: t.unique1, ((SubPlan 1))
-> Nested Loop
-> Parallel Index Only Scan using tenk1_unique1 on tenk1 t
-> Function Scan on generate_series
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on tenk1
- Index Cond: (unique1 = t.unique1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on tenk1
+ Index Cond: (unique1 = t.unique1)
(10 rows)
-- Parallel sort but with expression not available until the upper rel.
diff --git a/src/test/regress/expected/partition_prune.out b/src/test/regress/expected/partition_prune.out
index 9a4c48c055..c45590fdfe 100644
--- a/src/test/regress/expected/partition_prune.out
+++ b/src/test/regress/expected/partition_prune.out
@@ -1489,60 +1489,64 @@ EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM part p(x) ORDER BY x;
--
-- pruning won't work for mc3p, because some keys are Params
explain (costs off) select * from mc2p t1, lateral (select count(*) from mc3p t2 where t2.a = t1.b and abs(t2.b) = 1 and t2.c = 1) s where t1.a = 1;
- QUERY PLAN
------------------------------------------------------------------------
- Nested Loop
- -> Append
- -> Seq Scan on mc2p1 t1_1
- Filter: (a = 1)
- -> Seq Scan on mc2p2 t1_2
- Filter: (a = 1)
- -> Seq Scan on mc2p_default t1_3
- Filter: (a = 1)
- -> Aggregate
- -> Append
- -> Seq Scan on mc3p0 t2_1
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p1 t2_2
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p2 t2_3
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p3 t2_4
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p4 t2_5
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p5 t2_6
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p6 t2_7
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p7 t2_8
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p_default t2_9
- Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
-(28 rows)
+ QUERY PLAN
+-----------------------------------------------------------------------------
+ Gather
+ Workers Planned: 2
+ -> Nested Loop
+ -> Parallel Append
+ -> Parallel Seq Scan on mc2p1 t1_1
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p2 t1_2
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p_default t1_3
+ Filter: (a = 1)
+ -> Aggregate
+ -> Append
+ -> Seq Scan on mc3p0 t2_1
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p1 t2_2
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p2 t2_3
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p3 t2_4
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p4 t2_5
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p5 t2_6
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p6 t2_7
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p7 t2_8
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p_default t2_9
+ Filter: ((a = t1.b) AND (c = 1) AND (abs(b) = 1))
+(30 rows)
-- pruning should work fine, because values for a prefix of keys (a, b) are
-- available
explain (costs off) select * from mc2p t1, lateral (select count(*) from mc3p t2 where t2.c = t1.b and abs(t2.b) = 1 and t2.a = 1) s where t1.a = 1;
- QUERY PLAN
------------------------------------------------------------------------
- Nested Loop
- -> Append
- -> Seq Scan on mc2p1 t1_1
- Filter: (a = 1)
- -> Seq Scan on mc2p2 t1_2
- Filter: (a = 1)
- -> Seq Scan on mc2p_default t1_3
- Filter: (a = 1)
- -> Aggregate
- -> Append
- -> Seq Scan on mc3p0 t2_1
- Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p1 t2_2
- Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
- -> Seq Scan on mc3p_default t2_3
- Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
-(16 rows)
+ QUERY PLAN
+-----------------------------------------------------------------------------
+ Gather
+ Workers Planned: 2
+ -> Nested Loop
+ -> Parallel Append
+ -> Parallel Seq Scan on mc2p1 t1_1
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p2 t1_2
+ Filter: (a = 1)
+ -> Parallel Seq Scan on mc2p_default t1_3
+ Filter: (a = 1)
+ -> Aggregate
+ -> Append
+ -> Seq Scan on mc3p0 t2_1
+ Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p1 t2_2
+ Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
+ -> Seq Scan on mc3p_default t2_3
+ Filter: ((c = t1.b) AND (a = 1) AND (abs(b) = 1))
+(18 rows)
-- also here, because values for all keys are provided
explain (costs off) select * from mc2p t1, lateral (select count(*) from mc3p t2 where t2.a = 1 and abs(t2.b) = 1 and t2.c = 1) s where t1.a = 1;
diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
index b76132ffe4..a6c4d6bf3f 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -130,30 +130,48 @@ select sp_test_func() order by 1;
foo
(2 rows)
--- Parallel Append is not to be used when the subpath depends on the outer param
+-- Parallel Append is can be used when the subpath depends on the outer params
+-- when those params are consumed within the worker that generates them.
create table part_pa_test(a int, b int) partition by range(a);
create table part_pa_test_p1 partition of part_pa_test for values from (minvalue) to (0);
create table part_pa_test_p2 partition of part_pa_test for values from (0) to (maxvalue);
-explain (costs off)
+explain (costs off, verbose)
select (select max((select pa1.b from part_pa_test pa1 where pa1.a = pa2.a)))
from part_pa_test pa2;
- QUERY PLAN
---------------------------------------------------------------
- Aggregate
+ QUERY PLAN
+---------------------------------------------------------------------------
+ Finalize Aggregate
+ Output: (SubPlan 2)
-> Gather
+ Output: (PARTIAL max((SubPlan 1)))
Workers Planned: 3
- -> Parallel Append
- -> Parallel Seq Scan on part_pa_test_p1 pa2_1
- -> Parallel Seq Scan on part_pa_test_p2 pa2_2
+ -> Partial Aggregate
+ Output: PARTIAL max((SubPlan 1))
+ -> Parallel Append
+ -> Parallel Seq Scan on public.part_pa_test_p1 pa2_1
+ Output: pa2_1.a
+ -> Parallel Seq Scan on public.part_pa_test_p2 pa2_2
+ Output: pa2_2.a
+ SubPlan 1
+ -> Append
+ -> Seq Scan on public.part_pa_test_p1 pa1_1
+ Output: pa1_1.b
+ Filter: (pa1_1.a = pa2.a)
+ -> Seq Scan on public.part_pa_test_p2 pa1_2
+ Output: pa1_2.b
+ Filter: (pa1_2.a = pa2.a)
SubPlan 2
-> Result
- SubPlan 1
- -> Append
- -> Seq Scan on part_pa_test_p1 pa1_1
- Filter: (a = pa2.a)
- -> Seq Scan on part_pa_test_p2 pa1_2
- Filter: (a = pa2.a)
-(14 rows)
+ Output: max((SubPlan 1))
+(23 rows)
+
+insert into part_pa_test(a, b) values (-1, 1), (1, 3);
+select (select max((select pa1.b from part_pa_test pa1 where pa1.a = pa2.a)))
+from part_pa_test pa2;
+ max
+-----
+ 3
+(1 row)
drop table part_pa_test;
-- test with leader participation disabled
@@ -320,19 +338,19 @@ explain (costs off, verbose) select
QUERY PLAN
----------------------------------------------------------------------------
Gather
- Output: (SubPlan 1)
+ Output: ((SubPlan 1))
Workers Planned: 4
-> Nested Loop
- Output: t.unique1
+ Output: (SubPlan 1)
-> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
Output: t.unique1
-> Function Scan on pg_catalog.generate_series
Output: generate_series.generate_series
Function Call: generate_series(1, 10)
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(14 rows)
explain (costs off, verbose) select
@@ -341,63 +359,69 @@ explain (costs off, verbose) select
QUERY PLAN
----------------------------------------------------------------------
Gather
- Output: (SubPlan 1)
+ Output: ((SubPlan 1))
Workers Planned: 4
-> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
- Output: t.unique1
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ Output: (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(9 rows)
explain (costs off, verbose) select
(select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
from tenk1 t
limit 1;
- QUERY PLAN
--------------------------------------------------------------------
+ QUERY PLAN
+----------------------------------------------------------------------------
Limit
Output: ((SubPlan 1))
- -> Seq Scan on public.tenk1 t
- Output: (SubPlan 1)
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
-(8 rows)
+ -> Gather
+ Output: ((SubPlan 1))
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(11 rows)
explain (costs off, verbose) select t.unique1
from tenk1 t
where t.unique1 = (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
- QUERY PLAN
--------------------------------------------------------------
- Seq Scan on public.tenk1 t
+ QUERY PLAN
+----------------------------------------------------------------------
+ Gather
Output: t.unique1
- Filter: (t.unique1 = (SubPlan 1))
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
-(7 rows)
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ Filter: (t.unique1 = (SubPlan 1))
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(10 rows)
explain (costs off, verbose) select *
from tenk1 t
order by (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
- QUERY PLAN
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
- Sort
+ QUERY PLAN
+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Gather Merge
Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, ((SubPlan 1))
- Sort Key: ((SubPlan 1))
- -> Gather
- Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, (SubPlan 1)
- Workers Planned: 4
+ Workers Planned: 4
+ -> Sort
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, ((SubPlan 1))
+ Sort Key: ((SubPlan 1))
-> Parallel Seq Scan on public.tenk1 t
- Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(12 rows)
-- test subplan in join/lateral join
@@ -409,14 +433,14 @@ explain (costs off, verbose, timing off) select t.unique1, l.*
QUERY PLAN
----------------------------------------------------------------------
Gather
- Output: t.unique1, (SubPlan 1)
+ Output: t.unique1, ((SubPlan 1))
Workers Planned: 4
-> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
- Output: t.unique1
- SubPlan 1
- -> Index Only Scan using tenk1_unique1 on public.tenk1
- Output: t.unique1
- Index Cond: (tenk1.unique1 = t.unique1)
+ Output: t.unique1, (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
(9 rows)
-- can't put a gather at the top of a subplan that takes a param
@@ -1350,9 +1374,12 @@ SELECT 1 FROM tenk1_vw_sec
Workers Planned: 4
-> Parallel Index Only Scan using tenk1_unique1 on tenk1
SubPlan 1
- -> Aggregate
- -> Seq Scan on int4_tbl
- Filter: (f1 < tenk1_vw_sec.unique1)
-(9 rows)
+ -> Finalize Aggregate
+ -> Gather
+ Workers Planned: 1
+ -> Partial Aggregate
+ -> Parallel Seq Scan on int4_tbl
+ Filter: (f1 < tenk1_vw_sec.unique1)
+(12 rows)
rollback;
diff --git a/src/test/regress/sql/select_parallel.sql b/src/test/regress/sql/select_parallel.sql
index cc752a7d45..3fc24b2b94 100644
--- a/src/test/regress/sql/select_parallel.sql
+++ b/src/test/regress/sql/select_parallel.sql
@@ -53,13 +53,17 @@ $$ select 'foo'::varchar union all select 'bar'::varchar $$
language sql stable;
select sp_test_func() order by 1;
--- Parallel Append is not to be used when the subpath depends on the outer param
+-- Parallel Append is can be used when the subpath depends on the outer params
+-- when those params are consumed within the worker that generates them.
create table part_pa_test(a int, b int) partition by range(a);
create table part_pa_test_p1 partition of part_pa_test for values from (minvalue) to (0);
create table part_pa_test_p2 partition of part_pa_test for values from (0) to (maxvalue);
-explain (costs off)
+explain (costs off, verbose)
select (select max((select pa1.b from part_pa_test pa1 where pa1.a = pa2.a)))
from part_pa_test pa2;
+insert into part_pa_test(a, b) values (-1, 1), (1, 3);
+select (select max((select pa1.b from part_pa_test pa1 where pa1.a = pa2.a)))
+from part_pa_test pa2;
drop table part_pa_test;
-- test with leader participation disabled
--
2.39.3 (Apple Git-145)
[application/octet-stream] v12-0001-Add-tests-before-change.patch (9.3K, ../../CAAaqYe-Ld1q_b9pOf0PRjfQLWCgdkTvnopfeXgmrSvEQu0Ox=A@mail.gmail.com/3-v12-0001-Add-tests-before-change.patch)
download | inline diff:
From 3ba7b7b5a638cd74b5b8d546722514409d9e3fd3 Mon Sep 17 00:00:00 2001
From: jcoleman <[email protected]>
Date: Mon, 26 Sep 2022 20:30:23 -0400
Subject: [PATCH v12 1/2] Add tests before change
---
src/test/regress/expected/select_parallel.out | 137 ++++++++++++++++++
src/test/regress/sql/select_parallel.sql | 33 +++++
2 files changed, 170 insertions(+)
diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
index d88353d496..b76132ffe4 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -311,6 +311,143 @@ select count(*) from tenk1 where (two, four) not in
10000
(1 row)
+-- test parallel plans for queries containing correlated subplans
+-- where the subplan only needs params available from the current
+-- worker's scan.
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t, generate_series(1, 10);
+ QUERY PLAN
+----------------------------------------------------------------------------
+ Gather
+ Output: (SubPlan 1)
+ Workers Planned: 4
+ -> Nested Loop
+ Output: t.unique1
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ -> Function Scan on pg_catalog.generate_series
+ Output: generate_series.generate_series
+ Function Call: generate_series(1, 10)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(14 rows)
+
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t;
+ QUERY PLAN
+----------------------------------------------------------------------
+ Gather
+ Output: (SubPlan 1)
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(9 rows)
+
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t
+ limit 1;
+ QUERY PLAN
+-------------------------------------------------------------------
+ Limit
+ Output: ((SubPlan 1))
+ -> Seq Scan on public.tenk1 t
+ Output: (SubPlan 1)
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(8 rows)
+
+explain (costs off, verbose) select t.unique1
+ from tenk1 t
+ where t.unique1 = (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+ QUERY PLAN
+-------------------------------------------------------------
+ Seq Scan on public.tenk1 t
+ Output: t.unique1
+ Filter: (t.unique1 = (SubPlan 1))
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(7 rows)
+
+explain (costs off, verbose) select *
+ from tenk1 t
+ order by (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+ QUERY PLAN
+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Sort
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, ((SubPlan 1))
+ Sort Key: ((SubPlan 1))
+ -> Gather
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4, (SubPlan 1)
+ Workers Planned: 4
+ -> Parallel Seq Scan on public.tenk1 t
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(12 rows)
+
+-- test subplan in join/lateral join
+explain (costs off, verbose, timing off) select t.unique1, l.*
+ from tenk1 t
+ join lateral (
+ select (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1 offset 0)
+ ) l on true;
+ QUERY PLAN
+----------------------------------------------------------------------
+ Gather
+ Output: t.unique1, (SubPlan 1)
+ Workers Planned: 4
+ -> Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t
+ Output: t.unique1
+ SubPlan 1
+ -> Index Only Scan using tenk1_unique1 on public.tenk1
+ Output: t.unique1
+ Index Cond: (tenk1.unique1 = t.unique1)
+(9 rows)
+
+-- can't put a gather at the top of a subplan that takes a param
+-- (use a join in the subplan for the harder case)
+explain (costs off, verbose) select * from tenk1 t where t.two in (
+ select t.two
+ from tenk1
+ join tenk1 t3 on t3.stringu1 = tenk1.stringu1
+ where tenk1.four = t.four
+);
+ QUERY PLAN
+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Seq Scan on public.tenk1 t
+ Output: t.unique1, t.unique2, t.two, t.four, t.ten, t.twenty, t.hundred, t.thousand, t.twothousand, t.fivethous, t.tenthous, t.odd, t.even, t.stringu1, t.stringu2, t.string4
+ Filter: (SubPlan 1)
+ SubPlan 1
+ -> Hash Join
+ Output: t.two
+ Hash Cond: (tenk1.stringu1 = t3.stringu1)
+ -> Seq Scan on public.tenk1
+ Output: tenk1.unique1, tenk1.unique2, tenk1.two, tenk1.four, tenk1.ten, tenk1.twenty, tenk1.hundred, tenk1.thousand, tenk1.twothousand, tenk1.fivethous, tenk1.tenthous, tenk1.odd, tenk1.even, tenk1.stringu1, tenk1.stringu2, tenk1.string4
+ Filter: (tenk1.four = t.four)
+ -> Hash
+ Output: t3.stringu1
+ -> Gather
+ Output: t3.stringu1
+ Workers Planned: 4
+ -> Parallel Seq Scan on public.tenk1 t3
+ Output: t3.stringu1
+(17 rows)
+
-- this is not parallel-safe due to use of random() within SubLink's testexpr:
explain (costs off)
select * from tenk1 where (unique1 + random())::integer not in
diff --git a/src/test/regress/sql/select_parallel.sql b/src/test/regress/sql/select_parallel.sql
index 80c914dc02..cc752a7d45 100644
--- a/src/test/regress/sql/select_parallel.sql
+++ b/src/test/regress/sql/select_parallel.sql
@@ -111,6 +111,39 @@ explain (costs off)
(select hundred, thousand from tenk2 where thousand > 100);
select count(*) from tenk1 where (two, four) not in
(select hundred, thousand from tenk2 where thousand > 100);
+-- test parallel plans for queries containing correlated subplans
+-- where the subplan only needs params available from the current
+-- worker's scan.
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t, generate_series(1, 10);
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t;
+explain (costs off, verbose) select
+ (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+ from tenk1 t
+ limit 1;
+explain (costs off, verbose) select t.unique1
+ from tenk1 t
+ where t.unique1 = (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+explain (costs off, verbose) select *
+ from tenk1 t
+ order by (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1);
+-- test subplan in join/lateral join
+explain (costs off, verbose, timing off) select t.unique1, l.*
+ from tenk1 t
+ join lateral (
+ select (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1 offset 0)
+ ) l on true;
+-- can't put a gather at the top of a subplan that takes a param
+-- (use a join in the subplan for the harder case)
+explain (costs off, verbose) select * from tenk1 t where t.two in (
+ select t.two
+ from tenk1
+ join tenk1 t3 on t3.stringu1 = tenk1.stringu1
+ where tenk1.four = t.four
+);
-- this is not parallel-safe due to use of random() within SubLink's testexpr:
explain (costs off)
select * from tenk1 where (unique1 + random())::integer not in
--
2.39.3 (Apple Git-145)
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2024-01-31 02:56 James Coleman <[email protected]>
parent: Robert Haas <[email protected]>
0 siblings, 1 reply; 22+ messages in thread
From: James Coleman @ 2024-01-31 02:56 UTC (permalink / raw)
To: Robert Haas <[email protected]>; +Cc: Akshat Jaimini <[email protected]>; [email protected]
On Tue, Jan 30, 2024 at 11:54 AM Robert Haas <[email protected]> wrote:
>
> On Tue, Jan 30, 2024 at 11:17 AM Akshat Jaimini <[email protected]> wrote:
> > I think we should move this patch to the next CF as I believe that work is still going on resolving the last reported bug.
>
> We shouldn't just keep pushing this forward to the next CF. It's been
> idle since July. If it needs more work, mark it RwF and it can be
> reopened when there's something for a reviewer to do.
I don't follow the "Idle since July" since it just hasn't received
review since then, so there's been nothing to reply to.
That being said, Vignesh's note in January about a now-failing test is
relevant activity, and I've just today responded to that, so I'm
changing the status back from Waiting on Author to Needs Review.
Regards,
James Coleman
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2024-01-31 03:34 Tom Lane <[email protected]>
parent: James Coleman <[email protected]>
0 siblings, 1 reply; 22+ messages in thread
From: Tom Lane @ 2024-01-31 03:34 UTC (permalink / raw)
To: James Coleman <[email protected]>; +Cc: vignesh C <[email protected]>; Richard Guo <[email protected]>; Tomas Vondra <[email protected]>; Andres Freund <[email protected]>; Robert Haas <[email protected]>; pgsql-hackers; Amit Kapila <[email protected]>
James Coleman <[email protected]> writes:
> I've finally had a chance to look at this, and I don't believe there's
> any real failure here, merely drift of how the planner works on master
> resulting in this query now being eligible for a different plan shape.
> I was a bit wary at first because the changing test query is one I'd
> previously referenced in [1] as likely exposing the bug I'd fixed
> where params where being used across worker boundaries. However
> looking at the diff in the patch at that point (v10) that particular
> test query formed a different plan shape (there were two gather nodes
> being created, and params crossing between them).
> But in the current revision of master with the current patch applied
> that's no longer true: we have a Gather node, and the Subplan using
> the param is properly under that Gather node, and the param should be
> being both generated and consumed within the same worker process.
Hmm ... so the question this raises for me is: was that test intended
to verify behavior of params being passed across workers? If so,
haven't you broken the point of the test? This doesn't mean that
your code change is wrong; but I think maybe you need to find a way
to modify that test case so that it still tests what it's meant to.
This is a common hazard when changing the planner's behavior.
regards, tom lane
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2024-01-31 13:53 James Coleman <[email protected]>
parent: Tom Lane <[email protected]>
0 siblings, 0 replies; 22+ messages in thread
From: James Coleman @ 2024-01-31 13:53 UTC (permalink / raw)
To: Tom Lane <[email protected]>; +Cc: vignesh C <[email protected]>; Richard Guo <[email protected]>; Tomas Vondra <[email protected]>; Andres Freund <[email protected]>; Robert Haas <[email protected]>; pgsql-hackers; Amit Kapila <[email protected]>
On Tue, Jan 30, 2024 at 10:34 PM Tom Lane <[email protected]> wrote:
>
> James Coleman <[email protected]> writes:
> > I've finally had a chance to look at this, and I don't believe there's
> > any real failure here, merely drift of how the planner works on master
> > resulting in this query now being eligible for a different plan shape.
>
> > I was a bit wary at first because the changing test query is one I'd
> > previously referenced in [1] as likely exposing the bug I'd fixed
> > where params where being used across worker boundaries. However
> > looking at the diff in the patch at that point (v10) that particular
> > test query formed a different plan shape (there were two gather nodes
> > being created, and params crossing between them).
>
> > But in the current revision of master with the current patch applied
> > that's no longer true: we have a Gather node, and the Subplan using
> > the param is properly under that Gather node, and the param should be
> > being both generated and consumed within the same worker process.
>
> Hmm ... so the question this raises for me is: was that test intended
> to verify behavior of params being passed across workers? If so,
> haven't you broken the point of the test? This doesn't mean that
> your code change is wrong; but I think maybe you need to find a way
> to modify that test case so that it still tests what it's meant to.
> This is a common hazard when changing the planner's behavior.
I'd been thinking it was covered by another test I'd added in 0001,
but looking at it again that test doesn't exercise parallel append
(though it does exercise a different case of cross-worker param
usage), so I'll add another test for the parallel append behavior.
Regards,
James Coleman
^ permalink raw reply [nested|flat] 22+ messages in thread
* Re: Parallelize correlated subqueries that execute within each worker
@ 2024-01-31 20:18 Robert Haas <[email protected]>
parent: James Coleman <[email protected]>
0 siblings, 0 replies; 22+ messages in thread
From: Robert Haas @ 2024-01-31 20:18 UTC (permalink / raw)
To: James Coleman <[email protected]>; +Cc: Akshat Jaimini <[email protected]>; [email protected]
On Tue, Jan 30, 2024 at 9:56 PM James Coleman <[email protected]> wrote:
> I don't follow the "Idle since July" since it just hasn't received
> review since then, so there's been nothing to reply to.
It wasn't clear to me if you thought that the patch was ready for
review since July, or if it was waiting on you since July. Those are
quite different, IMV.
> That being said, Vignesh's note in January about a now-failing test is
> relevant activity, and I've just today responded to that, so I'm
> changing the status back from Waiting on Author to Needs Review.
Sounds good.
--
Robert Haas
EDB: http://www.enterprisedb.com
^ permalink raw reply [nested|flat] 22+ messages in thread
end of thread, other threads:[~2024-01-31 20:18 UTC | newest]
Thread overview: 22+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2023-01-23 14:00 Re: Parallelize correlated subqueries that execute within each worker James Coleman <[email protected]>
2023-02-06 16:40 ` Antonin Houska <[email protected]>
2023-02-08 13:38 ` James Coleman <[email protected]>
2023-03-08 18:07 ` Antonin Houska <[email protected]>
2023-06-06 09:35 ` Richard Guo <[email protected]>
2023-06-12 02:23 ` James Coleman <[email protected]>
2023-06-26 06:47 ` Richard Guo <[email protected]>
2023-07-04 01:26 ` James Coleman <[email protected]>
2024-01-09 07:08 ` vignesh C <[email protected]>
2024-01-23 14:33 ` Akshat Jaimini <[email protected]>
2024-01-23 20:20 ` Robert Haas <[email protected]>
2024-01-30 16:16 ` Akshat Jaimini <[email protected]>
2024-01-30 16:54 ` Robert Haas <[email protected]>
2024-01-31 02:56 ` James Coleman <[email protected]>
2024-01-31 20:18 ` Robert Haas <[email protected]>
2024-01-31 02:51 ` James Coleman <[email protected]>
2024-01-31 03:34 ` Tom Lane <[email protected]>
2024-01-31 13:53 ` James Coleman <[email protected]>
2023-02-24 20:07 [PATCH] f!fixes for LZ4 Justin Pryzby <[email protected]>
2023-02-24 20:07 [PATCH] f!fixes for LZ4 Justin Pryzby <[email protected]>
2023-02-24 20:07 [PATCH 1/4] f!fixes for LZ4 Justin Pryzby <[email protected]>
2023-02-24 20:07 [PATCH 1/4] f!fixes for LZ4 Justin Pryzby <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox