vortex-array 0.59.4

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

//! Encodings that enable zero-copy sharing of data with Arrow.

use std::sync::Arc;

use vortex_buffer::Buffer;
use vortex_dtype::DType;
use vortex_dtype::NativePType;
use vortex_error::VortexExpect;
use vortex_error::VortexResult;
use vortex_error::vortex_ensure;
use vortex_error::vortex_panic;

use crate::Array;
use crate::ArrayRef;
use crate::Columnar;
use crate::Executable;
use crate::ExecutionCtx;
use crate::IntoArray;
use crate::arrays::BoolArray;
use crate::arrays::BoolArrayParts;
use crate::arrays::BoolVTable;
use crate::arrays::DecimalArray;
use crate::arrays::DecimalArrayParts;
use crate::arrays::DecimalVTable;
use crate::arrays::ExtensionArray;
use crate::arrays::ExtensionVTable;
use crate::arrays::FixedSizeListArray;
use crate::arrays::FixedSizeListVTable;
use crate::arrays::ListViewArray;
use crate::arrays::ListViewArrayParts;
use crate::arrays::ListViewRebuildMode;
use crate::arrays::ListViewVTable;
use crate::arrays::NullArray;
use crate::arrays::NullVTable;
use crate::arrays::PrimitiveArray;
use crate::arrays::PrimitiveArrayParts;
use crate::arrays::PrimitiveVTable;
use crate::arrays::StructArray;
use crate::arrays::StructArrayParts;
use crate::arrays::StructVTable;
use crate::arrays::VarBinViewArray;
use crate::arrays::VarBinViewArrayParts;
use crate::arrays::VarBinViewVTable;
use crate::arrays::constant_canonicalize;
use crate::builders::builder_with_capacity;
use crate::matcher::Matcher;

/// An enum capturing the default uncompressed encodings for each [Vortex type](DType).
///
/// Any array can be decoded into canonical form via the [`to_canonical`](Array::to_canonical)
/// trait method. This is the simplest encoding for a type, and will not be compressed but may
/// contain compressed child arrays.
///
/// Canonical form is useful for doing type-specific compute where you need to know that all
/// elements are laid out decompressed and contiguous in memory.
///
/// Each `Canonical` variant has a corresponding [`DType`] variant, with the notable exception of
/// [`Canonical::VarBinView`], which is the canonical encoding for both [`DType::Utf8`] and
/// [`DType::Binary`].
///
/// # Laziness
///
/// Canonical form is not recursive, so while a `StructArray` is the canonical format for any
/// `Struct` type, individual column child arrays may still be compressed. This allows
/// compute over Vortex arrays to push decoding as late as possible, and ideally many child arrays
/// never need to be decoded into canonical form at all depending on the compute.
///
/// # Arrow interoperability
///
/// All of the Vortex canonical encodings have an equivalent Arrow encoding that can be built
/// zero-copy, and the corresponding Arrow array types can also be built directly.
///
/// The full list of canonical types and their equivalent Arrow array types are:
///
/// * `NullArray`: [`arrow_array::NullArray`]
/// * `BoolArray`: [`arrow_array::BooleanArray`]
/// * `PrimitiveArray`: [`arrow_array::PrimitiveArray`]
/// * `DecimalArray`: [`arrow_array::Decimal128Array`] and [`arrow_array::Decimal256Array`]
/// * `VarBinViewArray`: [`arrow_array::GenericByteViewArray`]
/// * `ListViewArray`: [`arrow_array::ListViewArray`]
/// * `FixedSizeListArray`: [`arrow_array::FixedSizeListArray`]
/// * `StructArray`: [`arrow_array::StructArray`]
///
/// Vortex uses a logical type system, unlike Arrow which uses physical encodings for its types.
/// As an example, there are at least six valid physical encodings for a `Utf8` array. This can
/// create ambiguity.
/// Thus, if you receive an Arrow array, compress it using Vortex, and then
/// decompress it later to pass to a compute kernel, there are multiple suitable Arrow array
/// variants to hold the data.
///
/// To disambiguate, we choose a canonical physical encoding for every Vortex [`DType`], which
/// will correspond to an arrow-rs [`arrow_schema::DataType`].
///
/// # Views support
///
/// Binary and String views, also known as "German strings" are a better encoding format for
/// nearly all use-cases. Variable-length binary views are part of the Apache Arrow spec, and are
/// fully supported by the Datafusion query engine. We use them as our canonical string encoding
/// for all `Utf8` and `Binary` typed arrays in Vortex. They provide considerably faster filter
/// execution than the core `StringArray` and `BinaryArray` types, at the expense of potentially
/// needing [garbage collection][arrow_array::GenericByteViewArray::gc] to clear unreferenced items
/// from memory.
///
/// # For Developers
///
/// If you add another variant to this enum, make sure to update `dyn Array::is_canonical`,
/// and the fuzzer in `fuzz/fuzz_targets/array_ops.rs`.
#[derive(Debug, Clone)]
pub enum Canonical {
    Null(NullArray),
    Bool(BoolArray),
    Primitive(PrimitiveArray),
    Decimal(DecimalArray),
    VarBinView(VarBinViewArray),
    List(ListViewArray),
    FixedSizeList(FixedSizeListArray),
    Struct(StructArray),
    Extension(ExtensionArray),
}

impl Canonical {
    // TODO(connor): This can probably be specialized for each of the canonical arrays.
    /// Create an empty canonical array of the given dtype.
    pub fn empty(dtype: &DType) -> Canonical {
        builder_with_capacity(dtype, 0).finish_into_canonical()
    }

    pub fn len(&self) -> usize {
        match self {
            Canonical::Null(c) => c.len(),
            Canonical::Bool(c) => c.len(),
            Canonical::Primitive(c) => c.len(),
            Canonical::Decimal(c) => c.len(),
            Canonical::VarBinView(c) => c.len(),
            Canonical::List(c) => c.len(),
            Canonical::FixedSizeList(c) => c.len(),
            Canonical::Struct(c) => c.len(),
            Canonical::Extension(c) => c.len(),
        }
    }

    pub fn dtype(&self) -> &DType {
        match self {
            Canonical::Null(c) => c.dtype(),
            Canonical::Bool(c) => c.dtype(),
            Canonical::Primitive(c) => c.dtype(),
            Canonical::Decimal(c) => c.dtype(),
            Canonical::VarBinView(c) => c.dtype(),
            Canonical::List(c) => c.dtype(),
            Canonical::FixedSizeList(c) => c.dtype(),
            Canonical::Struct(c) => c.dtype(),
            Canonical::Extension(c) => c.dtype(),
        }
    }

    pub fn is_empty(&self) -> bool {
        match self {
            Canonical::Null(c) => c.is_empty(),
            Canonical::Bool(c) => c.is_empty(),
            Canonical::Primitive(c) => c.is_empty(),
            Canonical::Decimal(c) => c.is_empty(),
            Canonical::VarBinView(c) => c.is_empty(),
            Canonical::List(c) => c.is_empty(),
            Canonical::FixedSizeList(c) => c.is_empty(),
            Canonical::Struct(c) => c.is_empty(),
            Canonical::Extension(c) => c.is_empty(),
        }
    }
}

impl Canonical {
    /// Performs a (potentially expensive) compaction operation on the array before it is complete.
    ///
    /// This is mostly relevant for the variable-length types such as Utf8, Binary or List where
    /// they can accumulate wasted space after slicing and taking operations.
    ///
    /// This operation is very expensive and can result in things like allocations, full-scans
    /// and copy operations.
    pub fn compact(&self) -> VortexResult<Canonical> {
        match self {
            Canonical::VarBinView(array) => Ok(Canonical::VarBinView(array.compact_buffers()?)),
            Canonical::List(array) => Ok(Canonical::List(
                array.rebuild(ListViewRebuildMode::TrimElements)?,
            )),
            _ => Ok(self.clone()),
        }
    }
}

// Unwrap canonical type back down to specialized type.
impl Canonical {
    pub fn as_null(&self) -> &NullArray {
        if let Canonical::Null(a) = self {
            a
        } else {
            vortex_panic!("Cannot get NullArray from {:?}", &self)
        }
    }

    pub fn into_null(self) -> NullArray {
        if let Canonical::Null(a) = self {
            a
        } else {
            vortex_panic!("Cannot unwrap NullArray from {:?}", &self)
        }
    }

    pub fn as_bool(&self) -> &BoolArray {
        if let Canonical::Bool(a) = self {
            a
        } else {
            vortex_panic!("Cannot get BoolArray from {:?}", &self)
        }
    }

    pub fn into_bool(self) -> BoolArray {
        if let Canonical::Bool(a) = self {
            a
        } else {
            vortex_panic!("Cannot unwrap BoolArray from {:?}", &self)
        }
    }

    pub fn as_primitive(&self) -> &PrimitiveArray {
        if let Canonical::Primitive(a) = self {
            a
        } else {
            vortex_panic!("Cannot get PrimitiveArray from {:?}", &self)
        }
    }

    pub fn into_primitive(self) -> PrimitiveArray {
        if let Canonical::Primitive(a) = self {
            a
        } else {
            vortex_panic!("Cannot unwrap PrimitiveArray from {:?}", &self)
        }
    }

    pub fn as_decimal(&self) -> &DecimalArray {
        if let Canonical::Decimal(a) = self {
            a
        } else {
            vortex_panic!("Cannot get DecimalArray from {:?}", &self)
        }
    }

    pub fn into_decimal(self) -> DecimalArray {
        if let Canonical::Decimal(a) = self {
            a
        } else {
            vortex_panic!("Cannot unwrap DecimalArray from {:?}", &self)
        }
    }

    pub fn as_varbinview(&self) -> &VarBinViewArray {
        if let Canonical::VarBinView(a) = self {
            a
        } else {
            vortex_panic!("Cannot get VarBinViewArray from {:?}", &self)
        }
    }

    pub fn into_varbinview(self) -> VarBinViewArray {
        if let Canonical::VarBinView(a) = self {
            a
        } else {
            vortex_panic!("Cannot unwrap VarBinViewArray from {:?}", &self)
        }
    }

    pub fn as_listview(&self) -> &ListViewArray {
        if let Canonical::List(a) = self {
            a
        } else {
            vortex_panic!("Cannot get ListArray from {:?}", &self)
        }
    }

    pub fn into_listview(self) -> ListViewArray {
        if let Canonical::List(a) = self {
            a
        } else {
            vortex_panic!("Cannot unwrap ListArray from {:?}", &self)
        }
    }

    pub fn as_fixed_size_list(&self) -> &FixedSizeListArray {
        if let Canonical::FixedSizeList(a) = self {
            a
        } else {
            vortex_panic!("Cannot get FixedSizeListArray from {:?}", &self)
        }
    }

    pub fn into_fixed_size_list(self) -> FixedSizeListArray {
        if let Canonical::FixedSizeList(a) = self {
            a
        } else {
            vortex_panic!("Cannot unwrap FixedSizeListArray from {:?}", &self)
        }
    }

    pub fn as_struct(&self) -> &StructArray {
        if let Canonical::Struct(a) = self {
            a
        } else {
            vortex_panic!("Cannot get StructArray from {:?}", &self)
        }
    }

    pub fn into_struct(self) -> StructArray {
        if let Canonical::Struct(a) = self {
            a
        } else {
            vortex_panic!("Cannot unwrap StructArray from {:?}", &self)
        }
    }

    pub fn as_extension(&self) -> &ExtensionArray {
        if let Canonical::Extension(a) = self {
            a
        } else {
            vortex_panic!("Cannot get ExtensionArray from {:?}", &self)
        }
    }

    pub fn into_extension(self) -> ExtensionArray {
        if let Canonical::Extension(a) = self {
            a
        } else {
            vortex_panic!("Cannot unwrap ExtensionArray from {:?}", &self)
        }
    }
}

impl AsRef<dyn Array> for Canonical {
    fn as_ref(&self) -> &(dyn Array + 'static) {
        match &self {
            Canonical::Null(a) => a.as_ref(),
            Canonical::Bool(a) => a.as_ref(),
            Canonical::Primitive(a) => a.as_ref(),
            Canonical::Decimal(a) => a.as_ref(),
            Canonical::Struct(a) => a.as_ref(),
            Canonical::List(a) => a.as_ref(),
            Canonical::FixedSizeList(a) => a.as_ref(),
            Canonical::VarBinView(a) => a.as_ref(),
            Canonical::Extension(a) => a.as_ref(),
        }
    }
}

impl IntoArray for Canonical {
    fn into_array(self) -> ArrayRef {
        match self {
            Canonical::Null(a) => a.into_array(),
            Canonical::Bool(a) => a.into_array(),
            Canonical::Primitive(a) => a.into_array(),
            Canonical::Decimal(a) => a.into_array(),
            Canonical::Struct(a) => a.into_array(),
            Canonical::List(a) => a.into_array(),
            Canonical::FixedSizeList(a) => a.into_array(),
            Canonical::VarBinView(a) => a.into_array(),
            Canonical::Extension(a) => a.into_array(),
        }
    }
}

/// Trait for types that can be converted from an owned type into an owned array variant.
///
/// # Canonicalization
///
/// This trait has a blanket implementation for all types implementing [ToCanonical].
pub trait ToCanonical {
    /// Canonicalize into a [`NullArray`] if the target is [`Null`](DType::Null) typed.
    fn to_null(&self) -> NullArray;

    /// Canonicalize into a [`BoolArray`] if the target is [`Bool`](DType::Bool) typed.
    fn to_bool(&self) -> BoolArray;

    /// Canonicalize into a [`PrimitiveArray`] if the target is [`Primitive`](DType::Primitive)
    /// typed.
    fn to_primitive(&self) -> PrimitiveArray;

    /// Canonicalize into a [`DecimalArray`] if the target is [`Decimal`](DType::Decimal)
    /// typed.
    fn to_decimal(&self) -> DecimalArray;

    /// Canonicalize into a [`StructArray`] if the target is [`Struct`](DType::Struct) typed.
    fn to_struct(&self) -> StructArray;

    /// Canonicalize into a [`ListViewArray`] if the target is [`List`](DType::List) typed.
    fn to_listview(&self) -> ListViewArray;

    /// Canonicalize into a [`FixedSizeListArray`] if the target is [`List`](DType::FixedSizeList)
    /// typed.
    fn to_fixed_size_list(&self) -> FixedSizeListArray;

    /// Canonicalize into a [`VarBinViewArray`] if the target is [`Utf8`](DType::Utf8)
    /// or [`Binary`](DType::Binary) typed.
    fn to_varbinview(&self) -> VarBinViewArray;

    /// Canonicalize into an [`ExtensionArray`] if the array is [`Extension`](DType::Extension)
    /// typed.
    fn to_extension(&self) -> ExtensionArray;
}

// Blanket impl for all Array encodings.
impl<A: Array + ?Sized> ToCanonical for A {
    fn to_null(&self) -> NullArray {
        self.to_canonical()
            .vortex_expect("to_canonical failed")
            .into_null()
    }

    fn to_bool(&self) -> BoolArray {
        self.to_canonical()
            .vortex_expect("to_canonical failed")
            .into_bool()
    }

    fn to_primitive(&self) -> PrimitiveArray {
        self.to_canonical()
            .vortex_expect("to_canonical failed")
            .into_primitive()
    }

    fn to_decimal(&self) -> DecimalArray {
        self.to_canonical()
            .vortex_expect("to_canonical failed")
            .into_decimal()
    }

    fn to_struct(&self) -> StructArray {
        self.to_canonical()
            .vortex_expect("to_canonical failed")
            .into_struct()
    }

    fn to_listview(&self) -> ListViewArray {
        self.to_canonical()
            .vortex_expect("to_canonical failed")
            .into_listview()
    }

    fn to_fixed_size_list(&self) -> FixedSizeListArray {
        self.to_canonical()
            .vortex_expect("to_canonical failed")
            .into_fixed_size_list()
    }

    fn to_varbinview(&self) -> VarBinViewArray {
        self.to_canonical()
            .vortex_expect("to_canonical failed")
            .into_varbinview()
    }

    fn to_extension(&self) -> ExtensionArray {
        self.to_canonical()
            .vortex_expect("to_canonical failed")
            .into_extension()
    }
}

impl From<Canonical> for ArrayRef {
    fn from(value: Canonical) -> Self {
        match value {
            Canonical::Null(a) => a.into_array(),
            Canonical::Bool(a) => a.into_array(),
            Canonical::Primitive(a) => a.into_array(),
            Canonical::Decimal(a) => a.into_array(),
            Canonical::Struct(a) => a.into_array(),
            Canonical::List(a) => a.into_array(),
            Canonical::FixedSizeList(a) => a.into_array(),
            Canonical::VarBinView(a) => a.into_array(),
            Canonical::Extension(a) => a.into_array(),
        }
    }
}

/// Recursively execute the array until it reaches canonical form.
///
/// Callers should prefer to execute into `Columnar` if they are able to optimize their use for
/// constant arrays.
impl Executable for Canonical {
    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
        if let Some(canonical) = array.as_opt::<AnyCanonical>() {
            return Ok(canonical.into());
        }

        // Invoke execute directly to avoid logging the call in the execution context.
        Ok(match Columnar::execute(array.clone(), ctx)? {
            Columnar::Canonical(c) => c,
            Columnar::Constant(s) => {
                let canonical = constant_canonicalize(&s)?;
                canonical
                    .as_ref()
                    .statistics()
                    .inherit_from(array.statistics());
                canonical
            }
        })
    }
}

/// Recursively execute the array until it reaches canonical form along with its validity.
///
/// Callers should prefer to execute into `Columnar` instead of this specific target.
/// This target is useful when preparing arrays for writing.
pub struct CanonicalValidity(pub Canonical);

impl Executable for CanonicalValidity {
    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
        match array.execute::<Canonical>(ctx)? {
            n @ Canonical::Null(_) => Ok(CanonicalValidity(n)),
            Canonical::Bool(b) => {
                let BoolArrayParts {
                    bits,
                    offset,
                    len,
                    validity,
                } = b.into_parts();
                Ok(CanonicalValidity(Canonical::Bool(
                    BoolArray::try_new_from_handle(bits, offset, len, validity.execute(ctx)?)?,
                )))
            }
            Canonical::Primitive(p) => {
                let PrimitiveArrayParts {
                    ptype,
                    buffer,
                    validity,
                } = p.into_parts();
                Ok(CanonicalValidity(Canonical::Primitive(unsafe {
                    PrimitiveArray::new_unchecked_from_handle(buffer, ptype, validity.execute(ctx)?)
                })))
            }
            Canonical::Decimal(d) => {
                let DecimalArrayParts {
                    decimal_dtype,
                    values,
                    values_type,
                    validity,
                } = d.into_parts();
                Ok(CanonicalValidity(Canonical::Decimal(unsafe {
                    DecimalArray::new_unchecked_handle(
                        values,
                        values_type,
                        decimal_dtype,
                        validity.execute(ctx)?,
                    )
                })))
            }
            Canonical::VarBinView(vbv) => {
                let VarBinViewArrayParts {
                    dtype,
                    buffers,
                    views,
                    validity,
                } = vbv.into_parts();
                Ok(CanonicalValidity(Canonical::VarBinView(unsafe {
                    VarBinViewArray::new_handle_unchecked(
                        views,
                        buffers,
                        dtype,
                        validity.execute(ctx)?,
                    )
                })))
            }
            Canonical::List(l) => {
                let ListViewArrayParts {
                    elements,
                    offsets,
                    sizes,
                    validity,
                    ..
                } = l.into_parts();
                Ok(CanonicalValidity(Canonical::List(unsafe {
                    ListViewArray::new_unchecked(elements, offsets, sizes, validity.execute(ctx)?)
                })))
            }
            Canonical::FixedSizeList(fsl) => {
                let list_size = fsl.list_size();
                let len = fsl.len();
                let (elements, validity, _) = fsl.into_parts();
                Ok(CanonicalValidity(Canonical::FixedSizeList(
                    FixedSizeListArray::new(elements, list_size, validity.execute(ctx)?, len),
                )))
            }
            Canonical::Struct(st) => {
                let len = st.len();
                let StructArrayParts {
                    struct_fields,
                    fields,
                    validity,
                } = st.into_parts();
                Ok(CanonicalValidity(Canonical::Struct(unsafe {
                    StructArray::new_unchecked(fields, struct_fields, len, validity.execute(ctx)?)
                })))
            }
            Canonical::Extension(ext) => Ok(CanonicalValidity(Canonical::Extension(
                ExtensionArray::new(
                    ext.ext_dtype().clone(),
                    ext.storage()
                        .clone()
                        .execute::<CanonicalValidity>(ctx)?
                        .0
                        .into_array(),
                ),
            ))),
        }
    }
}

/// Recursively execute the array until all of its children are canonical.
///
/// This method is useful to guarantee that all operators are fully executed,
/// callers should prefer an execution target that's suitable for their use case instead of this one.
pub struct RecursiveCanonical(pub Canonical);

impl Executable for RecursiveCanonical {
    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
        match array.execute::<Canonical>(ctx)? {
            n @ Canonical::Null(_) => Ok(RecursiveCanonical(n)),
            Canonical::Bool(b) => {
                let BoolArrayParts {
                    bits,
                    offset,
                    len,
                    validity,
                } = b.into_parts();
                Ok(RecursiveCanonical(Canonical::Bool(
                    BoolArray::try_new_from_handle(bits, offset, len, validity.execute(ctx)?)?,
                )))
            }
            Canonical::Primitive(p) => {
                let PrimitiveArrayParts {
                    ptype,
                    buffer,
                    validity,
                } = p.into_parts();
                Ok(RecursiveCanonical(Canonical::Primitive(unsafe {
                    PrimitiveArray::new_unchecked_from_handle(buffer, ptype, validity.execute(ctx)?)
                })))
            }
            Canonical::Decimal(d) => {
                let DecimalArrayParts {
                    decimal_dtype,
                    values,
                    values_type,
                    validity,
                } = d.into_parts();
                Ok(RecursiveCanonical(Canonical::Decimal(unsafe {
                    DecimalArray::new_unchecked_handle(
                        values,
                        values_type,
                        decimal_dtype,
                        validity.execute(ctx)?,
                    )
                })))
            }
            Canonical::VarBinView(vbv) => {
                let VarBinViewArrayParts {
                    dtype,
                    buffers,
                    views,
                    validity,
                } = vbv.into_parts();
                Ok(RecursiveCanonical(Canonical::VarBinView(unsafe {
                    VarBinViewArray::new_handle_unchecked(
                        views,
                        buffers,
                        dtype,
                        validity.execute(ctx)?,
                    )
                })))
            }
            Canonical::List(l) => {
                let ListViewArrayParts {
                    elements,
                    offsets,
                    sizes,
                    validity,
                    ..
                } = l.into_parts();
                Ok(RecursiveCanonical(Canonical::List(unsafe {
                    ListViewArray::new_unchecked(
                        elements.execute::<RecursiveCanonical>(ctx)?.0.into_array(),
                        offsets.execute::<RecursiveCanonical>(ctx)?.0.into_array(),
                        sizes.execute::<RecursiveCanonical>(ctx)?.0.into_array(),
                        validity.execute(ctx)?,
                    )
                })))
            }
            Canonical::FixedSizeList(fsl) => {
                let list_size = fsl.list_size();
                let len = fsl.len();
                let (elements, validity, _) = fsl.into_parts();
                Ok(RecursiveCanonical(Canonical::FixedSizeList(
                    FixedSizeListArray::new(
                        elements.execute::<RecursiveCanonical>(ctx)?.0.into_array(),
                        list_size,
                        validity.execute(ctx)?,
                        len,
                    ),
                )))
            }
            Canonical::Struct(st) => {
                let len = st.len();
                let StructArrayParts {
                    struct_fields,
                    fields,
                    validity,
                } = st.into_parts();
                let executed_fields = fields
                    .iter()
                    .map(|f| Ok(f.clone().execute::<RecursiveCanonical>(ctx)?.0.into_array()))
                    .collect::<VortexResult<Arc<[_]>>>()?;

                Ok(RecursiveCanonical(Canonical::Struct(unsafe {
                    StructArray::new_unchecked(
                        executed_fields,
                        struct_fields,
                        len,
                        validity.execute(ctx)?,
                    )
                })))
            }
            Canonical::Extension(ext) => Ok(RecursiveCanonical(Canonical::Extension(
                ExtensionArray::new(
                    ext.ext_dtype().clone(),
                    ext.storage()
                        .clone()
                        .execute::<RecursiveCanonical>(ctx)?
                        .0
                        .into_array(),
                ),
            ))),
        }
    }
}

/// Execute a primitive typed array into a buffer of native values, assuming all values are valid.
///
/// # Errors
///
/// Returns a `VortexError` if the array is not all-valid (has any nulls).
impl<T: NativePType> Executable for Buffer<T> {
    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
        let array = PrimitiveArray::execute(array, ctx)?;
        vortex_ensure!(
            array.all_valid()?,
            "Cannot execute to native buffer: array is not all-valid."
        );
        Ok(array.into_buffer())
    }
}

/// Execute the array to canonical form and unwrap as a [`PrimitiveArray`].
///
/// This will panic if the array's dtype is not primitive.
impl Executable for PrimitiveArray {
    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
        match array.try_into::<PrimitiveVTable>() {
            Ok(primitive) => Ok(primitive),
            Err(array) => Ok(Canonical::execute(array, ctx)?.into_primitive()),
        }
    }
}

/// Execute the array to canonical form and unwrap as a [`BoolArray`].
///
/// This will panic if the array's dtype is not bool.
impl Executable for BoolArray {
    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
        match array.try_into::<BoolVTable>() {
            Ok(bool_array) => Ok(bool_array),
            Err(array) => Ok(Canonical::execute(array, ctx)?.into_bool()),
        }
    }
}

/// Execute the array to canonical form and unwrap as a [`NullArray`].
///
/// This will panic if the array's dtype is not null.
impl Executable for NullArray {
    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
        match array.try_into::<NullVTable>() {
            Ok(null_array) => Ok(null_array),
            Err(array) => Ok(Canonical::execute(array, ctx)?.into_null()),
        }
    }
}

/// Execute the array to canonical form and unwrap as a [`VarBinViewArray`].
///
/// This will panic if the array's dtype is not utf8 or binary.
impl Executable for VarBinViewArray {
    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
        match array.try_into::<VarBinViewVTable>() {
            Ok(varbinview) => Ok(varbinview),
            Err(array) => Ok(Canonical::execute(array, ctx)?.into_varbinview()),
        }
    }
}

/// Execute the array to canonical form and unwrap as an [`ExtensionArray`].
///
/// This will panic if the array's dtype is not an extension type.
impl Executable for ExtensionArray {
    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
        match array.try_into::<ExtensionVTable>() {
            Ok(ext_array) => Ok(ext_array),
            Err(array) => Ok(Canonical::execute(array, ctx)?.into_extension()),
        }
    }
}

/// Execute the array to canonical form and unwrap as a [`DecimalArray`].
///
/// This will panic if the array's dtype is not decimal.
impl Executable for DecimalArray {
    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
        match array.try_into::<DecimalVTable>() {
            Ok(decimal) => Ok(decimal),
            Err(array) => Ok(Canonical::execute(array, ctx)?.into_decimal()),
        }
    }
}

/// Execute the array to canonical form and unwrap as a [`ListViewArray`].
///
/// This will panic if the array's dtype is not list.
impl Executable for ListViewArray {
    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
        match array.try_into::<ListViewVTable>() {
            Ok(list) => Ok(list),
            Err(array) => Ok(Canonical::execute(array, ctx)?.into_listview()),
        }
    }
}

/// Execute the array to canonical form and unwrap as a [`FixedSizeListArray`].
///
/// This will panic if the array's dtype is not fixed size list.
impl Executable for FixedSizeListArray {
    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
        match array.try_into::<FixedSizeListVTable>() {
            Ok(fsl) => Ok(fsl),
            Err(array) => Ok(Canonical::execute(array, ctx)?.into_fixed_size_list()),
        }
    }
}

/// Execute the array to canonical form and unwrap as a [`StructArray`].
///
/// This will panic if the array's dtype is not struct.
impl Executable for StructArray {
    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
        match array.try_into::<StructVTable>() {
            Ok(struct_array) => Ok(struct_array),
            Err(array) => Ok(Canonical::execute(array, ctx)?.into_struct()),
        }
    }
}

/// A view into a canonical array type.
#[derive(Debug, Clone)]
pub enum CanonicalView<'a> {
    Null(&'a NullArray),
    Bool(&'a BoolArray),
    Primitive(&'a PrimitiveArray),
    Decimal(&'a DecimalArray),
    VarBinView(&'a VarBinViewArray),
    List(&'a ListViewArray),
    FixedSizeList(&'a FixedSizeListArray),
    Struct(&'a StructArray),
    Extension(&'a ExtensionArray),
}

impl From<CanonicalView<'_>> for Canonical {
    fn from(value: CanonicalView<'_>) -> Self {
        match value {
            CanonicalView::Null(a) => Canonical::Null(a.clone()),
            CanonicalView::Bool(a) => Canonical::Bool(a.clone()),
            CanonicalView::Primitive(a) => Canonical::Primitive(a.clone()),
            CanonicalView::Decimal(a) => Canonical::Decimal(a.clone()),
            CanonicalView::VarBinView(a) => Canonical::VarBinView(a.clone()),
            CanonicalView::List(a) => Canonical::List(a.clone()),
            CanonicalView::FixedSizeList(a) => Canonical::FixedSizeList(a.clone()),
            CanonicalView::Struct(a) => Canonical::Struct(a.clone()),
            CanonicalView::Extension(a) => Canonical::Extension(a.clone()),
        }
    }
}

impl AsRef<dyn Array> for CanonicalView<'_> {
    fn as_ref(&self) -> &dyn Array {
        match self {
            CanonicalView::Null(a) => a.as_ref(),
            CanonicalView::Bool(a) => a.as_ref(),
            CanonicalView::Primitive(a) => a.as_ref(),
            CanonicalView::Decimal(a) => a.as_ref(),
            CanonicalView::VarBinView(a) => a.as_ref(),
            CanonicalView::List(a) => a.as_ref(),
            CanonicalView::FixedSizeList(a) => a.as_ref(),
            CanonicalView::Struct(a) => a.as_ref(),
            CanonicalView::Extension(a) => a.as_ref(),
        }
    }
}

/// A matcher for any canonical array type.
pub struct AnyCanonical;
impl Matcher for AnyCanonical {
    type Match<'a> = CanonicalView<'a>;

    fn matches(array: &dyn Array) -> bool {
        array.is::<NullVTable>()
            || array.is::<BoolVTable>()
            || array.is::<PrimitiveVTable>()
            || array.is::<DecimalVTable>()
            || array.is::<StructVTable>()
            || array.is::<ListViewVTable>()
            || array.is::<FixedSizeListVTable>()
            || array.is::<VarBinViewVTable>()
            || array.is::<ExtensionVTable>()
    }

    fn try_match<'a>(array: &'a dyn Array) -> Option<Self::Match<'a>> {
        if let Some(a) = array.as_opt::<NullVTable>() {
            Some(CanonicalView::Null(a))
        } else if let Some(a) = array.as_opt::<BoolVTable>() {
            Some(CanonicalView::Bool(a))
        } else if let Some(a) = array.as_opt::<PrimitiveVTable>() {
            Some(CanonicalView::Primitive(a))
        } else if let Some(a) = array.as_opt::<DecimalVTable>() {
            Some(CanonicalView::Decimal(a))
        } else if let Some(a) = array.as_opt::<StructVTable>() {
            Some(CanonicalView::Struct(a))
        } else if let Some(a) = array.as_opt::<ListViewVTable>() {
            Some(CanonicalView::List(a))
        } else if let Some(a) = array.as_opt::<FixedSizeListVTable>() {
            Some(CanonicalView::FixedSizeList(a))
        } else if let Some(a) = array.as_opt::<VarBinViewVTable>() {
            Some(CanonicalView::VarBinView(a))
        } else {
            array
                .as_opt::<ExtensionVTable>()
                .map(CanonicalView::Extension)
        }
    }
}

#[cfg(test)]
mod test {
    use std::sync::Arc;

    use arrow_array::Array as ArrowArray;
    use arrow_array::ArrayRef as ArrowArrayRef;
    use arrow_array::ListArray as ArrowListArray;
    use arrow_array::PrimitiveArray as ArrowPrimitiveArray;
    use arrow_array::StringArray;
    use arrow_array::StringViewArray;
    use arrow_array::StructArray as ArrowStructArray;
    use arrow_array::cast::AsArray;
    use arrow_array::types::Int32Type;
    use arrow_array::types::Int64Type;
    use arrow_array::types::UInt64Type;
    use arrow_buffer::NullBufferBuilder;
    use arrow_buffer::OffsetBuffer;
    use arrow_schema::DataType;
    use arrow_schema::Field;
    use vortex_buffer::buffer;

    use crate::ArrayRef;
    use crate::IntoArray;
    use crate::arrays::ConstantArray;
    use crate::arrays::StructArray;
    use crate::arrow::FromArrowArray;
    use crate::arrow::IntoArrowArray;

    #[test]
    fn test_canonicalize_nested_struct() {
        // Create a struct array with multiple internal components.
        let nested_struct_array = StructArray::from_fields(&[
            ("a", buffer![1u64].into_array()),
            (
                "b",
                StructArray::from_fields(&[(
                    "inner_a",
                    // The nested struct contains a ConstantArray representing the primitive array
                    //   [100i64]
                    // ConstantArray is not a canonical type, so converting `into_arrow()` should
                    // map this to the nearest canonical type (PrimitiveArray).
                    ConstantArray::new(100i64, 1).into_array(),
                )])
                .unwrap()
                .into_array(),
            ),
        ])
        .unwrap();

        let arrow_struct = nested_struct_array
            .into_array()
            .into_arrow_preferred()
            .unwrap()
            .as_any()
            .downcast_ref::<ArrowStructArray>()
            .cloned()
            .unwrap();

        assert!(
            arrow_struct
                .column(0)
                .as_any()
                .downcast_ref::<ArrowPrimitiveArray<UInt64Type>>()
                .is_some()
        );

        let inner_struct = arrow_struct
            .column(1)
            .clone()
            .as_any()
            .downcast_ref::<ArrowStructArray>()
            .cloned()
            .unwrap();

        let inner_a = inner_struct
            .column(0)
            .as_any()
            .downcast_ref::<ArrowPrimitiveArray<Int64Type>>();
        assert!(inner_a.is_some());

        assert_eq!(
            inner_a.cloned().unwrap(),
            ArrowPrimitiveArray::from_iter([100i64])
        );
    }

    #[test]
    fn roundtrip_struct() {
        let mut nulls = NullBufferBuilder::new(6);
        nulls.append_n_non_nulls(4);
        nulls.append_null();
        nulls.append_non_null();
        let names = Arc::new(StringViewArray::from_iter(vec![
            Some("Joseph"),
            None,
            Some("Angela"),
            Some("Mikhail"),
            None,
            None,
        ]));
        let ages = Arc::new(ArrowPrimitiveArray::<Int32Type>::from(vec![
            Some(25),
            Some(31),
            None,
            Some(57),
            None,
            None,
        ]));

        let arrow_struct = ArrowStructArray::new(
            vec![
                Arc::new(Field::new("name", DataType::Utf8View, true)),
                Arc::new(Field::new("age", DataType::Int32, true)),
            ]
            .into(),
            vec![names, ages],
            nulls.finish(),
        );

        let vortex_struct = ArrayRef::from_arrow(&arrow_struct, true).unwrap();

        assert_eq!(
            &arrow_struct,
            vortex_struct.into_arrow_preferred().unwrap().as_struct()
        );
    }

    #[test]
    fn roundtrip_list() {
        let names = Arc::new(StringArray::from_iter(vec![
            Some("Joseph"),
            Some("Angela"),
            Some("Mikhail"),
        ]));

        let arrow_list = ArrowListArray::new(
            Arc::new(Field::new_list_field(DataType::Utf8, true)),
            OffsetBuffer::from_lengths(vec![0, 2, 1]),
            names,
            None,
        );
        let list_data_type = arrow_list.data_type();

        let vortex_list = ArrayRef::from_arrow(&arrow_list, true).unwrap();

        let rt_arrow_list = vortex_list.into_arrow(list_data_type).unwrap();

        assert_eq!(
            (Arc::new(arrow_list.clone()) as ArrowArrayRef).as_ref(),
            rt_arrow_list.as_ref()
        );
    }
}