andygrove opened a new issue, #2081:
URL: https://github.com/apache/datafusion-ballista/issues/2081
**Describe the bug**
The adaptive (AQE) planner decides whether to broadcast a join's build side
in `DynamicJoinSelectionExec::supports_collect_by_thresholds`:
```rust
let Ok(stats) = plan.partition_statistics(None) else { return false; };
if let Some(byte_size) = stats.total_byte_size.get_value() {
*byte_size != 0 && *byte_size < threshold_byte_size // 10 MB
} else if let Some(num_rows) = stats.num_rows.get_value() {
*num_rows != 0 && *num_rows < threshold_num_rows // 1,000,000
} else {
false
}
```
When `total_byte_size` is known this is a size check, which is what we want.
But when it is `Absent` the decision falls back to a **row count** compared
against `hash_join_single_partition_threshold_rows` (default 1,000,000). A row
count says nothing about size, so a build side of up to 1M arbitrarily wide
rows can be collected and replicated to every probe task while the 10 MB guard
never gets a chance to apply.
This is not a rare corner. `total_byte_size` is `Absent` for exactly the
intermediates whose size we cannot guess:
1. **Join outputs, always.** DataFusion's `estimate_join_statistics`
hardcodes `total_byte_size: Precision::Absent` and keeps only an estimated row
count, so every join result has unknown size.
2. **Anything with a variable-width column.**
`Statistics::calculate_total_byte_size` sums `DataType::primitive_width()`
across the schema; that returns `None` for
`Utf8`/`LargeUtf8`/`Utf8View`/`Binary` (and `Boolean`), so a single string
column makes it bail out to `total_byte_size.to_inexact()`. `to_inexact()` on
`Absent` is still `Absent`, so once the size is lost a projection cannot
recover it.
In TPC-H that covers most dimension-side join results, since they carry
names, addresses, or comments.
**To Reproduce**
TPC-H SF1000, 2 executors x (8 cores / 56 GiB), `target_partitions=32`,
`prefer_hash_join=false`, adaptive planner on, with
`RUST_LOG=info,ballista_scheduler::state::aqe=debug`. Q2 alone produces:
```
to_actual_join - plan_id: 1, decision: CollectLeft left: (Inexact(799140)
| Absent), right: (Exact(10000000) | Absent)
to_actual_join - plan_id: 7, decision: CollectLeft left: (Inexact(639312)
| Absent), right: (Inexact(159969600) | Inexact(3839270400))
to_actual_join - plan_id: 5, decision: CollectLeft left: (Inexact(10000000)
| Inexact(320000000)), right: (Exact(25) | Exact(400))
to_actual_join - plan_id: 2, decision: CollectLeft left: (Inexact(3196560)
| Absent), right: (Exact(25) | Absent)
```
`plan_id: 1` and `plan_id: 7` are broadcast on the row rule alone (799,140
and 639,312 rows, size unknown). These are `part`/`supplier`-derived rows
carrying string columns, so the real payload replicated to all 32 tasks is
plausibly hundreds of MB rather than the 10 MB the byte threshold is meant to
cap.
The same run also shows how easily the size is lost. Two decisions over the
same 25-row `nation` table:
- `plan_id: 5`, right side `Exact(25) | Exact(400)` — an all-numeric
projection, so bytes are recomputed as rows x width.
- `plan_id: 2`, right side `Exact(25) | Absent` — the schema carries a
string column, so the recompute gives up.
**Expected behavior**
The broadcast decision should be driven by an estimate of the build side's
**size**. When `total_byte_size` is unavailable, Ballista should estimate it
(rows x estimated row width, from the schema's fixed-width columns plus
per-column statistics or a configurable default for variable-width columns) and
test that against the byte threshold, rather than treating "fewer than N rows"
as a proxy for "small".
For reference, Spark drives broadcast selection from
`spark.sql.autoBroadcastJoinThreshold` in bytes, and estimates a per-row size
from the schema (`EstimationUtils.getSizePerRow`) when it must; it has no
row-count threshold standing in for size.
**Additional context**
Found while benchmarking TPC-H SF1000 under both planners (2x8 cores,
node-local data, 1 iteration). AQE-on and AQE-off totalled 6320.7 s and 6378.2
s respectively — a wash overall, but the per-query results move in both
directions, and the queries where AQE loses most (Q5 700.5 vs 542.7, Q8 801.2
vs 657.1, Q9 1025.2 vs 891.8, Q18 820.9 vs 620.7) are join-heavy ones where
this fallback is most likely to be making the call. Whether over-eager
broadcasts explain those regressions is not yet established — this issue is
about the decision rule being size-blind, which is true regardless.
There are related gaps that are worth tracking separately and are not
proposed here:
- `stats_for_partitions`
(`ballista/core/src/execution_plans/shuffle_reader.rs`) reports shuffle
`num_rows`/`total_byte_size` as `Exact` from measured output, but sets
`column_statistics` to all-unknown (there is a `TODO stats: add column
statistics to PartitionStats`), and the wire `ColumnStats` message has no
`byte_size` field. So Ballista measures real sizes at every shuffle boundary
and then cannot carry them into downstream planning.
- The two DataFusion behaviours above (joins discarding `total_byte_size`;
`calculate_total_byte_size` ignoring the existing `ColumnStatistics::byte_size`
field for variable-width columns) would be better fixed upstream.
--
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]
