Optimize HashJoinStream for empty build side

Implement a staged mini-plan for HashJoinStream to
immediately exit when the build side is empty and
the join type's result is fully determined. This change
avoids unnecessary entry into FetchProbeBatch for
Inner, Left, LeftSemi, LeftAnti, LeftMark, and RightSemi
joins without filters.

Add tests to verify join behavior with empty build:
- join_does_not_consume_probe_when_empty_build_fixes_output
- join_still_consumes_probe_when_empty_build_needs_probe_rows
These use MockExec to distinguish between short-circuiting
and necessary probe row consumption.

Author: kosiew

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

@@ -4983,6 +4983,109 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn join_does_not_consume_probe_when_empty_build_fixes_output() {
+        let left_batch =
+            build_table_i32(("a1", &vec![]), ("b1", &vec![]), ("c1", &vec![]));
+        let left_schema = left_batch.schema();
+
+        let err = exec_err!("bad data error");
+        let right = build_table_i32(("a2", &vec![]), ("b1", &vec![]), ("c2", &vec![]));
+
+        let on = vec![(
+            Arc::new(Column::new_with_schema("b1", &left_schema).unwrap()) as _,
+            Arc::new(Column::new_with_schema("b1", &right.schema()).unwrap()) as _,
+        )];
+        let schema = right.schema();
+        let right_input =
+            Arc::new(MockExec::new(vec![Ok(right), err], schema).with_use_task(false));
+        let left: Arc<dyn ExecutionPlan> = TestMemoryExec::try_new_exec(
+            &[vec![left_batch]],
+            Arc::clone(&left_schema),
+            None,
+        )
+        .unwrap();
+
+        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) as Arc<dyn ExecutionPlan>,
+                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:?}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn join_still_consumes_probe_when_empty_build_needs_probe_rows() {
+        let left_batch =
+            build_table_i32(("a1", &vec![]), ("b1", &vec![]), ("c1", &vec![]));
+        let left_schema = left_batch.schema();
+
+        let err = exec_err!("bad data error");
+        let right = build_table_i32(("a2", &vec![]), ("b1", &vec![]), ("c2", &vec![]));
+
+        let on = vec![(
+            Arc::new(Column::new_with_schema("b1", &left_schema).unwrap()) as _,
+            Arc::new(Column::new_with_schema("b1", &right.schema()).unwrap()) as _,
+        )];
+        let schema = right.schema();
+        let right_input =
+            Arc::new(MockExec::new(vec![Ok(right), err], schema).with_use_task(false));
+        let left: Arc<dyn ExecutionPlan> = TestMemoryExec::try_new_exec(
+            &[vec![left_batch]],
+            Arc::clone(&left_schema),
+            None,
+        )
+        .unwrap();
+
+        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) as Arc<dyn ExecutionPlan>,
+                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}"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn join_split_batch() {
         let left = build_table(

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

@@ -406,6 +406,21 @@ impl HashJoinStream {
         }
     }
 
+    /// Returns true when an empty build side fully determines the join result,
+    /// so the probe side does not need to be consumed.
+    fn can_skip_probe_on_empty_build_side(&self) -> bool {
+        self.filter.is_none()
+            && matches!(
+                self.join_type,
+                JoinType::Inner
+                    | JoinType::Left
+                    | JoinType::LeftSemi
+                    | JoinType::LeftAnti
+                    | JoinType::LeftMark
+                    | JoinType::RightSemi
+            )
+    }
+
     /// Separate implementation function that unpins the [`HashJoinStream`] so
     /// that partial borrows work correctly
     fn poll_next_impl(
@@ -469,7 +484,14 @@ impl HashJoinStream {
         if let Some(ref mut fut) = self.build_waiter {
             ready!(fut.get_shared(cx))?;
         }
-        self.state = HashJoinStreamState::FetchProbeBatch;
+        let build_side = self.build_side.try_as_ready()?;
+        self.state = if build_side.left_data.map().is_empty()
+            && self.can_skip_probe_on_empty_build_side()
+        {
+            HashJoinStreamState::Completed
+        } else {
+            HashJoinStreamState::FetchProbeBatch
+        };
         Poll::Ready(Ok(StatefulStreamResult::Continue))
     }
 
@@ -540,7 +562,13 @@ impl HashJoinStream {
             }));
             self.state = HashJoinStreamState::WaitPartitionBoundsReport;
         } else {
-            self.state = HashJoinStreamState::FetchProbeBatch;
+            self.state = if left_data.map().is_empty()
+                && self.can_skip_probe_on_empty_build_side()
+            {
+                HashJoinStreamState::Completed
+            } else {
+                HashJoinStreamState::FetchProbeBatch
+            };
         }
 
         self.build_side = BuildSide::Ready(BuildSideReadyState { left_data });