andygrove opened a new issue, #2661:
URL: https://github.com/apache/datafusion-ballista/issues/2661
**Describe the bug**
AQE can promote a join to a broadcast (`CollectLeft`) based on an `Inexact`
row estimate that is off by orders of magnitude, and nothing re-checks the
build side once its stage has run and its real size is known. In the case below
a 152M-row build side was broadcast and the job failed with:
```
Job failed due to stage 3 failed: Task failed due to runtime execution error:
DataFusionError(Shared(ArrowError(OffsetOverflowError(2147543406),
Some(""))))
```
`CollectLeft` concatenates the build side into one batch, and its `s_name`
column (~18 bytes × 152M rows ≈ 2.7 GB) overflows `Utf8`'s i32 offsets. Below
the overflow point the same mistake would still replicate gigabytes to every
probe task.
The AQE debug log for the join (`ballista_scheduler::state::aqe=debug`):
```
AQE join decision plan_id=2 action=CollectLeft(broadcast)
partition_mode=CollectLeft build_side=left under_threshold=true
left=(rows=Inexact(399809), bytes=Absent) right=(rows=Inexact(300000000),
bytes=Absent)
byte_threshold=134217728 row_threshold=1000000
```
The left side is `(supplier ⋈ nation) ⋈ lineitem`. Its estimate of 399,809
rows is just the `supplier ⋈ nation` row count; it actually produced
151,688,609 rows.
Where the estimate comes from:
1. The resolved `ExchangeExec` reports exact `num_rows`/`total_byte_size`
but unknown column statistics (`stats_for_partition(s)` in `shuffle_reader.rs`:
`// TODO stats: add column statistics to PartitionStats`).
2. With no distinct counts, DataFusion's `estimate_inner_join_cardinality`
uses each side's row count as its NDV, so the estimate is `399,809 ×
3,793,363,900 / max(399,809, 3,793,363,900) = 399,809`. That treats the large
side as the unique-key side; for a PK–FK join where the large side is the FK
side, it underestimates by the fan-out (380× here).
3. `DynamicJoinSelectionExec::to_actual_join` takes the
`(JoinInputState::Unknown, CollectLeft)` path on that estimate (399,809 rows is
under the 1M row threshold, and the estimated bytes are under 128 MB).
4. When the broadcast stage completes with exact stats (151.7M rows), the
join is not reconsidered. `DemoteUnsafeBroadcastJoinRule` only demotes on join
type, not size.
**To Reproduce**
TPC-H SF1000, 2 executors, AQE enabled (default), TPC-H q21 with `nation`
moved second in the `FROM` list (the query is otherwise unchanged):
```sql
from
supplier,
nation,
lineitem l1,
orders
```
Stage 3 (`HashJoinExec: mode=CollectLeft ... ShuffleReaderExec:
upstream_stage: 2, broadcast: true`) fails with the error above on every run.
The original `FROM` order is not affected because its joins are all decided on
measured or large estimates.
Setting `ballista.optimizer.broadcast_join_threshold_rows` below the
estimate (e.g. 100000) avoids the broadcast.
**Expected behavior**
A join should not be broadcast when its build side turns out to exceed
`ballista.optimizer.broadcast_join_threshold_bytes`; at worst the query should
fall back to a partitioned join, not fail.
**Additional context**
Possible fixes, not mutually exclusive:
- Re-check the measured size of a broadcast build stage against the
threshold once it resolves, and fall back to a partitioned hash or sort-merge
join if it is over.
- Don't promote to `CollectLeft` on an `Inexact` estimate derived through a
join; stage the build side first (the existing `StageBuildSide` action) so the
decision is made on measured size.
- Carry column statistics (at least distinct counts for join keys) through
shuffle `PartitionStats`, so join cardinality estimates downstream of an
exchange aren't degenerate.
- As a backstop, refuse to concatenate a `CollectLeft` build side whose
variable-width columns exceed i32 offsets (or use large/view types).
--
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]
