perf: Optimize NULL handling in find_in_set (#21464)

## Which issue does this PR close?

- Closes #21463.

## Rationale for this change

`find_in_set` uses `PrimitiveArray::<T>::builder` to construct its
results, which means building the internal null buffer in an iterative
fashion (via repeated `append_null` calls). It is more efficient to
construct the null buffer directly, via `NullBuffer::union` (when
multiple arguments might be NULL) or just cloning the input null buffer
(when passed a single argument).

Benchmarks (ARM64):

```
  - find_in_set/string_len_8: 589.0 µs → 529.0 µs (-10.2%)
  - find_in_set/string_len_32: 736.2 µs → 660.5 µs (-10.3%)
  - find_in_set/string_len_1024: 7.5 ms → 7.4 ms (-1.3%)
  - find_in_set/string_view_len_8: 616.0 µs → 579.9 µs (-5.9%)
  - find_in_set/string_view_len_32: 748.0 µs → 701.9 µs (-6.2%)
  - find_in_set/string_view_len_1024: 7.6 ms → 7.6 ms (0.0%)
  - find_in_set_scalar/string_len_8: 76.7 µs → 48.0 µs (-37.4%)
  - find_in_set_scalar/string_len_32: 76.5 µs → 47.6 µs (-37.8%)
  - find_in_set_scalar/string_len_1024: 76.2 µs → 48.0 µs (-37.0%)
  - find_in_set_scalar/string_view_len_8: 81.9 µs → 55.7 µs (-32.0%)
  - find_in_set_scalar/string_view_len_32: 85.5 µs → 56.8 µs (-33.6%)
  - find_in_set_scalar/string_view_len_1024: 85.2 µs → 57.4 µs (-32.6%)
```

The change should be an improvement for both scalar and array cases. The
relative improvement is larger in the scalar case because the scalar
case is doing less work and so NULL handling was a larger fraction of
the total runtime.

## What changes are included in this PR?

* Optimize NULL handling for both scalar and array arg cases

## Are these changes tested?

Yes.

## Are there any user-facing changes?

No.

Author: neilconway

datafusion/functions/src/unicode/find_in_set.rs

@@ -18,10 +18,10 @@
 use std::sync::Arc;
 
 use arrow::array::{
-    ArrayAccessor, ArrayIter, ArrayRef, ArrowPrimitiveType, AsArray, OffsetSizeTrait,
-    PrimitiveArray,
+    ArrayAccessor, ArrayRef, ArrowPrimitiveType, AsArray, OffsetSizeTrait, PrimitiveArray,
 };
 use arrow::datatypes::{ArrowNativeType, DataType, Int32Type, Int64Type};
+use arrow_buffer::NullBuffer;
 
 use crate::utils::utf8_to_int_type;
 use datafusion_common::{
@@ -265,53 +265,55 @@ fn find_in_set_general<'a, T, V>(string_array: V, str_list_array: V) -> Result<A
 where
     T: ArrowPrimitiveType,
     T::Native: OffsetSizeTrait,
-    V: ArrayAccessor<Item = &'a str>,
+    V: ArrayAccessor<Item = &'a str> + Copy,
 {
-    let string_iter = ArrayIter::new(string_array);
-    let str_list_iter = ArrayIter::new(str_list_array);
-
-    let mut builder = PrimitiveArray::<T>::builder(string_iter.len());
-
-    string_iter
-        .zip(str_list_iter)
-        .for_each(
-            |(string_opt, str_list_opt)| match (string_opt, str_list_opt) {
-                (Some(string), Some(str_list)) => {
-                    let position = str_list
-                        .split(',')
-                        .position(|s| s == string)
-                        .map_or(0, |idx| idx + 1);
-                    builder.append_value(T::Native::from_usize(position).unwrap());
-                }
-                _ => builder.append_null(),
-            },
-        );
+    let len = string_array.len();
+    let nulls = NullBuffer::union(string_array.nulls(), str_list_array.nulls());
+    let zero = T::Native::from_usize(0).unwrap();
+
+    let values: Vec<T::Native> = (0..len)
+        .map(|i| {
+            if nulls.as_ref().is_some_and(|n| n.is_null(i)) {
+                return zero;
+            }
+            let string = string_array.value(i);
+            let str_list = str_list_array.value(i);
+            let position = str_list
+                .split(',')
+                .position(|s| s == string)
+                .map_or(0, |idx| idx + 1);
+            T::Native::from_usize(position).unwrap()
+        })
+        .collect();
 
-    Ok(Arc::new(builder.finish()) as ArrayRef)
+    Ok(Arc::new(PrimitiveArray::<T>::new(values.into(), nulls)) as ArrayRef)
 }
 
 fn find_in_set_left_literal<'a, T, V>(string: &str, str_list_array: V) -> Result<ArrayRef>
 where
     T: ArrowPrimitiveType,
     T::Native: OffsetSizeTrait,
-    V: ArrayAccessor<Item = &'a str>,
+    V: ArrayAccessor<Item = &'a str> + Copy,
 {
-    let mut builder = PrimitiveArray::<T>::builder(str_list_array.len());
-
-    let str_list_iter = ArrayIter::new(str_list_array);
-
-    str_list_iter.for_each(|str_list_opt| match str_list_opt {
-        Some(str_list) => {
+    let len = str_list_array.len();
+    let nulls = str_list_array.nulls().cloned();
+    let zero = T::Native::from_usize(0).unwrap();
+
+    let values: Vec<T::Native> = (0..len)
+        .map(|i| {
+            if nulls.as_ref().is_some_and(|n| n.is_null(i)) {
+                return zero;
+            }
+            let str_list = str_list_array.value(i);
             let position = str_list
                 .split(',')
                 .position(|s| s == string)
                 .map_or(0, |idx| idx + 1);
-            builder.append_value(T::Native::from_usize(position).unwrap());
-        }
-        None => builder.append_null(),
-    });
+            T::Native::from_usize(position).unwrap()
+        })
+        .collect();
 
-    Ok(Arc::new(builder.finish()) as ArrayRef)
+    Ok(Arc::new(PrimitiveArray::<T>::new(values.into(), nulls)) as ArrayRef)
 }
 
 fn find_in_set_right_literal<'a, T, V>(
@@ -321,24 +323,27 @@ fn find_in_set_right_literal<'a, T, V>(
 where
     T: ArrowPrimitiveType,
     T::Native: OffsetSizeTrait,
-    V: ArrayAccessor<Item = &'a str>,
+    V: ArrayAccessor<Item = &'a str> + Copy,
 {
-    let mut builder = PrimitiveArray::<T>::builder(string_array.len());
-
-    let string_iter = ArrayIter::new(string_array);
-
-    string_iter.for_each(|string_opt| match string_opt {
-        Some(string) => {
+    let len = string_array.len();
+    let nulls = string_array.nulls().cloned();
+    let zero = T::Native::from_usize(0).unwrap();
+
+    let values: Vec<T::Native> = (0..len)
+        .map(|i| {
+            if nulls.as_ref().is_some_and(|n| n.is_null(i)) {
+                return zero;
+            }
+            let string = string_array.value(i);
             let position = str_list
                 .iter()
                 .position(|s| *s == string)
                 .map_or(0, |idx| idx + 1);
-            builder.append_value(T::Native::from_usize(position).unwrap());
-        }
-        None => builder.append_null(),
-    });
+            T::Native::from_usize(position).unwrap()
+        })
+        .collect();
 
-    Ok(Arc::new(builder.finish()) as ArrayRef)
+    Ok(Arc::new(PrimitiveArray::<T>::new(values.into(), nulls)) as ArrayRef)
 }
 
 #[cfg(test)]