Follow-up from review of #2084 (both spotted by @phillipleblanc).
Two related imprecisions in DynamicJoinSelectionExec::to_actual_join
(ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs): the byte-size
checks are computed against inputs that may not match the build side the executor
actually runs.
1. Broadcast eligibility is OR-ed across both sides, but the build side can be chosen by row count
under_threshold is
supports_collect_by_thresholds(left) || supports_collect_by_thresholds(right), so the
join is promoted to CollectLeft if either side passes the byte threshold. Which side
is actually broadcast is then decided by the swap logic, which can fall back to row count.
If only the right side passes the byte estimate while the wider-by-bytes left side has
fewer rows, AQE can still broadcast the left — replicating more bytes to every probe task
than the byte threshold intended.
Direction: preserve per-side byte eligibility and consult it when selecting the build
input, rather than collapsing to a single OR-ed bool.
2. The hash-join build-fit check measures the pre-swap, pre-coalesce input
max_per_partition_build_bytes(&self.left) measures self.left's shuffle partition
sizes, but the eventual build task runs after (a) SelectJoinRule may swap the inputs and
(b) AQE partition coalescing groups partitions. The check can therefore validate a
different or smaller partition than the build actually consumes: a coalesced build
partition can exceed the hash_join_max_build_partition_bytes budget the check passed
(still OOM), or a swap can make self.right the real build side.
Direction: apply the budget to the post-swap, post-coalesce build input.
Both were raised as non-blocking suggestions on the approved PR #2084; filing here to
track.
Follow-up from review of #2084 (both spotted by @phillipleblanc).
Two related imprecisions in
DynamicJoinSelectionExec::to_actual_join(
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs): the byte-sizechecks are computed against inputs that may not match the build side the executor
actually runs.
1. Broadcast eligibility is OR-ed across both sides, but the build side can be chosen by row count
under_thresholdissupports_collect_by_thresholds(left) || supports_collect_by_thresholds(right), so thejoin is promoted to
CollectLeftif either side passes the byte threshold. Which sideis actually broadcast is then decided by the swap logic, which can fall back to row count.
If only the right side passes the byte estimate while the wider-by-bytes left side has
fewer rows, AQE can still broadcast the left — replicating more bytes to every probe task
than the byte threshold intended.
Direction: preserve per-side byte eligibility and consult it when selecting the build
input, rather than collapsing to a single OR-ed bool.
2. The hash-join build-fit check measures the pre-swap, pre-coalesce input
max_per_partition_build_bytes(&self.left)measuresself.left's shuffle partitionsizes, but the eventual build task runs after (a)
SelectJoinRulemay swap the inputs and(b) AQE partition coalescing groups partitions. The check can therefore validate a
different or smaller partition than the build actually consumes: a coalesced build
partition can exceed the
hash_join_max_build_partition_bytesbudget the check passed(still OOM), or a swap can make
self.rightthe real build side.Direction: apply the budget to the post-swap, post-coalesce build input.
Both were raised as non-blocking suggestions on the approved PR #2084; filing here to
track.