vortex_array/arrays/varbin/

array.rs

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright the Vortex contributors

use std::fmt::Display;
use std::fmt::Formatter;

use num_traits::AsPrimitive;
use vortex_array::arrays::PrimitiveArray;
use vortex_buffer::BufferAllocatorRef;
use vortex_buffer::ByteBuffer;
use vortex_error::VortexExpect;
use vortex_error::VortexResult;
use vortex_error::vortex_bail;
use vortex_error::vortex_ensure;
use vortex_error::vortex_ensure_eq;
use vortex_error::vortex_err;

use crate::ArrayRef;
use crate::ArraySlots;
use crate::ExecutionCtx;
use crate::aggregate_fn::NumericalAggregateOpts;
use crate::aggregate_fn::fns::min_max::min_max;
use crate::array::Array;
use crate::array::ArrayParts;
use crate::array::TypedArrayRef;
use crate::array::child_to_validity;
use crate::array::validity_to_child;
use crate::array_slots;
use crate::arrays::VarBin;
use crate::arrays::varbin::builder::VarBinBuilder;
use crate::buffer::BufferHandle;
use crate::dtype::DType;
use crate::dtype::Nullability;
use crate::dtype::OffsetBuilderPType;
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
}

/// Returns `true` if `offsets` never decrease, end within `bytes`, and delimit valid UTF-8
/// strings, including the strings at null rows.
///
/// 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. This costs one UTF-8 pass over the range instead of one call per string.
pub(crate) fn offsets_tile_utf8<O: AsPrimitive<usize> + PartialOrd>(
    offsets: &[O],
    bytes: &[u8],
) -> bool {
    let (Some(first), Some(last)) = (offsets.first(), offsets.last()) else {
        return false;
    };
    let (first, last): (usize, usize) = (first.as_(), last.as_());

    first <= last
        && last <= bytes.len()
        && offsets.windows(2).all(|pair| pair[0] <= pair[1])
        && simdutf8::basic::from_utf8(&bytes[first..last]).is_ok()
        && offsets
            .iter()
            .all(|&offset| is_char_boundary_at(bytes, offset.as_()))
}

#[array_slots(VarBin)]
pub struct VarBinSlots {
    /// The offsets array defining the start/end of each variable-length binary element.
    #[slot(0)]
    pub offsets: ArrayRef,
    /// The validity bitmap indicating which elements are non-null.
    #[slot(1)]
    pub validity: Option<ArrayRef>,
}

#[derive(Clone, Debug)]
pub struct VarBinData {
    pub(super) bytes: BufferHandle,
}

impl Display for VarBinData {
    fn fmt(&self, _f: &mut Formatter<'_>) -> std::fmt::Result {
        Ok(())
    }
}

pub struct VarBinDataParts {
    pub dtype: DType,
    pub bytes: BufferHandle,
    pub offsets: ArrayRef,
    pub validity: Validity,
}

impl VarBinData {
    /// Creates a new `VarBinArray`.
    ///
    /// # Panics
    ///
    /// 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,
        ctx: &mut ExecutionCtx,
    ) -> Self {
        Self::try_build(offsets, bytes, dtype, validity, ctx).vortex_expect("VarBinArray new")
    }

    /// Creates a new `VarBinArray`.
    ///
    /// # Panics
    ///
    /// Panics if the provided components do not satisfy the invariants documented
    /// in `VarBinArray::new_unchecked`.
    pub fn build_from_handle(
        offset: ArrayRef,
        bytes: BufferHandle,
        dtype: DType,
        validity: Validity,
        ctx: &mut ExecutionCtx,
    ) -> Self {
        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 {
        VarBinSlots {
            offsets,
            validity: validity_to_child(validity, len),
        }
        .into_slots()
    }

    /// Constructs a new `VarBinArray`.
    ///
    /// See `VarBinArray::new_unchecked` for more information.
    ///
    /// # Errors
    ///
    /// Returns an error if the provided components do not satisfy the invariants documented in
    /// `VarBinArray::new_unchecked`.
    pub fn try_build(
        offsets: ArrayRef,
        bytes: ByteBuffer,
        dtype: DType,
        validity: Validity,
        ctx: &mut ExecutionCtx,
    ) -> VortexResult<Self> {
        let bytes = BufferHandle::new_host(bytes);
        Self::validate(&offsets, &bytes, &dtype, &validity, ctx)?;

        // SAFETY: validate ensures all invariants are met.
        Ok(unsafe { Self::new_unchecked_from_handle(bytes) })
    }

    /// Constructs a new `VarBinArray` from a `BufferHandle` of memory that may exist
    /// on the CPU or GPU.
    ///
    /// See `VarBinArray::new_unchecked` for more information.
    ///
    /// # Errors
    ///
    /// Returns an error if the provided components do not satisfy the invariants documented in
    /// `VarBinArray::new_unchecked`.
    pub fn try_build_from_handle(
        offsets: ArrayRef,
        bytes: BufferHandle,
        dtype: DType,
        validity: Validity,
        ctx: &mut ExecutionCtx,
    ) -> VortexResult<Self> {
        Self::validate(&offsets, &bytes, &dtype, &validity, ctx)?;

        // SAFETY: validate ensures all invariants are met.
        Ok(unsafe { Self::new_unchecked_from_handle(bytes) })
    }

    /// Creates a new `VarBinArray` without validation from these components:
    ///
    /// * `offsets` is an array of byte offsets into the `bytes` buffer.
    /// * `bytes` is a buffer containing all the variable-length data concatenated.
    /// * `dtype` specifies whether this contains UTF-8 strings or binary data.
    /// * `validity` holds the null values.
    ///
    /// # Safety
    ///
    /// The caller must ensure all of the following invariants are satisfied:
    ///
    /// ## Offsets Requirements
    ///
    /// - `offsets` must be a non-nullable integer array.
    /// - `offsets` must contain at least 1 element (for empty array, it contains \[0\]).
    /// - All values in `offsets` must be monotonically non-decreasing.
    /// - All values in `offsets` must be non-negative.
    /// - No offset value may exceed `bytes.len()`.
    ///
    /// ## Type Requirements
    ///
    /// - `dtype` must be exactly [`DType::Binary`] or [`DType::Utf8`].
    /// - If `dtype` is [`DType::Utf8`], every byte slice `bytes[offsets[i]..offsets[i+1]]` must be valid UTF-8.
    /// - `dtype.is_nullable()` must match the nullability of `validity`.
    ///
    /// ## Validity Requirements
    ///
    /// - If `validity` is [`Validity::Array`], its length must exactly equal `offsets.len() - 1`.
    pub unsafe fn new_unchecked(bytes: ByteBuffer) -> Self {
        // SAFETY: `new_unchecked_from_handle` has same invariants which should be checked
        //  by caller.
        unsafe { Self::new_unchecked_from_handle(BufferHandle::new_host(bytes)) }
    }

    /// Creates a new `VarBinArray` without validation from its components, with string data
    /// stored in a `BufferHandle` (CPU or GPU).
    ///
    /// # Safety
    ///
    /// The caller must ensure all the invariants documented in `new_unchecked` are satisfied.
    pub unsafe fn new_unchecked_from_handle(bytes: BufferHandle) -> Self {
        Self { bytes }
    }

    /// Validates the components that would be used to create a `VarBinArray`.
    ///
    /// This function checks all the invariants required by `VarBinArray::new_unchecked`.
    pub fn validate(
        offsets: &ArrayRef,
        bytes: &BufferHandle,
        dtype: &DType,
        validity: &Validity,
        ctx: &mut ExecutionCtx,
    ) -> VortexResult<()> {
        // Check offsets are non-nullable integer
        vortex_ensure!(
            offsets.dtype().is_int() && !offsets.dtype().is_nullable(),
            MismatchedTypes: "non nullable int", offsets.dtype()
        );

        // Check dtype is Binary or Utf8
        vortex_ensure!(
            matches!(dtype, DType::Binary(_) | DType::Utf8(_)),
            MismatchedTypes: "utf8 or binary", dtype
        );

        // Check nullability matches
        vortex_ensure!(
            dtype.is_nullable() != matches!(validity, Validity::NonNullable),
            InvalidArgument: "incorrect validity {:?} for dtype {}",
            validity,
            dtype
        );

        // Check offsets has at least one element
        vortex_ensure!(
            !offsets.is_empty(),
            InvalidArgument: "Offsets must have at least one element"
        );

        // Check validity length
        if let Some(validity_len) = validity.maybe_len() {
            vortex_ensure_eq!(
                validity_len,
                offsets.len() - 1,
                "Validity length doesn't match array length",
            );
        }

        // Validate UTF-8 for Utf8 dtype, which also validates the offsets. Skip when offsets/bytes
        // are not host-resident. Offsets on a device cannot be read here.
        if offsets.is_host()
            && bytes.is_on_host()
            && matches!(dtype, DType::Utf8(_))
            && let Some(bytes) = bytes.as_host_opt()
        {
            Self::validate_utf8(offsets, bytes.as_ref(), validity, ctx)?;
        } else if offsets.is_host() {
            Self::validate_offsets(offsets, bytes.len(), ctx)?;
        }

        Ok(())
    }

    /// Validates that the offsets are sorted, non-negative and do not exceed `bytes_len`.
    fn validate_offsets(
        offsets: &ArrayRef,
        bytes_len: usize,
        ctx: &mut ExecutionCtx,
    ) -> VortexResult<()> {
        // Offsets must be sorted (but not strictly sorted, empty values are allowed)
        let Some(is_sorted) = offsets.statistics().compute_is_sorted(ctx) else {
            vortex_bail!(InvalidArgument: "offsets must report is_sorted statistic");
        };
        vortex_ensure!(is_sorted, InvalidArgument: "offsets must be sorted");

        // Validate that offsets min is non-negative, and max does not exceed the length of
        // the bytes buffer.
        let Some(min_max) = min_max(offsets, ctx, NumericalAggregateOpts::default())? else {
            vortex_bail!(
                InvalidArgument: "offsets array with encoding {} must support min_max compute function",
                offsets.encoding_id()
            );
        };

        match_each_integer_ptype!(offsets.dtype().as_ptype(), |P| {
            #[allow(clippy::absurd_extreme_comparisons, unused_comparisons)]
            {
                let max = min_max
                    .max
                    .as_primitive()
                    .as_::<P>()
                    .vortex_expect("offsets type must fit offsets values");
                let min = min_max
                    .min
                    .as_primitive()
                    .as_::<P>()
                    .vortex_expect("offsets type must fit offsets values");

                vortex_ensure!(
                    min >= 0,
                    InvalidArgument: "offsets minimum {min} outside valid range [0, {max}]"
                );

                // An offset type too narrow for the bytes length cannot exceed it.
                vortex_ensure!(
                    P::try_from(bytes_len).ok().is_none_or(|len| max <= len),
                    InvalidArgument: "Max offset {max} is beyond the length of the bytes buffer {bytes_len}"
                );
            }
        });

        Ok(())
    }

    /// Validates that every non-null value is valid UTF-8.
    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!(
                    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
                let err = simdutf8::compat::from_utf8(string_bytes).unwrap_err();
                vortex_err!("invalid utf-8: {err} at index {i}")
            })?;
            Ok(())
        };

        // TODO(joe): update the created VarBin with this decompressed Array.
        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), ctx)?)
            }
            _ => None,
        };
        let all_invalid = validity.definitely_all_null();

        match_each_integer_ptype!(primitive_offsets.dtype().as_ptype(), |O| {
            let offsets_slice = primitive_offsets.as_slice::<O>();

            let last_offset: usize = offsets_slice[offsets_slice.len() - 1].as_();
            vortex_ensure!(
                last_offset <= bytes.len(),
                InvalidArgument: "Last offset {} exceeds bytes length {}",
                last_offset,
                bytes.len()
            );

            if offsets_tile_utf8(offsets_slice, bytes) {
                return Ok(());
            }

            // 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(), 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
                .windows(2)
                .map(|o| (o[0].as_(), o[1].as_()))
                .enumerate()
            {
                let valid = mask.as_ref().map_or(!all_invalid, |mask| mask.value(i));
                if valid {
                    validate_at(i, start, end)?;
                }
            }
        });
        Ok(())
    }

    /// Access the value bytes child buffer
    ///
    /// # Note
    ///
    /// Bytes child buffer is never sliced when the array is sliced so this can include values
    /// that are not logically present in the array. Users should prefer `sliced_bytes`
    /// unless they're resolving values via the offset child array.
    #[inline]
    pub fn bytes(&self) -> &ByteBuffer {
        self.bytes.as_host()
    }

    /// Access the value bytes buffer handle.
    #[inline]
    pub fn bytes_handle(&self) -> &BufferHandle {
        &self.bytes
    }
}

pub trait VarBinArrayExt: VarBinArraySlotsExt {
    fn dtype_parts(&self) -> (bool, Nullability) {
        match self.dtype() {
            DType::Utf8(nullability) => (true, *nullability),
            DType::Binary(nullability) => (false, *nullability),
            _ => unreachable!("VarBinArrayExt requires a utf8 or binary dtype"),
        }
    }

    fn is_utf8(&self) -> bool {
        self.dtype_parts().0
    }

    fn nullability(&self) -> Nullability {
        self.dtype_parts().1
    }

    fn varbin_validity(&self) -> Validity {
        child_to_validity(
            self.slots()[VarBinSlots::VALIDITY].as_ref(),
            self.nullability(),
        )
    }

    fn offset_at(&self, index: usize, ctx: &mut ExecutionCtx) -> usize {
        assert!(
            index <= self.len(),
            "Index {index} out of bounds 0..={}",
            self.len()
        );

        (&self
            .offsets()
            .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, 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, 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)
    }
}
impl<T: TypedArrayRef<VarBin>> VarBinArrayExt for T {}

/// Forwarding constructors for `VarBinArray` (= `Array<VarBin>`).
impl Array<VarBin> {
    pub fn from_vec<T: AsRef<[u8]>>(vec: Vec<T>, dtype: DType) -> Self {
        let size: usize = vec.iter().map(|v| v.as_ref().len()).sum();
        if size < u32::MAX as usize {
            Self::from_vec_sized::<u32, T>(vec, dtype)
        } else {
            Self::from_vec_sized::<u64, T>(vec, dtype)
        }
    }

    #[expect(
        clippy::same_name_method,
        reason = "intentionally named from_iter like Iterator::from_iter"
    )]
    pub fn from_iter<T: AsRef<[u8]>, I: IntoIterator<Item = Option<T>>>(
        iter: I,
        dtype: DType,
    ) -> Self {
        let iter = iter.into_iter();
        let mut builder = VarBinBuilder::<u32>::with_capacity_in(
            dtype,
            iter.size_hint().0,
            BufferAllocatorRef::static_ref(),
        );
        for v in iter {
            builder.append(v.as_ref().map(|o| o.as_ref()));
        }
        builder.finish_into_varbin()
    }

    pub fn from_iter_nonnull<T: AsRef<[u8]>, I: IntoIterator<Item = T>>(
        iter: I,
        dtype: DType,
    ) -> Self {
        let iter = iter.into_iter();
        let mut builder = VarBinBuilder::<u32>::with_capacity_in(
            dtype,
            iter.size_hint().0,
            BufferAllocatorRef::static_ref(),
        );
        for v in iter {
            builder.append_value(v);
        }
        builder.finish_into_varbin()
    }

    fn from_vec_sized<O, T>(vec: Vec<T>, dtype: DType) -> Self
    where
        O: OffsetBuilderPType,
        T: AsRef<[u8]>,
    {
        let mut builder = VarBinBuilder::<O>::with_capacity_in(
            dtype,
            vec.len(),
            BufferAllocatorRef::static_ref(),
        );
        for v in vec {
            builder.append_value(v.as_ref());
        }
        builder.finish_into_varbin()
    }

    /// Create from a vector of string slices.
    pub fn from_strs(value: Vec<&str>) -> Self {
        Self::from_vec(value, DType::Utf8(Nullability::NonNullable))
    }

    /// Create from a vector of optional string slices.
    pub fn from_nullable_strs(value: Vec<Option<&str>>) -> Self {
        Self::from_iter(value, DType::Utf8(Nullability::Nullable))
    }

    /// Create from a vector of byte slices.
    pub fn from_bytes(value: Vec<&[u8]>) -> Self {
        Self::from_vec(value, DType::Binary(Nullability::NonNullable))
    }

    /// Create from a vector of optional byte slices.
    pub fn from_nullable_bytes(value: Vec<Option<&[u8]>>) -> Self {
        Self::from_iter(value, DType::Binary(Nullability::Nullable))
    }

    pub fn into_data_parts(self) -> VarBinDataParts {
        let dtype = self.dtype().clone();
        let validity = self.varbin_validity();
        let offsets = self.offsets().clone();
        let data = self.into_data();
        VarBinDataParts {
            dtype,
            bytes: data.bytes,
            offsets,
            validity,
        }
    }
}

impl Array<VarBin> {
    /// Creates a new `VarBinArray`.
    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(
            slots[VarBinSlots::OFFSETS]
                .as_ref()
                .vortex_expect("VarBinArray offsets slot")
                .clone(),
            bytes,
            dtype.clone(),
            validity,
            ctx,
        );
        unsafe { Array::from_parts_unchecked(ArrayParts::new(VarBin, dtype, len, data, slots)) }
    }

    /// Creates a new `VarBinArray` without validation.
    ///
    /// # Safety
    ///
    /// See [`VarBinData::new_unchecked`].
    pub unsafe fn new_unchecked(
        offsets: ArrayRef,
        bytes: ByteBuffer,
        dtype: DType,
        validity: Validity,
    ) -> Self {
        let len = offsets.len().saturating_sub(1);
        let slots = VarBinData::make_slots(offsets, &validity, len);
        let data = unsafe { VarBinData::new_unchecked(bytes) };
        unsafe { Array::from_parts_unchecked(ArrayParts::new(VarBin, dtype, len, data, slots)) }
    }

    /// Creates a new `VarBinArray` without validation from a [`BufferHandle`].
    ///
    /// # Safety
    ///
    /// See [`VarBinData::new_unchecked_from_handle`].
    pub unsafe fn new_unchecked_from_handle(
        offsets: ArrayRef,
        bytes: BufferHandle,
        dtype: DType,
        validity: Validity,
    ) -> Self {
        let len = offsets.len().saturating_sub(1);
        let slots = VarBinData::make_slots(offsets, &validity, len);
        let data = unsafe { VarBinData::new_unchecked_from_handle(bytes) };
        unsafe { Array::from_parts_unchecked(ArrayParts::new(VarBin, dtype, len, data, slots)) }
    }

    /// 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, 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) };
        Ok(
            unsafe {
                Array::from_parts_unchecked(ArrayParts::new(VarBin, dtype, len, data, slots))
            },
        )
    }
}

impl From<Vec<&[u8]>> for Array<VarBin> {
    fn from(value: Vec<&[u8]>) -> Self {
        Self::from_vec(value, DType::Binary(Nullability::NonNullable))
    }
}

impl From<Vec<Vec<u8>>> for Array<VarBin> {
    fn from(value: Vec<Vec<u8>>) -> Self {
        Self::from_vec(value, DType::Binary(Nullability::NonNullable))
    }
}

impl From<Vec<String>> for Array<VarBin> {
    fn from(value: Vec<String>) -> Self {
        Self::from_vec(value, DType::Utf8(Nullability::NonNullable))
    }
}

impl From<Vec<&str>> for Array<VarBin> {
    fn from(value: Vec<&str>) -> Self {
        Self::from_vec(value, DType::Utf8(Nullability::NonNullable))
    }
}

impl From<Vec<Option<&[u8]>>> for Array<VarBin> {
    fn from(value: Vec<Option<&[u8]>>) -> Self {
        Self::from_iter(value, DType::Binary(Nullability::Nullable))
    }
}

impl From<Vec<Option<Vec<u8>>>> for Array<VarBin> {
    fn from(value: Vec<Option<Vec<u8>>>) -> Self {
        Self::from_iter(value, DType::Binary(Nullability::Nullable))
    }
}

impl From<Vec<Option<String>>> for Array<VarBin> {
    fn from(value: Vec<Option<String>>) -> Self {
        Self::from_iter(value, DType::Utf8(Nullability::Nullable))
    }
}

impl From<Vec<Option<&str>>> for Array<VarBin> {
    fn from(value: Vec<Option<&str>>) -> Self {
        Self::from_iter(value, DType::Utf8(Nullability::Nullable))
    }
}

impl<'a> FromIterator<Option<&'a [u8]>> for Array<VarBin> {
    fn from_iter<T: IntoIterator<Item = Option<&'a [u8]>>>(iter: T) -> Self {
        Self::from_iter(iter, DType::Binary(Nullability::Nullable))
    }
}

impl FromIterator<Option<Vec<u8>>> for Array<VarBin> {
    fn from_iter<T: IntoIterator<Item = Option<Vec<u8>>>>(iter: T) -> Self {
        Self::from_iter(iter, DType::Binary(Nullability::Nullable))
    }
}

impl FromIterator<Option<String>> for Array<VarBin> {
    fn from_iter<T: IntoIterator<Item = Option<String>>>(iter: T) -> Self {
        Self::from_iter(iter, DType::Utf8(Nullability::Nullable))
    }
}

impl<'a> FromIterator<Option<&'a str>> for Array<VarBin> {
    fn from_iter<T: IntoIterator<Item = Option<&'a str>>>(iter: T) -> Self {
        Self::from_iter(iter, DType::Utf8(Nullability::Nullable))
    }
}