Simplify hash-join state machine and batch construction

Refactor stream.rs and utils.rs to streamline the hash-join
state machine. Compute the post-build state directly from
inputs, eliminating unnecessary indirection. Update the
empty-build-side batch construction to utilize early returns
and iterator-based collection for columns, replacing manual
Vec setup and push logic.

Author: kosiew

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

@@ -408,13 +408,14 @@ 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,
+    fn state_after_build_ready(
+        has_filter: bool,
+        join_type: JoinType,
         left_data: &JoinLeftData,
     ) -> HashJoinStreamState {
-        if left_data.map().is_empty()
-            && self.filter.is_none()
-            && empty_build_side_produces_empty_result(self.join_type)
+        if !has_filter
+            && left_data.map().is_empty()
+            && empty_build_side_produces_empty_result(join_type)
         {
             HashJoinStreamState::Completed
         } else {
@@ -486,7 +487,11 @@ impl HashJoinStream {
             ready!(fut.get_shared(cx))?;
         }
         let build_side = self.build_side.try_as_ready()?;
-        self.state = self.next_state_after_build_ready(build_side.left_data.as_ref());
+        self.state = Self::state_after_build_ready(
+            self.filter.is_some(),
+            self.join_type,
+            build_side.left_data.as_ref(),
+        );
         Poll::Ready(Ok(StatefulStreamResult::Continue))
     }
 
@@ -557,7 +562,11 @@ impl HashJoinStream {
             }));
             self.state = HashJoinStreamState::WaitPartitionBoundsReport;
         } else {
-            self.state = self.next_state_after_build_ready(left_data.as_ref());
+            self.state = Self::state_after_build_ready(
+                self.filter.is_some(),
+                self.join_type,
+                left_data.as_ref(),
+            );
         }
 
         self.build_side = BuildSide::Ready(BuildSideReadyState { left_data });

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

@@ -1078,37 +1078,33 @@ pub(crate) fn build_batch_empty_build_side(
 ) -> Result<RecordBatch> {
     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,
-                )),
-            };
-
-            columns.push(array);
-        }
+        return Ok(RecordBatch::new_empty(Arc::new(schema.clone())));
+    }
 
-        Ok(RecordBatch::try_new(Arc::new(schema.clone()), columns)?)
+    // 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 columns = column_indices
+        .iter()
+        .map(|column_index| 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))
+            }
+        })
+        .collect();
+
+    Ok(RecordBatch::try_new(Arc::new(schema.clone()), columns)?)
 }
 
 /// The input is the matched indices for left and right and