Refactor join type logic into utils.rs

Move the pure JoinType semantic rule to utils.rs, placing
it alongside the existing join behavior helpers. Update
HashJoinStream in stream.rs to focus solely on its
stream-specific execution concern by removing unnecessary
logic related to filtering.

Author: kosiew

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

@@ -42,7 +42,7 @@ use crate::{
         BuildProbeJoinMetrics, ColumnIndex, JoinFilter, JoinHashMapType,
         StatefulStreamResult, adjust_indices_by_join_type, apply_join_filter_to_indices,
         build_batch_empty_build_side, build_batch_from_indices,
-        need_produce_result_in_final,
+        can_skip_probe_on_empty_build_side, need_produce_result_in_final,
     },
 };
 
@@ -406,25 +406,13 @@ 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
-            )
-    }
-
     /// 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 {
-        if left_data.map().is_empty() && self.can_skip_probe_on_empty_build_side() {
+        if left_data.map().is_empty()
+            && self.filter.is_none()
+            && can_skip_probe_on_empty_build_side(self.join_type)
+        {
             HashJoinStreamState::Completed
         } else {
             HashJoinStreamState::FetchProbeBatch

datafusion/physical-plan/src/joins/utils.rs

@@ -855,6 +855,20 @@ pub(crate) fn need_produce_result_in_final(join_type: JoinType) -> bool {
     )
 }
 
+/// Returns true when an empty build side fully determines the join result,
+/// so the probe side does not need to be consumed.
+pub(crate) fn can_skip_probe_on_empty_build_side(join_type: JoinType) -> bool {
+    matches!(
+        join_type,
+        JoinType::Inner
+            | JoinType::Left
+            | JoinType::LeftSemi
+            | JoinType::LeftAnti
+            | JoinType::LeftMark
+            | JoinType::RightSemi
+    )
+}
+
 pub(crate) fn get_final_indices_from_shared_bitmap(
     shared_bitmap: &SharedBitmapBuilder,
     join_type: JoinType,