perf: Optimize substr_index to use bulk-NULL string builder (#21877)

## Which issue does this PR close?

- Closes #21876.

## Rationale for this change

As with other recent optimizations, we can optimize NULL handling in
`substr_index` by using the new bulk-NULL string builders.

Benchmarks:

Utf8
  - utf8_100_array_long_delimiter: 10.0 µs → 10.1 µs (+1.00%)
  - utf8_100_array_single_delimiter: 2.9 µs → 2.5 µs (−13.79%)
  - utf8_100_scalar_long_delimiter_neg: 4.1 µs → 3.5 µs (−14.63%)
  - utf8_100_scalar_long_delimiter_pos: 2.9 µs → 2.7 µs (−6.90%)
  - utf8_100_scalar_single_delimiter_neg: 2.2 µs → 1.993 µs (−9.41%)
  - utf8_100_scalar_single_delimiter_pos: 2.1 µs → 1.845 µs (−12.13%)
  - utf8_1000_array_long_delimiter: 101.0 µs → 101.1 µs (+0.10%)
  - utf8_1000_array_single_delimiter: 36.8 µs → 31.7 µs (−13.86%)
  - utf8_1000_scalar_long_delimiter_neg: 38.9 µs → 36.9 µs (−5.14%)
  - utf8_1000_scalar_long_delimiter_pos: 25.1 µs → 23.3 µs (−7.17%)
  - utf8_1000_scalar_single_delimiter_neg: 19.3 µs → 17.7 µs (−8.29%)
  - utf8_1000_scalar_single_delimiter_pos: 18.2 µs → 16.6 µs (−8.79%)
  - utf8_10000_array_long_delimiter: 1083.4 µs → 1038.2 µs (−4.17%)
  - utf8_10000_array_single_delimiter: 461.8 µs → 414.7 µs (−10.20%)
  - utf8_10000_scalar_long_delimiter_neg: 392.4 µs → 379.3 µs (−3.34%)
  - utf8_10000_scalar_long_delimiter_pos: 246.5 µs → 227.4 µs (−7.75%)
  - utf8_10000_scalar_single_delimiter_neg: 191.3 µs → 177.5 µs (−7.21%)
  - utf8_10000_scalar_single_delimiter_pos: 179.4 µs → 168.8 µs (−5.91%)

Utf8View
  - utf8view_100_array_long_delimiter: 9.5 µs → 9.8 µs (+3.16%)
  - utf8view_100_array_single_delimiter: 2.6 µs → 2.6 µs (0.00%)
  - utf8view_100_scalar_long_delimiter_neg: 4.0 µs → 4.0 µs (0.00%)
  - utf8view_100_scalar_long_delimiter_pos: 2.8 µs → 2.8 µs (0.00%)
  - utf8view_100_scalar_single_delimiter_neg: 2.3 µs → 2.3 µs (0.00%)
  - utf8view_100_scalar_single_delimiter_pos: 2.2 µs → 2.1 µs (−4.55%)
  - utf8view_1000_array_long_delimiter: 94.8 µs → 99.2 µs (+4.64%)
  - utf8view_1000_array_single_delimiter: 31.5 µs → 32.0 µs (+1.59%)
  - utf8view_1000_scalar_long_delimiter_neg: 38.7 µs → 39.0 µs (+0.78%)
  - utf8view_1000_scalar_long_delimiter_pos: 25.4 µs → 25.4 µs (0.00%)
- utf8view_1000_scalar_single_delimiter_neg: 21.4 µs → 21.8 µs (+1.87%)
- utf8view_1000_scalar_single_delimiter_pos: 20.8 µs → 20.9 µs (+0.48%)
  - utf8view_10000_array_long_delimiter: 998.4 µs → 1025.4 µs (+2.70%)
  - utf8view_10000_array_single_delimiter: 414.9 µs → 415.7 µs (+0.19%)
- utf8view_10000_scalar_long_delimiter_neg: 393.7 µs → 395.9 µs (+0.56%)
- utf8view_10000_scalar_long_delimiter_pos: 253.4 µs → 252.7 µs (−0.28%)
- utf8view_10000_scalar_single_delimiter_neg: 214.5 µs → 217.3 µs
(+1.31%)
- utf8view_10000_scalar_single_delimiter_pos: 207.9 µs → 208.7 µs
(+0.38%)

This PR doesn't touch the Utf8View code path, so the Utf8View
regressions above are likely measurement noise.

## What changes are included in this PR?

* Optimize `substr_index` by switching from Arrow string builders to
bulk-NULL string builders

## Are these changes tested?

Yes, covered by existing tests.

## Are there any user-facing changes?

No.

---------

Co-authored-by: Martin Grigorov <martin-g@users.noreply.github.com>

Author: neilconway

datafusion/functions/src/unicode/substrindex.rs

@@ -18,14 +18,14 @@
 use std::sync::Arc;
 
 use arrow::array::{
-    Array, ArrayRef, AsArray, ByteView, GenericStringArray, GenericStringBuilder,
-    OffsetSizeTrait, PrimitiveArray, StringArrayType, StringLikeArrayBuilder,
-    StringViewArray, make_view, new_null_array,
+    Array, ArrayRef, AsArray, ByteView, GenericStringArray, OffsetSizeTrait,
+    PrimitiveArray, StringArrayType, StringViewArray, make_view, new_null_array,
 };
 use arrow::buffer::ScalarBuffer;
 use arrow::datatypes::{DataType, Int64Type};
 use arrow_buffer::NullBuffer;
 
+use crate::strings::GenericStringArrayBuilder;
 use crate::utils::make_scalar_function;
 use datafusion_common::{
     Result, ScalarValue, exec_datafusion_err, exec_err, utils::take_function_args,
@@ -151,7 +151,7 @@ fn substr_index(args: &[ArrayRef]) -> Result<ArrayRef> {
                 string_array,
                 delimiter_array,
                 count_array,
-                GenericStringBuilder::<i32>::with_capacity(
+                GenericStringArrayBuilder::<i32>::with_capacity(
                     string_array.len(),
                     visible_string_bytes(string_array),
                 ),
@@ -165,7 +165,7 @@ fn substr_index(args: &[ArrayRef]) -> Result<ArrayRef> {
                 string_array,
                 delimiter_array,
                 count_array,
-                GenericStringBuilder::<i64>::with_capacity(
+                GenericStringArrayBuilder::<i64>::with_capacity(
                     string_array.len(),
                     visible_string_bytes(string_array),
                 ),
@@ -229,7 +229,7 @@ fn substr_index_scalar(
                 arr,
                 delimiter,
                 count,
-                GenericStringBuilder::<i32>::with_capacity(
+                GenericStringArrayBuilder::<i32>::with_capacity(
                     arr.len(),
                     visible_string_bytes(arr),
                 ),
@@ -241,7 +241,7 @@ fn substr_index_scalar(
                 arr,
                 delimiter,
                 count,
-                GenericStringBuilder::<i64>::with_capacity(
+                GenericStringArrayBuilder::<i64>::with_capacity(
                     arr.len(),
                     visible_string_bytes(arr),
                 ),
@@ -261,30 +261,37 @@ fn visible_string_bytes<T: OffsetSizeTrait>(
     offsets[offsets.len() - 1].as_usize() - offsets[0].as_usize()
 }
 
-fn substr_index_general<'a, S, B>(
+fn substr_index_general<'a, S, O>(
     string_array: S,
     delimiter_array: S,
     count_array: &PrimitiveArray<Int64Type>,
-    mut builder: B,
+    mut builder: GenericStringArrayBuilder<O>,
 ) -> Result<ArrayRef>
 where
     S: StringArrayType<'a> + Copy,
-    B: StringLikeArrayBuilder,
+    O: OffsetSizeTrait,
 {
-    for ((string, delimiter), n) in string_array
-        .iter()
-        .zip(delimiter_array.iter())
-        .zip(count_array.iter())
-    {
-        match (string, delimiter, n) {
-            (Some(string), Some(delimiter), Some(n)) => {
-                builder.append_value(substr_index_slice(string, delimiter, n));
-            }
-            _ => builder.append_null(),
+    let num_rows = string_array.len();
+    // Output is null IFF any input is null.
+    let nulls = NullBuffer::union(
+        NullBuffer::union(string_array.nulls(), delimiter_array.nulls()).as_ref(),
+        count_array.nulls(),
+    );
+
+    for i in 0..num_rows {
+        if nulls.as_ref().is_some_and(|n| n.is_null(i)) {
+            builder.append_placeholder();
+            continue;
         }
+        // SAFETY: `i < num_rows` and the union of input nulls is valid at i,
+        // so each input is also valid at i.
+        let string = unsafe { string_array.value_unchecked(i) };
+        let delimiter = unsafe { delimiter_array.value_unchecked(i) };
+        let n = unsafe { count_array.value_unchecked(i) };
+        builder.append_value(substr_index_slice(string, delimiter, n));
     }
 
-    Ok(Arc::new(builder.finish()) as ArrayRef)
+    Ok(Arc::new(builder.finish(nulls)?) as ArrayRef)
 }
 
 fn substr_index_view(
@@ -332,15 +339,15 @@ fn substr_index_view(
     }
 }
 
-fn substr_index_scalar_impl<'a, S, B>(
+fn substr_index_scalar_impl<'a, S, O>(
     string_array: S,
     delimiter: &str,
     count: i64,
-    builder: B,
+    builder: GenericStringArrayBuilder<O>,
 ) -> Result<ArrayRef>
 where
     S: StringArrayType<'a> + Copy,
-    B: StringLikeArrayBuilder,
+    O: OffsetSizeTrait,
 {
     if count == 0 || delimiter.is_empty() {
         return map_strings(string_array, builder, |string| &string[..0]);
@@ -465,19 +472,28 @@ fn substr_index_scalar_view(
     }
 }
 
-fn map_strings<'a, S, B, F>(string_array: S, mut builder: B, f: F) -> Result<ArrayRef>
+fn map_strings<'a, S, O, F>(
+    string_array: S,
+    mut builder: GenericStringArrayBuilder<O>,
+    f: F,
+) -> Result<ArrayRef>
 where
     S: StringArrayType<'a> + Copy,
-    B: StringLikeArrayBuilder,
+    O: OffsetSizeTrait,
     F: Fn(&'a str) -> &'a str,
 {
-    for string in string_array.iter() {
-        match string {
-            Some(s) => builder.append_value(f(s)),
-            None => builder.append_null(),
+    let nulls = string_array.nulls().cloned();
+    for i in 0..string_array.len() {
+        if nulls.as_ref().is_some_and(|n| n.is_null(i)) {
+            builder.append_placeholder();
+            continue;
         }
+        // SAFETY: `i < string_array.len()` and `nulls` is valid at i, so the
+        // input is also valid at i.
+        let s = unsafe { string_array.value_unchecked(i) };
+        builder.append_value(f(s));
     }
-    Ok(Arc::new(builder.finish()) as ArrayRef)
+    Ok(Arc::new(builder.finish(nulls)?) as ArrayRef)
 }
 
 #[inline]