andygrove commented on code in PR #2434:
URL:
https://github.com/apache/datafusion-ballista/pull/2434#discussion_r3993978902
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -212,6 +212,15 @@ impl DisplayAs for DynamicJoinSelectionExec {
}
pub enum JoinSelectionAction {
+ /// Shuffle only the prospective build side, leaving the probe side
+ /// untouched, and re-decide the join once the build side's *measured* size
+ /// is known. See [`should_stage_build_side`].
+ StageBuildSide {
+ join: Arc<DynamicJoinSelectionExec>,
+ /// Which child to give an exchange: `true` for `left`, `false` for
+ /// `right`.
+ build_is_left: bool,
Review Comment:
Agreed the bool was opaque. I went with DataFusion's existing
`datafusion::common::JoinSide` rather than defining a new `BuildSide` — it is
already the generic Left/Right/None enum for exactly this, and it is already
imported elsewhere in this crate (`physical_optimizer/join_selection.rs`), so
it is one less type to keep in sync.
The variant is now `build_side: JoinSide`, and `SelectJoinRule` matches on
it instead of indexing children with a bool. `JoinSide::None` is rejected with
an internal error, since the resolver only ever produces Left or Right.
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -606,6 +664,82 @@ impl DynamicJoinSelectionExec {
/// join always uses hash join); an unknown build size
(`build_max_partition_bytes
/// == None`) is treated as fitting too, since there is no evidence to force a
/// fallback.
+/// How much larger the probe side must be before staging the build side alone
+/// is worth an extra round trip.
+///
+/// Staging serialises the two shuffles that `Repartition` would otherwise run
+/// concurrently, so the deferral costs at most the *smaller* side's runtime.
+/// Requiring an order of magnitude keeps that cost well under the probe-side
+/// shuffle it stands to avoid entirely.
+const STAGE_BUILD_SIDE_MIN_PROBE_RATIO: usize = 10;
+
+/// How far over the broadcast budget an *estimated* build side may sit and
+/// still be worth measuring.
+///
+/// TPC-H q8's filtered `part` scan estimates roughly 10x over budget, because
+/// the planner falls back to `default_filter_selectivity` for
+/// `p_type = '...'`, and measures three orders of magnitude under it. A raw
+/// fact-table scan sits far beyond this multiple and is shuffled without the
+/// extra round trip.
+const STAGE_BUILD_SIDE_MAX_ESTIMATE_MULTIPLE: usize = 32;
+
+/// Whether to shuffle only the prospective build side now and re-decide the
+/// join once its measured size is known, rather than shuffling both sides.
+///
+/// [`JoinSelectionAction::Repartition`] shuffles both inputs at once, which
for
+/// a fact-table probe side commits to the single most expensive stage in the
+/// query *before* any measurement exists. When the build side's size is only a
+/// guess, that guess is the sole reason the join is not a broadcast, and the
+/// probe side dwarfs it, one cheap stage buys an exact number and often flips
+/// the join to `CollectLeft`, leaving the probe side never shuffled at all.
+///
+/// Each of the three conditions is necessary:
+///
+/// * the build estimate must be **inexact**. An exact size over budget is a
+/// fact rather than a guess, so measuring it again cannot change the outcome
+/// and would only serialise two shuffles that could run concurrently.
+/// * the estimate must be **plausibly wrong enough to flip**. Past
+/// [`STAGE_BUILD_SIDE_MAX_ESTIMATE_MULTIPLE`] the side is large on any
+/// reading, and no measurement brings it under budget.
+/// * the probe side must be **much larger**, per
+/// [`STAGE_BUILD_SIDE_MIN_PROBE_RATIO`], which is what bounds the cost of
+/// being wrong.
+///
+/// Both sides are compared in bytes. A missing probe-side size declines: with
+/// nothing to compare against there is no evidence the round trip pays off.
+fn should_stage_build_side(
Review Comment:
Renamed to `requires_build_staging`.
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -606,6 +664,82 @@ impl DynamicJoinSelectionExec {
/// join always uses hash join); an unknown build size
(`build_max_partition_bytes
/// == None`) is treated as fitting too, since there is no evidence to force a
/// fallback.
+/// How much larger the probe side must be before staging the build side alone
+/// is worth an extra round trip.
+///
+/// Staging serialises the two shuffles that `Repartition` would otherwise run
+/// concurrently, so the deferral costs at most the *smaller* side's runtime.
+/// Requiring an order of magnitude keeps that cost well under the probe-side
+/// shuffle it stands to avoid entirely.
+const STAGE_BUILD_SIDE_MIN_PROBE_RATIO: usize = 10;
+
+/// How far over the broadcast budget an *estimated* build side may sit and
+/// still be worth measuring.
+///
+/// TPC-H q8's filtered `part` scan estimates roughly 10x over budget, because
+/// the planner falls back to `default_filter_selectivity` for
+/// `p_type = '...'`, and measures three orders of magnitude under it. A raw
+/// fact-table scan sits far beyond this multiple and is shuffled without the
+/// extra round trip.
+const STAGE_BUILD_SIDE_MAX_ESTIMATE_MULTIPLE: usize = 32;
Review Comment:
Good call — both are config keys now, so they are tunable without a rebuild:
- `ballista.optimizer.stage_build_side_min_probe_ratio` (UInt64, default
`10`)
- `ballista.optimizer.stage_build_side_max_estimate_multiple` (UInt64,
default `32`)
They are read into a new `BuildStagingLimits` struct alongside
`broadcast_join_threshold_bytes`, which is also where the reasoning for each
figure now lives. Both are only consulted when `stage_build_side` is enabled.
Config docs regenerated via `./dev/update_config_docs.sh`, and
`stage_build_side_ships_the_documented_defaults` pins the shipped values, since
those are what the SF1000 numbers were measured under.
--
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]
