andygrove commented on code in PR #2084:
URL:
https://github.com/apache/datafusion-ballista/pull/2084#discussion_r3622221300
##########
ballista/core/src/execution_plans/sort_shuffle/writer.rs:
##########
@@ -567,7 +568,7 @@ impl SortShuffleWriterExec {
write_time,
);
} else {
- debug!(
+ info!(
Review Comment:
Fixed in b563844b — demoted to `debug!`. It fires once per shuffle
partition; the spill case stays at `warn!`.
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -315,7 +309,36 @@ impl DynamicJoinSelectionExec {
// this method calculates partition mode, and at the moment it
// can't calculate it as PartitionMode::Auto
(_, PartitionMode::Auto) => internal_err!("this case should not be
possible"),
- }
+ }?;
+
+ let action_label = match &action {
+ JoinSelectionAction::Repartition(_) => "Repartition",
+ JoinSelectionAction::CollectLeft(_) => "CollectLeft(broadcast)",
+ JoinSelectionAction::LateCollectLeft(_) =>
"LateCollectLeft(broadcast)",
+ JoinSelectionAction::Hash(_) => "Hash(Partitioned)",
+ JoinSelectionAction::Sort(_) => "SortMerge(Partitioned)",
+ };
+
+ info!(
Review Comment:
Applied. b563844b also switches the `use log::info` import to `use
log::debug` so the file still compiles after the change.
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -365,12 +388,29 @@ impl DynamicJoinSelectionExec {
};
if let Some(byte_size) = stats.total_byte_size.get_value() {
- *byte_size != 0 && *byte_size < threshold_byte_size
- } else if let Some(num_rows) = stats.num_rows.get_value() {
- *num_rows != 0 && *num_rows < threshold_num_rows
- } else {
- false
+ return *byte_size != 0 && *byte_size < threshold_byte_size;
}
+
+ // `total_byte_size` is unknown, which is the common case rather than
the
+ // exception: DataFusion discards it on every join, and rebuilding it
in
+ // `Statistics::calculate_total_byte_size` only works when every
column has
+ // a fixed width, so one `Utf8` column loses it for good.
+ //
+ // A row count on its own says nothing about how much data a broadcast
+ // would replicate to every probe task, so estimate the size and hold
it
+ // to the same byte threshold. The row threshold is kept as an
additional
+ // ceiling, so this can only ever reject a broadcast the row rule would
+ // have allowed, never introduce a new one.
+ let Some(num_rows) = stats.num_rows.get_value().copied() else {
+ return false;
+ };
+
+ if num_rows == 0 || num_rows >= threshold_num_rows {
+ return false;
+ }
+
+ estimate_output_byte_size(&plan.schema(), num_rows,
&stats.column_statistics)
Review Comment:
Confirmed — the per-side eligibility is lost once `under_threshold`
collapses to an OR of both sides. Tracking as a follow-up in #2121; it would
need a re-benchmark so keeping it out of this PR.
##########
ballista/scheduler/src/state/aqe/execution_plan/dynamic_join.rs:
##########
@@ -228,6 +229,8 @@ impl DynamicJoinSelectionExec {
.unwrap_or_default();
let threshold_collect_left_join_bytes =
bc.broadcast_join_threshold_bytes();
let threshold_collect_left_join_rows =
bc.broadcast_join_threshold_rows();
+ let max_build_bytes = bc.hash_join_max_build_partition_bytes();
+ let build_max_partition_bytes =
max_per_partition_build_bytes(&self.left);
Review Comment:
Confirmed — the check measures the pre-swap, pre-coalesce `self.left`, which
can differ from the partition the build task consumes. Tracking in #2121 to
apply the budget to the post-swap, post-coalesce build input.
--
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]
