[EPIC] Enable Adaptive Query Execution (AQE) by default

[andygrove]

Goal

Adaptive Query Execution (AQE) is currently EXPERIMENTAL and disabled by default. This epic tracks the correctness, fault-tolerance, performance, and feature work required to make the adaptive planner the default execution path and eventually remove the experimental warning.

The design and initial implementation of the adaptive planner are tracked in the design epic #1359. This epic is the "graduation" checklist: the remaining work to flip the default from off to on.

Current state

AQE is gated behind these configuration flags, all off by default:

• ballista.planner.adaptive.enabled — master switch (AdaptiveExecutionGraph / AdaptivePlanner)
• ballista.planner.coalesce.enabled — coalesce shuffle partitions rule
• ballista.planner.propagate_empty.enabled — propagate-empty rule
• ballista.planner.adaptive_join.enabled — dynamic join selection (SelectJoinRule / DynamicJoinSelectionExec)

The scheduler emits "Adaptive Query Planning is EXPERIMENTAL, should be used for testing purposes only!" when the master switch is on.

Exit criteria

Before flipping the default we want:

1. No AQE-specific correctness failures on the TPC-H and TPC-DS suites at scale.
2. AQE on is at least as fast as AQE off across the benchmark suites (no regressions).
3. The adaptive path has the same fault-tolerance guarantees as the static path (executor loss, shuffle-fetch retry, stage/task retry).
4. The adaptive feature set (coalesce, propagate-empty, dynamic join selection, runtime join/aggregate switching) is complete and enabled.
5. Operational parity with the static planner (intermediate-shuffle reclaim, stats plumbing).

Blocking issues

Correctness (AQE-specific)

• Adaptive (AQE) execution hangs when a replan cancels an in-flight stage #2063 — Adaptive execution hangs when a replan cancels an in-flight stage (stage cancellation unimplemented)
• TPCDS Q72 Fails with OffsetOverflowError(2147731589) (AQE) #1826 — TPC-DS Q72 fails with OffsetOverflowError (AQE)

Fault tolerance / high availability

• AQE: handle executor failure on the adaptive execution path #1986 — Handle executor failure on the adaptive execution path
• Job hangs indefinitely instead of failing when all executors are lost #2029 — Job hangs indefinitely instead of failing when all executors are lost
• Shuffle-fetch failures lose their type, so the map-stage resubmit never fires and the job fails #2027 — Shuffle-fetch failures lose their type, so the map-stage resubmit never fires
• Retryable IO errors are misclassified as non-retryable when wrapped in DataFusionError::Shared #2028 — Retryable IO errors misclassified as non-retryable when wrapped in DataFusionError::Shared
• Add an end-to-end HA chaos harness: the scheduler's fault-tolerance paths are never tested on a real cluster #2030 — Add an end-to-end HA chaos harness to exercise the fault-tolerance paths

Planning quality / performance regressions

• AQE regresses join-heavy TPC-H queries at SF1000 instead of promoting broadcasts #2024 — AQE regresses join-heavy TPC-H queries at SF1000 instead of promoting broadcasts
• AQE broadcast decision falls back to a size-blind row threshold when total_byte_size is unknown #2081 — Broadcast decision falls back to a size-blind row threshold when total_byte_size is unknown
• Hash-join build side OOMs on TPC-H Q18 at SF1000 and does not shrink with target_partitions #2025 — Hash-join build side OOMs on TPC-H Q18 at SF1000 and does not shrink with target_partitions

Statistics plumbing

• Surface collected column statistics to planners via ShuffleReaderExec #1960 — Surface collected column statistics to planners via ShuffleReaderExec

Feature completeness

• AQE: switch from hash join to sort-merge join based on runtime statistics #1988 — Switch hash join to sort-merge join based on runtime statistics
• AQE: switch from streaming aggregation to hash aggregation based on runtime statistics #1989 — Switch streaming aggregation to hash aggregation based on runtime statistics
• AQE coalesce rule bails the entire alignment group when any leaf is a broadcast exchange #2166 — Coalesce rule bails the entire alignment group when any leaf is a broadcast exchange
• AQE coalesce rule bails on heterogeneous leaf partition counts (per-M subgrouping TODO / Q22 panic guard) #2167 — Coalesce rule bails on heterogeneous leaf partition counts (per-M subgrouping / Q22 panic guard)

Operational

• AQE: immediate intermediate-shuffle reclaim on job success is a no-op under the adaptive planner #1996 — Immediate intermediate-shuffle reclaim on job success is a no-op under the adaptive planner

Already resolved

For context, these AQE blockers are already fixed:

• Adaptive (AQE) execution panics with EmptyExec invalid partition on many TPC-DS queries #2047 — Adaptive execution panics with EmptyExec invalid partition on many TPC-DS queries
• ballista.optimizer.broadcast_join_threshold_bytes has no effect when adaptive query planning is enabled #2085 — ballista.optimizer.broadcast_join_threshold_bytes has no effect when adaptive planning is enabled
• Adaptive query planning fails with remote object stores: "No suitable object store found" #1950 — Adaptive query planning fails with remote object stores
• Adaptive query planner does not support sort-based shuffle #1611 — Adaptive query planner does not support sort-based shuffle

Related

• [EPIC] Adaptive Query Execution (AQE) #1359 — [EPIC] Adaptive Query Execution (AQE) (design and implementation)