Consolidate empty-build probe tests and join setup

Refactor exec.rs by consolidating empty-build probe-behavior tests
into `assert_empty_build_probe_behavior(...)` and repeated dynamic
filter join setup into `hash_join_with_dynamic_filter(...)`.
Maintain existing runtime logic while reducing duplicate test
boilerplate and redundant local setup for improved clarity
and maintainability.

Author: kosiew

datafusion/physical-plan/src/joins/hash_join/exec.rs

@@ -2275,8 +2275,10 @@ mod tests {
         )
     }
 
-    fn empty_build_with_probe_error_inputs() -> (Arc<dyn ExecutionPlan>, Arc<dyn ExecutionPlan>, JoinOn) {
-        let left_batch = build_table_i32(("a1", &vec![]), ("b1", &vec![]), ("c1", &vec![]));
+    fn empty_build_with_probe_error_inputs()
+    -> (Arc<dyn ExecutionPlan>, Arc<dyn ExecutionPlan>, JoinOn) {
+        let left_batch =
+            build_table_i32(("a1", &vec![]), ("b1", &vec![]), ("c1", &vec![]));
         let left_schema = left_batch.schema();
         let left: Arc<dyn ExecutionPlan> =
             TestMemoryExec::try_new_exec(&[vec![left_batch]], left_schema.clone(), None)
@@ -2297,6 +2299,69 @@ mod tests {
         (left, right, on)
     }
 
+    async fn assert_empty_build_probe_behavior(
+        join_types: &[JoinType],
+        expect_probe_error: bool,
+    ) {
+        let (left, right, on) = empty_build_with_probe_error_inputs();
+
+        for join_type in join_types {
+            let join = join(
+                Arc::clone(&left),
+                Arc::clone(&right),
+                on.clone(),
+                join_type,
+                NullEquality::NullEqualsNothing,
+            )
+            .unwrap();
+
+            let result = common::collect(
+                join.execute(0, Arc::new(TaskContext::default())).unwrap(),
+            )
+            .await;
+
+            if expect_probe_error {
+                let result_string = result.unwrap_err().to_string();
+                assert!(
+                    result_string.contains("bad data error"),
+                    "actual: {result_string}"
+                );
+            } else {
+                let batches = result.unwrap();
+                assert!(
+                    batches.is_empty(),
+                    "expected no output batches for {join_type}, got {batches:?}"
+                );
+            }
+        }
+    }
+
+    fn hash_join_with_dynamic_filter(
+        left: Arc<dyn ExecutionPlan>,
+        right: Arc<dyn ExecutionPlan>,
+        on: JoinOn,
+        join_type: JoinType,
+    ) -> Result<(HashJoinExec, Arc<DynamicFilterPhysicalExpr>)> {
+        let dynamic_filter = HashJoinExec::create_dynamic_filter(&on);
+        let mut join = HashJoinExec::try_new(
+            left,
+            right,
+            on,
+            None,
+            &join_type,
+            None,
+            PartitionMode::CollectLeft,
+            NullEquality::NullEqualsNothing,
+            false,
+        )?;
+        join.dynamic_filter = Some(HashJoinExecDynamicFilter {
+            filter: Arc::clone(&dynamic_filter),
+            build_accumulator: OnceLock::new(),
+        });
+
+        Ok((join, dynamic_filter))
+    }
+
     async fn join_collect(
         left: Arc<dyn ExecutionPlan>,
         right: Arc<dyn ExecutionPlan>,
@@ -5007,67 +5072,32 @@ mod tests {
 
     #[tokio::test]
     async fn join_does_not_consume_probe_when_empty_build_fixes_output() {
-        let (left, right_input, on) = empty_build_with_probe_error_inputs();
-
-        let join_types = vec![
-            JoinType::Inner,
-            JoinType::Left,
-            JoinType::LeftSemi,
-            JoinType::LeftAnti,
-            JoinType::LeftMark,
-            JoinType::RightSemi,
-        ];
-
-        for join_type in join_types {
-            let join = join(
-                Arc::clone(&left),
-                Arc::clone(&right_input),
-                on.clone(),
-                &join_type,
-                NullEquality::NullEqualsNothing,
-            )
-            .unwrap();
-            let task_ctx = Arc::new(TaskContext::default());
-
-            let stream = join.execute(0, task_ctx).unwrap();
-            let batches = common::collect(stream).await.unwrap();
-
-            assert!(
-                batches.is_empty(),
-                "expected no output batches for {join_type}, got {batches:?}"
-            );
-        }
+        assert_empty_build_probe_behavior(
+            &[
+                JoinType::Inner,
+                JoinType::Left,
+                JoinType::LeftSemi,
+                JoinType::LeftAnti,
+                JoinType::LeftMark,
+                JoinType::RightSemi,
+            ],
+            false,
+        )
+        .await;
     }
 
     #[tokio::test]
     async fn join_still_consumes_probe_when_empty_build_needs_probe_rows() {
-        let (left, right_input, on) = empty_build_with_probe_error_inputs();
-
-        let join_types = vec![
-            JoinType::Right,
-            JoinType::Full,
-            JoinType::RightAnti,
-            JoinType::RightMark,
-        ];
-
-        for join_type in join_types {
-            let join = join(
-                Arc::clone(&left),
-                Arc::clone(&right_input),
-                on.clone(),
-                &join_type,
-                NullEquality::NullEqualsNothing,
-            )
-            .unwrap();
-            let task_ctx = Arc::new(TaskContext::default());
-
-            let stream = join.execute(0, task_ctx).unwrap();
-            let result_string = common::collect(stream).await.unwrap_err().to_string();
-            assert!(
-                result_string.contains("bad data error"),
-                "actual: {result_string}"
-            );
-        }
+        assert_empty_build_probe_behavior(
+            &[
+                JoinType::Right,
+                JoinType::Full,
+                JoinType::RightAnti,
+                JoinType::RightMark,
+            ],
+            true,
+        )
+        .await;
     }
 
     #[tokio::test]
@@ -5513,34 +5543,16 @@ mod tests {
             Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
         )];
 
-        // Create a dynamic filter manually
-        let dynamic_filter = HashJoinExec::create_dynamic_filter(&on);
-        let dynamic_filter_clone = Arc::clone(&dynamic_filter);
-
-        // Create HashJoinExec with the dynamic filter
-        let mut join = HashJoinExec::try_new(
-            left,
-            right,
-            on,
-            None,
-            &JoinType::Inner,
-            None,
-            PartitionMode::CollectLeft,
-            NullEquality::NullEqualsNothing,
-            false,
-        )?;
-        join.dynamic_filter = Some(HashJoinExecDynamicFilter {
-            filter: dynamic_filter,
-            build_accumulator: OnceLock::new(),
-        });
+        let (join, dynamic_filter) =
+            hash_join_with_dynamic_filter(left, right, on, JoinType::Inner)?;
 
         // Execute the join
         let stream = join.execute(0, task_ctx)?;
         let _batches = common::collect(stream).await?;
 
         // After the join completes, the dynamic filter should be marked as complete
         // wait_complete() should return immediately
-        dynamic_filter_clone.wait_complete().await;
+        dynamic_filter.wait_complete().await;
 
         Ok(())
     }
@@ -5562,71 +5574,37 @@ mod tests {
             Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
         )];
 
-        // Create a dynamic filter manually
-        let dynamic_filter = HashJoinExec::create_dynamic_filter(&on);
-        let dynamic_filter_clone = Arc::clone(&dynamic_filter);
-
-        // Create HashJoinExec with the dynamic filter
-        let mut join = HashJoinExec::try_new(
-            left,
-            right,
-            on,
-            None,
-            &JoinType::Inner,
-            None,
-            PartitionMode::CollectLeft,
-            NullEquality::NullEqualsNothing,
-            false,
-        )?;
-        join.dynamic_filter = Some(HashJoinExecDynamicFilter {
-            filter: dynamic_filter,
-            build_accumulator: OnceLock::new(),
-        });
+        let (join, dynamic_filter) =
+            hash_join_with_dynamic_filter(left, right, on, JoinType::Inner)?;
 
         // Execute the join
         let stream = join.execute(0, task_ctx)?;
         let _batches = common::collect(stream).await?;
 
         // Even with empty build side, the dynamic filter should be marked as complete
         // wait_complete() should return immediately
-        dynamic_filter_clone.wait_complete().await;
+        dynamic_filter.wait_complete().await;
 
         Ok(())
     }
 
     #[tokio::test]
-    async fn test_hash_join_skips_probe_on_empty_build_after_partition_bounds_report(
-    ) -> Result<()> {
+    async fn test_hash_join_skips_probe_on_empty_build_after_partition_bounds_report()
+    -> Result<()> {
         let task_ctx = Arc::new(TaskContext::default());
         let (left, right, on) = empty_build_with_probe_error_inputs();
 
         // Keep an extra consumer reference so execute() enables dynamic filter pushdown
         // and enters the WaitPartitionBoundsReport path before deciding whether to poll
         // the probe side.
-        let dynamic_filter = HashJoinExec::create_dynamic_filter(&on);
-        let dynamic_filter_clone = Arc::clone(&dynamic_filter);
-
-        let mut join = HashJoinExec::try_new(
-            left,
-            right,
-            on,
-            None,
-            &JoinType::Inner,
-            None,
-            PartitionMode::CollectLeft,
-            NullEquality::NullEqualsNothing,
-            false,
-        )?;
-        join.dynamic_filter = Some(HashJoinExecDynamicFilter {
-            filter: dynamic_filter,
-            build_accumulator: OnceLock::new(),
-        });
+        let (join, dynamic_filter) =
+            hash_join_with_dynamic_filter(left, right, on, JoinType::Inner)?;
 
         let stream = join.execute(0, task_ctx)?;
         let batches = common::collect(stream).await?;
         assert!(batches.is_empty());
 
-        dynamic_filter_clone.wait_complete().await;
+        dynamic_filter.wait_complete().await;
 
         Ok(())
     }

datafusion/physical-plan/src/joins/hash_join/stream.rs

@@ -408,7 +408,10 @@ impl HashJoinStream {
 
     /// Returns the next state after the build side has been fully collected
     /// and any required build-side coordination has completed.
-    fn next_state_after_build_ready(&self, left_data: &JoinLeftData) -> HashJoinStreamState {
+    fn next_state_after_build_ready(
+        &self,
+        left_data: &JoinLeftData,
+    ) -> HashJoinStreamState {
         if left_data.map().is_empty()
             && self.filter.is_none()
             && can_skip_probe_on_empty_build_side(self.join_type)