null_handling

Author: B Vadlamani

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

@@ -1993,9 +1993,7 @@ async fn collect_left_input(
     };
 
     let (join_hash_map, batch, left_values) =
-        if config.execution.perfect_hash_join_small_build_threshold > 0
-            && should_use_roaring_bitmap(&join_type, &on_left, &schema, has_filter)
-        {
+        if should_use_roaring_bitmap(&join_type, &on_left, &schema, has_filter) {
             let batch = concat_batches(&schema, &batches)?;
             let left_values = evaluate_expressions_to_arrays(&on_left, &batch)?;
             let key_col = &left_values[0];
@@ -2004,17 +2002,16 @@ async fn collect_left_input(
                     .as_any()
                     .downcast_ref::<UInt32Array>()
                     .unwrap()
-                    .values()
                     .iter()
-                    .map(|&v| v)
+                    .flatten()
                     .collect(),
                 DataType::Int32 => key_col
                     .as_any()
                     .downcast_ref::<Int32Array>()
                     .unwrap()
-                    .values()
                     .iter()
-                    .map(|&v| v as u32)
+                    .flatten()
+                    .map(|v| v as u32)
                     .collect(),
                 _ => return internal_err!("unsupported data type to build bitmap"),
             };

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

@@ -46,7 +46,8 @@ use crate::{
     },
 };
 
-use arrow::array::{Array, ArrayRef, Int32Array, UInt32Array, UInt64Array};
+use arrow::array::{Array, ArrayRef, BooleanArray, Int32Array, UInt32Array, UInt64Array};
+use arrow::compute::filter_record_batch;
 use arrow::datatypes::{Schema, SchemaRef};
 use arrow::record_batch::RecordBatch;
 use arrow_schema::DataType;
@@ -727,44 +728,30 @@ impl HashJoinStream {
             Map::RoaringMap(bitmap) => {
                 let key_col = &state.values[0];
                 let is_semi = matches!(self.join_type, JoinType::RightSemi);
-                let right_indices = match key_col.data_type() {
+                let mask: BooleanArray = match key_col.data_type() {
                     DataType::Int32 => {
                         let arr = key_col.as_any().downcast_ref::<Int32Array>().unwrap();
-                        arr.values()
-                            .iter()
-                            .enumerate()
-                            .filter_map(|(i, v)| {
-                                let contains = bitmap.contains(*v as u32);
-                                let emit = if is_semi { contains } else { !contains };
-                                emit.then_some(i as u32)
+                        arr.iter()
+                            .map(|v| match v {
+                                Some(v) => bitmap.contains(v as u32) == is_semi,
+                                None => !is_semi,
                             })
-                            .collect::<Vec<u32>>()
+                            .collect()
                     }
                     DataType::UInt32 => {
                         let arr = key_col.as_any().downcast_ref::<UInt32Array>().unwrap();
-                        arr.values()
-                            .iter()
-                            .enumerate()
-                            .filter_map(|(i, v)| {
-                                let contains = bitmap.contains(*v);
-                                let emit = if is_semi { contains } else { !contains };
-                                emit.then_some(i as u32)
+                        arr.iter()
+                            .map(|v| match v {
+                                Some(v) => bitmap.contains(v) == is_semi,
+                                None => !is_semi,
                             })
-                            .collect::<Vec<u32>>()
+                            .collect()
                     }
                     _ => {
                         return internal_err!("unsupported data type for roaring bitmap");
                     }
                 };
-                let indices = UInt32Array::from(right_indices);
-                let columns: Vec<ArrayRef> = state
-                    .batch
-                    .columns()
-                    .iter()
-                    .map(|col| arrow::compute::take(col, &indices, None))
-                    .collect::<Result<Vec<_>, _>>()?;
-
-                let batch = RecordBatch::try_new(self.schema.clone(), columns)?;
+                let batch = filter_record_batch(&state.batch, &mask)?;
                 self.output_buffer.push_batch(batch)?;
                 self.state = HashJoinStreamState::FetchProbeBatch;
                 return Ok(StatefulStreamResult::Continue);