milenkovicm commented on code in PR #2434:
URL:
https://github.com/apache/datafusion-ballista/pull/2434#discussion_r3995998749
##########
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.
Review Comment:
is there a difference if statictics are exact or estimated ?
##########
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:
perhaps we should mention that we're working with estimated statistics only
##########
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:
nit: consider adding `max_probe_ratio` & `max_estimate_multiple`, instead of
`BC`, "law of demeter" principle. at the moment we have mix and match of
paratemters as `broadcast_threshold_bytes` comes from BC as well
##########
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 {
+ Self {
+ broadcast_threshold_bytes,
+ min_probe_ratio: bc.stage_build_side_min_probe_ratio(),
+ max_estimate_multiple: bc.stage_build_side_max_estimate_multiple(),
+ }
+ }
+}
+
+/// 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
Review Comment:
this answers my question from, reasoning make sense to me.
--
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]
