Subject: Parallelize correlated subqueries that execute within each worker

From: James Coleman <[email-masked]>
Date: Fri, 7 May 2021 12:30:02 -0400

In a bug report back in November [1] a subthread explored why parallel
query is excluded any time we have "Plan nodes which reference a
correlated SubPlan". Amit's understanding was that the reasoning had
to do with inability to easily pass (potentially variable length)
Param values between workers.

However a decent-sized subset of this kind of query doesn't actually
require that we communicate between workers. If the Subplan executes
per-tuple within the worker then there's no reason I can see why it
needs to be marked parallel unsafe. Amit concurred but noted that
identifying that subset of plans is the difficult part (as is usually
the case!)

At the time I'd started work on an approach to handle this case and
hoped to "post about it in a new thread later this week." That didn't
happen, but here we are now, and I finally have this patch cleaned up
enough to share.

The basic idea is that we need to track (both on nodes and relations)
not only whether that node or rel is parallel safe but also whether
it's parallel safe assuming params are rechecked in the using context.
That allows us to delay making a final decision until we have
sufficient context to conclude that a given usage of a Param is
actually parallel safe or unsafe.

The first patch in this series was previously posted in the thread
"Consider parallel for lateral subqueries with limit" [2] and is
required as a precursor for various test cases to work here.

The second patch implements the core of the series. It results in
parallel query being possible for subplans that execute entirely
within the context of a parallel worker for cases where that subplan
is in the target, a LATERAL JOIN, or the WHERE and ORDER BY clauses.

The final patch notes several places where we set e.g.
rel->consider_parallel but setting the corresponding new value
rel->consider_parallel_recheckng_params wasn't yet necessary. It shows
opportunity either for further improvement or concluding certain cases
can't benefit and should be left unchanged.

James

1: https://www.postgresql.org/message-id/CAAaqYe_vihKjc%2B8LuQa49EHW4%2BKfefb3wHqPYFnCuUqozo%2BLFg%40mail.gmail.com
2: https://www.postgresql.org/message-id/flat/CAAaqYe_HEkmLwf_1iEHxXwQOWiRyiFd%3DuOu6kwj3sWYdVd1-zA%40mail.gmail.com

---

From: Daniel Gustafsson <[email-masked]>
Date: Wed, 1 Sep 2021 13:06:04 +0200

> On 7 May 2021, at 18:30, James Coleman <[email-masked]> wrote:

> ..here we are now, and I finally have this patch cleaned up
> enough to share.

This patch no longer applies to HEAD, can you please submit a rebased version?

--
Daniel Gustafsson		https://vmware.com/

---

From: James Coleman <[email-masked]>
Date: Tue, 7 Sep 2021 09:16:49 -0400

> This patch no longer applies to HEAD, can you please submit a rebased version?

See attached.

Thanks,
James

---

From: Zhihong Yu <[email-masked]>
Date: Tue, 7 Sep 2021 08:06:00 -0700

> > This patch no longer applies to HEAD, can you please submit a rebased
Hi,
For v2-0002-Parallel-query-support-for-basic-correlated-subqu.patch :

+    * is when we're going to execute multiple partial parths in parallel

parths -> paths

        if (index->amcanparallel &&
-           rel->consider_parallel && outer_relids == NULL &&
-           scantype != ST_BITMAPSCAN)
+               rel->consider_parallel && outer_relids == NULL &&
+               scantype != ST_BITMAPSCAN)

the change above seems unnecessary since the first line of if condition
doesn't change.
Similar comment for the next hunk.

+            * It's not a partial path; it'a a full path that is executed
as a subquery.

it'a a -> it's a

+           /* rel->consider_parallel_rechecking_params = false; */
+           /* rel->partial_pathlist = NIL; */

The commented code can be taken out.

Cheers

---

From: James Coleman <[email-masked]>
Date: Wed, 8 Sep 2021 08:47:35 -0400

On Tue, Sep 7, 2021 at 11:06 AM Zhihong Yu <[email-masked]> wrote:
> the change above seems unnecessary since the first line of if condition doesn't change.
> +            * It's not a partial path; it'a a full path that is executed as a subquery.
>
> it'a a -> it's a
>

Thanks for taking a look at this.

See updated patch series attached.

James Coleman

---

From: James Coleman <[email-masked]>
Date: Wed, 3 Nov 2021 09:46:33 -0400

> See updated patch series attached.

Jaime,

I noticed on 3-October you moved this into "waiting on author"; I
don't see anything waiting in this thread, however. Am I missing
something?

I'm planning to change it back to "needs review".

Thanks,
James

---

From: Robert Haas <[email-masked]>
Date: Wed, 3 Nov 2021 10:48:48 -0400

As a preliminary comment, it would be quite useful to get Tom Lane's
opinion on this, since it's not an area I understand especially well,
and I think he understands it better than anyone.

> actually parallel safe or unsafe.

I don't really understand what you mean by "assuming params are
rechecked in the using context." However, I think that a possibly
better approach to this whole area would be to try to solve the
problem by putting limits on where you can insert a Gather node.
Consider:

Nested Loop
-> Seq Scan on x
-> Index Scan on y
   Index Cond: y.q = x.q

If you insert a Gather node atop the Index Scan, necessarily changing
it to a Parallel Index Scan, then you need to pass values around. For
every value we get for x.q, we would need to start workers, sending
them the value of x.q, and they do a parallel index scan working
together to find all rows where y.q = x.q, and then exit. We repeat
this for every tuple from x.q. In the absence of infrastructure to
pass those parameters, we can't put the Gather there. We also don't
want to, because it would be really slow.

If you insert the Gather node atop the Seq Scan or the Nested Loop, in
either case necessarily changing the Seq Scan to a Parallel Seq Scan,
you have no problem. If you put it on top of the Nested Loop, the
parameter will be set in the workers and used in the workers and
everything is fine. If you put it on top of the Seq Scan, the
parameter will be set in the leader -- by the Nested Loop -- and used
in the leader, and again you have no problem.

So in my view of the world, the parameter just acts as an additional
constraint on where Gather nodes can be placed. I don't see that there
are any parameters that are unsafe categorically -- they're just
unsafe if the place where they are set is on a different side of the
Gather from the place where they are used. So I don't understand --
possibly just because I'm dumb -- the idea behind
consider_parallel_rechecking_params, because that seems to be making a
sort of overall judgement about the safety or unsafety of the
parameter on its own merits, rather than thinking about the Gather
placement.

When I last worked on this, I had hoped that extParam or allParam
would be the thing that would answer the question: are there any
parameters used under this node that are not also set under this node?
But I seem to recall that neither seemed to be answering precisely
that question, and the lousy naming of those fields and limited
documentation of their intended purpose did not help.

-- 
Robert Haas
EDB: http://www.enterprisedb.com

---

From: Tom Lane <[email-masked]>
Date: Wed, 03 Nov 2021 11:14:35 -0400

> documentation of their intended purpose did not help.

FWIW, I've never been very happy with those fields either.  IIRC the
design in that area was all Vadim's, but to the extent that there's
any usable documentation of extParam/allParam, it was filled in by me
while trying to understand what Vadim did.  If somebody wants to step
up and do a rewrite to make the planner's Param management more useful
or at least easier to understand, I think that'd be great.

But anyway: yeah, those fields as currently constituted don't help
much.  They tell you which Params are consumed by this node or its
subnodes, but not where those Params came from.  The planner's
plan_params and outer_params fields might be more nearly the right
thing, but I'm not sure they're spot-on either, nor that they're
up-to-date at the point where you'd want to make decisions about
Gather safety.

			regards, tom lane

---

From: Robert Haas <[email-masked]>
Date: Wed, 3 Nov 2021 12:27:42 -0400

> or at least easier to understand, I think that'd be great.

Good to know, thanks.

> up-to-date at the point where you'd want to make decisions about

One thing I discovered when I was looking at this a few years ago is
that there was only one query in the regression tests where extParam
and allParam were not the same. The offending query was select 1 =
all(select (select 1)), and the resulting plan has a Materialize node
with an attached InitPlan. For that Materialize node, extParam = {}
and allParam = {$0}, with $0 also being the output parameter of the
InitPlan attached that that Materialize node. In every other node in
that plan and in every node of every other plan generated by the
regression tests, the values were identical. So it's extremely niche
that these fields are even different from each other, and it's unclear
to me that we really need both of them.

What's also interesting is that extParam is computed (by
finalize_plan) as plan->extParam = bms_del_members(plan->extParam,
initSetParam). So I think it mostly ends up that extParam for a node
is not exactly all the parameters that anything under that node cares
about, but rather - approximately - all the things that anything under
that node cares about that aren't also set someplace under that node.
If it were exactly that, I think it would be perfect for our needs
here: if the set of things used but not set below the current level is
empty, it's OK to insert a Gather node; otherwise, it's not, at least,
not unless we find a way to pipe parameters from the leader into the
workers. But I think there's some reason that I no longer remember why
it's not exactly that, and therefore the idea doesn't work.

One problem I do remember is that attaching initplans at the top of
each subquery level as we presently do is really not good for this
kind of thing. Suppose you have several levels of Nested Loop and
someplace down in the plan you reference an InitPlan. The planner sees
no harm in attaching the InitPlan at the top level, which makes it
unsafe to put the Gather any place but at the top level. If you
attached the InitPlan to the lowest node in the plan tree that is high
enough to be above all the places that use the value from that
parameter, you could potentially shift the Gather down the plan tree,
which would be great if, for example, there's exactly one
parallel-restricted join and the rest are parallel-safe. The best plan
might be to do all the other joins under a Gather and then perform the
parallel-restricted join above it.

But I found it very hard to figure out how to rejigger the logic that
places InitPlans to be more intelligent, and eventually gave up.

---

From: Tom Lane <[email-masked]>
Date: Wed, 03 Nov 2021 12:52:16 -0400

> to me that we really need both of them.

Yeah, I've had that nagging feeling about them too.  But ISTR trying to
reduce them to one value years ago, and finding that it didn't quite work,
or at least would result in more subquery-re-evaluation than we do today.
You have to dig into what the executor uses these values for to really
grok them.  I'm afraid that that detail is all swapped out right now, so
I can't say much more.

			regards, tom lane

---

From: James Coleman <[email-masked]>
Date: Wed, 3 Nov 2021 13:34:39 -0400

Hi Robert, thanks for the detailed reply.

> parameter on its own merits, rather than thinking about the Gather

I had to read through this several times before I understood the point
(not your fault, this is, as you note, a complicated area). I *think*
if I grok it properly you're effectively describing what this patch
results in conceptually (but possibly solving it from a different
direction).

As I understand the current code, parallel plans are largely chosen
based not on where it's safe to insert a Gather node but rather by
determining if a given path is parallel safe. Through that lens params
are a bit of an odd man out -- they aren't inherently unsafe in the
way a parallel-unsafe function is, but they can only be used in
parallel plans under certain conditions (whether because of project
policy, performance, or missing infrastructure).

Under that paradigm the existing consider_parallel and parallel_safe
boolean values imply everything is about whether a plan is inherently
parallel safe. Thus the current doesn't have the context to handle the
nuance of params (as they are not inherently parallel-safe or unsafe).

Introducing consider_parallel_rechecking_params and
parallel_safe_ignoring_params allows us to keep more context on params
and make a more nuanced decision at the proper level of the plan. This
is what I mean by "rechecked in the using context", though I realize
now that both "recheck" and "context" are overloaded terms in the
project, so don't describe the concept particularly clearly. When a
path relies on params we can only make a final determination about its
parallel safety if we know whether or not the current parallel node
can provide the param's value. We don't necessarily know that
information until we attempt to generate a full parallel node in the
plan (I think what you're describing as "inserting a Gather node")
since the param may come from another node in the plan. These new
values allow us to do that by tracking tentatively parallel-safe
subplans (given proper Gather node placement) and delaying the
parallel-safety determination until the point at which a param is
available (or not).

Is that a more helpful framing of what my goal is here?

> documentation of their intended purpose did not help.

I don't really know anything about extParam or allParam, so I can't
offer any insight here.

Thanks,
James Coleman

---

From: Robert Haas <[email-masked]>
Date: Mon, 15 Nov 2021 10:01:37 -0500

> policy, performance, or missing infrastructure).

Right.

> parallel-safety determination until the point at which a param is

So I think I agree with you here. But I don't like all of this
"ignoring_params" stuff and I don't see why it's necessary. Say we
don't have both parallel_safe and parallel_safe_ignoring_params. Say
we just have parallel_safe. If the plan will be parallel safe if the
params are available, we label it parallel safe. If the plan will not
be parallel safe even if the params are available, we say it's not
parallel safe. Then, when we get to generate_gather_paths(), we don't
generate any paths if there are required parameters that are not
available. What's wrong with that approach?

Maybe it's clearer to say this: I feel like one extra Boolean is
either too much or too little. I think maybe it's not even needed. But
if it is needed, then why just a bool instead of, say, a Bitmapset of
params that are needed, or something?

I'm sort of speaking from intuition here rather than sure knowledge. I
might be totally wrong.

---

From: Michael Paquier <[email-masked]>
Date: Fri, 3 Dec 2021 16:35:08 +0900

>> policy, performance, or missing infrastructure).

Please note that the CF bot is complaining here, so I have moved this
patch to the next CF, but changed the status as waiting on author.
--
Michael

---

From: James Coleman <[email-masked]>
Date: Sat, 22 Jan 2022 20:25:19 -0500

On Fri, Jan 21, 2022 at 3:20 PM Robert Haas <[email-masked]> wrote:
>
> On Fri, Jan 14, 2022 at 2:25 PM James Coleman <[email-masked]> wrote:
> > I've been chewing on this a bit, and I was about to go re-read the
> > code and see how easy it'd be to do exactly what you're suggesting in
> > generate_gather_paths() (and verifying it doesn't need to happen in
> > other places). However there's one (I think large) gotcha with that
> > approach (assuming it otherwise makes sense): it means we do
> > unnecessary work. In the current patch series we only need to recheck
> > parallel safety if we're in a situation where we might actually
> > benefit from doing that work (namely when we have a correlated
> > subquery we might otherwise be able to execute in a parallel plan). If
> > we don't track that status we'd have to recheck the full parallel
> > safety of the path for all paths -- even without correlated
> > subqueries.
>
> I don't think there's an intrinsic problem with the idea of making a
> tentative determination about parallel safety and then refining it
> later, but I'm not sure why you think it would be a lot of work to
> figure this out at the point where we generate gather paths. I think
> it's just a matter of testing whether the set of parameters that the
> path needs as input is the empty set. It may be that neither extParam
> nor allParam are precisely that thing, but I think both are very
> close, and it seems to me that there's no theoretical reason why we
> can't know for every path the set of inputs that it requires "from the
> outside."

As I understand it now (not sure I realized this before) you're
suggesting that *even when there are required params* marking it as
parallel safe, and then checking the params for parallel safety later.
>From a purely theoretical perspective that seemed reasonable, so I
took a pass at that approach.

The first, and likely most interesting, thing I discovered was that
the vast majority of what the patch accomplishes it does so not via
the delayed params safety checking but rather via the required outer
relids checks I'd added to generate_useful_gather_paths.

For that to happen I did have to mark PARAM_EXEC params as presumed
parallel safe. That means that parallel_safe now doesn't strictly mean
"parallel safe in the current context" but "parallel safe as long as
any params are provided". That's a real change, but probably
acceptable as long as a project policy decision is made in that
direction.

There are a few concerns I have (and I'm not sure what level they rise to):

1. From what I can tell we don't have access on a path to the set of
params required by that path (I believe this is what Tom was
referencing in his sister reply at this point in the thread). That
means we have to rely on checking that the required outer relids are
provided by the current query level. I'm not quite sure yet whether or
not that guarantees (or if the rest of the path construction logic
guarantees for us) that the params provided by the outer rel are used
in a correlated way that isn't shared across workers. And because we
don't have the param information available we can't add additional
checks (that I can tell) to verify that.
2. Are we excluding any paths (by having one that will always be
invalid win the cost comparisons in add_partial_path)? I suppose this
danger actually exists in the previous version of the patch as well,
and I don't actually have any examples of this being a problem. Also
maybe this can only be a problem if (1) reveals a bug.
3. The new patch series actually ends up allowing parallelization of
correlated params in a few more places than the original patch series.
>From what I can tell all of the cases are in fact safe to execute in
parallel, which, if true, means this is a feature not a concern. The
changed query plans fall into two categories: a.) putting a gather
inside a subplan and b.) correlated param usages in a subquery scan
path on the inner side of a join. I've separated out those specific
changes in a separate patch to make it easier to tell which ones I'm
referencing.

On the other hand this is a dramatically simpler patch series.
Assuming the approach is sound, it should much easier to maintain than
the previous version.

The final patch in the series is a set of additional checks I could
imagine to try to be more explicit, but at least in the current test
suite there isn't anything at all they affect.

Does this look at least somewhat more like what you'd envisionsed
(granting the need to squint hard given the relids checks instead of
directly checking params)?

Thanks,
James Coleman
From 6596bb6909b9dfaadd6e6e834be4032e7903c4e9 Mon Sep 17 00:00:00 2001
From: jcoleman <[email-masked]>
Date: Mon, 30 Nov 2020 11:36:35 -0500
Subject: [PATCH v4 1/4] Allow parallel LATERAL subqueries with LIMIT/OFFSET

The code that determined whether or not a rel should be considered for
parallel query excluded subqueries with LIMIT/OFFSET. That's correct in
the general case: as the comment notes that'd mean we have to guarantee
ordering (and claims it's not worth checking that) for results to be
consistent across workers. However there's a simpler case that hasn't
been considered: LATERAL subqueries with LIMIT/OFFSET don't fall under
the same reasoning since they're executed (when not converted to a JOIN)
per tuple anyway, so consistency of results across workers isn't a
factor.
---
 src/backend/optimizer/path/allpaths.c         |  4 +++-
 src/test/regress/expected/select_parallel.out | 15 +++++++++++++++
 src/test/regress/sql/select_parallel.sql      |  6 ++++++
 3 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index 169b1d53fc..6a581e20fa 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -682,11 +682,13 @@ set_rel_consider_parallel(PlannerInfo *root, RelOptInfo *rel,
 			 * inconsistent results at the top-level.  (In some cases, where
 			 * the result is ordered, we could relax this restriction.  But it
 			 * doesn't currently seem worth expending extra effort to do so.)
+			 * LATERAL is an exception: LIMIT/OFFSET is safe to execute within
+			 * workers since the sub-select is executed per tuple
 			 */
 			{
 				Query	   *subquery = castNode(Query, rte->subquery);
 
-				if (limit_needed(subquery))
+				if (!rte->lateral && limit_needed(subquery))
 					return;
 			}
 			break;
diff --git a/src/test/regress/expected/select_parallel.out b/src/test/regress/expected/select_parallel.out
index 4ea1aa7dfd..2303f70d6e 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -1040,6 +1040,21 @@ explain (costs off)
                            Filter: (stringu1 ~~ '%AAAA'::text)
 (11 rows)
 
+-- ...unless it's LATERAL
+savepoint settings;
+set parallel_tuple_cost=0;
+explain (costs off) select t.unique1 from tenk1 t
+join lateral (select t.unique1 from tenk1 offset 0) l on true;
+                             QUERY PLAN                              
+---------------------------------------------------------------------
+ Gather
+   Workers Planned: 4
+   ->  Nested Loop
+         ->  Parallel Index Only Scan using tenk1_unique1 on tenk1 t
+         ->  Index Only Scan using tenk1_hundred on tenk1
+(5 rows)
+
+rollback to savepoint settings;
 -- to increase the parallel query test coverage
 SAVEPOINT settings;
 SET LOCAL force_parallel_mode = 1;
diff --git a/src/test/regress/sql/select_parallel.sql b/src/test/regress/sql/select_parallel.sql
index f924731248..019e17e751 100644
--- a/src/test/regress/sql/select_parallel.sql
+++ b/src/test/regress/sql/select_parallel.sql
@@ -390,6 +390,12 @@ explain (costs off, verbose)
 explain (costs off)
   select * from tenk1 a where two in
     (select two from tenk1 b where stringu1 like '%AAAA' limit 3);
+-- ...unless it's LATERAL
+savepoint settings;
+set parallel_tuple_cost=0;
+explain (costs off) select t.unique1 from tenk1 t
+join lateral (select t.unique1 from tenk1 offset 0) l on true;
+rollback to savepoint settings;
 
 -- to increase the parallel query test coverage
 SAVEPOINT settings;
-- 
2.17.1

From 200c4ed4118f92014253b49faa41a76ecec84c96 Mon Sep 17 00:00:00 2001
From: jcoleman <[email-masked]>
Date: Sat, 22 Jan 2022 18:34:54 -0500
Subject: [PATCH v4 3/4] Changed queries

---
 src/test/regress/expected/partition_prune.out | 104 +++++++++---------
 src/test/regress/expected/select_parallel.out |  26 +++--
 2 files changed, 69 insertions(+), 61 deletions(-)

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 124fe9fec5..9bb60c2c1e 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
@@ -1330,8 +1332,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.17.1

From c11d38b8a88e8f7a46528712fc9341de34171f8d Mon Sep 17 00:00:00 2001
From: jcoleman <[email-masked]>
Date: Sat, 22 Jan 2022 17:33:23 -0500
Subject: [PATCH v4 4/4] Possible additional checks

---
 src/backend/optimizer/path/allpaths.c | 33 ++++++++++++++++++++++-----
 1 file changed, 27 insertions(+), 6 deletions(-)

diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index 776f002054..7e30824320 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -2682,11 +2682,16 @@ 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;
 
+	if (!bms_is_subset(required_outer, rel->relids))
+		return;
+
+
 	/* Should we override the rel's rowcount estimate? */
 	if (override_rows)
 		rowsp = &rows;
@@ -2697,12 +2702,16 @@ 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,
-						   rel->lateral_relids, rowsp);
-	add_path(rel, simple_gather_path);
+	if (cheapest_partial_path->param_info == NULL ||
+			bms_is_subset(cheapest_partial_path->param_info->ppi_req_outer, 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,
+							   rel->lateral_relids, rowsp);
+		add_path(rel, simple_gather_path);
+	}
 
 	/*
 	 * For each useful ordering, we can consider an order-preserving Gather
@@ -2716,6 +2725,10 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
 		if (subpath->pathkeys == NIL)
 			continue;
 
+		if (subpath->param_info != NULL &&
+			!bms_is_subset(subpath->param_info->ppi_req_outer, rel->relids))
+			break;
+
 		rows = subpath->rows * subpath->parallel_workers;
 		path = create_gather_merge_path(root, rel, subpath, rel->reltarget,
 										subpath->pathkeys, rel->lateral_relids, rowsp);
@@ -2888,6 +2901,10 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
 			{
 				Path	   *tmp;
 
+				if (subpath->param_info != NULL &&
+					!bms_is_subset(subpath->param_info->ppi_req_outer, rel->relids))
+					break;
+
 				tmp = (Path *) create_sort_path(root,
 												rel,
 												subpath,
@@ -2916,6 +2933,10 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
 			{
 				Path	   *tmp;
 
+				if (subpath->param_info != NULL &&
+					!bms_is_subset(subpath->param_info->ppi_req_outer, rel->relids))
+					break;
+
 				/*
 				 * We should have already excluded pathkeys of length 1
 				 * because then presorted_keys > 0 would imply is_sorted was
-- 
2.17.1

From ba785f2abca7c9f5199f0fc27c3b71f6d1d8010b Mon Sep 17 00:00:00 2001
From: jcoleman <[email-masked]>
Date: Fri, 21 Jan 2022 22:38:46 -0500
Subject: [PATCH v4 2/4] 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         |  18 ++-
 src/backend/optimizer/path/joinpath.c         |  16 ++-
 src/backend/optimizer/util/clauses.c          |   3 +
 src/backend/optimizer/util/pathnode.c         |   2 +
 src/backend/utils/misc/guc.c                  |   1 +
 src/include/nodes/pathnodes.h                 |   2 +-
 .../regress/expected/incremental_sort.out     |  28 ++--
 src/test/regress/expected/select_parallel.out | 125 ++++++++++++++++++
 src/test/regress/sql/select_parallel.sql      |  25 ++++
 10 files changed, 197 insertions(+), 26 deletions(-)

diff --git a/doc/src/sgml/parallel.sgml b/doc/src/sgml/parallel.sgml
index 13479d7e5e..2d924dd2ac 100644
--- a/doc/src/sgml/parallel.sgml
+++ b/doc/src/sgml/parallel.sgml
@@ -517,7 +517,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 6a581e20fa..776f002054 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -556,7 +556,8 @@ set_rel_pathlist(PlannerInfo *root, RelOptInfo *rel,
 	 * (see grouping_planner).
 	 */
 	if (rel->reloptkind == RELOPT_BASEREL &&
-		bms_membership(root->all_baserels) != BMS_SINGLETON)
+		bms_membership(root->all_baserels) != BMS_SINGLETON
+		&& (rel->subplan_params == NIL || rte->rtekind != RTE_SUBQUERY))
 		generate_useful_gather_paths(root, rel, false);
 
 	/* Now find the cheapest of the paths for this rel */
@@ -2700,7 +2701,7 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_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);
+						   rel->lateral_relids, rowsp);
 	add_path(rel, simple_gather_path);
 
 	/*
@@ -2717,7 +2718,7 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
 
 		rows = subpath->rows * subpath->parallel_workers;
 		path = create_gather_merge_path(root, rel, subpath, rel->reltarget,
-										subpath->pathkeys, NULL, rowsp);
+										subpath->pathkeys, rel->lateral_relids, rowsp);
 		add_path(rel, &path->path);
 	}
 }
@@ -2819,11 +2820,15 @@ 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;
 
+	if (!bms_is_subset(required_outer, rel->relids))
+		return;
+
 	/* Should we override the rel's rowcount estimate? */
 	if (override_rows)
 		rowsp = &rows;
@@ -2895,7 +2900,7 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
 												tmp,
 												rel->reltarget,
 												tmp->pathkeys,
-												NULL,
+												required_outer,
 												rowsp);
 
 				add_path(rel, &path->path);
@@ -2929,7 +2934,7 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
 												tmp,
 												rel->reltarget,
 												tmp->pathkeys,
-												NULL,
+												required_outer,
 												rowsp);
 
 				add_path(rel, &path->path);
@@ -3108,7 +3113,8 @@ standard_join_search(PlannerInfo *root, int levels_needed, List *initial_rels)
 			/*
 			 * Except for the topmost scan/join rel, consider gathering
 			 * partial paths.  We'll do the same for the topmost scan/join rel
-			 * once we know the final targetlist (see grouping_planner).
+			 * once we know the final targetlist (see
+			 * apply_scanjoin_target_to_paths).
 			 */
 			if (lev < levels_needed)
 				generate_useful_gather_paths(root, rel, false);
diff --git a/src/backend/optimizer/path/joinpath.c b/src/backend/optimizer/path/joinpath.c
index f96fc9fd28..e85b5449ea 100644
--- a/src/backend/optimizer/path/joinpath.c
+++ b/src/backend/optimizer/path/joinpath.c
@@ -1791,16 +1791,24 @@ match_unsorted_outer(PlannerInfo *root,
 	 * 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.
+	 * we handle JOIN_FULL and JOIN_RIGHT, because they can produce false null
+	 * extended rows.
+	 *
+	 * Partial paths may only have parameters in limited cases
+	 * where the parameterization is fully satisfied without sharing state
+	 * between workers, so we only allow lateral rels on inputs to the join
+	 * if the resulting join contains no lateral rels, the inner rel's laterals
+	 * are fully satisfied by the outer rel, and the outer rel doesn't depend
+	 * on the inner rel to produce any laterals.
 	 */
 	if (joinrel->consider_parallel &&
 		save_jointype != JOIN_UNIQUE_OUTER &&
 		save_jointype != JOIN_FULL &&
 		save_jointype != JOIN_RIGHT &&
 		outerrel->partial_pathlist != NIL &&
-		bms_is_empty(joinrel->lateral_relids))
+		bms_is_empty(joinrel->lateral_relids) &&
+		bms_is_subset(innerrel->lateral_relids, outerrel->relids) &&
+		(bms_is_empty(outerrel->lateral_relids) || !bms_is_subset(outerrel->lateral_relids, innerrel->relids)))
 	{
 		if (nestjoinOK)
 			consider_parallel_nestloop(root, joinrel, outerrel, innerrel,
diff --git a/src/backend/optimizer/util/clauses.c b/src/backend/optimizer/util/clauses.c
index a707dc9f26..f0002f6887 100644
--- a/src/backend/optimizer/util/clauses.c
+++ b/src/backend/optimizer/util/clauses.c
@@ -822,6 +822,9 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context)
 		if (param->paramkind == PARAM_EXTERN)
 			return false;
 
+		if (param->paramkind == PARAM_EXEC)
+			return false;
+
 		if (param->paramkind != PARAM_EXEC ||
 			!list_member_int(context->safe_param_ids, param->paramid))
 		{
diff --git a/src/backend/optimizer/util/pathnode.c b/src/backend/optimizer/util/pathnode.c
index 5c32c96b71..144b2c485d 100644
--- a/src/backend/optimizer/util/pathnode.c
+++ b/src/backend/optimizer/util/pathnode.c
@@ -2418,6 +2418,8 @@ create_nestloop_path(PlannerInfo *root,
 	NestPath   *pathnode = makeNode(NestPath);
 	Relids		inner_req_outer = PATH_REQ_OUTER(inner_path);
 
+	/* TODO: Assert lateral relids subset safety? */
+
 	/*
 	 * If the inner path is parameterized by the outer, we must drop any
 	 * restrict_clauses that are due to be moved into the inner path.  We have
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index effb9d03a0..52cd3512b3 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -58,6 +58,7 @@
 #include "libpq/libpq.h"
 #include "libpq/pqformat.h"
 #include "miscadmin.h"
+#include "optimizer/clauses.h"
 #include "optimizer/cost.h"
 #include "optimizer/geqo.h"
 #include "optimizer/optimizer.h"
diff --git a/src/include/nodes/pathnodes.h b/src/include/nodes/pathnodes.h
index 1f33fe13c1..75681d6fb9 100644
--- a/src/include/nodes/pathnodes.h
+++ b/src/include/nodes/pathnodes.h
@@ -2478,7 +2478,7 @@ typedef struct MinMaxAggInfo
  * for conflicting purposes.
  *
  * In addition, PARAM_EXEC slots are assigned for Params representing outputs
- * from subplans (values that are setParam items for those subplans).  These
+ * from subplans (values that are setParam items for those subplans). [TODO: is this true, or only for init plans?]  These
  * IDs need not be tracked via PlannerParamItems, since we do not need any
  * duplicate-elimination nor later processing of the represented expressions.
  * Instead, we just record the assignment of the slot number by appending to
diff --git a/src/test/regress/expected/incremental_sort.out b/src/test/regress/expected/incremental_sort.out
index 545e301e48..8f9ca05e60 100644
--- a/src/test/regress/expected/incremental_sort.out
+++ b/src/test/regress/expected/incremental_sort.out
@@ -1614,16 +1614,16 @@ from tenk1 t, generate_series(1, 1000);
                                    QUERY PLAN                                    
 ---------------------------------------------------------------------------------
  Unique
-   ->  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)
 (11 rows)
 
 explain (costs off) select
@@ -1633,16 +1633,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/select_parallel.out b/src/test/regress/expected/select_parallel.out
index 2303f70d6e..124fe9fec5 100644
--- a/src/test/regress/expected/select_parallel.out
+++ b/src/test/regress/expected/select_parallel.out
@@ -311,6 +311,131 @@ 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: (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)
+(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: (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 (analyze, costs off, summary off, verbose, timing off) select
+  (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
+  from tenk1 t
+  limit 1;
+                                             QUERY PLAN                                             
+----------------------------------------------------------------------------------------------------
+ Limit (actual rows=1 loops=1)
+   Output: ((SubPlan 1))
+   ->  Gather (actual rows=1 loops=1)
+         Output: ((SubPlan 1))
+         Workers Planned: 4
+         Workers Launched: 4
+         ->  Parallel Index Only Scan using tenk1_unique1 on public.tenk1 t (actual rows=1 loops=5)
+               Output: (SubPlan 1)
+               Heap Fetches: 0
+               Worker 0:  actual rows=1 loops=1
+               Worker 1:  actual rows=1 loops=1
+               Worker 2:  actual rows=1 loops=1
+               Worker 3:  actual rows=1 loops=1
+               SubPlan 1
+                 ->  Index Only Scan using tenk1_unique1 on public.tenk1 (actual rows=1 loops=5)
+                       Output: t.unique1
+                       Index Cond: (tenk1.unique1 = t.unique1)
+                       Heap Fetches: 0
+                       Worker 0:  actual rows=1 loops=1
+                       Worker 1:  actual rows=1 loops=1
+                       Worker 2:  actual rows=1 loops=1
+                       Worker 3:  actual rows=1 loops=1
+(22 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                              
+----------------------------------------------------------------------
+ Gather
+   Output: t.unique1
+   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                                                                                                
+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ 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))
+   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)
+               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)
+         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 019e17e751..c49799b6d4 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 (analyze, costs off, summary off, verbose, timing off) 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.17.1

---

From: Andres Freund <[email-masked]>
Date: Mon, 21 Mar 2022 17:48:05 -0700

Hi,

> directly checking params)?

This fails on freebsd (so likely a timing issue): https://cirrus-ci.com/task/4758411492458496?logs=test_world#L2225

Marked as waiting on author.

Greetings,

Andres Freund

---

From: Jacob Champion <[email-masked]>
Date: Tue, 2 Aug 2022 13:57:42 -0700

This entry has been waiting on author input for a while (our current
threshold is roughly two weeks), so I've marked it Returned with
Feedback.

Once you think the patchset is ready for review again, you (or any
interested party) can resurrect the patch entry by visiting

    https://commitfest.postgresql.org/38/3246/

and changing the status to "Needs Review", and then changing the
status again to "Move to next CF". (Don't forget the second step;
hopefully we will have streamlined this in the near future!)

Thanks,
--Jacob

---

From: James Coleman <[email-masked]>
Date: Mon, 26 Sep 2022 22:56:11 -0400

> Marked as waiting on author.

I've finally gotten around to checking this out, and the issue was an
"explain analyze" test that had actual loops different on FreeBSD.
There doesn't seem to be a way to disable loop output, but instead of
processing the explain output with e.g. a function (as we do some
other places) to remove the offending and unnecessary output I've just
removed the "analyze" (as I don't believe it was actually necessary).

Attached is an updated patch series. In this version I've removed the
"parallelize some subqueries with limit" patch since discussion is
proceeding in the spun off thread. The first patch adds additional
tests so that you can see how those new tests change with the code
changes in the 2nd patch in the series. As before the final patch in
the series includes changes where we may also want to verify
correctness but don't have a test demonstrating the need.

Thanks,
James Coleman

---

From: vignesh C <[email-masked]>
Date: Wed, 4 Jan 2023 15:19:24 +0530

> correctness but don't have a test demonstrating the need.

The patch does not apply on top of HEAD as in [1], please post a rebased patch:

=== Applying patches on top of PostgreSQL commit ID
bf03cfd162176d543da79f9398131abc251ddbb9 ===
=== applying patch ./v5-0002-Parallelize-correlated-subqueries.patch
patching file src/backend/optimizer/path/allpaths.c
...
Hunk #5 FAILED at 3225.
Hunk #6 FAILED at 3259.
Hunk #7 succeeded at 3432 (offset -6 lines).
2 out of 7 hunks FAILED -- saving rejects to file
src/backend/optimizer/path/allpaths.c.rej

[1] - http://cfbot.cputube.org/patch_41_3246.log

Regards,
Vignesh

---

From: Tomas Vondra <[email-masked]>
Date: Wed, 18 Jan 2023 20:09:21 +0100

Hi,

This patch hasn't been updated since September, and it got broken by
4a29eabd1d91c5484426bc5836e0a7143b064f5a which the incremental sort
stuff a little bit. But the breakage was rather limited, so I took a
stab at fixing it - attached is the result, hopefully correct.

I also added a couple minor comments about stuff I noticed while
rebasing and skimming the patch, I kept those in separate commits.
There's also a couple pre-existing TODOs.

James, what's your plan with this patch. Do you intend to work on it for
PG16, or are there some issues I missed in the thread?

One of the queries in in incremental_sort changed plans a little bit:

explain (costs off) select distinct
  unique1,
  (select t.unique1 from tenk1 where tenk1.unique1 = t.unique1)
from tenk1 t, generate_series(1, 1000);

switched from

 Unique  (cost=18582710.41..18747375.21 rows=10000 width=8)
   ->  Gather Merge  (cost=18582710.41..18697375.21 rows=10000000 ...)
         Workers Planned: 2
         ->  Sort  (cost=18582710.39..18593127.06 rows=4166667 ...)
               Sort Key: t.unique1, ((SubPlan 1))
             ...

to

 Unique  (cost=18582710.41..18614268.91 rows=10000 ...)
   ->  Gather Merge  (cost=18582710.41..18614168.91 rows=20000 ...)
         Workers Planned: 2
         ->  Unique  (cost=18582710.39..18613960.39 rows=10000 ...)
               ->  Sort  (cost=18582710.39..18593127.06 ...)
                     Sort Key: t.unique1, ((SubPlan 1))
                   ...

which probably makes sense, as the cost estimate decreases a bit.

-- 
Tomas Vondra
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

From 95db15fe16303ed3f4fdea52af3b8d6d05a8d7a6 Mon Sep 17 00:00:00 2001
From: jcoleman <[email-masked]>
Date: Mon, 26 Sep 2022 20:30:23 -0400
Subject: [PATCH 1/5] 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 91f74fe47a3..9b4d7dd44a4 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)
 
+-------------------------------------------------------------------
+ Limit
+   ->  Seq Scan on public.tenk1 t
+(8 rows)
+-------------------------------------------------------------
+ Seq Scan on public.tenk1 t
+(7 rows)
+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ 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
index 62fb68c7a04..21c2f1c7424 100644
2.39.0

From cd379afbe22a9b41ff2b5dd5d86719a19f15301f Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email-masked]>
Date: Wed, 18 Jan 2023 19:15:06 +0100
Subject: [PATCH 2/5] Parallelize correlated subqueries

Alternative approach using just relids subset check
---
 doc/src/sgml/parallel.sgml                    |   3 +-
 src/backend/optimizer/path/allpaths.c         |  16 ++-
 src/backend/optimizer/path/joinpath.c         |  16 ++-
 src/backend/optimizer/util/clauses.c          |   3 +
 src/backend/optimizer/util/pathnode.c         |   2 +
 src/include/nodes/pathnodes.h                 |   2 +-
 .../regress/expected/incremental_sort.out     |  28 ++--
 src/test/regress/expected/partition_prune.out | 104 +++++++-------
 src/test/regress/expected/select_parallel.out | 128 ++++++++++--------
 9 files changed, 168 insertions(+), 134 deletions(-)

diff --git a/doc/src/sgml/parallel.sgml b/doc/src/sgml/parallel.sgml
index 5acc9537d6f..fd32572ec8b 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 c2fc568dc8a..7108501b9a7 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -558,7 +558,8 @@ set_rel_pathlist(PlannerInfo *root, RelOptInfo *rel,
 	 * the final scan/join targetlist is available (see grouping_planner).
 	 */
 	if (rel->reloptkind == RELOPT_BASEREL &&
-		!bms_equal(rel->relids, root->all_baserels))
+		!bms_equal(rel->relids, root->all_baserels)
+		&& (rel->subplan_params == NIL || rte->rtekind != RTE_SUBQUERY))
 		generate_useful_gather_paths(root, rel, false);
 
@@ -3037,7 +3038,7 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
 
 	/*
@@ -3054,7 +3055,7 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
 
@@ -3156,11 +3157,15 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r

+	if (!bms_is_subset(required_outer, rel->relids))
+		return;
+
 	/* Should we override the rel's rowcount estimate? */
 	if (override_rows)
 		rowsp = &rows;
@@ -3249,7 +3254,7 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
 											subpath,
 											rel->reltarget,
 											subpath->pathkeys,
-											NULL,
+											required_outer,
 											rowsp);
 
@@ -3427,7 +3432,8 @@ standard_join_search(PlannerInfo *root, int levels_needed, List *initial_rels)
 			if (!bms_equal(rel->relids, root->all_baserels))
index d345c0437a4..000e3ca9a25 100644
@@ -1792,16 +1792,24 @@ match_unsorted_outer(PlannerInfo *root,
index aa584848cf9..035471d05d0 100644
@@ -816,6 +816,9 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context)
 
index 4478036bb6a..5667222e925 100644
@@ -2437,6 +2437,8 @@ create_nestloop_path(PlannerInfo *root,
 
index c20b7298a3d..a5ba65a5616 100644
@@ -2962,7 +2962,7 @@ typedef struct MinMaxAggInfo
index 0c3433f8e58..0cb7c1a49c0 100644
@@ -1600,16 +1600,16 @@ from tenk1 t, generate_series(1, 1000);
 
@@ -1619,16 +1619,16 @@ from tenk1 t, generate_series(1, 1000)
 
index 7555764c779..5c45f9c0a50 100644

index 9b4d7dd44a4..01443e2ffbe 100644
 
 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)
 
@@ -1322,8 +1330,10 @@ SELECT 1 FROM tenk1_vw_sec
 
 rollback;
-- 
2.39.0

From 1aa279e8e82a4e9771fb1a5068e9a5792a824a54 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email-masked]>
Date: Wed, 18 Jan 2023 19:51:38 +0100
Subject: [PATCH 3/5] review comments

---
 src/backend/optimizer/path/allpaths.c | 5 +++--
 src/backend/optimizer/util/clauses.c  | 5 +++++
 2 files changed, 8 insertions(+), 2 deletions(-)

diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index 7108501b9a7..357dfaab3e8 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -558,8 +558,8 @@ set_rel_pathlist(PlannerInfo *root, RelOptInfo *rel,
 	 * the final scan/join targetlist is available (see grouping_planner).
 	 */
 	if (rel->reloptkind == RELOPT_BASEREL &&
-		!bms_equal(rel->relids, root->all_baserels)
-		&& (rel->subplan_params == NIL || rte->rtekind != RTE_SUBQUERY))
+		!bms_equal(rel->relids, root->all_baserels) &&
+		(rel->subplan_params == NIL || rte->rtekind != RTE_SUBQUERY))
 		generate_useful_gather_paths(root, rel, false);
 
 	/* Now find the cheapest of the paths for this rel */
@@ -3163,6 +3163,7 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
 	if (rel->partial_pathlist == NIL)
 		return;
 
+	/* FIXME ??? */
 	if (!bms_is_subset(required_outer, rel->relids))
 		return;
 
diff --git a/src/backend/optimizer/util/clauses.c b/src/backend/optimizer/util/clauses.c
index 035471d05d0..9585acf8e69 100644
--- a/src/backend/optimizer/util/clauses.c
+++ b/src/backend/optimizer/util/clauses.c
@@ -819,6 +819,11 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context)
 		if (param->paramkind == PARAM_EXEC)
 			return false;
 
+		/*
+		 * XXX The first condition is certainly true, thanks to the preceding
+		 * check. The comment above should be updated to reflect this change,
+		 * probably.
+		 */
 		if (param->paramkind != PARAM_EXEC ||
 			!list_member_int(context->safe_param_ids, param->paramid))
 		{
-- 
2.39.0

From 45da97c1141ead1fe757fcff549e68fbd4c1ded9 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email-masked]>
Date: Wed, 18 Jan 2023 19:21:24 +0100
Subject: [PATCH 4/5] Possible additional checks

---
 src/backend/optimizer/path/allpaths.c | 29 +++++++++++++++++++++------
 1 file changed, 23 insertions(+), 6 deletions(-)

diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index 357dfaab3e8..eaa972ec64b 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -3019,11 +3019,16 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
 	ListCell   *lc;
 	double		rows;
 	double	   *rowsp = NULL;
+	Relids		required_outer = rel->lateral_relids;

@@ -3034,12 +3039,16 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
 
 	/*
 	 * For each useful ordering, we can consider an order-preserving Gather
@@ -3053,6 +3062,10 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
 		if (subpath->pathkeys == NIL)
 			continue;
 
+		if (subpath->param_info != NULL &&
+			!bms_is_subset(subpath->param_info->ppi_req_outer, rel->relids))
+			break;
+
 		rows = subpath->rows * subpath->parallel_workers;
 		path = create_gather_merge_path(root, rel, subpath, rel->reltarget,
 										subpath->pathkeys, rel->lateral_relids, rowsp);
@@ -3223,6 +3236,10 @@ generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_r
 				(presorted_keys == 0 || !enable_incremental_sort))
 				continue;
 
+			if (subpath->param_info != NULL &&
+				!bms_is_subset(subpath->param_info->ppi_req_outer, rel->relids))
+				break;
+
 			/*
 			 * Consider regular sort for any path that's not presorted or if
 			 * incremental sort is disabled.  We've no need to consider both
-- 
2.39.0

From 2ca25bd5bbe1aa6849c55cab58dafe202adacce9 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email-masked]>
Date: Wed, 18 Jan 2023 19:52:46 +0100
Subject: [PATCH 5/5] review comments

---
 src/backend/optimizer/path/allpaths.c | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index eaa972ec64b..9dd10a1274c 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -3025,10 +3025,10 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
 	if (rel->partial_pathlist == NIL)
 		return;
 
+	/* FIXME ??? */
 	if (!bms_is_subset(required_outer, rel->relids))
 		return;
 
-
 	/* Should we override the rel's rowcount estimate? */
 	if (override_rows)
 		rowsp = &rows;
@@ -3062,6 +3062,7 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
 		if (subpath->pathkeys == NIL)
 			continue;
 
+		/* FIXME ??? */
 		if (subpath->param_info != NULL &&
 			!bms_is_subset(subpath->param_info->ppi_req_outer, rel->relids))
 			break;
-- 
2.39.0

---

From: James Coleman <[email-masked]>
Date: Wed, 18 Jan 2023 21:34:21 -0500

> stab at fixing it - attached is the result, hopefully correct.

Thanks for fixing this up; the changes look correct to me.

> There's also a couple pre-existing TODOs.

I started work on some of these, but wasn't able to finish this
evening, so I don't have an updated series yet.

> PG16, or are there some issues I missed in the thread?

I'd love to see it get into PG16. I don't have any known issues, but
reviewing activity has been light. Originally Robert had had some
concerns about my original approach; I think my updated approach
resolves those issues, but it'd be good to have that sign-off.

Beyond that I'm mostly looking for review and evaluation of the
approach I've taken; of note is my description of that in [1].

> which probably makes sense, as the cost estimate decreases a bit.

Off the cuff that seems fine. I'll read it over again when I send the
updated series.

James Coleman

1: https://www.postgresql.org/message-id/CAAaqYe8m0DHUWk7gLKb_C4abTD4nMkU26ErE%3Dahow4zNMZbzPQ%40mail.gmail.com

---

From: Akshat Jaimini <[email-masked]>
Date: Tue, 23 Jan 2024 14:33:22 +0000

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.

---

From: Robert Haas <[email-masked]>
Date: Tue, 23 Jan 2024 15:20:31 -0500

> 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.

---

From: Akshat Jaimini <[email-masked]>
Date: Tue, 30 Jan 2024 16:16:20 +0000

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.

---

From: Robert Haas <[email-masked]>
Date: Tue, 30 Jan 2024 11:54:34 -0500

> 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.

---

From: James Coleman <[email-masked]>
Date: Tue, 30 Jan 2024 21:51:09 -0500

On Tue, Jan 9, 2024 at 2:09 AM vignesh C <[email-masked]> 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)
> - Aggregate
> +--------------------------------------------------------------------
> + 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
> +(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_upgrade/log/regress_log_002_pg_upgrade

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.gmail.com

---

From: James Coleman <[email-masked]>
Date: Tue, 30 Jan 2024 21:56:20 -0500

> 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

---

From: Tom Lane <[email-masked]>
Date: Tue, 30 Jan 2024 22:34:01 -0500

> 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

---

From: James Coleman <[email-masked]>
Date: Wed, 31 Jan 2024 08:53:39 -0500

> 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

---

From: Robert Haas <[email-masked]>
Date: Wed, 31 Jan 2024 15:18:43 -0500

> 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.

> changing the status back from Waiting on Author to Needs Review.

Sounds good.

---

From: Andy Fan <[email-masked]>
Date: Thu, 03 Jul 2025 14:04:22 +0000

Hi James:

Very nice to find this topic, I am recently working on this topic [1] as 
well without finding this topic has been discussed before. I just go
through this thread and find it goes with a different direction with mine.
would you mind to check my soluation to see is there any case I can't
cover? I suggested this because my soluation should be much easier than
yours. But I'm not suprised to know I'm miss some obvious keypoint. 

> Off the cuff that seems fine. I'll read it over again when I send the

I had a detailed explaination for this plan change in [1] and I think
this could be amazing gain no matter which way we go finally. 

[1] https://www.postgresql.org/message-id/[email-masked] 

-- 
Best Regards
Andy Fan