Zoltán Borók-Nagy created IMPALA-15676:
------------------------------------------
Summary: Streaming pre-aggregation: bypass hashing when the
measured reduction is low, and fix the integer division in
ShouldExpandPreaggHashTables()
Key: IMPALA-15676
URL: https://issues.apache.org/jira/browse/IMPALA-15676
Project: IMPALA
Issue Type: Improvement
Components: Backend
Reporter: Zoltán Borók-Nagy
Whether a streaming pre-aggregation keeps growing its hash tables depends on
GroupingAggregator::ShouldExpandPreaggHashTables() (master 1a26e1f778). It
extrapolates the reduction seen so far to the whole input with R = 1 + (N / n)
* (r - 1), where N comes from the planner's input estimate. Two problems:
* N / n is an integer division (both are int64_t), so it truncates.
* The extrapolation trusts the planner's estimate. At large scale,
pre-aggregations that reduce almost nothing build huge hash tables.
TPC-DS at SF3000 (8 executors, MT_DOP=8, PLANNER=CALCITE):
* In q78, q97 and q65 the streaming pre-aggregation keeps 92-99% of its rows
but builds hash tables of up to 1.2 GB per instance.
* q65: the planner estimates 129M input rows per instance (8.25B / 64, from the
default range selectivity, see IMPALA-14867); the actual is 25.5M. The measured
reduction 1.13 extrapolates to 1.95 against a threshold of 2.0, the real
reduction is 1.09, and the pre-aggregation builds a 786 MB table at 166 ns/row.
With PREAGG_BYTES_LIMIT=16777216 q65 goes from 13.0 s to 9.9 s.
* q4 goes the other way: the planner expects a 1.4x reduction, the real one is
11.6x. So better estimates cannot replace a runtime decision.
*Prototype:* fix the division, and once a pre-aggregation instance has hashed
1M rows, measure the reduction over the second half of them. If it is below
1.2, pass rows through without hashing them, and re-sample periodically (the
bypass window doubles after each re-sample that confirms the low reduction, up
to 16M rows). Results are not affected.
Measured on the 103 TPC-DS queries at SF3000: 644.8 s -> 621.9 s, with
identical results.
* Gains: q67a 61.9 s -> 56.1 s, q65 13.0 -> 8.0, q97 17.9 -> 13.0, q78 51.5 ->
47.7.
* Losses: q4 18.5 -> 19.5, and a few within 0.4 s.
At SF100 on 5 executors (MT_DOP=0) the same change gave -2.1% (geomean 0.97),
mixed per query. Input sorted on the group key can look non-reducing early; the
periodic re-sampling limits the damage.
Prototype: https://gerrit.cloudera.org/#/c/25099/ (flag
--preagg_adaptive_bypass, default true)
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
