Refactor empty build side handling

Remove duplication by introducing a shared helper,
empty_build_side_produces_empty_result, in utils.rs.
Update build_batch_empty_build_side to use this helper
directly, ensuring alignment in the short-circuit and
batch-construction logic within the hash join state
transition in stream.rs.

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,
-        can_skip_probe_on_empty_build_side, need_produce_result_in_final,
+        empty_build_side_produces_empty_result, need_produce_result_in_final,
     },
 };
 
@@ -414,7 +414,7 @@ impl HashJoinStream {
     ) -> HashJoinStreamState {
         if left_data.map().is_empty()
             && self.filter.is_none()
-            && can_skip_probe_on_empty_build_side(self.join_type)
+            && empty_build_side_produces_empty_result(self.join_type)
         {
             HashJoinStreamState::Completed
         } else {

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

@@ -855,9 +855,11 @@ 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 {
+/// 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
@@ -1074,46 +1076,38 @@ pub(crate) fn build_batch_empty_build_side(
     column_indices: &[ColumnIndex],
     join_type: JoinType,
 ) -> Result<RecordBatch> {
-    match join_type {
-        // these join types only return data if the left side is not empty, so we return an
-        // empty RecordBatch
-        JoinType::Inner
-        | JoinType::Left
-        | JoinType::LeftSemi
-        | JoinType::RightSemi
-        | JoinType::LeftAnti
-        | JoinType::LeftMark => Ok(RecordBatch::new_empty(Arc::new(schema.clone()))),
-
-        // the remaining joins will return data for the right columns and null for the left ones
-        JoinType::Right | JoinType::Full | JoinType::RightAnti | JoinType::RightMark => {
-            let num_rows = probe_batch.num_rows();
-            if schema.fields().is_empty() {
-                return new_empty_schema_batch(schema, num_rows);
-            }
-            let mut columns: Vec<Arc<dyn Array>> =
-                Vec::with_capacity(schema.fields().len());
-
-            for column_index in column_indices {
-                let array = match column_index.side {
-                    // left -> null array
-                    JoinSide::Left => new_null_array(
-                        build_batch.column(column_index.index).data_type(),
-                        num_rows,
-                    ),
-                    // right -> respective right array
-                    JoinSide::Right => Arc::clone(probe_batch.column(column_index.index)),
-                    // right mark -> unset boolean array as there are no matches on the left side
-                    JoinSide::None => Arc::new(BooleanArray::new(
-                        BooleanBuffer::new_unset(num_rows),
-                        None,
-                    )),
-                };
-
-                columns.push(array);
-            }
+    if empty_build_side_produces_empty_result(join_type) {
+        // These join types only return data if the left side is not empty.
+        Ok(RecordBatch::new_empty(Arc::new(schema.clone())))
+    } else {
+        // The remaining joins return right-side rows and nulls for the left side.
+        let num_rows = probe_batch.num_rows();
+        if schema.fields().is_empty() {
+            return new_empty_schema_batch(schema, num_rows);
+        }
+        let mut columns: Vec<Arc<dyn Array>> =
+            Vec::with_capacity(schema.fields().len());
+
+        for column_index in column_indices {
+            let array = match column_index.side {
+                // left -> null array
+                JoinSide::Left => new_null_array(
+                    build_batch.column(column_index.index).data_type(),
+                    num_rows,
+                ),
+                // right -> respective right array
+                JoinSide::Right => Arc::clone(probe_batch.column(column_index.index)),
+                // right mark -> unset boolean array as there are no matches on the left side
+                JoinSide::None => Arc::new(BooleanArray::new(
+                    BooleanBuffer::new_unset(num_rows),
+                    None,
+                )),
+            };
 
-            Ok(RecordBatch::try_new(Arc::new(schema.clone()), columns)?)
+            columns.push(array);
         }
+
+        Ok(RecordBatch::try_new(Arc::new(schema.clone()), columns)?)
     }
 }