search

trinodb / trino

Official repository of Trino, the distributed SQL query engine for big data, formerly known as PrestoSQL (https://trino.io)
https://trino.io
Apache License 2.0
10.18k stars 2.93k forks source link

Performance regression around MarkDistinct operator #3503

Open xerial opened 4 years ago

xerial commented 4 years ago

We are not yet 100% sure about the cause of performance regression, but a query which could finish in 1.96m with Presto 0.205 takes more than 15 minutes with Presto 317.

Here are screenshots of Presto query stages: Presto 0.205 (Finished in 1.96m) image

Presto 317: (Takes more than 16 minutes. Unnest operator is found at Stage 1): image

The only difference we found so far is the position of unnest operator, which is pulled up to Stage 1 in Presto 317, so we suspect performance regression in this PR: https://github.com/prestosql/presto/pull/901

It looks like markDistinct operator (count(distinct x)) or join processing needs to wait for the whole inputs from the child stages because unnest operation is evaluated later.

We will try to extract SQL subquery that will be relevant to this change.

Updates:

A rough SQL statement of this query plan (around stage 1-5) are like this:

with calendar as (
  select date from values 
 (sequence(from_unix_time(...), from_unix_time(...), …)) as 
 t(date_array) cross join unnest(date_array) as t(date)
),
m as (select t.date, count(distinct id) as x, count(distinct if …) as y, count(...) as z 
from t join calendar on t.date < calendar.date group by 1)
select f1, f2 from m cross join 
unnest (array[‘a’, ‘b’, ‘c’], array[m.x, m.y, m.z]) as t(f1, f2)

Here is the difference of query plans: image

martint commented 4 years ago

That PR only changed the implementation of the operator. It didn't touch planning or optimization. Since the unnest operation is taking place at a different part in the plan, this is likely a different problem.

phd3 commented 4 years ago

Agreed with Martin, the difference is likely coming from planner side. Also, looks like the linked PR from prestodb (for revert) was closed without merging.

xerial commented 4 years ago

ok. Do you have any idea which optimizer rule or planner pulls up unnest operator like this example?

martint commented 4 years ago

What makes you think the problem is in unnest? In the screenshot, the stage that contains that unnest operation you highlighted has seen 0 rows after 16 minutes. That would seem to indicate the problem is upstream of it.

BTW, can you describe how to interpret the structure laid out in the screenshots? What does [1] REPARTITION mean in the first one? Why does stage [4] appear to have three upstream stages given that there's only one join involved?

Maybe, it'd be easier if you could post the output of EXPLAIN for both versions (after anonymizing any sensitive data).

martint commented 4 years ago

Can you provide more details about stage [2], which has a markDistinct and an aggregation? Are both versions using the same fields for markDistinct? What functions are involved in the aggregation? What's in the project nodes in the first version?

xerial commented 4 years ago

ok. Showing explain (type distributed) will be a good idea as the other operators are also involved not only unnest

We will work on that (need some anonymization)

xerial commented 4 years ago

@martint @phd3 Uploaded a screenshot that shows the difference of EXPLAIN results.

A notable difference is where combine_hash is used around MarkDistinct operators. Unnest seems unrelated, so I'll change the ticket title.

tooptoop4 commented 4 years ago

@xerial just doing smaller query like below do u face perf regression?

select t.date, count(distinct id) as x, count(distinct if …) as y, count(...) as z from t join calendar on t.date < calendar.date group by 1

tooptoop4 commented 4 years ago

similar to https://github.com/prestosql/presto/issues/3031 which has DISTINCT

pangyifish commented 4 years ago

similar to #3031 which has DISTINCT

Thanks! not resolved yet:(

archongum commented 2 years ago

Any update? I hit on this issue.

Without count(distinct store_id) execution time is 10s. With count(distinct store_id) execution time is 2.94m.

Without count(distinct store_id) plan:

Fragment 1 [HASH]
    CPU: 5.95s, Scheduled: 6.25s, Input: 63 rows (198.80MB); per task: avg.: 7.88 std.dev.: 13.66, Output: 2 rows (120B)
    Output layout: [array_agg_distinct, sum_593, sum_592, expr_610, expr_399, sum]
    Output partitioning: SINGLE []
    Stage Execution Strategy: UNGROUPED_EXECUTION
    Project[]
    │   Layout: [array_agg_distinct:integer, sum_593:bigint, sum_592:bigint, expr_610:decimal(15,2), expr_399:varchar, sum:decimal(38,4)]
    │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
    │   CPU: 2.00ms (0.00%), Scheduled: 2.00ms (0.00%), Output: 2 rows (120B)
    │   Input avg.: 0.02 rows, Input std.dev.: 793.73%
    │   expr_610 := (CASE WHEN ("array_agg_distinct_594" > 0) THEN ((CAST("expr_595" AS decimal(10, 0)) * CAST(DECIMAL '100.00' AS decimal(5, 2))) / CAST("expr_596" AS decimal(10, 0))) ELSE CAST(DECIMAL '0.00' AS decimal(15, 2)) END)
    └─ Aggregate(FINAL)[expr_399][$hashvalue]
       │   Layout: [expr_399:varchar, $hashvalue:bigint, array_agg_distinct:integer, sum_593:bigint, expr_595:integer, sum_592:bigint, array_agg_distinct_594:integer, sum:decimal(38,4), expr_596:integer]
       │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
       │   CPU: 5.82s (1.49%), Scheduled: 6.11s (1.16%), Output: 2 rows (150B)
       │   Input avg.: 0.49 rows, Input std.dev.: 794.64%
       │   array_agg_distinct := array_agg_distinct("array_agg_distinct_676")
       │   sum_593 := sum("sum_675")
       │   expr_595 := array_agg_distinct("array_agg_distinct_678")
       │   sum_592 := sum("sum_674")
       │   array_agg_distinct_594 := array_agg_distinct("array_agg_distinct_677")
       │   sum := sum("sum_673")
       │   expr_596 := array_agg_distinct("array_agg_distinct_679")
       └─ LocalExchange[HASH][$hashvalue] ("expr_399")
          │   Layout: [expr_399:varchar, sum_674:row(bigint, boolean, bigint, boolean), array_agg_distinct_678:varchar, sum_673:varbinary, array_agg_distinct_679:varchar, array_agg_distinct_676:varchar, sum_675:row(bigint, boolean, bigint, boolean), array_agg_distinct_677:varchar, $hashvalue:bigint]
          │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
          │   CPU: 72.00ms (0.02%), Scheduled: 73.00ms (0.01%), Output: 63 rows (198.80MB)
          │   Input avg.: 0.25 rows, Input std.dev.: 497.35%
          ├─ Project[]
          │  │   Layout: [expr_399:varchar, sum_674:row(bigint, boolean, bigint, boolean), array_agg_distinct_678:varchar, sum_673:varbinary, array_agg_distinct_679:varchar, array_agg_distinct_676:varchar, sum_675:row(bigint, boolean, bigint, boolean), array_agg_distinct_677:varchar, $hashvalue_680:bigint]
          │  │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
          │  │   CPU: 12.00ms (0.00%), Scheduled: 26.00ms (0.00%), Output: 63 rows (198.80MB)
          │  │   Input avg.: 0.49 rows, Input std.dev.: 344.50%
          │  │   expr_399 := "time_period"
          │  └─ RemoteSource[2]
          │         Layout: [time_period:varchar, sum_674:row(bigint, boolean, bigint, boolean), array_agg_distinct_678:varchar, sum_673:varbinary, array_agg_distinct_679:varchar, array_agg_distinct_676:varchar, sum_675:row(bigint, boolean, bigint, boolean), array_agg_distinct_677:varchar, $hashvalue_680:bigint]
          │         CPU: 4.00ms (0.00%), Scheduled: 4.00ms (0.00%), Output: 63 rows (198.80MB)
          │         Input avg.: 0.49 rows, Input std.dev.: 344.50%
          └─ Project[]
             │   Layout: [expr_399:varchar, sum_674:row(bigint, boolean, bigint, boolean), array_agg_distinct_678:varchar, sum_673:varbinary, array_agg_distinct_679:varchar, array_agg_distinct_676:varchar, sum_675:row(bigint, boolean, bigint, boolean), array_agg_distinct_677:varchar, $hashvalue_682:bigint]
             │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
             │   CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Output: 0 rows (0B)
             │   Input avg.: 0.00 rows, Input std.dev.: ?%
             │   expr_399 := "time_period_211"
             └─ RemoteSource[3]
                    Layout: [time_period_211:varchar, sum_674:row(bigint, boolean, bigint, boolean), array_agg_distinct_678:varchar, sum_673:varbinary, array_agg_distinct_679:varchar, array_agg_distinct_676:varchar, sum_675:row(bigint, boolean, bigint, boolean), array_agg_distinct_677:varchar, $hashvalue_682:bigint]
                    CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Output: 0 rows (0B)
                    Input avg.: 0.00 rows, Input std.dev.: ?%

Fragment 2 [SOURCE]
    CPU: 6.40m, Scheduled: 7.61m, Input: 597047 rows (5.00GB); per task: avg.: 74630.88 std.dev.: 14964.00, Output: 63 rows (198.80MB)
    Output layout: [time_period, sum_674, array_agg_distinct_678, sum_673, array_agg_distinct_679, array_agg_distinct_676, sum_675, array_agg_distinct_677, $hashvalue_681]
    Output partitioning: HASH [time_period][$hashvalue_681]
    Stage Execution Strategy: UNGROUPED_EXECUTION
    Aggregate(PARTIAL)[time_period][$hashvalue_681]
    │   Layout: [time_period:varchar, $hashvalue_681:bigint, sum_674:row(bigint, boolean, bigint, boolean), array_agg_distinct_678:varchar, sum_673:varbinary, array_agg_distinct_679:varchar, array_agg_distinct_676:varchar, sum_675:row(bigint, boolean, bigint, boolean), array_agg_distinct_677:varchar]
    │   CPU: 5.00m (77.03%), Scheduled: 5.14m (58.55%), Output: 63 rows (198.80MB)
    │   Input avg.: 3772.22 rows, Input std.dev.: 45.50%
    │   sum_674 := sum("expr_661")
    │   array_agg_distinct_678 := array_agg_distinct("item_amt_ids")
    │   sum_673 := sum("sls_amt")
    │   array_agg_distinct_679 := array_agg_distinct("item_ids")
    │   array_agg_distinct_676 := array_agg_distinct("second_class_arr")
    │   sum_675 := sum("expr_662")
    │   array_agg_distinct_677 := array_agg_distinct("item_ids")
    └─ Project[]
       │   Layout: [time_period:varchar, sls_amt:decimal(20,4), expr_661:bigint, expr_662:bigint, second_class_arr:array(varchar), item_amt_ids:array(varchar), item_ids:array(varchar), $hashvalue_681:bigint]
       │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
       │   CPU: 481.00ms (0.12%), Scheduled: 488.00ms (0.09%), Output: 260283 rows (3.23GB)
       │   Input avg.: 3772.22 rows, Input std.dev.: 45.50%
       │   expr_661 := CAST("sls_qty" AS bigint)
       │   expr_662 := CAST("sls_order_cnt" AS bigint)
       └─ ScanFilterProject[table = hive:ads:ads_store_sales_detail_merge, grouped = false, filterPredicate = ((("province" IN (CAST(U&'\4E0A\6D77\5E02' AS varchar), CAST(U&'\4E91\5357\7701' AS varchar), CAST(U&'\5185\8499\53E4\81EA\6CBB\533A' AS varchar), CAST(U&'\5317\4EAC\5E02' AS varchar), CAST(U&'\5409\6797\7701' AS varchar), CAST(U&'\56DB\5DDD\7701' AS varchar), CAST(U&'\5929\6D25\5E02' AS varchar), CAST(U&'\5B81\590F\56DE\65CF\81EA\6CBB\533A' AS varchar), CAST(U&'\5B89\5FBD\7701' AS varchar), CAST(U&'\5C71\4E1C\7701' AS varchar), CAST(U&'\5C71\897F\7701' AS varchar), CAST(U&'\5E7F\4E1C\7701' AS varchar), CAST(U&'\5E7F\897F\58EE\65CF\81EA\6CBB\533A' AS varchar), CAST(U&'\65B0\7586\7EF4\543E\5C14\81EA\6CBB\533A' AS varchar), CAST(U&'\6C5F\82CF\7701' AS varchar), CAST(U&'\6C5F\897F\7701' AS varchar), CAST(U&'\6CB3\5317\7701' AS varchar), CAST(U&'\6CB3\5357\7701' AS varchar), CAST(U&'\6D59\6C5F\7701' AS varchar), CAST(U&'\6D77\5357\7701' AS varchar), CAST(U&'\6E56\5317\7701' AS varchar), CAST(U&'\6E56\5357\7701' AS varchar), CAST(U&'\7518\8083\7701' AS varchar), CAST(U&'\798F\5EFA\7701' AS varchar), CAST(U&'\897F\85CF\81EA\6CBB\533A' AS varchar), CAST(U&'\8D35\5DDE\7701' AS varchar), CAST(U&'\8FBD\5B81\7701' AS varchar), CAST(U&'\91CD\5E86\5E02' AS varchar), CAST(U&'\9655\897F\7701' AS varchar), CAST(U&'\9752\6D77\7701' AS varchar), CAST(U&'\9ED1\9F99\6C5F\7701' AS varchar))) AND ("platform_name" IN (CAST(U&'\4EAC\4E1C\5230\5BB6' AS varchar), CAST(U&'\7F8E\56E2\5916\5356' AS varchar), CAST(U&'\997F\4E86\4E48' AS varchar)))) AND ("store_brand_name" IN (CAST(U&'111\533B\836F\9986' AS varchar), CAST(U&'1\53F7\5FEB\836F' AS varchar), CAST(U&'\4E00\54C1\836F\4E1A' AS varchar), CAST(U&'\4E00\5FC3\533B\836F' AS varchar), CAST(U&'\4E00\5FC3\5802' AS varchar), CAST(U&'\4E00\6811\5409\5927\592B\836F\4E1A' AS varchar), CAST(U&'\4E00\6811\8001\767E\59D3' AS varchar), CAST(U&'\4E00\6811\836F\4E1A' AS varchar), CAST(U&'\4E00\6811\957F\5BFF\836F\4E1A' AS varchar), CAST(U&'\4E07\5B9D\5802\836F\5E97' AS varchar), CAST(U&'\4E07\5BB6\71D5\5927\836F\623F' AS varchar), CAST(U&'\4E07\82B8\5065\5EB7\836F\623F' AS varchar), CAST(U&'\4E09\5143\81F3\76DB\836F\623F' AS varchar), CAST(U&'\4E2D\667A\5927\836F\623F' AS varchar), CAST(U&'\4E5D\6D32\5927\836F\623F' AS varchar), CAST(U&'\4E5D\829D\5802\5927\836F\623F' AS varchar), CAST(U&'\4EC1\548C\5802\836F\5E97' AS varchar), CAST(U&'\4F1F\5FB7\533B\836F' AS varchar), CAST(U&'\4FDD\5174\5927\836F\623F' AS varchar), CAST(U&'\5065\4E4B\4F73\5065\5EB7\836F\623F' AS varchar), CAST(U&'\5065\751F\6E90\533B\836F' AS varchar), CAST(U&'\5143\521D\836F\623F' AS varchar), CAST(U&'\5148\58F0\518D\5EB7\5927\836F\623F' AS varchar), CAST(U&'\51EF\5FB7\836F\54C1' AS varchar), CAST(U&'\5317\4EAC\9996\90FD\673A\573A\5927\836F\623F' AS varchar), CAST(U&'\533B\521B\5FEB\836F' AS varchar), CAST(U&'\534E\6C0F\5927\836F\623F' AS varchar), CAST(U&'\534E\901A\533B\836F' AS varchar), CAST(U&'\53CB\548C\53E4\57CE\5927\836F\623F' AS varchar), CAST(U&'\53EE\5F53\5FEB\836F' AS varchar), CAST(U&'\540C\4EC1\5802' AS varchar), CAST(U&'\540C\65B9\836F\4E1A' AS varchar), CAST(U&'\548C\5E73\836F\623F' AS varchar), CAST(U&'\5510\4EBA\533B\836F' AS varchar), CAST(U&'\5609\5B9D\534E' AS varchar), CAST(U&'\5609\5B9D\5802\5927\836F\623F' AS varchar), CAST(U&'\56FD\5927\836F\623F' AS varchar), CAST(U&'\56FD\80DC\5927\836F\623F' AS varchar), CAST(U&'\5723\6770\836F\4E1A' AS varchar), CAST(U&'\5927\53C2\6797' AS varchar), CAST(U&'\5929\4FDD\5802\5927\836F\623F' AS varchar), CAST(U&'\5929\5929\4E50\5927\836F\623F' AS varchar), CAST(U&'\5929\6D4E\5927\836F\623F' AS varchar), CAST(U&'\5947\51A0\836F\5E97' AS varchar), CAST(U&'\5B8F\4EC1\5802\5927\836F\5E97' AS varchar), CAST(U&'\5B9C\53C8\4F73\533B\836F' AS varchar), CAST(U&'\5B9D\548C\5802\836F\4E1A' AS varchar), CAST(U&'\5E03\8863\5927\836F\623F' AS varchar), CAST(U&'\5E7F\6D4E\5927\836F\623F' AS varchar), CAST(U&'\5E7F\6D4E\8FDE\9501\836F\5E97' AS varchar), CAST(U&'\5EB7\4F70\5BB6\5927\836F\623F' AS varchar), CAST(U&'\5EB7\76DB\5802\5927\836F\5E97' AS varchar), CAST(U&'\5EB7\8D5B\5927\836F\623F' AS varchar), CAST(U&'\5F00\5FC3\4EBA\5927\836F\623F' AS varchar), CAST(U&'\5F18\5FB7\5927\836F\623F' AS varchar), CAST(U&'\5F18\6D4E\5802\5927\836F\623F' AS varchar), CAST(U&'\5F20\4EF2\666F\5927\836F\623F' AS varchar), CAST(U&'\5FB7\751F\5802' AS varchar), CAST(U&'\5FC3\8FDE\5FC3\5927\836F\623F' AS varchar), CAST(U&'\6021\5EB7\533B\836F' AS varchar), CAST(U&'\6052\6CF0\4EBA\6C11\5927\836F\623F' AS varchar), CAST(U&'\6052\9752\533B\836F' AS varchar), CAST(U&'\60E0\4EC1\5802\836F\4E1A' AS varchar), CAST(U&'\65B0\5174\836F\623F' AS varchar), CAST(U&'\65B0\5229\5B89\5FB7\836F\623F' AS varchar), CAST(U&'\65B0\836F\5927\836F\623F' AS varchar), CAST(U&'\6842\4E2D\5927\836F\623F' AS varchar), CAST(U&'\6B63\4E00\533B\836F' AS varchar), CAST(U&'\6B63\4EAC\5143\5927\836F\623F' AS varchar), CAST(U&'\6B63\548C\7965\5065\5EB7\836F\623F' AS varchar), CAST(U&'\6C11\5FC3\533B\836F\8FDE\9501' AS varchar), CAST(U&'\6C47\5EB7\533B\836F' AS varchar), CAST(U&'\6CB3\5357\7701\533B\836F\8D85\5E02' AS varchar), CAST(U&'\6CC9\6E90\5802\836F\623F' AS varchar), CAST(U&'\6CF0\9633\5927\836F\623F' AS varchar), CAST(U&'\6CFD\5F3A\836F\5E97' AS varchar), CAST(U&'\6D77\738B\661F\8FB0' AS varchar), CAST(U&'\6F31\7389\5E73\6C11\5927\836F\623F' AS varchar), CAST(U&'\71D5\559C\5802' AS varchar), CAST(U&'\745E\4EBA\5802' AS varchar), CAST(U&'\745E\6F84\5927\836F\623F' AS varchar), CAST(U&'\767E\4F73\60E0\5927\836F\623F' AS varchar), CAST(U&'\767E\4FE1\7F18' AS varchar), CAST(U&'\767E\59D3\4EBA\5BB6\5927\836F\623F' AS varchar), CAST(U&'\767E\59D3\5927\836F\623F' AS varchar), CAST(U&'\767E\59D3\7F18\5927\836F\623F' AS varchar), CAST(U&'\767E\5EB7\5927\836F\623F' AS varchar), CAST(U&'\767E\6B23\5802' AS varchar), CAST(U&'\767E\6CF0\533B\836F' AS varchar), CAST(U&'\76CA\4E30\5927\836F\623F' AS varchar), CAST(U&'\76CA\6625\5802\836F\623F' AS varchar), CAST(U&'\76CA\751F\5802\5927\836F\623F' AS varchar), CAST(U&'\7ACB\65B9\5927\836F\623F' AS varchar), CAST(U&'\7D2B\5149\836F\4E1A' AS varchar), CAST(U&'\8001\767E\59D3\5927\836F\623F' AS varchar), CAST(U&'\91D1\901A\5927\836F\5E97' AS varchar), CAST(U&'\9633\5149\5927\836F\623F' AS varchar), CAST(U&'\9AD8\6D4E\836F\623F' AS varchar), CAST(U&'\9E6D\71D5\5927\836F\623F' AS varchar), CAST(U&'\9F50\6CF0\533B\836F' AS varchar))))]
              Layout: [item_amt_ids:array(varchar), second_class_arr:array(varchar), item_ids:array(varchar), sls_qty:integer, sls_amt:decimal(20,4), sls_order_cnt:integer, time_period:varchar, $hashvalue_681:bigint]
              Estimates: {rows: 597047 (120.71MB), cpu: 209.53M, memory: 0B, network: 0B}/{rows: ? (?), cpu: 419.07M, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B}
              CPU: 1.39m (21.33%), Scheduled: 3.52m (40.18%), Output: 260283 rows (3.22GB)
              Input avg.: 8652.86 rows, Input std.dev.: 44.70%
              $hashvalue_681 := combine_hash(bigint '0', COALESCE("$operator$hash_code"("time_period"), 0))
              platform_name := platform_name:string:REGULAR
              store_brand_name := store_brand_name:string:REGULAR
              province := province:string:REGULAR
              second_class_arr := second_class_arr:array<string>:REGULAR
              item_amt_ids := item_amt_ids:array<string>:REGULAR
              item_ids := item_ids:array<string>:REGULAR
              sls_order_cnt := sls_order_cnt:int:REGULAR
              sls_qty := sls_qty:int:REGULAR
              sls_amt := sls_amt:decimal(20,4):REGULAR
              time_period := time_period:string:PARTITION_KEY
                  :: [[202108], [202109]]
              platform_id:string:PARTITION_KEY
                  :: [[ele], [meituanwm], [pdj]]
              Input: 597047 rows (5.00GB), Filtered: 56.40%

Fragment 3 [SINGLE]
    CPU: 950.41us, Scheduled: 993.11us, Input: 0 rows (0B); per task: avg.: 0.00 std.dev.: 0.00, Output: 0 rows (0B)
    Output layout: [time_period_211, sum_674, array_agg_distinct_678, sum_673, array_agg_distinct_679, array_agg_distinct_676, sum_675, array_agg_distinct_677, $hashvalue_683]
    Output partitioning: HASH [time_period_211][$hashvalue_683]
    Stage Execution Strategy: UNGROUPED_EXECUTION
    Aggregate(PARTIAL)[time_period_211][$hashvalue_683]
    │   Layout: [time_period_211:varchar, $hashvalue_683:bigint, sum_674:row(bigint, boolean, bigint, boolean), array_agg_distinct_678:varchar, sum_673:varbinary, array_agg_distinct_679:varchar, array_agg_distinct_676:varchar, sum_675:row(bigint, boolean, bigint, boolean), array_agg_distinct_677:varchar]
    │   CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Output: 0 rows (0B)
    │   Input avg.: 0.00 rows, Input std.dev.: ?%
    │   Collisions avg.: ? (?% est.), Collisions std.dev.: ?%
    │   sum_674 := sum("expr_668")
    │   array_agg_distinct_678 := array_agg_distinct("item_amt_ids_199")
    │   sum_673 := sum("sls_amt_206")
    │   array_agg_distinct_679 := array_agg_distinct("item_ids_200")
    │   array_agg_distinct_676 := array_agg_distinct("second_class_arr_198")
    │   sum_675 := sum("expr_669")
    │   array_agg_distinct_677 := array_agg_distinct("item_ids_200")
    └─ Project[]
       │   Layout: [time_period_211:varchar, sls_amt_206:decimal(20,4), expr_668:bigint, expr_669:bigint, second_class_arr_198:array(varchar), item_amt_ids_199:array(varchar), item_ids_200:array(varchar), $hashvalue_683:bigint]
       │   Estimates: {rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}
       │   CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Output: 0 rows (0B)
       │   Input avg.: 0.00 rows, Input std.dev.: ?%
       │   expr_668 := CAST("sls_qty_205" AS bigint)
       │   expr_669 := CAST("sls_order_cnt_203" AS bigint)
       └─ LocalExchange[ROUND_ROBIN] ()
          │   Layout: [second_class_arr_198:array(varchar), item_amt_ids_199:array(varchar), item_ids_200:array(varchar), sls_order_cnt_203:integer, sls_qty_205:integer, sls_amt_206:decimal(20,4), time_period_211:varchar, $hashvalue_683:bigint]
          │   Estimates: {rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}
          │   CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Output: 0 rows (0B)
          │   Input avg.: 0.00 rows, Input std.dev.: ?%
          └─ Project[]
             │   Layout: [second_class_arr_198:array(varchar), item_amt_ids_199:array(varchar), item_ids_200:array(varchar), sls_order_cnt_203:integer, sls_qty_205:integer, sls_amt_206:decimal(20,4), time_period_211:varchar, $hashvalue_684:bigint]
             │   Estimates: {rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}
             │   CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Output: 0 rows (0B)
             │   Input avg.: 0.00 rows, Input std.dev.: ?%
             │   $hashvalue_684 := combine_hash(bigint '0', COALESCE("$operator$hash_code"("time_period_211"), 0))
             └─ Values
                    Layout: [second_class_arr_198:array(varchar), item_amt_ids_199:array(varchar), item_ids_200:array(varchar), sls_order_cnt_203:integer, sls_qty_205:integer, sls_amt_206:decimal(20,4), time_period_211:varchar]
                    Estimates: {rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}
                    CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Output: 0 rows (0B)
                    Input avg.: 0.00 rows, Input std.dev.: ?%

With count(distinct store_id) plan:

Fragment 1 [HASH]
    CPU: 5.30m, Scheduled: 5.39m, Input: 260283 rows (8.44GB); per task: avg.: 32535.38 std.dev.: 56367.75, Output: 2 rows (138B)
    Output layout: [array_agg_distinct, sum_594, sum_593, count, expr_399, sum, expr_612]
    Output partitioning: SINGLE []
    Stage Execution Strategy: UNGROUPED_EXECUTION
    Project[]
    │   Layout: [array_agg_distinct:integer, sum_594:bigint, sum_593:bigint, count:bigint, expr_399:varchar, sum:decimal(38,4), expr_612:decimal(15,2)]
    │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
    │   CPU: 99.00ms (0.02%), Scheduled: 99.00ms (0.02%), Output: 2 rows (138B)
    │   Input avg.: 0.02 rows, Input std.dev.: 793.73%
    │   expr_612 := (CASE WHEN ("array_agg_distinct_595" > 0) THEN ((CAST("expr_596" AS decimal(10, 0)) * CAST(DECIMAL '100.00' AS decimal(5, 2))) / CAST("expr_597" AS decimal(10, 0))) ELSE CAST(DECIMAL '0.00' AS decimal(15, 2)) END)
    └─ Aggregate[expr_399][$hashvalue]
       │   Layout: [expr_399:varchar, $hashvalue:bigint, array_agg_distinct:integer, sum_594:bigint, sum_593:bigint, count:bigint, sum:decimal(38,4), array_agg_distinct_595:integer, expr_597:integer, expr_596:integer]
       │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
       │   CPU: 4.77m (60.77%), Scheduled: 4.83m (47.77%), Output: 2 rows (168B)
       │   Input avg.: 2033.46 rows, Input std.dev.: 793.88%
       │   array_agg_distinct := array_agg_distinct("expr_386")
       │   sum_594 := sum("sls_order_cnt_587")
       │   sum_593 := sum("sls_qty_586")
       │   count := count("expr_381") (mask = store_id$distinct)
       │   sum := sum("expr_394")
       │   array_agg_distinct_595 := array_agg_distinct("expr_388")
       │   expr_597 := array_agg_distinct("expr_388")
       │   expr_596 := array_agg_distinct("expr_387")
       └─ MarkDistinct[distinct=expr_399:varchar, expr_381:varchar marker=store_id$distinct][$hashvalue_686]
          │   Layout: [expr_399:varchar, expr_394:decimal(20,4), sls_qty_586:bigint, sls_order_cnt_587:bigint, expr_386:array(varchar), expr_387:array(varchar), expr_388:array(varchar), expr_381:varchar, $hashvalue:bigint, $hashvalue_686:bigint, store_id$distinct:boolean]
          │   CPU: 410.00ms (0.09%), Scheduled: 412.00ms (0.07%), Output: 260283 rows (8.44GB)
          │   Input avg.: 2033.46 rows, Input std.dev.: 793.88%
          └─ LocalExchange[HASH][$hashvalue] ("expr_399")
             │   Layout: [expr_399:varchar, expr_394:decimal(20,4), sls_qty_586:bigint, sls_order_cnt_587:bigint, expr_386:array(varchar), expr_387:array(varchar), expr_388:array(varchar), expr_381:varchar, $hashvalue:bigint, $hashvalue_686:bigint]
             │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?}
             │   CPU: 30.33s (6.44%), Scheduled: 31.20s (5.14%), Output: 260283 rows (8.44GB)
             │   Input avg.: 1016.73 rows, Input std.dev.: 321.13%
             ├─ RemoteSource[2]
             │      Layout: [time_period:varchar, sls_amt:decimal(20,4), expr_672:bigint, expr_673:bigint, second_class_arr:array(varchar), item_amt_ids:array(varchar), item_ids:array(varchar), store_id:varchar, $hashvalue_687:bigint, $hashvalue_688:bigint]
             │      CPU: 1.39s (0.30%), Scheduled: 1.57s (0.26%), Output: 260283 rows (8.44GB)
             │      Input avg.: 2033.46 rows, Input std.dev.: 215.78%
             └─ RemoteSource[3]
                    Layout: [time_period_211:varchar, sls_amt_206:decimal(20,4), expr_680:bigint, expr_681:bigint, second_class_arr_198:array(varchar), item_amt_ids_199:array(varchar), item_ids_200:array(varchar), store_id_186:varchar, $hashvalue_691:bigint, $hashvalue_692:bigint]
                    CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Output: 0 rows (0B)
                    Input avg.: 0.00 rows, Input std.dev.: ?%

Fragment 2 [SOURCE]
    CPU: 2.55m, Scheduled: 3.72m, Input: 597047 rows (5.01GB); per task: avg.: 74630.88 std.dev.: 9431.65, Output: 260283 rows (8.44GB)
    Output layout: [time_period, sls_amt, expr_672, expr_673, second_class_arr, item_amt_ids, item_ids, store_id, $hashvalue_689, $hashvalue_690]
    Output partitioning: HASH [time_period][$hashvalue_689]
    Stage Execution Strategy: UNGROUPED_EXECUTION
    Project[]
    │   Layout: [time_period:varchar, sls_amt:decimal(20,4), expr_672:bigint, expr_673:bigint, second_class_arr:array(varchar), item_amt_ids:array(varchar), item_ids:array(varchar), store_id:varchar, $hashvalue_689:bigint, $hashvalue_690:bigint]
    │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
    │   CPU: 1.17m (14.89%), Scheduled: 1.30m (12.84%), Output: 260283 rows (8.44GB)
    │   Input avg.: 3772.22 rows, Input std.dev.: 45.50%
    │   expr_672 := CAST("sls_qty" AS bigint)
    │   expr_673 := CAST("sls_order_cnt" AS bigint)
    └─ ScanFilterProject[table = hive:ads:ads_store_sales_detail_merge, grouped = false, filterPredicate = ((("province" IN (CAST(U&'\4E0A\6D77\5E02' AS varchar), CAST(U&'\4E91\5357\7701' AS varchar), CAST(U&'\5185\8499\53E4\81EA\6CBB\533A' AS varchar), CAST(U&'\5317\4EAC\5E02' AS varchar), CAST(U&'\5409\6797\7701' AS varchar), CAST(U&'\56DB\5DDD\7701' AS varchar), CAST(U&'\5929\6D25\5E02' AS varchar), CAST(U&'\5B81\590F\56DE\65CF\81EA\6CBB\533A' AS varchar), CAST(U&'\5B89\5FBD\7701' AS varchar), CAST(U&'\5C71\4E1C\7701' AS varchar), CAST(U&'\5C71\897F\7701' AS varchar), CAST(U&'\5E7F\4E1C\7701' AS varchar), CAST(U&'\5E7F\897F\58EE\65CF\81EA\6CBB\533A' AS varchar), CAST(U&'\65B0\7586\7EF4\543E\5C14\81EA\6CBB\533A' AS varchar), CAST(U&'\6C5F\82CF\7701' AS varchar), CAST(U&'\6C5F\897F\7701' AS varchar), CAST(U&'\6CB3\5317\7701' AS varchar), CAST(U&'\6CB3\5357\7701' AS varchar), CAST(U&'\6D59\6C5F\7701' AS varchar), CAST(U&'\6D77\5357\7701' AS varchar), CAST(U&'\6E56\5317\7701' AS varchar), CAST(U&'\6E56\5357\7701' AS varchar), CAST(U&'\7518\8083\7701' AS varchar), CAST(U&'\798F\5EFA\7701' AS varchar), CAST(U&'\897F\85CF\81EA\6CBB\533A' AS varchar), CAST(U&'\8D35\5DDE\7701' AS varchar), CAST(U&'\8FBD\5B81\7701' AS varchar), CAST(U&'\91CD\5E86\5E02' AS varchar), CAST(U&'\9655\897F\7701' AS varchar), CAST(U&'\9752\6D77\7701' AS varchar), CAST(U&'\9ED1\9F99\6C5F\7701' AS varchar))) AND ("platform_name" IN (CAST(U&'\4EAC\4E1C\5230\5BB6' AS varchar), CAST(U&'\7F8E\56E2\5916\5356' AS varchar), CAST(U&'\997F\4E86\4E48' AS varchar)))) AND ("store_brand_name" IN (CAST(U&'111\533B\836F\9986' AS varchar), CAST(U&'1\53F7\5FEB\836F' AS varchar), CAST(U&'\4E00\54C1\836F\4E1A' AS varchar), CAST(U&'\4E00\5FC3\533B\836F' AS varchar), CAST(U&'\4E00\5FC3\5802' AS varchar), CAST(U&'\4E00\6811\5409\5927\592B\836F\4E1A' AS varchar), CAST(U&'\4E00\6811\8001\767E\59D3' AS varchar), CAST(U&'\4E00\6811\836F\4E1A' AS varchar), CAST(U&'\4E00\6811\957F\5BFF\836F\4E1A' AS varchar), CAST(U&'\4E07\5B9D\5802\836F\5E97' AS varchar), CAST(U&'\4E07\5BB6\71D5\5927\836F\623F' AS varchar), CAST(U&'\4E07\82B8\5065\5EB7\836F\623F' AS varchar), CAST(U&'\4E09\5143\81F3\76DB\836F\623F' AS varchar), CAST(U&'\4E2D\667A\5927\836F\623F' AS varchar), CAST(U&'\4E5D\6D32\5927\836F\623F' AS varchar), CAST(U&'\4E5D\829D\5802\5927\836F\623F' AS varchar), CAST(U&'\4EC1\548C\5802\836F\5E97' AS varchar), CAST(U&'\4F1F\5FB7\533B\836F' AS varchar), CAST(U&'\4FDD\5174\5927\836F\623F' AS varchar), CAST(U&'\5065\4E4B\4F73\5065\5EB7\836F\623F' AS varchar), CAST(U&'\5065\751F\6E90\533B\836F' AS varchar), CAST(U&'\5143\521D\836F\623F' AS varchar), CAST(U&'\5148\58F0\518D\5EB7\5927\836F\623F' AS varchar), CAST(U&'\51EF\5FB7\836F\54C1' AS varchar), CAST(U&'\5317\4EAC\9996\90FD\673A\573A\5927\836F\623F' AS varchar), CAST(U&'\533B\521B\5FEB\836F' AS varchar), CAST(U&'\534E\6C0F\5927\836F\623F' AS varchar), CAST(U&'\534E\901A\533B\836F' AS varchar), CAST(U&'\53CB\548C\53E4\57CE\5927\836F\623F' AS varchar), CAST(U&'\53EE\5F53\5FEB\836F' AS varchar), CAST(U&'\540C\4EC1\5802' AS varchar), CAST(U&'\540C\65B9\836F\4E1A' AS varchar), CAST(U&'\548C\5E73\836F\623F' AS varchar), CAST(U&'\5510\4EBA\533B\836F' AS varchar), CAST(U&'\5609\5B9D\534E' AS varchar), CAST(U&'\5609\5B9D\5802\5927\836F\623F' AS varchar), CAST(U&'\56FD\5927\836F\623F' AS varchar), CAST(U&'\56FD\80DC\5927\836F\623F' AS varchar), CAST(U&'\5723\6770\836F\4E1A' AS varchar), CAST(U&'\5927\53C2\6797' AS varchar), CAST(U&'\5929\4FDD\5802\5927\836F\623F' AS varchar), CAST(U&'\5929\5929\4E50\5927\836F\623F' AS varchar), CAST(U&'\5929\6D4E\5927\836F\623F' AS varchar), CAST(U&'\5947\51A0\836F\5E97' AS varchar), CAST(U&'\5B8F\4EC1\5802\5927\836F\5E97' AS varchar), CAST(U&'\5B9C\53C8\4F73\533B\836F' AS varchar), CAST(U&'\5B9D\548C\5802\836F\4E1A' AS varchar), CAST(U&'\5E03\8863\5927\836F\623F' AS varchar), CAST(U&'\5E7F\6D4E\5927\836F\623F' AS varchar), CAST(U&'\5E7F\6D4E\8FDE\9501\836F\5E97' AS varchar), CAST(U&'\5EB7\4F70\5BB6\5927\836F\623F' AS varchar), CAST(U&'\5EB7\76DB\5802\5927\836F\5E97' AS varchar), CAST(U&'\5EB7\8D5B\5927\836F\623F' AS varchar), CAST(U&'\5F00\5FC3\4EBA\5927\836F\623F' AS varchar), CAST(U&'\5F18\5FB7\5927\836F\623F' AS varchar), CAST(U&'\5F18\6D4E\5802\5927\836F\623F' AS varchar), CAST(U&'\5F20\4EF2\666F\5927\836F\623F' AS varchar), CAST(U&'\5FB7\751F\5802' AS varchar), CAST(U&'\5FC3\8FDE\5FC3\5927\836F\623F' AS varchar), CAST(U&'\6021\5EB7\533B\836F' AS varchar), CAST(U&'\6052\6CF0\4EBA\6C11\5927\836F\623F' AS varchar), CAST(U&'\6052\9752\533B\836F' AS varchar), CAST(U&'\60E0\4EC1\5802\836F\4E1A' AS varchar), CAST(U&'\65B0\5174\836F\623F' AS varchar), CAST(U&'\65B0\5229\5B89\5FB7\836F\623F' AS varchar), CAST(U&'\65B0\836F\5927\836F\623F' AS varchar), CAST(U&'\6842\4E2D\5927\836F\623F' AS varchar), CAST(U&'\6B63\4E00\533B\836F' AS varchar), CAST(U&'\6B63\4EAC\5143\5927\836F\623F' AS varchar), CAST(U&'\6B63\548C\7965\5065\5EB7\836F\623F' AS varchar), CAST(U&'\6C11\5FC3\533B\836F\8FDE\9501' AS varchar), CAST(U&'\6C47\5EB7\533B\836F' AS varchar), CAST(U&'\6CB3\5357\7701\533B\836F\8D85\5E02' AS varchar), CAST(U&'\6CC9\6E90\5802\836F\623F' AS varchar), CAST(U&'\6CF0\9633\5927\836F\623F' AS varchar), CAST(U&'\6CFD\5F3A\836F\5E97' AS varchar), CAST(U&'\6D77\738B\661F\8FB0' AS varchar), CAST(U&'\6F31\7389\5E73\6C11\5927\836F\623F' AS varchar), CAST(U&'\71D5\559C\5802' AS varchar), CAST(U&'\745E\4EBA\5802' AS varchar), CAST(U&'\745E\6F84\5927\836F\623F' AS varchar), CAST(U&'\767E\4F73\60E0\5927\836F\623F' AS varchar), CAST(U&'\767E\4FE1\7F18' AS varchar), CAST(U&'\767E\59D3\4EBA\5BB6\5927\836F\623F' AS varchar), CAST(U&'\767E\59D3\5927\836F\623F' AS varchar), CAST(U&'\767E\59D3\7F18\5927\836F\623F' AS varchar), CAST(U&'\767E\5EB7\5927\836F\623F' AS varchar), CAST(U&'\767E\6B23\5802' AS varchar), CAST(U&'\767E\6CF0\533B\836F' AS varchar), CAST(U&'\76CA\4E30\5927\836F\623F' AS varchar), CAST(U&'\76CA\6625\5802\836F\623F' AS varchar), CAST(U&'\76CA\751F\5802\5927\836F\623F' AS varchar), CAST(U&'\7ACB\65B9\5927\836F\623F' AS varchar), CAST(U&'\7D2B\5149\836F\4E1A' AS varchar), CAST(U&'\8001\767E\59D3\5927\836F\623F' AS varchar), CAST(U&'\91D1\901A\5927\836F\5E97' AS varchar), CAST(U&'\9633\5149\5927\836F\623F' AS varchar), CAST(U&'\9AD8\6D4E\836F\623F' AS varchar), CAST(U&'\9E6D\71D5\5927\836F\623F' AS varchar), CAST(U&'\9F50\6CF0\533B\836F' AS varchar))))]
           Layout: [store_id:varchar, item_amt_ids:array(varchar), second_class_arr:array(varchar), item_ids:array(varchar), sls_qty:integer, sls_amt:decimal(20,4), sls_order_cnt:integer, time_period:varchar, $hashvalue_689:bigint, $hashvalue_690:bigint]
           Estimates: {rows: 597047 (157.15MB), cpu: 240.85M, memory: 0B, network: 0B}/{rows: ? (?), cpu: 481.70M, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B}
           CPU: 1.37m (17.51%), Scheduled: 3.43m (33.90%), Output: 260283 rows (3.23GB)
           Input avg.: 8652.86 rows, Input std.dev.: 44.70%
           $hashvalue_689 := combine_hash(bigint '0', COALESCE("$operator$hash_code"("time_period"), 0))
           $hashvalue_690 := combine_hash(combine_hash(bigint '0', COALESCE("$operator$hash_code"("time_period"), 0)), COALESCE("$operator$hash_code"("store_id"), 0))
           platform_name := platform_name:string:REGULAR
           store_id := store_id:string:REGULAR
           store_brand_name := store_brand_name:string:REGULAR
           province := province:string:REGULAR
           second_class_arr := second_class_arr:array<string>:REGULAR
           item_amt_ids := item_amt_ids:array<string>:REGULAR
           item_ids := item_ids:array<string>:REGULAR
           sls_order_cnt := sls_order_cnt:int:REGULAR
           sls_qty := sls_qty:int:REGULAR
           sls_amt := sls_amt:decimal(20,4):REGULAR
           time_period := time_period:string:PARTITION_KEY
               :: [[202108], [202109]]
           platform_id:string:PARTITION_KEY
               :: [[ele], [meituanwm], [pdj]]
           Input: 597047 rows (5.01GB), Filtered: 56.40%

Fragment 3 [SINGLE]
    CPU: 700.43us, Scheduled: 742.25us, Input: 0 rows (0B); per task: avg.: 0.00 std.dev.: 0.00, Output: 0 rows (0B)
    Output layout: [time_period_211, sls_amt_206, expr_680, expr_681, second_class_arr_198, item_amt_ids_199, item_ids_200, store_id_186, $hashvalue_693, $hashvalue_694]
    Output partitioning: HASH [time_period_211][$hashvalue_693]
    Stage Execution Strategy: UNGROUPED_EXECUTION
    Project[]
    │   Layout: [time_period_211:varchar, sls_amt_206:decimal(20,4), expr_680:bigint, expr_681:bigint, second_class_arr_198:array(varchar), item_amt_ids_199:array(varchar), item_ids_200:array(varchar), store_id_186:varchar, $hashvalue_693:bigint, $hashvalue_694:bigint]
    │   Estimates: {rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}
    │   CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Output: 0 rows (0B)
    │   Input avg.: 0.00 rows, Input std.dev.: ?%
    │   expr_680 := CAST("sls_qty_205" AS bigint)
    │   expr_681 := CAST("sls_order_cnt_203" AS bigint)
    └─ LocalExchange[ROUND_ROBIN] ()
       │   Layout: [store_id_186:varchar, second_class_arr_198:array(varchar), item_amt_ids_199:array(varchar), item_ids_200:array(varchar), sls_order_cnt_203:integer, sls_qty_205:integer, sls_amt_206:decimal(20,4), time_period_211:varchar, $hashvalue_693:bigint, $hashvalue_694:bigint]
       │   Estimates: {rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}
       │   CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Output: 0 rows (0B)
       │   Input avg.: 0.00 rows, Input std.dev.: ?%
       └─ Project[]
          │   Layout: [store_id_186:varchar, second_class_arr_198:array(varchar), item_amt_ids_199:array(varchar), item_ids_200:array(varchar), sls_order_cnt_203:integer, sls_qty_205:integer, sls_amt_206:decimal(20,4), time_period_211:varchar, $hashvalue_695:bigint, $hashvalue_696:bigint]
          │   Estimates: {rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}
          │   CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Output: 0 rows (0B)
          │   Input avg.: 0.00 rows, Input std.dev.: ?%
          │   $hashvalue_695 := combine_hash(bigint '0', COALESCE("$operator$hash_code"("time_period_211"), 0))
          │   $hashvalue_696 := combine_hash(combine_hash(bigint '0', COALESCE("$operator$hash_code"("time_period_211"), 0)), COALESCE("$operator$hash_code"("store_id_186"), 0))
          └─ Values
                 Layout: [store_id_186:varchar, second_class_arr_198:array(varchar), item_amt_ids_199:array(varchar), item_ids_200:array(varchar), sls_order_cnt_203:integer, sls_qty_205:integer, sls_amt_206:decimal(20,4), time_period_211:varchar]
                 Estimates: {rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}
                 CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Output: 0 rows (0B)
                 Input avg.: 0.00 rows, Input std.dev.: ?%