[I] AQE broadcast decision falls back to a size-blind row threshold when total_byte_size is unknown [datafusion-ballista]

["andygrove (via GitHub)" <gi...@apache.org>]

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: github-unsubscribe@datafusion.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [I] AQE broadcast decision falls back to a size-blind row threshold when total_byte_size is unknown [datafusion-ballista]

["andygrove (via GitHub)" <gi...@apache.org>]

andygrove closed issue #2081: AQE broadcast decision falls back to a size-blind row threshold when total_byte_size is unknown
URL: https://github.com/apache/datafusion-ballista/issues/2081


-- 
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org