Coalesce build-side batches in hash join and fix BatchCoalescer min size

Coalesce small build-side batches into target_batch_size in
collect_left_input, so build_batch_from_indices hits the fast
SingleBatch/take path more often instead of MultiBatch/interleave.

Also add .max(8) guards to all LimitedBatchCoalescer call sites to
avoid debug_assert panic in arrow-select BatchCoalescer when
target_batch_size < array capacity (upstream arrow bug).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Dandandan

datafusion/physical-plan/src/coalesce_partitions.rs

@@ -252,7 +252,7 @@ impl ExecutionPlan for CoalescePartitionsExec {
                     )),
                     coalescer: LimitedBatchCoalescer::new(
                         self.schema(),
-                        batch_size,
+                        batch_size.max(8),
                         None, // fetch is already handled by ObservedStream
                     ),
                     completed: false,
@@ -401,15 +401,13 @@ impl Stream for CoalescedStream {
                     self.completed = true;
                     self.coalescer.finish()?;
                 }
-                Some(Ok(batch)) => {
-                    match self.coalescer.push_batch(batch)? {
-                        PushBatchStatus::Continue => {}
-                        PushBatchStatus::LimitReached => {
-                            self.completed = true;
-                            self.coalescer.finish()?;
-                        }
+                Some(Ok(batch)) => match self.coalescer.push_batch(batch)? {
+                    PushBatchStatus::Continue => {}
+                    PushBatchStatus::LimitReached => {
+                        self.completed = true;
+                        self.coalescer.finish()?;
                     }
-                }
+                },
                 other => return Poll::Ready(other),
             }
         }

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

@@ -66,6 +66,7 @@ use crate::{
 };
 
 use arrow::array::{ArrayRef, BooleanBufferBuilder, UInt64Array};
+use arrow::compute::BatchCoalescer;
 use arrow::datatypes::SchemaRef;
 use arrow::record_batch::RecordBatch;
 use arrow::util::bit_util;
@@ -2001,6 +2002,25 @@ async fn collect_left_input(
         bounds_accumulators,
     } = state;
 
+    // Coalesce small batches into larger ones to reduce the number of
+    // build-side batches. Fewer (larger) batches mean that
+    // `build_batch_from_indices` is more likely to take the fast
+    // `SingleBatch`/`take` path instead of the slower
+    // `MultiBatch`/`interleave` path.
+    let batches = {
+        let target_batch_size = config.execution.batch_size.max(8);
+        let mut coalescer = BatchCoalescer::new(Arc::clone(&schema), target_batch_size);
+        for batch in batches {
+            coalescer.push_batch(batch)?;
+        }
+        coalescer.finish_buffered_batch()?;
+        let mut coalesced = Vec::new();
+        while let Some(batch) = coalescer.next_completed_batch() {
+            coalesced.push(batch);
+        }
+        coalesced
+    };
+
     // Compute bounds
     let mut bounds = match bounds_accumulators {
         Some(accumulators) if num_rows > 0 => {

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

@@ -382,7 +382,7 @@ impl HashJoinStream {
     ) -> Self {
         // Create output buffer with coalescing and optional fetch limit.
         let output_buffer =
-            LimitedBatchCoalescer::new(Arc::clone(&schema), batch_size, fetch);
+            LimitedBatchCoalescer::new(Arc::clone(&schema), batch_size.max(8), fetch);
 
         Self {
             partition,

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

@@ -1984,16 +1984,14 @@ fn and_bitmap_with_boolean_buffer(
 macro_rules! dispatch_elementwise {
     ($data_type:expr, $equal_bits:expr, $left_indices:expr, $null_equality:expr, $action:ident) => {
         match $data_type {
-            DataType::Null => {
-                match $null_equality {
-                    NullEquality::NullEqualsNothing => {
-                        for i in 0..$left_indices.len() {
-                            $equal_bits.set_bit(i, false);
-                        }
+            DataType::Null => match $null_equality {
+                NullEquality::NullEqualsNothing => {
+                    for i in 0..$left_indices.len() {
+                        $equal_bits.set_bit(i, false);
                     }
-                    NullEquality::NullEqualsNull => {}
                 }
-            }
+                NullEquality::NullEqualsNull => {}
+            },
             DataType::Boolean => $action!(BooleanArray),
             DataType::Int8 => $action!(Int8Array),
             DataType::Int16 => $action!(Int16Array),
@@ -2059,7 +2057,13 @@ fn compare_rows_elementwise(
         }};
     }
 
-    dispatch_elementwise!(left_array.data_type(), equal_bits, left_indices, null_equality, compare_elementwise);
+    dispatch_elementwise!(
+        left_array.data_type(),
+        equal_bits,
+        left_indices,
+        null_equality,
+        compare_elementwise
+    );
     true
 }
 
@@ -2152,7 +2156,13 @@ fn compare_rows_elementwise_multi(
         }};
     }
 
-    dispatch_elementwise!(right_array.data_type(), equal_bits, left_indices, null_equality, compare_multi);
+    dispatch_elementwise!(
+        right_array.data_type(),
+        equal_bits,
+        left_indices,
+        null_equality,
+        compare_multi
+    );
     true
 }
 
@@ -2172,10 +2182,8 @@ fn do_compare_elementwise_multi<A: ArrayAccessor>(
     A::Item: PartialEq,
 {
     let right_nulls = right.nulls();
-    let has_nulls = right.null_count() > 0
-        || left_arrays
-            .iter()
-            .any(|a| a.null_count() > 0);
+    let has_nulls =
+        right.null_count() > 0 || left_arrays.iter().any(|a| a.null_count() > 0);
     let num_rows = left_indices.len();
 
     if !has_nulls {
@@ -2204,8 +2212,8 @@ fn do_compare_elementwise_multi<A: ArrayAccessor>(
             let batch_idx = (packed >> 32) as usize;
             let row_idx = (packed & 0xFFFFFFFF) as usize;
             let r_idx = right_indices[i] as usize;
-            let l_null = left_nulls_per_batch[batch_idx]
-                .is_some_and(|n| !n.is_valid(row_idx));
+            let l_null =
+                left_nulls_per_batch[batch_idx].is_some_and(|n| !n.is_valid(row_idx));
             let r_null = right_nulls.is_some_and(|n| !n.is_valid(r_idx));
 
             let is_equal = match (l_null, r_null) {

datafusion/physical-plan/src/repartition/mod.rs

@@ -1595,8 +1595,8 @@ impl PerPartitionStream {
         baseline_metrics: BaselineMetrics,
         batch_size: Option<usize>,
     ) -> Self {
-        let batch_coalescer =
-            batch_size.map(|s| LimitedBatchCoalescer::new(Arc::clone(&schema), s, None));
+        let batch_coalescer = batch_size
+            .map(|s| LimitedBatchCoalescer::new(Arc::clone(&schema), s.max(8), None));
         Self {
             schema,
             receiver,