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).
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.
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.
Follow-up to #377.
#377 laid the foundation for Adaptive Query Execution: stages are now planned dynamically, and runtime statistics gathered by the
SamplerExecare 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:
broadcast_joinsfollow-up).Early exploration lives on the
gabrielmusat/aqebranch.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:
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.
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.