Refactor join type handling for empty builds

Move empty-build predicate to JoinType. Implement
JoinType::empty_build_side_produces_empty_result method
and update relevant call sites. Stream state machine
now directly uses this method, simplifying logic.
Remove old utility predicate and update tests,
including new filtered regression tests.

Author: kosiew

datafusion/common/src/join_type.rs

@@ -142,6 +142,20 @@ impl JoinType {
                 | JoinType::RightMark
         )
     }
+
+    /// Returns true when an empty build side necessarily produces an empty
+    /// result for this join type.
+    pub fn empty_build_side_produces_empty_result(self) -> bool {
+        matches!(
+            self,
+            JoinType::Inner
+                | JoinType::Left
+                | JoinType::LeftSemi
+                | JoinType::LeftAnti
+                | JoinType::LeftMark
+                | JoinType::RightSemi
+        )
+    }
 }
 
 impl Display for JoinType {

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

@@ -2305,18 +2305,32 @@ mod tests {
     async fn assert_empty_build_probe_behavior(
         join_types: &[JoinType],
         expect_probe_error: bool,
+        with_filter: bool,
     ) {
         let (left, right, on) = empty_build_with_probe_error_inputs();
+        let filter = prepare_join_filter();
 
         for join_type in join_types {
-            let join = join(
-                Arc::clone(&left),
-                Arc::clone(&right),
-                on.clone(),
-                join_type,
-                NullEquality::NullEqualsNothing,
-            )
-            .unwrap();
+            let join = if with_filter {
+                join_with_filter(
+                    Arc::clone(&left),
+                    Arc::clone(&right),
+                    on.clone(),
+                    filter.clone(),
+                    join_type,
+                    NullEquality::NullEqualsNothing,
+                )
+                .unwrap()
+            } else {
+                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(),
@@ -5085,6 +5099,24 @@ mod tests {
                 JoinType::RightSemi,
             ],
             false,
+            false,
+        )
+        .await;
+    }
+
+    #[tokio::test]
+    async fn join_does_not_consume_probe_when_empty_build_fixes_output_with_filter() {
+        assert_empty_build_probe_behavior(
+            &[
+                JoinType::Inner,
+                JoinType::Left,
+                JoinType::LeftSemi,
+                JoinType::LeftAnti,
+                JoinType::LeftMark,
+                JoinType::RightSemi,
+            ],
+            false,
+            true,
         )
         .await;
     }
@@ -5099,6 +5131,22 @@ mod tests {
                 JoinType::RightMark,
             ],
             true,
+            false,
+        )
+        .await;
+    }
+
+    #[tokio::test]
+    async fn join_still_consumes_probe_when_empty_build_needs_probe_rows_with_filter() {
+        assert_empty_build_probe_behavior(
+            &[
+                JoinType::Right,
+                JoinType::Full,
+                JoinType::RightAnti,
+                JoinType::RightMark,
+            ],
+            true,
+            true,
         )
         .await;
     }

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,
-        empty_build_side_produces_empty_result, need_produce_result_in_final,
+        need_produce_result_in_final,
     },
 };
 
@@ -409,13 +409,11 @@ impl HashJoinStream {
     /// Returns the next state after the build side has been fully collected
     /// and any required build-side coordination has completed.
     fn state_after_build_ready(
-        has_filter: bool,
         join_type: JoinType,
         left_data: &JoinLeftData,
     ) -> HashJoinStreamState {
-        if !has_filter
-            && left_data.map().is_empty()
-            && empty_build_side_produces_empty_result(join_type)
+        if left_data.map().is_empty()
+            && join_type.empty_build_side_produces_empty_result()
         {
             HashJoinStreamState::Completed
         } else {
@@ -487,11 +485,8 @@ impl HashJoinStream {
             ready!(fut.get_shared(cx))?;
         }
         let build_side = self.build_side.try_as_ready()?;
-        self.state = Self::state_after_build_ready(
-            self.filter.is_some(),
-            self.join_type,
-            build_side.left_data.as_ref(),
-        );
+        self.state =
+            Self::state_after_build_ready(self.join_type, build_side.left_data.as_ref());
         Poll::Ready(Ok(StatefulStreamResult::Continue))
     }
 
@@ -562,11 +557,8 @@ impl HashJoinStream {
             }));
             self.state = HashJoinStreamState::WaitPartitionBoundsReport;
         } else {
-            self.state = Self::state_after_build_ready(
-                self.filter.is_some(),
-                self.join_type,
-                left_data.as_ref(),
-            );
+            self.state =
+                Self::state_after_build_ready(self.join_type, left_data.as_ref());
         }
 
         self.build_side = BuildSide::Ready(BuildSideReadyState { left_data });

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

@@ -855,22 +855,6 @@ pub(crate) fn need_produce_result_in_final(join_type: JoinType) -> bool {
     )
 }
 
-/// Returns true when an empty build side necessarily produces an empty result.
-///
-/// This is the shared source of truth for both state-machine short-circuiting
-/// and `build_batch_empty_build_side`.
-pub(crate) fn empty_build_side_produces_empty_result(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,
@@ -1076,7 +1060,7 @@ pub(crate) fn build_batch_empty_build_side(
     column_indices: &[ColumnIndex],
     join_type: JoinType,
 ) -> Result<RecordBatch> {
-    if empty_build_side_produces_empty_result(join_type) {
+    if join_type.empty_build_side_produces_empty_result() {
         // These join types only return data if the left side is not empty.
         return Ok(RecordBatch::new_empty(Arc::new(schema.clone())));
     }