Perform validation and null view replacement in two passes

vortex-array/Cargo.toml

@@ -238,6 +238,10 @@ harness = false
 name = "varbinview_try_new"
 harness = false
 
+[[bench]]
+name = "varbin_validate"
+harness = false
+
 [[bench]]
 name = "bool_zip"
 harness = false

vortex-array/benches/varbin_validate.rs

@@ -0,0 +1,152 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Compares `VarBinViewArray` and `VarBinArray` UTF-8 validation across string shapes and null
+//! densities.
+
+#![allow(clippy::unwrap_used)]
+
+use std::sync::Arc;
+use std::sync::LazyLock;
+
+use divan::Bencher;
+use mimalloc::MiMalloc;
+use vortex_array::IntoArray;
+use vortex_array::VortexSessionExecute;
+use vortex_array::array_session;
+use vortex_array::arrays::VarBinArray;
+use vortex_array::arrays::VarBinViewArray;
+use vortex_array::arrays::varbinview::BinaryView;
+use vortex_array::dtype::DType;
+use vortex_array::dtype::Nullability;
+use vortex_array::validity::Validity;
+use vortex_buffer::BitBuffer;
+use vortex_buffer::Buffer;
+use vortex_buffer::ByteBuffer;
+use vortex_session::VortexSession;
+
+#[global_allocator]
+static GLOBAL: MiMalloc = MiMalloc;
+
+fn main() {
+    divan::main();
+}
+
+const N: usize = 1 << 16;
+static SESSION: LazyLock<VortexSession> = LazyLock::new(array_session);
+
+/// `n` strings of `len` bytes, concatenated, plus their offsets.
+fn strings(n: usize, len: usize, multibyte: bool) -> (Vec<u8>, Vec<u32>) {
+    let unit = if multibyte {
+        "zażółć gęślą jaźń "
+    } else {
+        "abcdefghijklmnopqrstuvwxyz"
+    };
+    let mut bytes = Vec::new();
+    let mut offsets = vec![0u32];
+    for i in 0..n {
+        let s: String = unit.chars().cycle().skip(i % 7).take(len).collect();
+        bytes.extend_from_slice(s.as_bytes());
+        offsets.push(u32::try_from(bytes.len()).unwrap());
+    }
+    (bytes, offsets)
+}
+
+fn views(bytes: &[u8], offsets: &[u32], count: usize) -> Vec<BinaryView> {
+    offsets
+        .windows(2)
+        .take(count)
+        .map(|o| BinaryView::make_view(&bytes[o[0] as usize..o[1] as usize], 0, o[0]))
+        .collect()
+}
+
+/// `null_every == 0` means non-nullable.
+fn validity(n: usize, null_every: usize) -> Validity {
+    if null_every == 0 {
+        Validity::NonNullable
+    } else {
+        Validity::from_bit_buffer(
+            BitBuffer::from_iter((0..n).map(|i| i % null_every != 0)),
+            Nullability::Nullable,
+        )
+    }
+}
+
+fn nullability(null_every: usize) -> Nullability {
+    if null_every == 0 {
+        Nullability::NonNullable
+    } else {
+        Nullability::Nullable
+    }
+}
+
+fn run_vbv(bencher: Bencher, len: usize, multibyte: bool, null_every: usize, count: usize) {
+    let (bytes, offsets) = strings(N, len, multibyte);
+    let v = views(&bytes, &offsets, count);
+    let buffers: Arc<[ByteBuffer]> = Arc::new([ByteBuffer::from(bytes)]);
+    let dtype = DType::Utf8(nullability(null_every));
+    let validity = validity(count, null_every);
+    bencher
+        .with_inputs(|| {
+            (
+                Buffer::copy_from(&v),
+                Arc::clone(&buffers),
+                dtype.clone(),
+                validity.clone(),
+                SESSION.create_execution_ctx(),
+            )
+        })
+        .bench_values(|(views, buffers, dtype, validity, mut ctx)| {
+            VarBinViewArray::try_new(views, buffers, dtype, validity, &mut ctx).unwrap()
+        });
+}
+
+#[divan::bench(args = [0, 2, 10])]
+fn vbv_short20(bencher: Bencher, null_every: usize) {
+    run_vbv(bencher, 20, false, null_every, N);
+}
+
+#[divan::bench(args = [0, 2, 10])]
+fn vbv_long200(bencher: Bencher, null_every: usize) {
+    run_vbv(bencher, 200, false, null_every, N);
+}
+
+#[divan::bench(args = [0, 2])]
+fn vbv_multibyte24(bencher: Bencher, null_every: usize) {
+    run_vbv(bencher, 24, true, null_every, N);
+}
+
+/// 1024 views into the buffer of 65536 strings, as after a slice.
+#[divan::bench(args = [0, 2])]
+fn vbv_sliced_short20(bencher: Bencher, null_every: usize) {
+    run_vbv(bencher, 20, false, null_every, 1024);
+}
+
+fn run_varbin(bencher: Bencher, len: usize, null_every: usize) {
+    let (bytes, offsets) = strings(N, len, false);
+    let offsets: Vec<i32> = offsets
+        .into_iter()
+        .map(|o| i32::try_from(o).unwrap())
+        .collect();
+    let bytes = ByteBuffer::from(bytes);
+    let offsets = Buffer::from(offsets).into_array();
+    let dtype = DType::Utf8(nullability(null_every));
+    let validity = validity(N, null_every);
+    bencher
+        .with_inputs(|| {
+            (
+                offsets.clone(),
+                bytes.clone(),
+                dtype.clone(),
+                validity.clone(),
+            )
+        })
+        .bench_values(|(offsets, bytes, dtype, validity)| {
+            VarBinArray::try_new(offsets, bytes, dtype, validity).unwrap()
+        });
+}
+
+#[divan::bench(args = [0, 2, 10])]
+fn varbin_short20(bencher: Bencher, null_every: usize) {
+    run_varbin(bencher, 20, null_every);
+}

vortex-array/src/arrays/validation_tests.rs

@@ -223,4 +223,36 @@ mod tests {
         assert!(matches!(result, Err(VortexError::InvalidArgument(_, _))));
         assert!(result.is_err());
     }
+
+    #[rstest::rstest]
+    #[case::whole(vec![0, 2, 3], "éa".as_bytes().to_vec(), Validity::NonNullable, true)]
+    #[case::inside_a_char(vec![0, 1, 3], "éa".as_bytes().to_vec(), Validity::NonNullable, false)]
+    #[case::invalid_bytes(vec![0, 1, 2], vec![b'a', 0xFF], Validity::NonNullable, false)]
+    #[case::invalid_bytes_at_a_null(
+        vec![0, 1, 2, 3],
+        vec![b'a', 0xFF, b'b'],
+        Validity::from_iter([true, false, true]),
+        true
+    )]
+    #[case::decreasing_offsets(vec![0, 2, 1], "abc".as_bytes().to_vec(), Validity::NonNullable, false)]
+    #[case::offset_past_the_end(vec![0, 5, 1], "abc".as_bytes().to_vec(), Validity::NonNullable, false)]
+    fn test_varbin_utf8_validation(
+        #[case] offsets: Vec<i32>,
+        #[case] bytes: Vec<u8>,
+        #[case] validity: Validity,
+        #[case] ok: bool,
+    ) {
+        let nullability = if matches!(validity, Validity::NonNullable) {
+            Nullability::NonNullable
+        } else {
+            Nullability::Nullable
+        };
+        let result = VarBinArray::try_new(
+            Buffer::from(offsets).into_array(),
+            ByteBuffer::from(bytes),
+            DType::Utf8(nullability),
+            validity,
+        );
+        assert_eq!(result.is_ok(), ok);
+    }
 }

vortex-array/src/arrays/varbin/array.rs

@@ -32,6 +32,12 @@ use crate::legacy_session;
 use crate::match_each_integer_ptype;
 use crate::validity::Validity;
 
+/// Returns `true` if `index` is the end of `bytes` or the start of a UTF-8 char in it.
+fn is_char_boundary_at(bytes: &[u8], index: usize) -> bool {
+    // Continuation bytes have the form `0b10xx_xxxx`.
+    index == bytes.len() || bytes[index] & 0b1100_0000 != 0b1000_0000
+}
+
 #[array_slots(VarBin)]
 pub struct VarBinSlots {
     /// The offsets array defining the start/end of each variable-length binary element.
@@ -241,7 +247,12 @@ impl VarBinData {
     #[allow(clippy::disallowed_methods)]
     fn validate_utf8(offsets: &ArrayRef, bytes: &[u8], validity: &Validity) -> VortexResult<()> {
         let validate_at = |i: usize, start: usize, end: usize| -> VortexResult<()> {
-            let string_bytes = &bytes[start..end];
+            let string_bytes = bytes.get(start..end).ok_or_else(|| {
+                vortex_err!(
+                    InvalidArgument: "offsets {start}..{end} at index {i} are out of order or out of bounds for bytes of length {}",
+                    bytes.len()
+                )
+            })?;
             simdutf8::basic::from_utf8(string_bytes).map_err(|_| {
                 #[expect(clippy::unwrap_used)]
                 // run validation using `compat` package to get more detailed error message
@@ -277,6 +288,21 @@ impl VarBinData {
                 bytes.len()
             );
 
+            // When the offsets never decrease, the strings tile `bytes[first..last]`. If that range
+            // is valid UTF-8 as a whole, every string is valid UTF-8 if and only if every offset
+            // falls on a char boundary. Otherwise, for example for invalid bytes at a null, check
+            // the strings one by one.
+            let first_offset: usize = offsets_slice[0].as_();
+            if offsets_slice.windows(2).all(|o| o[0] <= o[1])
+                && first_offset <= last_offset
+                && simdutf8::basic::from_utf8(&bytes[first_offset..last_offset]).is_ok()
+                && offsets_slice
+                    .iter()
+                    .all(|&o| is_char_boundary_at(bytes, o.as_()))
+            {
+                return Ok(());
+            }
+
             for (i, (start, end)) in offsets_slice
                 .windows(2)
                 .map(|o| (o[0].as_(), o[1].as_()))