andygrove opened a new issue, #2665:
URL: https://github.com/apache/datafusion-ballista/issues/2665
**Is your feature request related to a problem or challenge? Please describe
what you are trying to do.**
Join cardinality estimates downstream of a shuffle are degenerate because a
resolved exchange carries no distinct counts. Without NDV, DataFusion's
`estimate_inner_join_cardinality` uses each side's row count as its NDV, so an
inner join is estimated at the smaller input's row count. For a key/foreign-key
join whose large side is the foreign key, that undercounts by the join's
fan-out.
In TPC-H q21 at SF1000 (#2662), `(supplier ⋈ nation) ⋈ lineitem` was
estimated at 399,809 rows and produced 151,688,609. AQE broadcast it on that
estimate and the job failed (#2661, fixed for that case by #2663 by measuring
before broadcasting). The same blind spot will mislead any join reordering,
which is the biggest q21 win.
#2664 (closing #1960) carries per-column statistics from completed stages
onto the AQE `ExchangeExec`, but only null counts are collected today.
**Describe the solution you'd like**
Collect approximate distinct counts with HyperLogLog on the shuffle write
path and report them as `ColumnStatistics::distinct_count`
(`Precision::Inexact`) on resolved exchanges.
The protocol already reserves the slot: `TaskColumnStats.sketches` is a
`repeated ColumnSketch` with `SketchType::SKETCH_TYPE_HLL`, currently always
empty.
1. **Executor**: in the three shuffle writers (passthrough, range and sort),
update an HLL per column alongside the null counts (`ColumnNullCounts` in
`shuffle_writer.rs`, `column_null_counts` in `sort_shuffle/writer.rs`), and
serialize its registers into `ColumnSketch { type: HLL, data }` in
`DefaultQueryStageExec::collect_column_stats`.
2. **Scheduler**: in `RunningStage::output_column_stats`, merge each
column's sketches across successful tasks (HLL registers merge by element-wise
max) and set `distinct_count` to `Inexact(count)`. Leave it `Absent` if any
successful task didn't report a sketch for the column, as is done for
`null_count`.
3. Nothing else is needed downstream: #2664 already passes
`output_column_stats` through `resolve_stage_partitions` to
`ExchangeExec::partition_statistics`.
Open questions:
- **Which columns.** A precision-14 HLL is 16 KB per column per task, and
task completion is a hot message (see the TODO on `SuccessfulTask` in
`ballista.proto`). Sketching only the columns that can be join or grouping keys
downstream (the shuffle's hash-partitioning expressions at least) keeps that
bounded. Lower precision (e.g. p=12, 4 KB, ~1.6% error) is another lever.
- **Implementation.** DataFusion's `HyperLogLog`
(`datafusion-functions-aggregate/src/hyperloglog.rs`) is `pub(crate)`, so
Ballista either needs a small local implementation (hash with
`datafusion_common::hash_utils::HLL_RANDOM_STATE` for consistency with
`approx_distinct`), an upstream PR to make it public, or a crate.
- **Cost.** Hashing every value of the sketched columns on the write path.
It should be measured on TPC-H SF1000 before being on by default; a config flag
to disable it is probably worth having.
**Describe alternatives you've considered**
- Exact distinct counts: unbounded memory on the write path.
- Per-partition sketches on `PartitionStats.column_stats` (the proto has
`distinct_count` there too). For a hash-partitioned key, per-partition NDVs
would sum exactly, but this multiplies the payload by the partition count.
Stage-level sketches are enough for join estimation.
**Additional context**
Part of #2662. Builds on #2664 / #1960 and the original collection work in
#1916 (#1827).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
