feat: broadcast small build sides via non-zero CollectLeft thresholds

ballista/client/tests/context_checks.rs

@@ -520,17 +520,14 @@ mod supported {
         Ok(())
     }
 
-    // As mentioned in https://github.com/apache/datafusion-ballista/issues/1055
-    // "Left/full outer join incorrect for CollectLeft / broadcast"
-    //
-    // In order to make correct results (decreasing performance) CollectLeft
-    // has been disabled until fixed
-
+    // Ballista raises these DataFusion thresholds above zero so the adaptive
+    // join resolver can collect a small build side into a broadcast
+    // (CollectLeft) hash join instead of repartitioning it.
     #[rstest]
     #[case::standalone(standalone_context())]
     #[case::remote(remote_context())]
     #[tokio::test]
-    async fn should_disable_collect_left(
+    async fn should_set_collect_left_thresholds(
         #[future(awt)]
         #[case]
         ctx: SessionContext,
@@ -542,12 +539,12 @@ mod supported {
             .await?;
 
         let expected = [
-            "+----------------------------------------------------------------+-------+",
-            "| name                                                           | value |",
-            "+----------------------------------------------------------------+-------+",
-            "| datafusion.optimizer.hash_join_single_partition_threshold      | 0     |",
-            "| datafusion.optimizer.hash_join_single_partition_threshold_rows | 0     |",
-            "+----------------------------------------------------------------+-------+",
+            "+----------------------------------------------------------------+----------+",
+            "| name                                                           | value    |",
+            "+----------------------------------------------------------------+----------+",
+            "| datafusion.optimizer.hash_join_single_partition_threshold      | 10485760 |",
+            "| datafusion.optimizer.hash_join_single_partition_threshold_rows | 1000000  |",
+            "+----------------------------------------------------------------+----------+",
         ];
 
         assert_batches_eq!(expected, &result);

ballista/core/src/extension.rs

@@ -767,18 +767,15 @@ impl SessionConfigHelperExt for SessionConfig {
             // same like previous comment
             .set_bool("datafusion.sql_parser.map_string_types_to_utf8view", false)
             //
-            // As mentioned in https://github.com/apache/datafusion-ballista/issues/1055
-            // "Left/full outer join incorrect for CollectLeft / broadcast"
-            //
-            // In order to make correct results (decreasing performance) CollectLeft
-            // has been disabled until fixed
+            // A build side smaller than these thresholds is collected into a
+            // CollectLeft (broadcast) hash join rather than being repartitioned.
             .set_u64(
                 "datafusion.optimizer.hash_join_single_partition_threshold",
-                0,
+                10 * 1024 * 1024,
             )
             .set_u64(
                 "datafusion.optimizer.hash_join_single_partition_threshold_rows",
-                0,
+                1_000_000,
             )
             //
             // DataFusion's hash join has no spill support, so each parallel
@@ -805,6 +802,19 @@ impl SessionConfigHelperExt for SessionConfig {
                 "datafusion.optimizer.enable_physical_uncorrelated_scalar_subquery",
                 false,
             )
+            //
+            // DataFusion's dynamic filters are populated at runtime by an
+            // upstream operator (a hash join build side, a TopK heap, a partial
+            // aggregate) and read by a downstream scan within the same plan.
+            // Ballista splits a plan into stages at shuffle and broadcast
+            // boundaries that run as independent tasks, so when the producing
+            // operator and the consuming scan land in different stages the
+            // filter is never populated across the boundary and the scan blocks
+            // forever. Disable dynamic filter pushdown until Ballista can carry
+            // dynamic filters across stage boundaries.
+            //
+            // See https://github.com/apache/datafusion-ballista/issues/1375
+            .set_bool("datafusion.optimizer.enable_dynamic_filter_pushdown", false)
     }
 }
 
@@ -1046,7 +1056,9 @@ mod test {
     // re-applying the defaults would discard them. See #1901.
     #[test]
     fn should_preserve_user_overrides_on_upgrade() {
-        // Ballista defaults these to prefer_hash_join=false and threshold=0.
+        // Ballista defaults these to prefer_hash_join=false and the threshold to
+        // 10 MB. The overrides below differ from those defaults so the assertions
+        // prove the user's values survived `upgrade_for_ballista`.
         let mut config = SessionConfig::new_with_ballista();
         config
             .options_mut()
@@ -1056,7 +1068,7 @@ mod test {
             .options_mut()
             .set(
                 "datafusion.optimizer.hash_join_single_partition_threshold",
-                "10485760",
+                "5242880",
             )
             .unwrap();
 
@@ -1068,7 +1080,7 @@ mod test {
                 .options()
                 .optimizer
                 .hash_join_single_partition_threshold,
-            10485760
+            5242880
         );
     }
 
@@ -1084,7 +1096,14 @@ mod test {
                 .options()
                 .optimizer
                 .hash_join_single_partition_threshold,
-            0
+            10 * 1024 * 1024
+        );
+        assert_eq!(
+            config
+                .options()
+                .optimizer
+                .hash_join_single_partition_threshold_rows,
+            1_000_000
         );
     }
 

ballista/scheduler/src/physical_optimizer/join_selection.rs

@@ -111,6 +111,31 @@ fn supports_collect_by_thresholds(
     }
 }
 
+/// Whether a `CollectLeft` (broadcast) hash join is correct for `join_type` in
+/// Ballista's distributed execution.
+///
+/// `CollectLeft` replicates the build (left) side to every probe task, and each
+/// task processes only its own slice of the probe (right) side. Output rows that
+/// are determined by a single probe-side row are therefore produced exactly
+/// once. Join types that emit rows on behalf of the build side — unmatched outer
+/// rows, or semi/anti/mark rows that depend on a build row's global match status
+/// — would instead be emitted independently by every task, producing duplicate
+/// or spurious rows. Only join types whose output is driven entirely by the
+/// probe side are safe to broadcast.
+///
+/// `join_type` must be the join type *after* any build-side swap, since the
+/// build side is always the left input of the resulting `HashJoinExec`.
+pub(crate) fn collect_left_broadcast_safe(join_type: JoinType) -> bool {
+    matches!(
+        join_type,
+        JoinType::Inner
+            | JoinType::Right
+            | JoinType::RightSemi
+            | JoinType::RightAnti
+            | JoinType::RightMark
+    )
+}
+
 /// Predicate that checks whether the given join type supports input swapping.
 #[deprecated(since = "45.0.0", note = "use JoinType::supports_swap instead")]
 #[allow(dead_code)]
@@ -617,6 +642,7 @@ fn apply_subrules(
 mod test {
     use std::sync::Arc;
 
+    use super::collect_left_broadcast_safe;
     use datafusion::{
         arrow::datatypes::{DataType, Field, Schema},
         common::{ColumnStatistics, JoinSide, JoinType, Statistics, stats::Precision},
@@ -629,6 +655,41 @@ mod test {
             test::exec::StatisticsExec,
         },
     };
+
+    // A CollectLeft broadcast replicates the build (left) side to every probe
+    // task. Only join types whose output is driven entirely by the probe (right)
+    // side may be broadcast; any type that emits rows on behalf of the build side
+    // would duplicate them across tasks. This locks down the full set so a new or
+    // reclassified `JoinType` can't silently become broadcastable.
+    #[test]
+    fn collect_left_broadcast_safe_matches_probe_driven_join_types() {
+        let safe = [
+            JoinType::Inner,
+            JoinType::Right,
+            JoinType::RightSemi,
+            JoinType::RightAnti,
+            JoinType::RightMark,
+        ];
+        let unsafe_types = [
+            JoinType::Left,
+            JoinType::Full,
+            JoinType::LeftSemi,
+            JoinType::LeftAnti,
+            JoinType::LeftMark,
+        ];
+        for join_type in safe {
+            assert!(
+                collect_left_broadcast_safe(join_type),
+                "{join_type:?} should be safe to broadcast"
+            );
+        }
+        for join_type in unsafe_types {
+            assert!(
+                !collect_left_broadcast_safe(join_type),
+                "{join_type:?} must not be broadcast"
+            );
+        }
+    }
     //
     // join selection should not change order of joins for
     // nested loop join as we're not able to insert new CoalescePartitions