Zoltán Borók-Nagy created IMPALA-15666:
------------------------------------------
Summary: Calcite: TPC-DS q72 joins date_dim d1 to d2 before
inventory and expands inventory 7x (SF3000: 19.4 s vs 7.6 s)
Key: IMPALA-15666
URL: https://issues.apache.org/jira/browse/IMPALA-15666
Project: IMPALA
Issue Type: Improvement
Components: Frontend
Reporter: Zoltán Borók-Nagy
At SF3000 on 8 executors (PLANNER=CALCITE, CTE_THRESHOLD=1, MT_DOP=8, with the
min-rows join order of IMPALA-15665), q72 takes 19.4 s:
* inventory (199.5M rows after runtime partition pruning) is joined to
warehouse and to a broadcast d2 x d1 join on d_week_seq (2,555 rows: 365 days
of d1 x 7). That expands it 7x to 1.37B rows.
* The 1.37B rows are shuffled on (d1.d_date_sk, inv_item_sk), about 53 GB, to a
partitioned join with the catalog_sales side.
* d1 (d_year = 2001) sits on the inventory side, so catalog_sales gets no date
pruning and reads all 4.3B rows.
* Estimates: the expanding join is estimated at 981.88M rows (actual 1.37B);
the (date, item) fact join at 39.2B (actual 29.9M).
A hand-written version with two inline views, inventory joined with d2 and
catalog_sales joined with d1, gives Calcite the better shape: each fact joins
its own date dimension first, and the facts join on (item, week). It takes 7.6
s with identical results (with BROADCAST_BYTES_LIMIT=100000000; otherwise the
catalog_sales side, estimated at 878K rows but about 27M actual, is broadcast).
*Why Calcite misses it:*
* ImpalaLoptOptimizeJoinRule.pushDownFactor() pushes a factor only into a
subtree that holds all factors it references. d2 references inventory
(inv_date_sk) and d1 (d_week_seq), so it is never grouped with inventory before
the fact-fact join.
* ImpalaRelMdDistinctRowCount.getDistinctRowCount() for a Join caps
NDV(d2.d_date_sk) by the join's row count (2,555) instead of about 371, so the
7x duplication is invisible to the |L| x |R| / max(NDV) estimate.
*Proposals:*
# Let the join ordering consider pushing a dimension into the fact side it
filters through an equi-join, even if it also references another factor.
# Compute the distinct count of a join output column from the column's distinct
count on its own side (bounded by the join rows), not from the join row count
alone.
# The composite-key overestimate of the (date, item) join (also 11.6B vs 9.1M
at SF1000) needs a correlation-aware estimate; today it makes large broadcasts
look cheap.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
