Validate VarBin arrays with an explicit execution context (#10593)

- The VarBin constructors take an `ExecutionCtx` to validate offsets and
UTF-8 data. This covers `VarBinArray::{new, try_new}` and
`VarBinData::{build, build_from_handle, try_build,
try_build_from_handle, validate}`.
- UTF-8 validation stays complete, because other code reads these bytes
with `from_utf8_unchecked`.
- Deserialization validates with a context from its session.
- `offset_at`, `bytes_at` and `sliced_bytes` take a ctx.
- The VarBin and FSST nullability casts don't change offsets or bytes,
so they rebuild with `new_unchecked_from_handle` instead of
revalidating.
- The builder's debug sortedness check reads its own offsets slice
directly.

part of #9446

Author: robert3005

encodings/fsst/benches/fsst_url_compare.rs

@@ -70,7 +70,7 @@ fn pick_url_with_domain(data: &VarBinArray, domain: &str) -> String {
         .unwrap();
     (0..data.len())
         .filter(|&i| mask.value(i))
-        .map(|i| unsafe { String::from_utf8_unchecked(data.bytes_at(i).to_vec()) })
+        .map(|i| unsafe { String::from_utf8_unchecked(data.bytes_at(i, &mut ctx).to_vec()) })
         .find(|u| u.contains(domain))
         .unwrap_or_else(|| format!("http://{domain}/missing"))
 }

encodings/fsst/src/canonical.rs

@@ -301,7 +301,7 @@ mod tests {
             let arr = builder.finish_into_varbin();
             let mask = arr.validity()?.execute_mask(arr.len(), &mut ctx)?;
             let actual = (0..arr.len())
-                .map(|i| mask.value(i).then(|| arr.bytes_at(i).to_vec()))
+                .map(|i| mask.value(i).then(|| arr.bytes_at(i, &mut ctx).to_vec()))
                 .collect::<Vec<_>>();
             assert_eq!(data, actual);
         }
@@ -368,6 +368,7 @@ mod tests {
             encoded.codes_bytes().clone(),
             DType::Binary(Nullability::NonNullable),
             Validity::NonNullable,
+            &mut ctx,
         );
         let sliced = FSST::try_new_with_symbol_table(
             encoded.dtype().clone(),

encodings/fsst/src/compute/cast.rs

@@ -23,12 +23,16 @@ fn build_with_codes_validity(
     new_codes_validity: Validity,
 ) -> VortexResult<ArrayRef> {
     let codes = array.codes();
-    let new_codes = VarBinArray::try_new(
-        codes.offsets().clone(),
-        codes.bytes().clone(),
-        codes.dtype().with_nullability(dtype.nullability()),
-        new_codes_validity,
-    )?;
+    // SAFETY: the offsets and bytes are unchanged from valid codes, and the validity was cast
+    // from the codes' own validity, so it has the same length.
+    let new_codes = unsafe {
+        VarBinArray::new_unchecked_from_handle(
+            codes.offsets().clone(),
+            codes.bytes_handle().clone(),
+            codes.dtype().with_nullability(dtype.nullability()),
+            new_codes_validity,
+        )
+    };
 
     Ok(unsafe {
         FSST::new_unchecked_with_symbol_table(

encodings/zstd/src/array.rs

@@ -1610,6 +1610,8 @@ impl OperationsVTable<Zstd> for Zstd {
 #[expect(clippy::cast_possible_truncation)]
 mod tests {
     use rstest::rstest;
+    use vortex_array::VortexSessionExecute;
+    use vortex_array::array_session;
     use vortex_array::arrays::varbin::VarBinArrayExt as _;
     use vortex_array::builders::VarBinBuilder;
     use vortex_array::dtype::DType;
@@ -1779,8 +1781,9 @@ mod tests {
         append_slice_to_varbin(&slice, &Mask::new_true(2), &mut builder)?;
 
         let appended = builder.finish_into_varbin();
-        assert_eq!(appended.bytes_at(0).as_slice(), b"hello");
-        assert_eq!(appended.bytes_at(1).as_slice(), b"world");
+        let mut ctx = array_session().create_execution_ctx();
+        assert_eq!(appended.bytes_at(0, &mut ctx).as_slice(), b"hello");
+        assert_eq!(appended.bytes_at(1, &mut ctx).as_slice(), b"world");
         Ok(())
     }
 

vortex-array/benches/dict_compare.rs

@@ -77,7 +77,7 @@ fn bench_compare_varbin(bencher: divan::Bencher, (len, uniqueness): (usize, usiz
     let mut ctx = SESSION.create_execution_ctx();
     let varbin_arr = VarBinArray::from(gen_varbin_words(len, uniqueness));
     let dict = dict_encode(&varbin_arr.clone().into_array(), &mut ctx).unwrap();
-    let const_bytes = varbin_arr.bytes_at(0);
+    let const_bytes = varbin_arr.bytes_at(0, &mut ctx);
     let value = unsafe { str::from_utf8_unchecked(const_bytes.as_slice()) };
 
     bencher
@@ -155,7 +155,7 @@ fn bench_compare_sliced_dict_varbinview(
     let varbin_arr = VarBinArray::from(gen_varbin_words(codes_len.max(values_len), values_len));
     let dict = dict_encode(&varbin_arr.clone().into_array(), &mut ctx).unwrap();
     let dict = dict.into_array().slice(0..codes_len).unwrap();
-    let const_bytes = varbin_arr.bytes_at(0);
+    let const_bytes = varbin_arr.bytes_at(0, &mut ctx);
     let value = unsafe { str::from_utf8_unchecked(const_bytes.as_slice()) };
 
     bencher

vortex-array/src/arrays/validation_tests.rs

@@ -151,6 +151,7 @@ mod tests {
         #[case] validity: Validity,
         #[case] ok: bool,
     ) {
+        let mut ctx = array_session().create_execution_ctx();
         let nullability = if matches!(validity, Validity::NonNullable) {
             Nullability::NonNullable
         } else {
@@ -161,6 +162,7 @@ mod tests {
             ByteBuffer::from(bytes),
             DType::Utf8(nullability),
             validity,
+            &mut ctx,
         );
         assert_eq!(result.is_ok(), ok);
     }

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

@@ -18,7 +18,6 @@ use vortex_error::vortex_err;
 use crate::ArrayRef;
 use crate::ArraySlots;
 use crate::ExecutionCtx;
-use crate::VortexSessionExecute;
 use crate::aggregate_fn::NumericalAggregateOpts;
 use crate::aggregate_fn::fns::min_max::min_max;
 use crate::array::Array;
@@ -33,7 +32,6 @@ use crate::buffer::BufferHandle;
 use crate::dtype::DType;
 use crate::dtype::Nullability;
 use crate::dtype::OffsetBuilderPType;
-use crate::legacy_session;
 use crate::match_each_integer_ptype;
 use crate::validity::Validity;
 
@@ -102,8 +100,14 @@ impl VarBinData {
     ///
     /// Panics if the provided components do not satisfy the invariants documented
     /// in `VarBinArray::new_unchecked`.
-    pub fn build(offsets: ArrayRef, bytes: ByteBuffer, dtype: DType, validity: Validity) -> Self {
-        Self::try_build(offsets, bytes, dtype, validity).vortex_expect("VarBinArray new")
+    pub fn build(
+        offsets: ArrayRef,
+        bytes: ByteBuffer,
+        dtype: DType,
+        validity: Validity,
+        ctx: &mut ExecutionCtx,
+    ) -> Self {
+        Self::try_build(offsets, bytes, dtype, validity, ctx).vortex_expect("VarBinArray new")
     }
 
     /// Creates a new `VarBinArray`.
@@ -117,8 +121,10 @@ impl VarBinData {
         bytes: BufferHandle,
         dtype: DType,
         validity: Validity,
+        ctx: &mut ExecutionCtx,
     ) -> Self {
-        Self::try_build_from_handle(offset, bytes, dtype, validity).vortex_expect("VarBinArray new")
+        Self::try_build_from_handle(offset, bytes, dtype, validity, ctx)
+            .vortex_expect("VarBinArray new")
     }
 
     pub(crate) fn make_slots(offsets: ArrayRef, validity: &Validity, len: usize) -> ArraySlots {
@@ -142,9 +148,10 @@ impl VarBinData {
         bytes: ByteBuffer,
         dtype: DType,
         validity: Validity,
+        ctx: &mut ExecutionCtx,
     ) -> VortexResult<Self> {
         let bytes = BufferHandle::new_host(bytes);
-        Self::validate(&offsets, &bytes, &dtype, &validity)?;
+        Self::validate(&offsets, &bytes, &dtype, &validity, ctx)?;
 
         // SAFETY: validate ensures all invariants are met.
         Ok(unsafe { Self::new_unchecked_from_handle(bytes) })
@@ -164,8 +171,9 @@ impl VarBinData {
         bytes: BufferHandle,
         dtype: DType,
         validity: Validity,
+        ctx: &mut ExecutionCtx,
     ) -> VortexResult<Self> {
-        Self::validate(&offsets, &bytes, &dtype, &validity)?;
+        Self::validate(&offsets, &bytes, &dtype, &validity, ctx)?;
 
         // SAFETY: validate ensures all invariants are met.
         Ok(unsafe { Self::new_unchecked_from_handle(bytes) })
@@ -218,12 +226,12 @@ impl VarBinData {
     /// Validates the components that would be used to create a `VarBinArray`.
     ///
     /// This function checks all the invariants required by `VarBinArray::new_unchecked`.
-    #[allow(clippy::disallowed_methods)]
     pub fn validate(
         offsets: &ArrayRef,
         bytes: &BufferHandle,
         dtype: &DType,
         validity: &Validity,
+        ctx: &mut ExecutionCtx,
     ) -> VortexResult<()> {
         // Check offsets are non-nullable integer
         vortex_ensure!(
@@ -267,10 +275,9 @@ impl VarBinData {
             && matches!(dtype, DType::Utf8(_))
             && let Some(bytes) = bytes.as_host_opt()
         {
-            Self::validate_utf8(offsets, bytes.as_ref(), validity)?;
+            Self::validate_utf8(offsets, bytes.as_ref(), validity, ctx)?;
         } else if offsets.is_host() {
-            let mut ctx = legacy_session().create_execution_ctx();
-            Self::validate_offsets(offsets, bytes.len(), &mut ctx)?;
+            Self::validate_offsets(offsets, bytes.len(), ctx)?;
         }
 
         Ok(())
@@ -328,8 +335,12 @@ impl VarBinData {
     }
 
     /// Validates that every non-null value is valid UTF-8.
-    #[allow(clippy::disallowed_methods)]
-    fn validate_utf8(offsets: &ArrayRef, bytes: &[u8], validity: &Validity) -> VortexResult<()> {
+    fn validate_utf8(
+        offsets: &ArrayRef,
+        bytes: &[u8],
+        validity: &Validity,
+        ctx: &mut ExecutionCtx,
+    ) -> VortexResult<()> {
         let validate_at = |i: usize, start: usize, end: usize| -> VortexResult<()> {
             let string_bytes = bytes.get(start..end).ok_or_else(|| {
                 vortex_err!(
@@ -346,16 +357,15 @@ impl VarBinData {
             Ok(())
         };
 
-        let mut ctx = legacy_session().create_execution_ctx();
         // TODO(joe): update the created VarBin with this decompressed Array.
-        let primitive_offsets = offsets.clone().execute::<PrimitiveArray>(&mut ctx)?;
+        let primitive_offsets = offsets.clone().execute::<PrimitiveArray>(ctx)?;
 
         // Array-backed validity is the only variant that needs an execution context: execute it into
         // a mask once. The constant variants resolve null-ness without one. Resolving this before
         // the per-type dispatch keeps the dtype loop simple.
         let mask = match validity {
             Validity::Array(_) => {
-                Some(validity.execute_mask(primitive_offsets.len().saturating_sub(1), &mut ctx)?)
+                Some(validity.execute_mask(primitive_offsets.len().saturating_sub(1), ctx)?)
             }
             _ => None,
         };
@@ -378,7 +388,7 @@ impl VarBinData {
 
             // The fast path proves the offsets are valid. The per-string check below skips null
             // rows, so it cannot find bad offsets at those rows.
-            Self::validate_offsets(offsets, bytes.len(), &mut ctx)?;
+            Self::validate_offsets(offsets, bytes.len(), ctx)?;
 
             // Invalid bytes at a null row fail the check above, so check valid strings one by one.
             for (i, (start, end)) in offsets_slice
@@ -438,8 +448,7 @@ pub trait VarBinArrayExt: VarBinArraySlotsExt {
         )
     }
 
-    #[allow(clippy::disallowed_methods)]
-    fn offset_at(&self, index: usize) -> usize {
+    fn offset_at(&self, index: usize, ctx: &mut ExecutionCtx) -> usize {
         assert!(
             index <= self.len(),
             "Index {index} out of bounds 0..={}",
@@ -448,21 +457,21 @@ pub trait VarBinArrayExt: VarBinArraySlotsExt {
 
         (&self
             .offsets()
-            .execute_scalar(index, &mut legacy_session().create_execution_ctx())
+            .execute_scalar(index, ctx)
             .vortex_expect("offsets must support execute_scalar"))
             .try_into()
             .vortex_expect("Failed to convert offset to usize")
     }
 
-    fn bytes_at(&self, index: usize) -> ByteBuffer {
-        let start = self.offset_at(index);
-        let end = self.offset_at(index + 1);
+    fn bytes_at(&self, index: usize, ctx: &mut ExecutionCtx) -> ByteBuffer {
+        let start = self.offset_at(index, ctx);
+        let end = self.offset_at(index + 1, ctx);
         self.bytes().slice(start..end)
     }
 
-    fn sliced_bytes(&self) -> ByteBuffer {
-        let first_offset: usize = self.offset_at(0);
-        let last_offset = self.offset_at(self.len());
+    fn sliced_bytes(&self, ctx: &mut ExecutionCtx) -> ByteBuffer {
+        let first_offset: usize = self.offset_at(0, ctx);
+        let last_offset = self.offset_at(self.len(), ctx);
         self.bytes().slice(first_offset..last_offset)
     }
 }
@@ -567,7 +576,13 @@ impl Array<VarBin> {
 
 impl Array<VarBin> {
     /// Creates a new `VarBinArray`.
-    pub fn new(offsets: ArrayRef, bytes: ByteBuffer, dtype: DType, validity: Validity) -> Self {
+    pub fn new(
+        offsets: ArrayRef,
+        bytes: ByteBuffer,
+        dtype: DType,
+        validity: Validity,
+        ctx: &mut ExecutionCtx,
+    ) -> Self {
         let len = offsets.len().saturating_sub(1);
         let slots = VarBinData::make_slots(offsets, &validity, len);
         let data = VarBinData::build(
@@ -578,6 +593,7 @@ impl Array<VarBin> {
             bytes,
             dtype.clone(),
             validity,
+            ctx,
         );
         unsafe { Array::from_parts_unchecked(ArrayParts::new(VarBin, dtype, len, data, slots)) }
     }
@@ -616,16 +632,18 @@ impl Array<VarBin> {
         unsafe { Array::from_parts_unchecked(ArrayParts::new(VarBin, dtype, len, data, slots)) }
     }
 
-    /// Constructs a new `VarBinArray`.
+    /// Constructs a new `VarBinArray`, using `ctx` to validate the offsets and, for UTF-8 arrays,
+    /// the string data.
     pub fn try_new(
         offsets: ArrayRef,
         bytes: ByteBuffer,
         dtype: DType,
         validity: Validity,
+        ctx: &mut ExecutionCtx,
     ) -> VortexResult<Self> {
         let len = offsets.len() - 1;
         let bytes = BufferHandle::new_host(bytes);
-        VarBinData::validate(&offsets, &bytes, &dtype, &validity)?;
+        VarBinData::validate(&offsets, &bytes, &dtype, &validity, ctx)?;
         let slots = VarBinData::make_slots(offsets, &validity, len);
         // SAFETY: validate ensures all invariants are met.
         let data = unsafe { VarBinData::new_unchecked_from_handle(bytes) };

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

@@ -25,8 +25,6 @@ use crate::ArrayView;
 use crate::Canonical;
 use crate::ExecutionCtx;
 use crate::IntoArray;
-#[cfg(debug_assertions)]
-use crate::VortexSessionExecute;
 use crate::arrays::PrimitiveArray;
 use crate::arrays::VarBin;
 use crate::arrays::VarBinArray;
@@ -39,8 +37,6 @@ use crate::dtype::DType;
 use crate::dtype::OffsetBuilderPType;
 use crate::expr::stats::Precision;
 use crate::expr::stats::Stat;
-#[cfg(debug_assertions)]
-use crate::legacy_session;
 use crate::match_each_integer_ptype;
 use crate::scalar::Scalar;
 use crate::validity::Validity;
@@ -352,7 +348,7 @@ impl<O: OffsetBuilderPType> VarBinBuilder<O> {
         usize: AsPrimitive<O>,
     {
         let offsets = array.offsets().clone().execute::<PrimitiveArray>(ctx)?;
-        let bytes: ByteBuffer = array.sliced_bytes();
+        let bytes: ByteBuffer = array.sliced_bytes(ctx);
         let validity = array
             .varbin_validity()
             .execute_mask(array.as_ref().len(), ctx)?;
@@ -406,7 +402,6 @@ impl<O: OffsetBuilderPType> VarBinBuilder<O> {
     }
 
     /// Finishes the appended values into a [`VarBinArray`] and resets the builder.
-    #[allow(clippy::disallowed_methods)]
     pub fn finish_into_varbin(&mut self) -> VarBinArray {
         assert_eq!(
             self.offsets.len() - 1,
@@ -424,14 +419,10 @@ impl<O: OffsetBuilderPType> VarBinBuilder<O> {
 
         // The builder adds offsets in monotonically increasing order. Store this statistic to
         // prevent VarBinArray::validate from recomputing it after deserialization.
-        #[cfg(debug_assertions)]
-        {
-            let offsets_are_sorted = offsets
-                .statistics()
-                .compute_is_sorted(&mut legacy_session().create_execution_ctx())
-                .unwrap_or(false);
-            debug_assert!(offsets_are_sorted, "VarBinBuilder offsets must be sorted");
-        }
+        debug_assert!(
+            offsets.as_slice::<O>().is_sorted(),
+            "VarBinBuilder offsets must be sorted"
+        );
         offsets
             .statistics()
             .set(Stat::IsSorted, Precision::Exact(true.into()));