phillipleblanc commented on code in PR #2084:
URL:
https://github.com/apache/datafusion-ballista/pull/2084#discussion_r3619307765
##########
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:
```suggestion
debug!(
```
##########
ballista/core/src/execution_plans/sort_shuffle/writer.rs:
##########
@@ -567,7 +568,7 @@ impl SortShuffleWriterExec {
write_time,
);
} else {
- debug!(
+ info!(
Review Comment:
Looks like this got flipped back to info
##########
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:
`self.left` is measured before SelectJoinRule may swap the hash inputs and
before AQE coalescing groups partitions. The eventual build task can therefore
consume a different or larger partition than this check validated. It might be
better to apply the budget to the post-swap, post-coalesce input.
##########
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:
This new byte estimate is not used when choosing the build side:
`to_actual_join` ORs both sides, then build-side selection can fall back to row
count. If only the right side passes this estimate but the wider left has fewer
rows, AQE still broadcasts the left. It might be better to preserve the
per-side eligibility when selecting the 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]
