Zoltán Borók-Nagy created IMPALA-15687:
------------------------------------------
Summary: Calcite: nested UNION DISTINCTs are not merged into one
n-ary union (TPC-DS q75: an extra de-duplication and a 0.8B-row shuffle per
year)
Key: IMPALA-15687
URL: https://issues.apache.org/jira/browse/IMPALA-15687
Project: IMPALA
Issue Type: Improvement
Components: Frontend
Reporter: Zoltán Borók-Nagy
With PLANNER=CALCITE, {{A UNION B UNION C}} is planned as two nested unions,
each with its own de-duplication. TPC-DS q75 unions catalog, store and web
sales for two years. On master (TPC-DS SF10000, 12 executors, MT_DOP=8,
PLANNER=CALCITE, best of 3: 62.3 s), the first year runs:
* {{01:UNION}} (catalog, store) -> {{62:EXCHANGE}} (835.69M rows) ->
{{63:AGGREGATE}} FINALIZE (743.06M rows, 3.37 s avg per instance);
* {{00:UNION}} (63, web) -> {{68:EXCHANGE}} (887.46M rows) -> {{69:AGGREGATE}}
FINALIZE.
The other year shuffles 820.61M rows into its first de-duplication ({{81}},
729.68M rows). The union opens its children one at a time
({{union-node.cc:232}}), so the web probe ({{23}}) starts only at 53.4-53.9 s,
after 63 has emitted its rows (from 49.0 s). A flat union does not avoid that
wait: with the legacy planner the web probe starts at 52.1-53.2 s (q75: 61.2 s,
best of 3), so merging mainly saves the extra shuffle and de-duplication.
The legacy planner already flattens the chain ({{SetOperationStmt.java:329}}
propagateDistinct(), {{:340}} unnestOperands()). Its q75 plan has one
three-operand UNION and one de-duplication per year (compare
{{tpcds/tpcds-q75.test}} with {{calcite_tpcds/tpcds-q75.test}}). q49 has the
same nested shape, on small inputs.
*Cause* (master c580d36e61): {{CalciteOptimizer.runExpandNodesProgram()}}
({{CalciteOptimizer.java:212-221}}) applies {{UNION_TO_DISTINCT}} bottom-up, so
{{Union(Union(cs, ss), ws)}} becomes
{{Aggregate(UnionAll(Aggregate(UnionAll(cs, ss)), ws))}}.
{{ImpalaCoreRules.UNION_MERGE}} ({{ImpalaCoreRules.java:123}}) wraps Calcite's
UnionMergeRule, which merges a DISTINCT union with a nested union, but no
program uses it.
*Proposal:*
* Run {{UNION_MERGE}} before {{UNION_TO_DISTINCT}} as a separate step. In the
same bottom-up collection, the inner union is rewritten first and can no longer
be merged.
* Consider {{INTERSECT_MERGE}} and {{MINUS_MERGE}} too: the nested INTERSECT of
q38 and EXCEPT of q87 also get two levels of aggregation.
* Test mixed UNION/INTERSECT chains (IMPALA-13540) and UNION ALL under UNION
DISTINCT.
The gain has not been measured with a prototype.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
