Zoltán Borók-Nagy created IMPALA-15669:
------------------------------------------
Summary: Calcite: eager aggregation below filter-only dimension
joins (TPC-DS q70) and eager DISTINCT before the shuffle (q87, q38)
Key: IMPALA-15669
URL: https://issues.apache.org/jira/browse/IMPALA-15669
Project: IMPALA
Issue Type: Improvement
Components: Frontend
Reporter: Zoltán Borók-Nagy
Two TPC-DS shapes aggregate large fact inputs only after joins that do not
reduce them. Measured at SF3000 on 8 executors (PLANNER=CALCITE,
CTE_THRESHOLD=1, MT_DOP=8).
*q70 (7.8 s; 1.8 s hand-written).* The "results" subquery runs three times
(once per rollup level). Each copy scans 1.6B store_sales rows, broadcast-joins
store and date_dim, and aggregates on (s_county, s_state) strings at about 210
ns/row. A hand-written plan that aggregates by (ss_store_sk, ss_sold_date_sk)
at the scan, joins afterwards, and shares one scan, takes 1.8 s with identical
results.
* The aggregate-join transpose (IMPALA-15668) does not fire: the top join of
"results" is the date join, and date_dim contributes no group-by column, so the
rule finds no unique side with grouping attributes. It only matches an
Aggregate directly on a Join, so it never reaches the store join below.
* A prototype that treats a unique, filter-only FK->PK side as transparent
gained only 0.7 s on q70, while q77 and q86 lost 4.9 s and 2.2 s. It needs a
tighter condition, e.g. an RF-able key plus the existing ratio check.
* At SF10000 (12 executors, best of 3), a prototype that joins filter-only
dimensions before attribute dimensions, so the join directly under the
aggregate provides the group-by columns and the transpose fires, cut q70 from
18.1 to 3.9 s and q14_2 from 47.6 to 39.1 s.
* The three copies of "results" are a separate problem (IMPALA-15661).
*q87 (7.9 s; 6.3 s hand-written).* Each of the three branches shuffles its
whole fact table on the customer key (1.62B store_sales rows, 9.7 GB) before
joining customer and taking DISTINCT (names, d_date). An eager DISTINCT
(ss_customer_sk, d_date) right after the broadcast date join cuts the rows
before the shuffle. The hand-written version takes 6.3 s with identical
results. q38 has the same shape.
* The transpose ratio gate does not fire because NDV(ss_customer_sk,
ss_sold_date_sk) is estimated as a product capped at the row count, so the
estimated reduction is about 1; the actual ratio is about 0.09.
*Proposal:* for duplicate-insensitive aggregates (DISTINCT, MIN, MAX) over
FK->PK joins, push a streaming-only pre-aggregation on (FK, other keys) to the
fact side, whatever the NDV estimate says. It needs no merge or shuffle,
because the upper aggregation removes duplicates, and the streaming
pre-aggregation already passes rows through when it does not reduce them.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
