Zoltán Borók-Nagy created IMPALA-15524:
------------------------------------------
Summary: Calcite planner: ProjectToSemiJoinRule drops a selective
DISTINCT, so TPC-DS q14 shuffles 43M rows 15 times
Key: IMPALA-15524
URL: https://issues.apache.org/jira/browse/IMPALA-15524
Project: IMPALA
Issue Type: Improvement
Components: Frontend
Reporter: Zoltán Borók-Nagy
CalciteOptimizer adds ProjectToSemiJoinRule (CalciteOptimizer.java:253,
ImpalaCoreRules.PROJECT_TO_SEMI_JOIN). The rule rewrites
Project(InnerJoin(left, Aggregate(input))) into Project(SemiJoin(left, input))
when the join keys are the aggregate's group keys. A semi-join does not need
the uniqueness that the Aggregate provides, so Calcite drops it.
For Impala this is a bad trade when the Aggregate reduces its input a lot.
Without the Aggregate, the whole raw input is shuffled and built into the hash
table. With it, Impala runs a streaming pre-aggregation before the exchange.
*Example: TPC-DS q14_1 (SF100, master 1f34ea915e, PLANNER=CALCITE).*
cross_items is an INTERSECT of three DISTINCT subqueries. For the web_sales
branch, the rule drops the DISTINCT:
* All 43M web_sales rows are shuffled, 15 times (once per inlined copy).
Senders take 4.3-4.6 s on average and 7.4-8.7 s at most.
* They are built into 702 MB hash tables, 11 GB in total, and those joins take
1.6 s each on average.
* The original planner pre-aggregates the same branch to 87.75K rows before the
exchange.
* As a result q14_1 runs 44.7 s with Calcite against 27.0 s with the original
planner (MT_DOP=0). Planning time accounts for part of the gap (IMPALA-XXXXX,
HepPlanner noDag).
*Proposal:* keep the aggregate under the semi-join, giving
Project(SemiJoin(left, Aggregate(input))), when it is a plain DISTINCT whose
estimated input is at least 1M rows and more than 10x its estimated output.
Otherwise behave as today.
* The join condition can be reused verbatim, because the right input's row type
does not change.
* The rewrite is correct because every left row matches at most one Aggregate
row.
* The 1M-row floor avoids pointless pre-aggregation of small inputs, such as a
filtered dimension table. Without it, q33 would pre-aggregate a 36K-row item
build down to 957 rows.
*Prototype result:*
* Among the 104 TPC-DS queries, only q14_1's plan changes.
* Together with IMPALA-XXXXX (HepPlanner DAG mode), q14_1 went from 22.3 s to
7.0 s (MT_DOP=8, CTE_THRESHOLD=1).
Alternatively, Impala could add a pre-aggregation on the build side of a
semi-join when the build input has many duplicate keys. That would also cover
plans from the original planner.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
