[Epic][part 2] Adaptive Query Execution, runtime plan optimization

Follow-up to #377.

#377 laid the foundation for Adaptive Query Execution: stages are now planned dynamically, and runtime statistics gathered by the SamplerExec are used to make decisions like the number of tasks assigned to each stage (see #432).

Now that accurate runtime statistics flow through the coordinator as stages execute, the next step is to re-optimize the not-yet-executed stages on the fly based on those stats, rather than only sizing tasks. For example:

• Swapping join orders based on observed cardinalities.
• Switching join implementations (e.g. broadcast vs shuffle) once the build-side size is known (relates to the broadcast_joins follow-up).
• Coalescing/splitting partitions to handle data skew discovered at runtime.
• Other physical optimizations that become possible once real stats replace planning-time estimates.

Early exploration lives on the gabrielmusat/aqe branch.

Prior art

This kind of execution-time replanning is well established in other distributed engines. Their designs are a good reference for what it unlocks and how it's delivered:

Apache Spark — Adaptive Query Execution (AQE) (enabled by default since 3.2). At each shuffle stage boundary, Spark re-runs the optimizer with statistics from completed stages. Three headline optimizations:

• coalescing post-shuffle partitions,
• converting sort-merge joins to broadcast hash joins once a side is known to be small,
• skew-join handling (splitting/replicating skewed partitions into evenly sized tasks).
  • Official docs: https://spark.apache.org/docs/latest/sql-performance-tuning.html#adaptive-query-execution
  • Databricks deep-dive: https://www.databricks.com/blog/2020/05/29/adaptive-query-execution-speeding-up-spark-sql-at-runtime.html

Trino — Adaptive plan optimizations (since Trino 457, requires fault-tolerant execution). Because intermediate exchange data is spooled, Trino can reorder partitioned joins based on the actual build/probe sizes observed mid-query, rather than pre-computed connector statistics.

• Adaptive plan optimizations: https://trino.io/docs/current/optimizer/adaptive-plan-optimizations.html
• Fault-tolerant execution (the architecture that enables it): https://trino.io/docs/current/admin/fault-tolerant-execution.html

A recurring theme across both: a materialization/exchange boundary is what makes safe replanning possible, and join strategy/order + partition sizing are the highest-value decisions to defer to runtime — which maps directly onto the stage boundaries this project already has.

This work will be broken down into independently contributable sub-issues:

• https://github.com/datafusion-contrib/datafusion-distributed/issues/641

Start by reviewing the foundation in #377, the runtime statistics flow through SamplerExec and the coordinator, and the exploration on the gabrielmusat/aqe branch. Use the Spark and Trino references to narrow this epic into a concrete sub-issue, then check #641 for the planned breakdown. Done should be defined by an independently contributable runtime re-optimization task with explicit behavior and validation.