parallel safety of correlated subquery (was: parallel_safe)

Andy Fan <zhihuifan1213@163.com> writes:

Hi,

After some coding with this subject, I think it is better redefining
the problem and solution.

Problem:
--------

Supplan is common to be ineffective *AND* recently I find it is hard to
work with parallel framework. e.g.

create table bigt (a int, b int, c int);
insert into bigt select i, i, i from generate_series(1, 1000000)i;
analyze bigt;

q1:
select * from bigt o where b = 1
and c > (select avg(c) from bigt i where c = o.c);

We get plan:

                QUERY PLAN                 
-------------------------------------------
 Seq Scan on bigt o
   Filter: ((b = 1) AND (c > (SubPlan 1)))
   SubPlan 1
     ->  Aggregate
           ->  Seq Scan on bigt i
                 Filter: (c = o.c)
(6 rows)

Here we can see there is no parallel at all. However if split the query
q1 into queries q2 and q3, both of them can be parallelized.

q2:
explain (costs off) select * from bigt o where b = 1 and c > 2;
              QUERY PLAN               
---------------------------------------
 Gather
   Workers Planned: 2
   ->  Parallel Seq Scan on bigt o
         Filter: ((c > 2) AND (b = 1))
(4 rows)

q3:
explain (costs off) select avg(c) from bigt o where c = 2;
               QUERY PLAN                
-----------------------------------------
 Aggregate
   ->  Gather
         Workers Planned: 2
         ->  Parallel Seq Scan on bigt o
               Filter: (c = 2)
(5 rows)


Analysis
--------

The major reason of q1 can't be paralleled is the subplan is parameterized. 

the comment from add_partial_path:

 *	  We don't generate parameterized partial paths for several reasons.  Most
 *	  importantly, they're not safe to execute, because there's nothing to
 *	  make sure that a parallel scan within the parameterized portion of the
 *	  plan is running with the same value in every worker at the same time.

the comment from max_parallel_hazard_walker:

 * We can't pass Params to workers at the moment either .. unless
 * they are 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.

Solutions
----------

two foundations for this solution in my mind:

1. It is not safe to execute a partial parameterized plan with different
   parameter value, as what we have well done and documented. But this
   doesn't apply to a parameterized completed plan, in this case each
   worker runs a completed plan, they always generate the same result
   no matter it runs in parallel worker or leader.

2. The subplan never be a partial Plan. in make_subplan:

    best_path = get_cheapest_fractional_path(final_rel, tuple_fraction);

	plan = create_plan(subroot, best_path);

	/* And convert to SubPlan or InitPlan format. */
	result = build_subplan(root, plan, best_path,
						   subroot, plan_params,
						   subLinkType, subLinkId,
						   testexpr, NIL, isTopQual);

    get_cheapest_fractional_path never read rel->partial_pathlist.                        

So I think it is safe to ignore the PARAM_EXEC check in
max_parallel_hazard_context.safe_param_ids) for subplan. See attached 
patch 1.  

Benefit:
--------

After this patch, we could get the below plan -- the correlated subplan
is parallelized. 

explain (costs off) select * from bigt o where b = 1
    and c > (select avg(c) from bigt i where c = o.c);
                      QUERY PLAN                      
------------------------------------------------------
 Seq Scan on bigt o
   Filter: ((b = 1) AND ((c)::numeric > (SubPlan 1)))
   SubPlan 1
     ->  Aggregate
           ->  Gather
                 Workers Planned: 2
                 ->  Parallel Seq Scan on bigt i
                       Filter: (c = o.c)
(8 rows)

Continue the test to prove the impact of this patch by removing the
"Gather" in SubPlan, we could get the below plan -- scan with
parallel-safe SubPlan is parallelized.  

create table t (a int, b int);
explain (costs off) select * from bigt o where b = 1
    and c > (select avg(a) from t i where b = o.c);
                         QUERY PLAN                         
------------------------------------------------------------
 Gather
   Workers Planned: 2
   ->  Parallel Seq Scan on bigt o
         Filter: ((b = 1) AND ((c)::numeric > (SubPlan 1)))
         SubPlan 1
           ->  Aggregate
                 ->  Seq Scan on t i
                       Filter: (b = o.c)
(8 rows)


incremental_sort.sql provides another impacts of this patch. It is
helpful for parallel sort. 

Query:

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

>From (master)

                                       QUERY PLAN                                       
----------------------------------------------------------------------------------------
 Unique
   Output: t.unique1, ((SubPlan 1))
   ->  Sort
         Output: t.unique1, ((SubPlan 1))
         Sort Key: t.unique1, ((SubPlan 1))
         ->  Gather
               Output: t.unique1, (SubPlan 1)
               Workers Planned: 2
               ->  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, 1000)
               SubPlan 1
                 ->  Index Only Scan using tenk1_unique1 on public.tenk1
                       Output: t.unique1
                       Index Cond: (tenk1.unique1 = t.unique1)
(19 rows)

To (patched)

                                          QUERY PLAN                                          
----------------------------------------------------------------------------------------------
 Unique
   Output: t.unique1, ((SubPlan 1))
   ->  Gather Merge  * Merge gather at last *
         Output: t.unique1, ((SubPlan 1))
         Workers Planned: 2
         ->  Unique
               Output: t.unique1, ((SubPlan 1))
               ->  Sort ** Sort In worker *
                     Output: t.unique1, ((SubPlan 1))
                     Sort Key: t.unique1, ((SubPlan 1))
                     ->  Nested Loop
                            *SubPlan in Worker.**
                           Output: t.unique1, (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, 1000)
                           SubPlan 1
                             ->  Index Only Scan using tenk1_unique1 on public.tenk1
                                   Output: t.unique1
                                   Index Cond: (tenk1.unique1 = t.unique1)
(21 rows)

The execution time for the above query also decreased from 13351.928 ms
to 4814.043 ms, by 64%. The major difference is:

(1) master: correlated subquery is parallel unsafe, so it runs in leader
only, and then sort.
(2) patched: correlated subquery is parallel safe, so it run in worker
(Nested Loop) and then *sort in parallel worker* and then run "merge
gather".

About the implementation, I know 2 issues at least (the state is PoC
now). 

1. Query.is_in_sublink should be set in parser and keep unchanged later.
2. The below comment increment_sort.sql should be changed, it is just
   conflicted with this patch.

   """
   -- Parallel sort but with expression (correlated subquery) that
   -- is prohibited in parallel plans.
   """

Hope I have made myself clear, any feedback is welcome!

-- 
Best Regards
Andy Fan