andygrove commented on code in PR #2434:
URL:
https://github.com/apache/datafusion-ballista/pull/2434#discussion_r3996846235
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -601,6 +657,107 @@ impl DynamicJoinSelectionExec {
}
}
+/// The configured limits that decide whether staging a join's build side on
its
+/// own is worth an extra round trip. See [`requires_build_staging`].
+#[derive(Debug, Clone, Copy)]
+struct BuildStagingLimits {
+ /// `ballista.optimizer.broadcast_join_threshold_bytes`: the budget a build
+ /// side must come in under to be broadcast. `0` disables broadcast
+ /// promotion, and with it any reason to measure the build side again.
+ broadcast_threshold_bytes: usize,
+ /// `ballista.optimizer.stage_build_side_min_probe_ratio`: how many times
+ /// larger the probe side must be before staging pays off.
+ ///
+ /// 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.
+ min_probe_ratio: usize,
+ /// `ballista.optimizer.stage_build_side_max_estimate_multiple`: 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.
+ max_estimate_multiple: usize,
+}
+
+impl BuildStagingLimits {
+ fn new(bc: &BallistaConfig, broadcast_threshold_bytes: usize) -> Self {
Review Comment:
Taken. `BuildStagingLimits` is gone: `requires_build_staging` now takes
`min_probe_ratio` and `max_estimate_multiple` as plain `usize` params rather
than reaching through `&BallistaConfig`, and `broadcast_threshold_bytes` is
passed the same way — so the mix of "two read off BC, one passed by hand" is
gone with it.
This also lines up with comphead's separate point that the three sibling
predicates in this same file (`supports_collect_by_thresholds`,
`hash_build_fits`, `broadcast_build_side_pays_off`) all take plain params, so
the bundle was the odd one out.
##########
ballista/core/src/config.rs:
##########
@@ -294,6 +319,31 @@ static CONFIG_ENTRIES: LazyLock<HashMap<String,
ConfigEntry>> = LazyLock::new(||
which makes AQE use a hash join regardless of build
size.".to_string(),
DataType::UInt64,
Some((64 * 1024 * 1024).to_string())),
+ ConfigEntry::new(BALLISTA_STAGE_BUILD_SIDE.to_string(),
+ "Stages a join's prospective build side as its own
shuffle before \
+ choosing the join strategy, when that side's size is
only an \
+ estimate and the probe side is far larger. Lets AQE
measure the \
Review Comment:
Taken. All three `stage_build_side*` entries now say it explicitly — staging
applies only where the build side's size is an estimate rather than a
measurement, because an exact size over the broadcast budget is a fact that no
further measurement can change.
The code says the same thing in one place now too: `requires_build_staging`
returns early unless `total_byte_size` is `Precision::Inexact`.
Config docs regenerated via `./dev/update_config_docs.sh`.
--
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]
