Re: Volatile Functions in Parallel Plans - Mailing list pgsql-hackers

Hi Zhenghua,

On Wed, Jul 15, 2020 at 5:44 AM Zhenghua Lyu wrote:
>
> Hi,
>     I test some SQL in the latest Postgres master branch code (we find these issues when
> developing Greenplum database in the PR https://nam04.safelinks.protection.outlook.com/?url=https%3A%2F%2Fgithub.com%2Fgreenplum-db%2Fgpdb%2Fpull%2F10418&data=02%7C01%7Czlyu%40vmware.com%7C41eeef401fb746757bc108d8294fe8d5%7Cb39138ca3cee4b4aa4d6cd83d9dd62f0%7C0%7C0%7C637304770468321177&sdata=XTxtrbJMO2d15WwQH9q4tXMzLPGFyk5hF8Tzs%2Bj3KlA%3D&reserved=0,
> and my colleague come up with the following cases in Postgres):
>
>
> create table t3 (c1 text, c2 text);
> CREATE TABLE
> insert into t3
> select
>   'fhufehwiohewiuewhuhwiufhwifhweuhfwu', --random data
>   'fiowehufwhfyegygfewpfwwfeuhwhufwh' --random data
> from generate_series(1, 10000000) i;
> INSERT 0 10000000
> analyze t3;
> ANALYZE
> create table t4 (like t3);
> CREATE TABLE
> insert into t4 select * from t4;
> INSERT 0 0
> insert into t4 select * from t3;
> INSERT 0 10000000
> analyze t4;
> ANALYZE
> set enable_hashjoin to off;
> SET
> explain (costs off)
> select count(*) from t3, t4
> where t3.c1 like '%sss'
>       and timeofday() = t4.c1 and t3.c1 = t4.c1;
>                        QUERY PLAN
> --------------------------------------------------------
>  Finalize Aggregate
>    ->  Gather
>          Workers Planned: 2
>          ->  Partial Aggregate
>                ->  Nested Loop
>                      Join Filter: (t3.c1 = t4.c1)
>                      ->  Parallel Seq Scan on t3
>                            Filter: (c1 ~~ '%sss'::text)
>                      ->  Seq Scan on t4
>                            Filter: (timeofday() = c1)
> (10 rows)
>
> explain (verbose, costs off)
> select count(*)
> from
>   t3,
>   (select *, timeofday() as x from t4 ) t4
> where t3.c1 like '%sss' and
>       timeofday() = t4.c1 and t3.c1 = t4.c1;
>                             QUERY PLAN
> ------------------------------------------------------------------
>  Finalize Aggregate
>    Output: count(*)
>    ->  Gather
>          Output: (PARTIAL count(*))
>          Workers Planned: 2
>          ->  Partial Aggregate
>                Output: PARTIAL count(*)
>                ->  Nested Loop
>                      Join Filter: (t3.c1 = t4.c1)
>                      ->  Parallel Seq Scan on public.t3
>                            Output: t3.c1, t3.c2
>                            Filter: (t3.c1 ~~ '%sss'::text)
>                      ->  Seq Scan on public.t4
>                            Output: t4.c1, NULL::text, timeofday()
>                            Filter: (timeofday() = t4.c1)
> (15 rows)
>
>
>
> Focus on the last two plans, the function timeofday is
> volatile but paralle-safe. And Postgres outputs two parallel
> plan.
>
>
> The first plan:
>
>  Finalize Aggregate
>    ->  Gather
>          Workers Planned: 2
>          ->  Partial Aggregate
>                ->  Nested Loop
>                      Join Filter: (t3.c1 = t4.c1)
>                      ->  Parallel Seq Scan on t3
>                            Filter: (c1 ~~ '%sss'::text)
>                      ->  Seq Scan on t4
>                            Filter: (timeofday() = c1)
>
> The join's left tree is parallel scan and the right tree is seq scan.
> This algorithm is correct using the distribute distributive law of
> distributed join:
>        A = [A1 A2 A3...An], B then A join B = gather( (A1 join B) (A2 join B) ... (An join B) )
>
> The correctness of the above law should have a pre-assumption:
>       The data set of B is the same in each join: (A1 join B) (A2 join B) ... (An join B)
>
> But things get complicated when volatile functions come in. Timeofday is just
> an example to show the idea. The core is volatile functions  can return different
> results on successive calls with the same arguments. Thus the following piece,
> the right tree of the join
>                      ->  Seq Scan on t4
>                            Filter: (timeofday() = c1)
> can not be considered consistent everywhere in the scan workers.
>
> The second plan
>
>  Finalize Aggregate
>    Output: count(*)
>    ->  Gather
>          Output: (PARTIAL count(*))
>          Workers Planned: 2
>          ->  Partial Aggregate
>                Output: PARTIAL count(*)
>                ->  Nested Loop
>                      Join Filter: (t3.c1 = t4.c1)
>                      ->  Parallel Seq Scan on public.t3
>                            Output: t3.c1, t3.c2
>                            Filter: (t3.c1 ~~ '%sss'::text)
>                      ->  Seq Scan on public.t4
>                            Output: t4.c1, NULL::text, timeofday()
>                            Filter: (timeofday() = t4.c1)
>
>
> have voltile projections in the right tree of the nestloop:
>
>                      ->  Seq Scan on public.t4
>                            Output: t4.c1, NULL::text, timeofday()
>                            Filter: (timeofday() = t4.c1)
>
> It should not be taken as consistent in different workers.

You are right, no they are not consistent. But Neither plans is
incorrect:

1. In the first query, it's semantically permissible to evaluate
timeofday() for each pair of (c1, c2), and the plan reflects that.
(Notice that the parallel nature of the plan is just noise here, the
planner could have gone with a Nested Loop of which the inner side is
_not_ materialized).

2. In the second query -- again -- in a canonical "outside-in"
evaluation, it's perfectly permissible to evaluate the subquery for each
value of t3. Again, the parallelism here is hardly relevant, a serial
plan without a material node on the inner side of a nested loop would
just as well (or as badly as you would feel) project different
timeofday() values for the same tuple from t4.

In short, the above plans seem fine.

P.S. the two plans you posted look identical to me, maybe I'm blind late
at night?

Cheers,
Jesse