perf: Optimize reverse using bulk-NULL string builders (#21991)

## Which issue does this PR close?

- Closes #21990.

## Rationale for this change

Rather than computing NULLs on a per-row basis, use the bulk-NULL string
builders to compute them just by copying the input NULL bitmap. See
recent related PRs (#21990, #21853, etc.) for context.

Benchmarks:

  StringArray, ASCII
  - str_len=8:    61.16 µs → 50.33 µs   (−17.7%)
  - str_len=32:   73.36 µs → 68.78 µs   (−6.9%)
  - str_len=128:  93.82 µs → 94.03 µs   (+0.7%)
  - str_len=4096: 3.614 ms → 2.422 ms   (−33.0%)

  StringArray, UTF-8 (0.8)
  - str_len=8:    195.97 µs → 169.22 µs   (−13.4%)
  - str_len=32:   818.77 µs → 848.41 µs   (+4.1%)
  - str_len=128:  3.387 ms → 3.472 ms     (+2.5%)
  - str_len=4096: 116.88 ms → 117.57 ms   (+0.6%)

  StringViewArray, ASCII
  - str_len=8:    62.67 µs → 51.00 µs    (−19.0%)
  - str_len=32:   84.74 µs → 76.05 µs    (−9.8%)
  - str_len=128:  103.49 µs → 89.50 µs   (−13.0%)
  - str_len=4096: 2.011 ms → 2.021 ms    (+0.5%)

  StringViewArray, UTF-8 (0.8)
  - str_len=8:    207.43 µs → 183.39 µs  (−11.2%)
  - str_len=32:   840.96 µs → 850.70 µs  (+2.4%)
  - str_len=128:  3.408 ms → 3.462 ms    (+1.6%)
  - str_len=4096: 115.38 ms → 117.01 ms  (+1.4%)

## What changes are included in this PR?

* Switch to bulk-NULL string builders
* Add a unit test for `reverse` with array inputs and some NULL values,
to ensure that this case is handled correctly

## Are these changes tested?

Yes.

## Are there any user-facing changes?

No.

Author: neilconway

datafusion/functions/src/unicode/reverse.rs

@@ -15,14 +15,12 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::sync::Arc;
-
+use crate::strings::{
+    BulkNullStringArrayBuilder, GenericStringArrayBuilder, StringViewArrayBuilder,
+};
 use crate::utils::make_scalar_function;
 use DataType::{LargeUtf8, Utf8, Utf8View};
-use arrow::array::{
-    Array, ArrayRef, AsArray, LargeStringBuilder, StringArrayType, StringBuilder,
-    StringLikeArrayBuilder, StringViewBuilder,
-};
+use arrow::array::{Array, ArrayRef, AsArray, StringArrayType};
 use arrow::datatypes::DataType;
 use datafusion_common::{Result, exec_err};
 use datafusion_expr::{
@@ -103,56 +101,79 @@ fn reverse(args: &[ArrayRef]) -> Result<ArrayRef> {
     let len = args[0].len();
 
     match args[0].data_type() {
+        LargeUtf8 => reverse_impl(
+            &args[0].as_string::<i64>(),
+            GenericStringArrayBuilder::<i64>::with_capacity(len, 1024),
+        ),
         Utf8 => reverse_impl(
             &args[0].as_string::<i32>(),
-            StringBuilder::with_capacity(len, 1024),
+            GenericStringArrayBuilder::<i32>::with_capacity(len, 1024),
         ),
         Utf8View => reverse_impl(
             &args[0].as_string_view(),
-            StringViewBuilder::with_capacity(len),
-        ),
-        LargeUtf8 => reverse_impl(
-            &args[0].as_string::<i64>(),
-            LargeStringBuilder::with_capacity(len, 1024),
+            StringViewArrayBuilder::with_capacity(len),
         ),
         _ => unreachable!(
             "Reverse can only be applied to Utf8View, Utf8 and LargeUtf8 types"
         ),
     }
 }
 
-fn reverse_impl<'a, StringArrType, StringBuilderType>(
+fn reverse_impl<'a, StringArrType, B>(
     string_array: &StringArrType,
-    mut array_builder: StringBuilderType,
+    mut array_builder: B,
 ) -> Result<ArrayRef>
 where
     StringArrType: StringArrayType<'a>,
-    StringBuilderType: StringLikeArrayBuilder,
+    B: BulkNullStringArrayBuilder,
 {
+    let item_len = string_array.len();
+    // Null-preserving: reuse the input null buffer as the output null buffer.
+    let nulls = string_array.nulls().cloned();
     let mut string_buf = String::new();
     let mut byte_buf = Vec::<u8>::new();
 
-    for string in string_array.iter() {
-        if let Some(s) = string {
-            if s.is_ascii() {
-                // reverse bytes directly since ASCII characters are single bytes
-                byte_buf.extend(s.as_bytes());
-                byte_buf.reverse();
-                // SAFETY: Since the original string was ASCII, reversing the bytes still results in valid UTF-8.
-                let reversed = unsafe { std::str::from_utf8_unchecked(&byte_buf) };
-                array_builder.append_value(reversed);
-                byte_buf.clear();
+    if let Some(ref n) = nulls {
+        for i in 0..item_len {
+            if n.is_null(i) {
+                array_builder.append_placeholder();
             } else {
-                string_buf.extend(s.chars().rev());
-                array_builder.append_value(&string_buf);
-                string_buf.clear();
+                // SAFETY: `n.is_null(i)` was false in the branch above.
+                let s = unsafe { string_array.value_unchecked(i) };
+                append_reversed(s, &mut array_builder, &mut byte_buf, &mut string_buf);
             }
-        } else {
-            array_builder.append_null();
+        }
+    } else {
+        for i in 0..item_len {
+            // SAFETY: no null buffer means every index is valid.
+            let s = unsafe { string_array.value_unchecked(i) };
+            append_reversed(s, &mut array_builder, &mut byte_buf, &mut string_buf);
         }
     }
 
-    Ok(Arc::new(array_builder.finish()) as ArrayRef)
+    array_builder.finish(nulls)
+}
+
+#[inline]
+fn append_reversed<B: BulkNullStringArrayBuilder>(
+    s: &str,
+    builder: &mut B,
+    byte_buf: &mut Vec<u8>,
+    string_buf: &mut String,
+) {
+    if s.is_ascii() {
+        // reverse bytes directly since ASCII characters are single bytes
+        byte_buf.extend(s.as_bytes());
+        byte_buf.reverse();
+        // SAFETY: input was ASCII, so reversed bytes are still valid UTF-8.
+        let reversed = unsafe { std::str::from_utf8_unchecked(byte_buf) };
+        builder.append_value(reversed);
+        byte_buf.clear();
+    } else {
+        string_buf.extend(s.chars().rev());
+        builder.append_value(string_buf);
+        string_buf.clear();
+    }
 }
 
 #[cfg(test)]
@@ -213,4 +234,58 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_array_with_nulls() {
+        use crate::unicode::reverse::reverse;
+        use arrow::array::ArrayRef;
+        use std::sync::Arc;
+
+        let input_values = vec![Some("abcd"), None, Some("XYZ"), Some("héllo"), None];
+        let expected: Vec<Option<&str>> =
+            vec![Some("dcba"), None, Some("ZYX"), Some("olléh"), None];
+
+        let cases: Vec<(&str, ArrayRef)> = vec![
+            (
+                "StringArray",
+                Arc::new(StringArray::from(input_values.clone())),
+            ),
+            (
+                "LargeStringArray",
+                Arc::new(LargeStringArray::from(input_values.clone())),
+            ),
+            (
+                "StringViewArray",
+                Arc::new(StringViewArray::from(input_values.clone())),
+            ),
+        ];
+
+        for (label, input) in cases {
+            let out = reverse(&[input]).unwrap();
+            assert_eq!(out.len(), expected.len(), "{label}: length mismatch");
+
+            let actual: Vec<Option<&str>> = match out.data_type() {
+                Utf8 => out
+                    .as_any()
+                    .downcast_ref::<StringArray>()
+                    .unwrap()
+                    .iter()
+                    .collect(),
+                LargeUtf8 => out
+                    .as_any()
+                    .downcast_ref::<LargeStringArray>()
+                    .unwrap()
+                    .iter()
+                    .collect(),
+                Utf8View => out
+                    .as_any()
+                    .downcast_ref::<StringViewArray>()
+                    .unwrap()
+                    .iter()
+                    .collect(),
+                other => panic!("{label}: unexpected output type {other:?}"),
+            };
+            assert_eq!(actual, expected, "{label}: value mismatch");
+        }
+    }
 }