Dandandan opened a new pull request, #2315: URL: https://github.com/apache/datafusion-ballista/pull/2315
## Summary Fixes five defects in adaptive query planning (AQE) and enables it by default. Before this change the TPC-DS suite hung at q23 with AQE on, and two queries returned wrong data. It now verifies clean against single-process DataFusion across all 99 queries. ## Bugs fixed **`InterleaveExec` invariant broken by per-branch rewrites** (fixes #2047). `InterleaveExec` requires all children to share one `Hash`/`Range` partitioning and asserts on rebuild. AQE rewrites union branches independently, and `TreeNode` rebuilds a parent as soon as a child changes, so the assertion fired before any rule could inspect the parent and aborted the whole replan. A new `NormalizeInterleaveRule` rewrites interleaves to `UnionExec` at the head of each replan; `EnforceDistribution` re-forms them later through its `can_interleave`-guarded constructor. **Stale stage completions hung the job.** A replan can drop a stage that still has tasks in flight. Their completions returned an error that the task-status handler only logs, so the stage never advanced and the query blocked forever. Late completions for superseded stages are now ignored. **Broadcast-unsafe `CollectLeft` joins produced duplicate rows.** DataFusion's `join_selection` promotes a small build side to `CollectLeft` without restricting by join type. That is correct in one process, but Ballista runs one task per probe partition, each with a full copy of the build side, so a `Left` join emitted its unmatched rows once per task. TPC-DS q77 reported a 16x inflated `sales` column as a result. The static planner already guards this in `maybe_promote_to_broadcast`; `DemoteUnsafeBroadcastJoinRule` applies the same guard to AQE, between `join_selection` and `EnsureRequirements`. **`ExchangeExec` leaked per-partition constants across a shuffle.** It inherited its input's equivalence properties, including constants that only hold within a partition. A `UNION ALL` branch projecting a literal marks that column constant, and keeping it across a hash repartition on the same column let `EnforceSorting` drop it from a sort whose `SortPreservingMergeExec` still merged on it, so q76 returned the wrong rows. `RepartitionExec` clears these for the same reason. **`LateCollectLeft` broadcast after shuffling.** It paid for the shuffle and the broadcast, and re-read the build side into every probe task. It is now restricted to null-aware anti joins, which require single-task `CollectLeft`. ## Verification `benchmarks/src/bin/tpcds.rs` previously skipped q31 and q71 because `ORDER BY` leaves the order of rows with equal sort keys unspecified, so a positional diff is unstable. The comparison now falls back to a canonical row-sorted check only after the positional one fails, so a wrong value still fails but a pure permutation passes. The skip list is empty and all 99 queries are verified. TPC-DS SF1, AQE on, each query diffed against single-process DataFusion: | | before | after | |---|---|---| | queries completing | hangs at q23 | 99 | | verified correct | – | 99 (97 exact, 2 order-only) | | skipped | 2 | 0 | ## Performance TPC-H SF10, one executor, 10 vcores, 16 GB pool, `--partitions 10`, best of 3. | configuration | total | |---|---| | static planner (previous default) | 62.2 s | | AQE (new default) | 29.6 s | 52% faster overall. Largest gains are q8 -91%, q9 -87%, q11 -86%, q2 -80%, q17 -73%. The static planner path is unchanged (+1.6%, within noise), and AQE performance is unchanged by the correctness fixes (-0.2%). One regression: q18 is 65% slower under AQE (4.8 s to 8.0 s). Its build side measures 83 MiB per partition against the 64 MiB `ballista.optimizer.hash_join_max_build_partition_bytes` default, so AQE falls back to a sort-merge join. Raising the threshold to 128 MiB brings it to 5.7 s. Tracked separately; the default is not changed here. ## Notes Physical-plan submissions now fall back to the static planner rather than erroring, since AQE plans from the logical plan and that path would otherwise break on the default configuration. `SessionConfigExt::with_ballista_adaptive_query_planner` selects the static planner in code, and scheduler tests that assert the static planner's stage layout pin it explicitly. 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
