vortex_arrow/

convert.rs

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

//! Canonical (non-plugin) conversions of Arrow arrays into Vortex arrays.
//!
//! Each `from_arrow_*` function converts one Arrow array shape; [`from_arrow_dyn`] dispatches on
//! the Arrow [`DataType`]. These functions have no awareness of Arrow extension types — the
//! plugin-aware entry points are the [`ArrowSession`](crate::ArrowSession) methods, which fall
//! back to these conversions for non-extension data. The deprecated [`FromArrowArray`] impls are
//! thin shims over these functions and will eventually be removed.
#![allow(deprecated)]

use std::ops::Range;
use std::sync::Arc;

use arrow_array::AnyDictionaryArray;
use arrow_array::Array as ArrowArray;
use arrow_array::ArrowPrimitiveType;
use arrow_array::BooleanArray as ArrowBooleanArray;
use arrow_array::DictionaryArray;
use arrow_array::FixedSizeListArray as ArrowFixedSizeListArray;
use arrow_array::GenericByteArray;
use arrow_array::GenericByteViewArray;
use arrow_array::GenericListArray;
use arrow_array::GenericListViewArray;
use arrow_array::MapArray as ArrowMapArray;
use arrow_array::NullArray as ArrowNullArray;
use arrow_array::OffsetSizeTrait;
use arrow_array::PrimitiveArray as ArrowPrimitiveArray;
use arrow_array::RecordBatch;
use arrow_array::StructArray as ArrowStructArray;
use arrow_array::cast::AsArray;
use arrow_array::cast::as_null_array;
use arrow_array::make_array;
use arrow_array::types::ArrowDictionaryKeyType;
use arrow_array::types::ByteArrayType;
use arrow_array::types::ByteViewType;
use arrow_array::types::Date32Type;
use arrow_array::types::Date64Type;
use arrow_array::types::Decimal32Type;
use arrow_array::types::Decimal64Type;
use arrow_array::types::Decimal128Type;
use arrow_array::types::Decimal256Type;
use arrow_array::types::Float16Type;
use arrow_array::types::Float32Type;
use arrow_array::types::Float64Type;
use arrow_array::types::Int8Type;
use arrow_array::types::Int16Type;
use arrow_array::types::Int32Type;
use arrow_array::types::Int64Type;
use arrow_array::types::Time32MillisecondType;
use arrow_array::types::Time32SecondType;
use arrow_array::types::Time64MicrosecondType;
use arrow_array::types::Time64NanosecondType;
use arrow_array::types::TimestampMicrosecondType;
use arrow_array::types::TimestampMillisecondType;
use arrow_array::types::TimestampNanosecondType;
use arrow_array::types::TimestampSecondType;
use arrow_array::types::UInt8Type;
use arrow_array::types::UInt16Type;
use arrow_array::types::UInt32Type;
use arrow_array::types::UInt64Type;
use arrow_buffer::ArrowNativeType;
use arrow_buffer::BooleanBuffer;
use arrow_buffer::Buffer as ArrowBuffer;
use arrow_buffer::ScalarBuffer;
use arrow_buffer::buffer::NullBuffer;
use arrow_buffer::buffer::OffsetBuffer;
use arrow_schema::DataType;
use arrow_schema::TimeUnit as ArrowTimeUnit;
use vortex_array::ArrayRef;
use vortex_array::IntoArray;
use vortex_array::arrays::BoolArray;
use vortex_array::arrays::DecimalArray;
use vortex_array::arrays::DictArray;
use vortex_array::arrays::FixedSizeListArray;
use vortex_array::arrays::ListArray;
use vortex_array::arrays::ListViewArray;
use vortex_array::arrays::MapArray;
use vortex_array::arrays::NullArray;
use vortex_array::arrays::PrimitiveArray;
use vortex_array::arrays::Struct;
use vortex_array::arrays::StructArray;
use vortex_array::arrays::TemporalArray;
use vortex_array::arrays::VarBinArray;
use vortex_array::arrays::VarBinViewArray;
use vortex_array::arrays::struct_::StructArrayExt;
use vortex_array::dtype::DType;
use vortex_array::dtype::DecimalDType;
use vortex_array::dtype::FieldNames;
use vortex_array::dtype::IntegerPType;
use vortex_array::dtype::MapDType;
use vortex_array::dtype::NativePType;
use vortex_array::dtype::Nullability;
use vortex_array::dtype::PType;
use vortex_array::dtype::i256;
use vortex_array::extension::datetime::TimeUnit;
use vortex_array::validity::Validity;
use vortex_buffer::Alignment;
use vortex_buffer::BitBuffer;
use vortex_buffer::Buffer;
use vortex_buffer::ByteBuffer;
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 vortex_error::vortex_panic;

use crate::FromArrowArray;
use crate::dtype::from_arrow_time_unit;

/// Zero-copy conversion of Arrow buffers into non-nullable Vortex arrays.
///
/// This mirrors [`IntoArray`] for Arrow buffer types; a separate trait is required because both
/// [`IntoArray`] and the Arrow buffer types are foreign to this crate.
pub trait IntoVortexArray {
    /// Convert this Arrow buffer into a non-nullable Vortex array without copying.
    fn into_array(self) -> ArrayRef;
}

impl IntoVortexArray for ArrowBuffer {
    fn into_array(self) -> ArrayRef {
        PrimitiveArray::from_byte_buffer(
            ByteBuffer::from_arrow_buffer(self, Alignment::of::<u8>()),
            PType::U8,
            Validity::NonNullable,
        )
        .into_array()
    }
}

impl IntoVortexArray for BooleanBuffer {
    fn into_array(self) -> ArrayRef {
        BoolArray::new(self.into(), Validity::NonNullable).into_array()
    }
}

impl<T> IntoVortexArray for ScalarBuffer<T>
where
    T: ArrowNativeType + NativePType,
{
    fn into_array(self) -> ArrayRef {
        PrimitiveArray::new(
            Buffer::<T>::from_arrow_scalar_buffer(self),
            Validity::NonNullable,
        )
        .into_array()
    }
}

impl<O> IntoVortexArray for OffsetBuffer<O>
where
    O: IntegerPType + OffsetSizeTrait,
{
    fn into_array(self) -> ArrayRef {
        let primitive = PrimitiveArray::new(
            Buffer::from_arrow_scalar_buffer(self.into_inner()),
            Validity::NonNullable,
        );

        primitive.into_array()
    }
}

/// Zero-copy conversion of an Arrow numeric primitive array into a Vortex array.
///
/// Use [`from_arrow_temporal`] for Arrow temporal arrays, which carry a logical dtype beyond
/// their physical primitive storage.
pub fn from_arrow_primitive<T: ArrowPrimitiveType>(
    value: &ArrowPrimitiveArray<T>,
    nullable: bool,
) -> VortexResult<ArrayRef>
where
    T::Native: NativePType,
{
    let buffer = Buffer::from_arrow_scalar_buffer(value.values().clone());
    let validity = nulls(value.nulls(), nullable)?;
    Ok(PrimitiveArray::new(buffer, validity).into_array())
}

macro_rules! impl_from_arrow_primitive {
    ($T:path) => {
        impl FromArrowArray<&ArrowPrimitiveArray<$T>> for ArrayRef {
            fn from_arrow(value: &ArrowPrimitiveArray<$T>, nullable: bool) -> VortexResult<Self> {
                from_arrow_primitive(value, nullable)
            }
        }
    };
}

impl_from_arrow_primitive!(Int8Type);
impl_from_arrow_primitive!(Int16Type);
impl_from_arrow_primitive!(Int32Type);
impl_from_arrow_primitive!(Int64Type);
impl_from_arrow_primitive!(UInt8Type);
impl_from_arrow_primitive!(UInt16Type);
impl_from_arrow_primitive!(UInt32Type);
impl_from_arrow_primitive!(UInt64Type);
impl_from_arrow_primitive!(Float16Type);
impl_from_arrow_primitive!(Float32Type);
impl_from_arrow_primitive!(Float64Type);

/// Zero-copy conversion of an Arrow `Decimal32` array into a Vortex decimal array.
pub fn from_arrow_decimal32(
    array: &ArrowPrimitiveArray<Decimal32Type>,
    nullable: bool,
) -> VortexResult<ArrayRef> {
    let decimal_type = DecimalDType::new(array.precision(), array.scale());
    let buffer = Buffer::from_arrow_scalar_buffer(array.values().clone());
    let validity = nulls(array.nulls(), nullable)?;
    Ok(DecimalArray::new(buffer, decimal_type, validity).into_array())
}

impl FromArrowArray<&ArrowPrimitiveArray<Decimal32Type>> for ArrayRef {
    fn from_arrow(
        array: &ArrowPrimitiveArray<Decimal32Type>,
        nullable: bool,
    ) -> VortexResult<Self> {
        from_arrow_decimal32(array, nullable)
    }
}

/// Zero-copy conversion of an Arrow `Decimal64` array into a Vortex decimal array.
pub fn from_arrow_decimal64(
    array: &ArrowPrimitiveArray<Decimal64Type>,
    nullable: bool,
) -> VortexResult<ArrayRef> {
    let decimal_type = DecimalDType::new(array.precision(), array.scale());
    let buffer = Buffer::from_arrow_scalar_buffer(array.values().clone());
    let validity = nulls(array.nulls(), nullable)?;
    Ok(DecimalArray::new(buffer, decimal_type, validity).into_array())
}

impl FromArrowArray<&ArrowPrimitiveArray<Decimal64Type>> for ArrayRef {
    fn from_arrow(
        array: &ArrowPrimitiveArray<Decimal64Type>,
        nullable: bool,
    ) -> VortexResult<Self> {
        from_arrow_decimal64(array, nullable)
    }
}

/// Zero-copy conversion of an Arrow `Decimal128` array into a Vortex decimal array.
pub fn from_arrow_decimal128(
    array: &ArrowPrimitiveArray<Decimal128Type>,
    nullable: bool,
) -> VortexResult<ArrayRef> {
    let decimal_type = DecimalDType::new(array.precision(), array.scale());
    let buffer = Buffer::from_arrow_scalar_buffer(array.values().clone());
    let validity = nulls(array.nulls(), nullable)?;
    Ok(DecimalArray::new(buffer, decimal_type, validity).into_array())
}

impl FromArrowArray<&ArrowPrimitiveArray<Decimal128Type>> for ArrayRef {
    fn from_arrow(
        array: &ArrowPrimitiveArray<Decimal128Type>,
        nullable: bool,
    ) -> VortexResult<Self> {
        from_arrow_decimal128(array, nullable)
    }
}

/// Zero-copy conversion of an Arrow `Decimal256` array into a Vortex decimal array.
pub fn from_arrow_decimal256(
    array: &ArrowPrimitiveArray<Decimal256Type>,
    nullable: bool,
) -> VortexResult<ArrayRef> {
    let decimal_type = DecimalDType::new(array.precision(), array.scale());
    let buffer = Buffer::from_arrow_scalar_buffer(array.values().clone());
    // SAFETY: Our i256 implementation has the same bit-pattern representation of the
    //  arrow_buffer::i256 type. It is safe to treat values held inside the buffer as values
    //  of either type.
    let buffer = unsafe { std::mem::transmute::<Buffer<arrow_buffer::i256>, Buffer<i256>>(buffer) };
    let validity = nulls(array.nulls(), nullable)?;
    Ok(DecimalArray::new(buffer, decimal_type, validity).into_array())
}

impl FromArrowArray<&ArrowPrimitiveArray<Decimal256Type>> for ArrayRef {
    fn from_arrow(
        array: &ArrowPrimitiveArray<Decimal256Type>,
        nullable: bool,
    ) -> VortexResult<Self> {
        from_arrow_decimal256(array, nullable)
    }
}

macro_rules! impl_from_arrow_temporal {
    ($T:path) => {
        impl FromArrowArray<&ArrowPrimitiveArray<$T>> for ArrayRef {
            fn from_arrow(
                value: &ArrowPrimitiveArray<$T>,
                nullable: bool,
            ) -> vortex_error::VortexResult<Self> {
                from_arrow_temporal(value, nullable)
            }
        }
    };
}

// timestamp
impl_from_arrow_temporal!(TimestampSecondType);
impl_from_arrow_temporal!(TimestampMillisecondType);
impl_from_arrow_temporal!(TimestampMicrosecondType);
impl_from_arrow_temporal!(TimestampNanosecondType);

// time
impl_from_arrow_temporal!(Time32SecondType);
impl_from_arrow_temporal!(Time32MillisecondType);
impl_from_arrow_temporal!(Time64MicrosecondType);
impl_from_arrow_temporal!(Time64NanosecondType);

// date
impl_from_arrow_temporal!(Date32Type);
impl_from_arrow_temporal!(Date64Type);

/// Conversion of an Arrow temporal array (timestamp/date/time) into a Vortex temporal
/// extension array.
pub fn from_arrow_temporal<T: ArrowPrimitiveType>(
    value: &ArrowPrimitiveArray<T>,
    nullable: bool,
) -> VortexResult<ArrayRef>
where
    T::Native: NativePType,
{
    let arr = PrimitiveArray::new(
        Buffer::from_arrow_scalar_buffer(value.values().clone()),
        nulls(value.nulls(), nullable)?,
    )
    .into_array();

    Ok(match value.data_type() {
        DataType::Timestamp(time_unit, tz) => {
            TemporalArray::new_timestamp(arr, from_arrow_time_unit(*time_unit), tz.clone()).into()
        }
        DataType::Time32(time_unit) => {
            TemporalArray::new_time(arr, from_arrow_time_unit(*time_unit)).into()
        }
        DataType::Time64(time_unit) => {
            TemporalArray::new_time(arr, from_arrow_time_unit(*time_unit)).into()
        }
        DataType::Date32 => TemporalArray::new_date(arr, TimeUnit::Days).into(),
        DataType::Date64 => TemporalArray::new_date(arr, TimeUnit::Milliseconds).into(),
        DataType::Duration(_) | DataType::Interval(_) => {
            vortex_bail!("Arrow {} arrays are not supported", value.data_type())
        }
        _ => vortex_panic!("Invalid temporal type: {}", value.data_type()),
    })
}

/// Rebase a (possibly sliced) Arrow offset buffer to start at zero.
///
/// A sliced Arrow string, binary or list array keeps its parent's whole child,
/// so the returned range selects just the child data its rows reference.
pub(crate) fn trim_offsets<O: OffsetSizeTrait>(
    offsets: &OffsetBuffer<O>,
) -> (OffsetBuffer<O>, Range<usize>) {
    let first = offsets[0];
    let referenced = first.as_usize()..offsets[offsets.len() - 1].as_usize();
    if first.as_usize() == 0 {
        return (offsets.clone(), referenced);
    }
    let rebased = OffsetBuffer::new(offsets.iter().map(|&offset| offset - first).collect());
    (rebased, referenced)
}

/// Conversion of an Arrow (large) string/binary array into a Vortex `VarBin` array, sharing its
/// bytes.
pub fn from_arrow_bytes<T: ByteArrayType>(
    value: &GenericByteArray<T>,
    nullable: bool,
) -> VortexResult<ArrayRef>
where
    <T as ByteArrayType>::Offset: IntegerPType,
{
    let dtype = match T::DATA_TYPE {
        DataType::Binary | DataType::LargeBinary => DType::Binary(nullable.into()),
        DataType::Utf8 | DataType::LargeUtf8 => DType::Utf8(nullable.into()),
        dt => vortex_panic!("Invalid data type for ByteArray: {dt}"),
    };
    let (offsets, referenced) = trim_offsets(value.offsets());
    let bytes = value
        .values()
        .slice_with_length(referenced.start, referenced.len());
    // SAFETY: Arrow arrays are already validated (valid UTF-8, valid offsets, correct validity).
    Ok(unsafe {
        VarBinArray::new_unchecked(
            offsets.into_array(),
            ByteBuffer::from_arrow_buffer(bytes, Alignment::of::<u8>()),
            dtype,
            nulls(value.nulls(), nullable)?,
        )
    }
    .into_array())
}

impl<T: ByteArrayType> FromArrowArray<&GenericByteArray<T>> for ArrayRef
where
    <T as ByteArrayType>::Offset: IntegerPType,
{
    fn from_arrow(value: &GenericByteArray<T>, nullable: bool) -> VortexResult<Self> {
        from_arrow_bytes(value, nullable)
    }
}

/// Zero-copy conversion of an Arrow string/binary view array into a Vortex `VarBinView` array.
pub fn from_arrow_byte_view<T: ByteViewType>(
    value: &GenericByteViewArray<T>,
    nullable: bool,
) -> VortexResult<ArrayRef> {
    let dtype = match T::DATA_TYPE {
        DataType::BinaryView => DType::Binary(nullable.into()),
        DataType::Utf8View => DType::Utf8(nullable.into()),
        dt => vortex_panic!("Invalid data type for ByteViewArray: {dt}"),
    };

    let views_buffer = Buffer::from_byte_buffer(
        Buffer::from_arrow_scalar_buffer(value.views().clone()).into_byte_buffer(),
    );

    // SAFETY: arrow-rs ByteViewArray already checks the same invariants, we inherit those
    //  guarantees by zero-copy constructing from one.
    Ok(unsafe {
        VarBinViewArray::new_unchecked(
            views_buffer,
            Arc::from(
                value
                    .data_buffers()
                    .iter()
                    .map(|b| ByteBuffer::from_arrow_buffer(b.clone(), Alignment::of::<u8>()))
                    .collect::<Vec<_>>(),
            ),
            dtype,
            nulls(value.nulls(), nullable)?,
        )
        .into_array()
    })
}

impl<T: ByteViewType> FromArrowArray<&GenericByteViewArray<T>> for ArrayRef {
    fn from_arrow(value: &GenericByteViewArray<T>, nullable: bool) -> VortexResult<Self> {
        from_arrow_byte_view(value, nullable)
    }
}

/// Zero-copy conversion of an Arrow boolean array into a Vortex `Bool` array.
pub fn from_arrow_boolean(value: &ArrowBooleanArray, nullable: bool) -> VortexResult<ArrayRef> {
    Ok(BoolArray::new(
        value.values().clone().into(),
        nulls(value.nulls(), nullable)?,
    )
    .into_array())
}

impl FromArrowArray<&ArrowBooleanArray> for ArrayRef {
    fn from_arrow(value: &ArrowBooleanArray, nullable: bool) -> VortexResult<Self> {
        from_arrow_boolean(value, nullable)
    }
}

/// Strip out the nulls from this array and return a new array without nulls.
pub(crate) fn remove_nulls(data: arrow_data::ArrayData) -> VortexResult<arrow_data::ArrayData> {
    if data.null_count() == 0 {
        // No nulls to remove, return the array as is
        return Ok(data);
    }

    let children = match data.data_type() {
        DataType::Struct(fields) => Some(
            fields
                .iter()
                .zip(data.child_data().iter())
                .map(|(field, child_data)| {
                    if field.is_nullable() {
                        Ok(child_data.clone())
                    } else {
                        remove_nulls(child_data.clone())
                    }
                })
                .collect::<VortexResult<Vec<_>>>()?,
        ),
        DataType::List(f)
        | DataType::LargeList(f)
        | DataType::ListView(f)
        | DataType::LargeListView(f)
        | DataType::FixedSizeList(f, _)
            if !f.is_nullable() =>
        {
            // All list types only have one child
            vortex_ensure_eq!(
                data.child_data().len(),
                1,
                "List types should have one child"
            );
            Some(vec![remove_nulls(data.child_data()[0].clone())?])
        }
        _ => None,
    };

    let mut builder = data.into_builder().nulls(None);
    if let Some(children) = children {
        builder = builder.child_data(children);
    }
    builder
        .build()
        .map_err(|e| vortex_err!("Failed to reconstruct Arrow array without nulls: {e}"))
}

/// Conversion of an Arrow struct array into a Vortex `Struct` array, converting each column.
pub fn from_arrow_struct(value: &ArrowStructArray, nullable: bool) -> VortexResult<ArrayRef> {
    Ok(StructArray::try_new(
        value.column_names().iter().copied().collect(),
        value
            .columns()
            .iter()
            .zip(value.fields())
            .map(|(c, field)| {
                // Arrow pushes down nulls, even into non-nullable fields. So we strip them
                // out here because Vortex is a little more strict.
                if c.null_count() > 0 && !field.is_nullable() {
                    let stripped = make_array(remove_nulls(c.into_data())?);
                    from_arrow_dyn(stripped.as_ref(), false)
                } else {
                    from_arrow_dyn(c.as_ref(), field.is_nullable())
                }
            })
            .collect::<VortexResult<Vec<_>>>()?,
        value.len(),
        nulls(value.nulls(), nullable)?,
    )?
    .into_array())
}

impl FromArrowArray<&ArrowStructArray> for ArrayRef {
    fn from_arrow(value: &ArrowStructArray, nullable: bool) -> VortexResult<Self> {
        from_arrow_struct(value, nullable)
    }
}

/// Conversion of an Arrow (large) list array into a Vortex `List` array.
pub fn from_arrow_list<O: IntegerPType + OffsetSizeTrait>(
    value: &GenericListArray<O>,
    nullable: bool,
) -> VortexResult<ArrayRef> {
    // Extract the validity of the underlying element array.
    let elements_are_nullable = match value.data_type() {
        DataType::List(field) => field.is_nullable(),
        DataType::LargeList(field) => field.is_nullable(),
        dt => vortex_panic!("Invalid data type for ListArray: {dt}"),
    };

    let (offsets, referenced) = trim_offsets(value.offsets());
    let elements = from_arrow_dyn(
        value
            .values()
            .slice(referenced.start, referenced.len())
            .as_ref(),
        elements_are_nullable,
    )?;
    let nulls = nulls(value.nulls(), nullable)?;

    // `offsets` are always non-nullable.
    Ok(ListArray::try_new(elements, offsets.into_array(), nulls)?.into_array())
}

impl<O: IntegerPType + OffsetSizeTrait> FromArrowArray<&GenericListArray<O>> for ArrayRef {
    fn from_arrow(value: &GenericListArray<O>, nullable: bool) -> VortexResult<Self> {
        from_arrow_list(value, nullable)
    }
}

/// Conversion of an Arrow (large) list-view array into a Vortex `ListView` array.
pub fn from_arrow_list_view<O: OffsetSizeTrait + NativePType>(
    array: &GenericListViewArray<O>,
    nullable: bool,
) -> VortexResult<ArrayRef> {
    // Extract the validity of the underlying element array.
    let elements_are_nullable = match array.data_type() {
        DataType::ListView(field) => field.is_nullable(),
        DataType::LargeListView(field) => field.is_nullable(),
        dt => vortex_panic!("Invalid data type for ListViewArray: {dt}"),
    };

    let elements = from_arrow_dyn(array.values().as_ref(), elements_are_nullable)?;

    // `offsets` and `sizes` are always non-nullable.
    let offsets = array.offsets().clone().into_array();
    let sizes = array.sizes().clone().into_array();
    let nulls = nulls(array.nulls(), nullable)?;

    Ok(ListViewArray::try_new(elements, offsets, sizes, nulls)?.into_array())
}

impl<O: OffsetSizeTrait + NativePType> FromArrowArray<&GenericListViewArray<O>> for ArrayRef {
    fn from_arrow(array: &GenericListViewArray<O>, nullable: bool) -> VortexResult<Self> {
        from_arrow_list_view(array, nullable)
    }
}

/// Conversion of an Arrow fixed-size list array into a Vortex `FixedSizeList` array.
pub fn from_arrow_fixed_size_list(
    array: &ArrowFixedSizeListArray,
    nullable: bool,
) -> VortexResult<ArrayRef> {
    let DataType::FixedSizeList(field, list_size) = array.data_type() else {
        vortex_panic!("Invalid data type for ListArray: {}", array.data_type());
    };

    Ok(FixedSizeListArray::try_new(
        from_arrow_dyn(array.values().as_ref(), field.is_nullable())?,
        *list_size as u32,
        nulls(array.nulls(), nullable)?,
        array.len(),
    )?
    .into_array())
}

impl FromArrowArray<&ArrowFixedSizeListArray> for ArrayRef {
    fn from_arrow(array: &ArrowFixedSizeListArray, nullable: bool) -> VortexResult<Self> {
        from_arrow_fixed_size_list(array, nullable)
    }
}

pub(crate) fn map_from_arrow_parts(
    entries: ArrayRef,
    offsets: &OffsetBuffer<i32>,
    arrow_nulls: Option<&NullBuffer>,
    keys_sorted: bool,
    nullable: bool,
) -> VortexResult<ArrayRef> {
    let DType::Struct(struct_dtype, Nullability::NonNullable) = entries.dtype() else {
        vortex_bail!(
            "Arrow map entries must import as non-nullable struct, got {}",
            entries.dtype()
        );
    };
    vortex_ensure_eq!(
        struct_dtype.nfields(),
        2,
        "Arrow map entries struct must contain exactly two fields"
    );

    let key_dtype = struct_dtype
        .field_by_index(0)
        .ok_or_else(|| vortex_err!("Arrow map entries struct missing key field"))?;
    vortex_ensure!(
        !key_dtype.is_nullable(),
        "Arrow map key field must be non-nullable"
    );
    let value_dtype = struct_dtype
        .field_by_index(1)
        .ok_or_else(|| vortex_err!("Arrow map entries struct missing value field"))?;
    let map_dtype = MapDType::try_new(key_dtype, value_dtype, keys_sorted)?;
    let entries_struct = entries.as_::<Struct>();
    let entries = StructArray::try_new(
        FieldNames::from(["key", "value"]),
        vec![
            entries_struct.unmasked_field(0).clone(),
            entries_struct.unmasked_field(1).clone(),
        ],
        entries.len(),
        Validity::NonNullable,
    )?
    .into_array();

    let len = offsets.len() - 1;
    let sizes = Buffer::<i32>::from_iter(offsets.windows(2).map(|window| window[1] - window[0]))
        .into_array();
    let offsets = offsets.inner().slice(0, len).into_array();
    let validity = nulls(arrow_nulls, nullable)?;
    let entries = ListViewArray::try_new(entries, offsets, sizes, validity)?;
    // SAFETY: Arrow Map offsets are sorted list offsets. Vortex sizes are adjacent offset
    // differences, so every view's end equals the next view's start. Sliced Arrow arrays may
    // leave leading or trailing entries unreferenced, which zero-copy-to-list permits.
    let entries = unsafe { entries.with_zero_copy_to_list(true) };

    Ok(MapArray::try_new(map_dtype, entries)?.into_array())
}

/// Conversion of an Arrow map array into a Vortex `Map` array.
pub fn from_arrow_map(array: &ArrowMapArray, nullable: bool) -> VortexResult<ArrayRef> {
    let DataType::Map(_, keys_sorted) = array.data_type() else {
        vortex_panic!("Invalid data type for MapArray: {}", array.data_type());
    };
    let entries = from_arrow_struct(array.entries(), false)?;
    map_from_arrow_parts(
        entries,
        array.offsets(),
        array.nulls(),
        *keys_sorted,
        nullable,
    )
}

impl FromArrowArray<&ArrowMapArray> for ArrayRef {
    fn from_arrow(array: &ArrowMapArray, nullable: bool) -> VortexResult<Self> {
        from_arrow_map(array, nullable)
    }
}

/// Conversion of an Arrow null array into a Vortex `Null` array.
pub fn from_arrow_null(value: &ArrowNullArray, nullable: bool) -> VortexResult<ArrayRef> {
    vortex_ensure!(
        nullable,
        "Cannot convert an Arrow NullArray into a non-nullable Vortex array"
    );
    Ok(NullArray::new(value.len()).into_array())
}

impl FromArrowArray<&ArrowNullArray> for ArrayRef {
    fn from_arrow(value: &ArrowNullArray, nullable: bool) -> VortexResult<Self> {
        from_arrow_null(value, nullable)
    }
}

/// Conversion of an Arrow dictionary array into a Vortex `Dict` array.
pub fn from_arrow_dictionary<K: ArrowDictionaryKeyType>(
    array: &DictionaryArray<K>,
    nullable: bool,
) -> VortexResult<DictArray> {
    let keys = AnyDictionaryArray::keys(array);
    let keys = from_arrow_dyn(keys, keys.is_nullable())?;
    let values = from_arrow_dyn(array.values().as_ref(), nullable)?;
    // SAFETY: we assume that Arrow has checked the invariants on construction.
    Ok(unsafe { DictArray::new_unchecked(keys, values) })
}

impl<K: ArrowDictionaryKeyType> FromArrowArray<&DictionaryArray<K>> for DictArray {
    fn from_arrow(array: &DictionaryArray<K>, nullable: bool) -> VortexResult<Self> {
        from_arrow_dictionary(array, nullable)
    }
}

/// Convert an Arrow [`NullBuffer`] into a Vortex [`Validity`] for an array of the given
/// nullability.
///
/// # Errors
///
/// Returns an error if `nullable` is false but the null buffer contains nulls.
pub fn nulls(nulls: Option<&NullBuffer>, nullable: bool) -> VortexResult<Validity> {
    if nullable {
        Ok(nulls
            .map(|nulls| {
                if nulls.null_count() == nulls.len() {
                    Validity::AllInvalid
                } else {
                    Validity::from(BitBuffer::from(nulls.inner().clone()))
                }
            })
            .unwrap_or(Validity::AllValid))
    } else {
        let null_count = nulls.map(NullBuffer::null_count).unwrap_or(0);
        vortex_ensure_eq!(
            null_count,
            0,
            "Cannot convert an Arrow array containing nulls into a non-nullable Vortex array"
        );
        Ok(Validity::NonNullable)
    }
}

/// Canonical conversion of any supported Arrow array into a Vortex array, dispatching on the
/// Arrow [`DataType`].
pub fn from_arrow_dyn(array: &dyn ArrowArray, nullable: bool) -> VortexResult<ArrayRef> {
    match array.data_type() {
        DataType::Boolean => from_arrow_boolean(array.as_boolean(), nullable),
        DataType::UInt8 => from_arrow_primitive(array.as_primitive::<UInt8Type>(), nullable),
        DataType::UInt16 => from_arrow_primitive(array.as_primitive::<UInt16Type>(), nullable),
        DataType::UInt32 => from_arrow_primitive(array.as_primitive::<UInt32Type>(), nullable),
        DataType::UInt64 => from_arrow_primitive(array.as_primitive::<UInt64Type>(), nullable),
        DataType::Int8 => from_arrow_primitive(array.as_primitive::<Int8Type>(), nullable),
        DataType::Int16 => from_arrow_primitive(array.as_primitive::<Int16Type>(), nullable),
        DataType::Int32 => from_arrow_primitive(array.as_primitive::<Int32Type>(), nullable),
        DataType::Int64 => from_arrow_primitive(array.as_primitive::<Int64Type>(), nullable),
        DataType::Float16 => from_arrow_primitive(array.as_primitive::<Float16Type>(), nullable),
        DataType::Float32 => from_arrow_primitive(array.as_primitive::<Float32Type>(), nullable),
        DataType::Float64 => from_arrow_primitive(array.as_primitive::<Float64Type>(), nullable),
        DataType::Utf8 => from_arrow_bytes(array.as_string::<i32>(), nullable),
        DataType::LargeUtf8 => from_arrow_bytes(array.as_string::<i64>(), nullable),
        DataType::Binary => from_arrow_bytes(array.as_binary::<i32>(), nullable),
        DataType::LargeBinary => from_arrow_bytes(array.as_binary::<i64>(), nullable),
        DataType::BinaryView => from_arrow_byte_view(array.as_binary_view(), nullable),
        DataType::Utf8View => from_arrow_byte_view(array.as_string_view(), nullable),
        DataType::Struct(_) => from_arrow_struct(array.as_struct(), nullable),
        DataType::List(_) => from_arrow_list(array.as_list::<i32>(), nullable),
        DataType::LargeList(_) => from_arrow_list(array.as_list::<i64>(), nullable),
        DataType::ListView(_) => from_arrow_list_view(array.as_list_view::<i32>(), nullable),
        DataType::LargeListView(_) => from_arrow_list_view(array.as_list_view::<i64>(), nullable),
        DataType::FixedSizeList(..) => {
            from_arrow_fixed_size_list(array.as_fixed_size_list(), nullable)
        }
        DataType::Map(..) => from_arrow_map(array.as_map(), nullable),
        DataType::Null => from_arrow_null(as_null_array(array), nullable),
        DataType::Timestamp(u, _) => match u {
            ArrowTimeUnit::Second => {
                from_arrow_temporal(array.as_primitive::<TimestampSecondType>(), nullable)
            }
            ArrowTimeUnit::Millisecond => {
                from_arrow_temporal(array.as_primitive::<TimestampMillisecondType>(), nullable)
            }
            ArrowTimeUnit::Microsecond => {
                from_arrow_temporal(array.as_primitive::<TimestampMicrosecondType>(), nullable)
            }
            ArrowTimeUnit::Nanosecond => {
                from_arrow_temporal(array.as_primitive::<TimestampNanosecondType>(), nullable)
            }
        },
        DataType::Date32 => from_arrow_temporal(array.as_primitive::<Date32Type>(), nullable),
        DataType::Date64 => from_arrow_temporal(array.as_primitive::<Date64Type>(), nullable),
        DataType::Time32(u) => match u {
            ArrowTimeUnit::Second => {
                from_arrow_temporal(array.as_primitive::<Time32SecondType>(), nullable)
            }
            ArrowTimeUnit::Millisecond => {
                from_arrow_temporal(array.as_primitive::<Time32MillisecondType>(), nullable)
            }
            ArrowTimeUnit::Microsecond | ArrowTimeUnit::Nanosecond => unreachable!(),
        },
        DataType::Time64(u) => match u {
            ArrowTimeUnit::Microsecond => {
                from_arrow_temporal(array.as_primitive::<Time64MicrosecondType>(), nullable)
            }
            ArrowTimeUnit::Nanosecond => {
                from_arrow_temporal(array.as_primitive::<Time64NanosecondType>(), nullable)
            }
            ArrowTimeUnit::Second | ArrowTimeUnit::Millisecond => unreachable!(),
        },
        DataType::Decimal32(..) => {
            from_arrow_decimal32(array.as_primitive::<Decimal32Type>(), nullable)
        }
        DataType::Decimal64(..) => {
            from_arrow_decimal64(array.as_primitive::<Decimal64Type>(), nullable)
        }
        DataType::Decimal128(..) => {
            from_arrow_decimal128(array.as_primitive::<Decimal128Type>(), nullable)
        }
        DataType::Decimal256(..) => {
            from_arrow_decimal256(array.as_primitive::<Decimal256Type>(), nullable)
        }
        DataType::Dictionary(key_type, _) => match key_type.as_ref() {
            DataType::Int8 => Ok(from_arrow_dictionary(
                array.as_dictionary::<Int8Type>(),
                nullable,
            )?
            .into_array()),
            DataType::Int16 => Ok(from_arrow_dictionary(
                array.as_dictionary::<Int16Type>(),
                nullable,
            )?
            .into_array()),
            DataType::Int32 => Ok(from_arrow_dictionary(
                array.as_dictionary::<Int32Type>(),
                nullable,
            )?
            .into_array()),
            DataType::Int64 => Ok(from_arrow_dictionary(
                array.as_dictionary::<Int64Type>(),
                nullable,
            )?
            .into_array()),
            DataType::UInt8 => Ok(from_arrow_dictionary(
                array.as_dictionary::<UInt8Type>(),
                nullable,
            )?
            .into_array()),
            DataType::UInt16 => Ok(from_arrow_dictionary(
                array.as_dictionary::<UInt16Type>(),
                nullable,
            )?
            .into_array()),
            DataType::UInt32 => Ok(from_arrow_dictionary(
                array.as_dictionary::<UInt32Type>(),
                nullable,
            )?
            .into_array()),
            DataType::UInt64 => Ok(from_arrow_dictionary(
                array.as_dictionary::<UInt64Type>(),
                nullable,
            )?
            .into_array()),
            key_dt => vortex_bail!("Unsupported dictionary key type: {key_dt}"),
        },
        dt => vortex_bail!("Array encoding not implemented for Arrow data type {dt}"),
    }
}

impl FromArrowArray<&dyn ArrowArray> for ArrayRef {
    fn from_arrow(array: &dyn ArrowArray, nullable: bool) -> VortexResult<Self> {
        from_arrow_dyn(array, nullable)
    }
}

/// Canonical conversion of an Arrow [`RecordBatch`] into a Vortex struct array.
pub fn from_arrow_batch(batch: &RecordBatch, nullable: bool) -> VortexResult<ArrayRef> {
    from_arrow_struct(&arrow_array::StructArray::from(batch.clone()), nullable)
}

impl FromArrowArray<RecordBatch> for ArrayRef {
    fn from_arrow(array: RecordBatch, nullable: bool) -> VortexResult<Self> {
        from_arrow_batch(&array, nullable)
    }
}

impl FromArrowArray<&RecordBatch> for ArrayRef {
    fn from_arrow(array: &RecordBatch, nullable: bool) -> VortexResult<Self> {
        from_arrow_batch(array, nullable)
    }
}

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

    use arrow_array::Array as ArrowArray;
    use arrow_array::BinaryArray;
    use arrow_array::BooleanArray;
    use arrow_array::Date32Array;
    use arrow_array::Date64Array;
    use arrow_array::FixedSizeListArray as ArrowFixedSizeListArray;
    use arrow_array::Float32Array;
    use arrow_array::Float64Array;
    use arrow_array::GenericListViewArray;
    use arrow_array::Int8Array;
    use arrow_array::Int16Array;
    use arrow_array::Int32Array;
    use arrow_array::Int64Array;
    use arrow_array::LargeBinaryArray;
    use arrow_array::LargeStringArray;
    use arrow_array::NullArray;
    use arrow_array::RecordBatch;
    use arrow_array::StringArray;
    use arrow_array::StructArray;
    use arrow_array::Time32MillisecondArray;
    use arrow_array::Time32SecondArray;
    use arrow_array::Time64MicrosecondArray;
    use arrow_array::Time64NanosecondArray;
    use arrow_array::TimestampMicrosecondArray;
    use arrow_array::TimestampMillisecondArray;
    use arrow_array::TimestampNanosecondArray;
    use arrow_array::TimestampSecondArray;
    use arrow_array::UInt8Array;
    use arrow_array::UInt16Array;
    use arrow_array::UInt32Array;
    use arrow_array::UInt64Array;
    use arrow_array::builder::BinaryViewBuilder;
    use arrow_array::builder::Decimal128Builder;
    use arrow_array::builder::Decimal256Builder;
    use arrow_array::builder::Int32Builder;
    use arrow_array::builder::Int64Builder;
    use arrow_array::builder::LargeListBuilder;
    use arrow_array::builder::ListBuilder;
    use arrow_array::builder::MapBuilder as ArrowMapBuilder;
    use arrow_array::builder::StringBuilder;
    use arrow_array::builder::StringViewBuilder;
    use arrow_array::new_null_array;
    use arrow_array::types::ArrowPrimitiveType;
    use arrow_array::types::Float16Type;
    use arrow_buffer::BooleanBuffer;
    use arrow_buffer::Buffer as ArrowBuffer;
    use arrow_buffer::OffsetBuffer;
    use arrow_buffer::ScalarBuffer;
    use arrow_schema::DataType;
    use arrow_schema::Field;
    use arrow_schema::Fields;
    use arrow_schema::Schema;
    use rstest::rstest;
    use vortex_array::ArrayRef;
    use vortex_array::IntoArray;
    use vortex_array::VortexSessionExecute as _;
    use vortex_array::array_session;
    use vortex_array::arrays::BoolArray;
    use vortex_array::arrays::Decimal;
    use vortex_array::arrays::DecimalArray;
    use vortex_array::arrays::FixedSizeList;
    use vortex_array::arrays::FixedSizeListArray as VortexFixedSizeListArray;
    use vortex_array::arrays::List;
    use vortex_array::arrays::ListArray;
    use vortex_array::arrays::ListView;
    use vortex_array::arrays::ListViewArray;
    use vortex_array::arrays::NullArray as VortexNullArray;
    use vortex_array::arrays::Primitive;
    use vortex_array::arrays::PrimitiveArray;
    use vortex_array::arrays::Struct;
    use vortex_array::arrays::StructArray as VortexStructArray;
    use vortex_array::arrays::TemporalArray;
    use vortex_array::arrays::VarBinArray;
    use vortex_array::arrays::VarBinView;
    use vortex_array::arrays::fixed_size_list::FixedSizeListArrayExt;
    use vortex_array::arrays::fixed_size_list::FixedSizeListArraySlotsExt;
    use vortex_array::arrays::list::ListArraySlotsExt;
    use vortex_array::arrays::listview::ListViewArraySlotsExt;
    use vortex_array::arrays::struct_::StructArrayExt;
    use vortex_array::assert_arrays_eq;
    use vortex_array::dtype::DType;
    use vortex_array::dtype::DecimalDType;
    use vortex_array::dtype::Nullability;
    use vortex_array::dtype::PType;
    use vortex_array::dtype::i256;
    use vortex_array::extension::datetime::TimeUnit;
    use vortex_array::extension::datetime::Timestamp;
    use vortex_array::scalar::Scalar;
    use vortex_array::validity::Validity;
    use vortex_buffer::buffer;
    use vortex_error::VortexResult;

    use crate::FromArrowArray as _;
    use crate::IntoVortexArray as _;
    use crate::convert::from_arrow_batch;
    use crate::convert::from_arrow_dyn;

    #[rstest]
    #[case::i8(
        Arc::new(Int8Array::from(vec![Some(1), None, Some(3), Some(4)])),
        Arc::new(Int8Array::from(vec![1, 2, 3, 4])),
        PType::I8,
    )]
    #[case::i16(
        Arc::new(Int16Array::from(vec![Some(100), None, Some(300), Some(400)])),
        Arc::new(Int16Array::from(vec![100, 200, 300, 400])),
        PType::I16,
    )]
    #[case::i32(
        Arc::new(Int32Array::from(vec![Some(1000), None, Some(3000), Some(4000)])),
        Arc::new(Int32Array::from(vec![1000, 2000, 3000, 4000])),
        PType::I32,
    )]
    #[case::i64(
        Arc::new(Int64Array::from(vec![Some(10000), None, Some(30000), Some(40000)])),
        Arc::new(Int64Array::from(vec![10000_i64, 20000, 30000, 40000])),
        PType::I64,
    )]
    #[case::u8(
        Arc::new(UInt8Array::from(vec![Some(1), None, Some(3), Some(4)])),
        Arc::new(UInt8Array::from(vec![1_u8, 2, 3, 4])),
        PType::U8,
    )]
    #[case::u16(
        Arc::new(UInt16Array::from(vec![Some(100), None, Some(300), Some(400)])),
        Arc::new(UInt16Array::from(vec![100_u16, 200, 300, 400])),
        PType::U16,
    )]
    #[case::u32(
        Arc::new(UInt32Array::from(vec![Some(1000), None, Some(3000), Some(4000)])),
        Arc::new(UInt32Array::from(vec![1000_u32, 2000, 3000, 4000])),
        PType::U32,
    )]
    #[case::u64(
        Arc::new(UInt64Array::from(vec![Some(10000), None, Some(30000), Some(40000)])),
        Arc::new(UInt64Array::from(vec![10000_u64, 20000, 30000, 40000])),
        PType::U64,
    )]
    #[case::f32(
        Arc::new(Float32Array::from(vec![Some(1.5), None, Some(3.5), Some(4.5)])),
        Arc::new(Float32Array::from(vec![1.5_f32, 2.5, 3.5, 4.5])),
        PType::F32,
    )]
    #[case::f64(
        Arc::new(Float64Array::from(vec![Some(1.5), None, Some(3.5), Some(4.5)])),
        Arc::new(Float64Array::from(vec![1.5_f64, 2.5, 3.5, 4.5])),
        PType::F64,
    )]
    fn test_primitive_array_conversion(
        #[case] nullable: Arc<dyn ArrowArray>,
        #[case] non_nullable: Arc<dyn ArrowArray>,
        #[case] expected_ptype: PType,
    ) {
        let v_null = ArrayRef::from_arrow(nullable.as_ref(), true).unwrap();
        let v_non_null = ArrayRef::from_arrow(non_nullable.as_ref(), false).unwrap();
        assert_eq!(v_null.len(), 4);
        assert_eq!(v_non_null.len(), 4);
        assert_eq!(v_null.as_::<Primitive>().ptype(), expected_ptype);
        assert_eq!(v_non_null.as_::<Primitive>().ptype(), expected_ptype);
        let (expected_nullable, expected_non_nullable) = match expected_ptype {
            PType::I8 => (
                PrimitiveArray::from_option_iter([Some(1i8), None, Some(3), Some(4)]),
                PrimitiveArray::from_iter([1i8, 2, 3, 4]),
            ),
            PType::I16 => (
                PrimitiveArray::from_option_iter([Some(100i16), None, Some(300), Some(400)]),
                PrimitiveArray::from_iter([100i16, 200, 300, 400]),
            ),
            PType::I32 => (
                PrimitiveArray::from_option_iter([Some(1000i32), None, Some(3000), Some(4000)]),
                PrimitiveArray::from_iter([1000i32, 2000, 3000, 4000]),
            ),
            PType::I64 => (
                PrimitiveArray::from_option_iter([Some(10000i64), None, Some(30000), Some(40000)]),
                PrimitiveArray::from_iter([10000i64, 20000, 30000, 40000]),
            ),
            PType::U8 => (
                PrimitiveArray::from_option_iter([Some(1u8), None, Some(3), Some(4)]),
                PrimitiveArray::from_iter([1u8, 2, 3, 4]),
            ),
            PType::U16 => (
                PrimitiveArray::from_option_iter([Some(100u16), None, Some(300), Some(400)]),
                PrimitiveArray::from_iter([100u16, 200, 300, 400]),
            ),
            PType::U32 => (
                PrimitiveArray::from_option_iter([Some(1000u32), None, Some(3000), Some(4000)]),
                PrimitiveArray::from_iter([1000u32, 2000, 3000, 4000]),
            ),
            PType::U64 => (
                PrimitiveArray::from_option_iter([Some(10000u64), None, Some(30000), Some(40000)]),
                PrimitiveArray::from_iter([10000u64, 20000, 30000, 40000]),
            ),
            PType::F32 => (
                PrimitiveArray::from_option_iter([Some(1.5f32), None, Some(3.5), Some(4.5)]),
                PrimitiveArray::from_iter([1.5f32, 2.5, 3.5, 4.5]),
            ),
            PType::F64 => (
                PrimitiveArray::from_option_iter([Some(1.5f64), None, Some(3.5), Some(4.5)]),
                PrimitiveArray::from_iter([1.5f64, 2.5, 3.5, 4.5]),
            ),
            _ => unreachable!(),
        };
        assert_arrays_eq!(
            v_null,
            expected_nullable,
            &mut array_session().create_execution_ctx()
        );
        assert_arrays_eq!(
            v_non_null,
            expected_non_nullable,
            &mut array_session().create_execution_ctx()
        );
    }

    #[test]
    fn test_float16_array_conversion() {
        let values = vec![
            Some(<Float16Type as ArrowPrimitiveType>::Native::from_f32(1.5)),
            None,
            Some(<Float16Type as ArrowPrimitiveType>::Native::from_f32(3.5)),
        ];
        let arrow_array = arrow_array::PrimitiveArray::<Float16Type>::from(values);
        let vortex_array = ArrayRef::from_arrow(&arrow_array, true).unwrap();

        let non_null_values = vec![
            <Float16Type as ArrowPrimitiveType>::Native::from_f32(1.5),
            <Float16Type as ArrowPrimitiveType>::Native::from_f32(2.5),
        ];
        let arrow_array_non_null =
            arrow_array::PrimitiveArray::<Float16Type>::from(non_null_values);
        let vortex_array_non_null = ArrayRef::from_arrow(&arrow_array_non_null, false).unwrap();

        assert_eq!(vortex_array.len(), 3);
        assert_eq!(vortex_array_non_null.len(), 2);

        // Verify metadata - should be PrimitiveArray with F16 ptype
        let primitive_array = vortex_array.as_::<Primitive>();
        assert_eq!(primitive_array.ptype(), PType::F16);

        let primitive_array_non_null = vortex_array_non_null.as_::<Primitive>();
        assert_eq!(primitive_array_non_null.ptype(), PType::F16);
        assert_arrays_eq!(
            vortex_array,
            PrimitiveArray::from_option_iter([
                Some(<Float16Type as ArrowPrimitiveType>::Native::from_f32(1.5)),
                None,
                Some(<Float16Type as ArrowPrimitiveType>::Native::from_f32(3.5))
            ]),
            &mut array_session().create_execution_ctx()
        );
        assert_arrays_eq!(
            vortex_array_non_null,
            PrimitiveArray::from_iter([
                <Float16Type as ArrowPrimitiveType>::Native::from_f32(1.5),
                <Float16Type as ArrowPrimitiveType>::Native::from_f32(2.5)
            ]),
            &mut array_session().create_execution_ctx()
        );
    }

    // Test decimal array conversions
    #[test]
    fn test_decimal128_array_conversion() {
        let mut builder = Decimal128Builder::with_capacity(4);
        builder.append_value(12345);
        builder.append_null();
        builder.append_value(67890);
        builder.append_value(11111);
        let decimal_array = builder.finish().with_precision_and_scale(10, 2).unwrap();

        let vortex_array = ArrayRef::from_arrow(&decimal_array, true).unwrap();
        assert_eq!(vortex_array.len(), 4);

        let mut builder_non_null = Decimal128Builder::with_capacity(3);
        builder_non_null.append_value(12345);
        builder_non_null.append_value(67890);
        builder_non_null.append_value(11111);
        let decimal_array_non_null = builder_non_null
            .finish()
            .with_precision_and_scale(10, 2)
            .unwrap();

        let vortex_array_non_null = ArrayRef::from_arrow(&decimal_array_non_null, false).unwrap();
        assert_eq!(vortex_array_non_null.len(), 3);

        // Verify metadata - should be DecimalArray with correct precision and scale
        let decimal_vortex_array = vortex_array.as_::<Decimal>();
        assert_eq!(decimal_vortex_array.decimal_dtype().precision(), 10);
        assert_eq!(decimal_vortex_array.decimal_dtype().scale(), 2);

        let decimal_vortex_array_non_null = vortex_array_non_null.as_::<Decimal>();
        assert_eq!(
            decimal_vortex_array_non_null.decimal_dtype().precision(),
            10
        );
        assert_eq!(decimal_vortex_array_non_null.decimal_dtype().scale(), 2);
        assert_arrays_eq!(
            vortex_array,
            DecimalArray::from_option_iter(
                [Some(12345i128), None, Some(67890i128), Some(11111i128)],
                DecimalDType::new(10, 2)
            ),
            &mut array_session().create_execution_ctx()
        );
        assert_arrays_eq!(
            vortex_array_non_null,
            DecimalArray::from_iter([12345i128, 67890i128, 11111i128], DecimalDType::new(10, 2)),
            &mut array_session().create_execution_ctx()
        );
    }

    #[test]
    fn test_decimal256_array_conversion() {
        let mut builder = Decimal256Builder::with_capacity(4);
        builder.append_value(arrow_buffer::i256::from_i128(12345));
        builder.append_null();
        builder.append_value(arrow_buffer::i256::from_i128(67890));
        builder.append_value(arrow_buffer::i256::from_i128(11111));
        let decimal_array = builder.finish().with_precision_and_scale(38, 10).unwrap();

        let vortex_array = ArrayRef::from_arrow(&decimal_array, true).unwrap();
        assert_eq!(vortex_array.len(), 4);

        let mut builder_non_null = Decimal256Builder::with_capacity(3);
        builder_non_null.append_value(arrow_buffer::i256::from_i128(12345));
        builder_non_null.append_value(arrow_buffer::i256::from_i128(67890));
        builder_non_null.append_value(arrow_buffer::i256::from_i128(11111));
        let decimal_array_non_null = builder_non_null
            .finish()
            .with_precision_and_scale(38, 10)
            .unwrap();

        let vortex_array_non_null = ArrayRef::from_arrow(&decimal_array_non_null, false).unwrap();
        assert_eq!(vortex_array_non_null.len(), 3);

        // Verify metadata - should be DecimalArray with correct precision and scale
        let decimal_vortex_array = vortex_array.as_::<Decimal>();
        assert_eq!(decimal_vortex_array.decimal_dtype().precision(), 38);
        assert_eq!(decimal_vortex_array.decimal_dtype().scale(), 10);

        let decimal_vortex_array_non_null = vortex_array_non_null.as_::<Decimal>();
        assert_eq!(
            decimal_vortex_array_non_null.decimal_dtype().precision(),
            38
        );
        assert_eq!(decimal_vortex_array_non_null.decimal_dtype().scale(), 10);
        assert_arrays_eq!(
            vortex_array,
            DecimalArray::from_option_iter(
                [
                    Some(i256::from_i128(12345)),
                    None,
                    Some(i256::from_i128(67890)),
                    Some(i256::from_i128(11111))
                ],
                DecimalDType::new(38, 10)
            ),
            &mut array_session().create_execution_ctx()
        );
        assert_arrays_eq!(
            vortex_array_non_null,
            DecimalArray::from_iter(
                [
                    i256::from_i128(12345),
                    i256::from_i128(67890),
                    i256::from_i128(11111)
                ],
                DecimalDType::new(38, 10)
            ),
            &mut array_session().create_execution_ctx()
        );
    }

    // Test temporal array conversions
    #[rstest]
    #[case::timestamp_second(
        Arc::new(TimestampSecondArray::from(vec![Some(1000), None, Some(3000), Some(4000)])),
        Arc::new(TimestampSecondArray::from(vec![1000_i64, 2000, 3000, 4000])),
    )]
    #[case::timestamp_millisecond(
        Arc::new(TimestampMillisecondArray::from(vec![Some(1000), None, Some(3000), Some(4000)])),
        Arc::new(TimestampMillisecondArray::from(vec![1000_i64, 2000, 3000, 4000])),
    )]
    #[case::timestamp_microsecond(
        Arc::new(TimestampMicrosecondArray::from(vec![Some(1000), None, Some(3000), Some(4000)])),
        Arc::new(TimestampMicrosecondArray::from(vec![1000_i64, 2000, 3000, 4000])),
    )]
    #[case::timestamp_nanosecond(
        Arc::new(TimestampNanosecondArray::from(vec![Some(1000), None, Some(3000), Some(4000)])),
        Arc::new(TimestampNanosecondArray::from(vec![1000_i64, 2000, 3000, 4000])),
    )]
    #[case::time32_second(
        Arc::new(Time32SecondArray::from(vec![Some(1000), None, Some(3000), Some(4000)])),
        Arc::new(Time32SecondArray::from(vec![1000_i32, 2000, 3000, 4000])),
    )]
    #[case::time32_millisecond(
        Arc::new(Time32MillisecondArray::from(vec![Some(1000), None, Some(3000), Some(4000)])),
        Arc::new(Time32MillisecondArray::from(vec![1000_i32, 2000, 3000, 4000])),
    )]
    #[case::time64_microsecond(
        Arc::new(Time64MicrosecondArray::from(vec![Some(1000), None, Some(3000), Some(4000)])),
        Arc::new(Time64MicrosecondArray::from(vec![1000_i64, 2000, 3000, 4000])),
    )]
    #[case::time64_nanosecond(
        Arc::new(Time64NanosecondArray::from(vec![Some(1000), None, Some(3000), Some(4000)])),
        Arc::new(Time64NanosecondArray::from(vec![1000_i64, 2000, 3000, 4000])),
    )]
    #[case::date32(
        Arc::new(Date32Array::from(vec![Some(18000), None, Some(18002), Some(18003)])),
        Arc::new(Date32Array::from(vec![18000_i32, 18001, 18002, 18003])),
    )]
    #[case::date64(
        Arc::new(Date64Array::from(vec![Some(1555200000000), None, Some(1555286400000), Some(1555372800000)]
        )),
        Arc::new(Date64Array::from(vec![1555200000000_i64, 1555213600000, 1555286400000, 1555372800000]
        )),
    )]
    fn test_temporal_array_conversion(
        #[case] nullable: Arc<dyn ArrowArray>,
        #[case] non_nullable: Arc<dyn ArrowArray>,
    ) {
        let v_null = ArrayRef::from_arrow(nullable.as_ref(), true).unwrap();
        let v_non_null = ArrayRef::from_arrow(non_nullable.as_ref(), false).unwrap();
        assert_eq!(v_null.len(), 4);
        assert_eq!(v_non_null.len(), 4);
        let expected = |nullable: bool| {
            let i32_values = if nullable {
                PrimitiveArray::from_option_iter([Some(1000i32), None, Some(3000), Some(4000)])
            } else {
                PrimitiveArray::from_iter([1000i32, 2000, 3000, 4000])
            }
            .into_array();
            let i64_values = if nullable {
                PrimitiveArray::from_option_iter([Some(1000i64), None, Some(3000), Some(4000)])
            } else {
                PrimitiveArray::from_iter([1000i64, 2000, 3000, 4000])
            }
            .into_array();
            match non_nullable.data_type() {
                DataType::Timestamp(unit, _) => {
                    let unit = match unit {
                        arrow_schema::TimeUnit::Second => TimeUnit::Seconds,
                        arrow_schema::TimeUnit::Millisecond => TimeUnit::Milliseconds,
                        arrow_schema::TimeUnit::Microsecond => TimeUnit::Microseconds,
                        arrow_schema::TimeUnit::Nanosecond => TimeUnit::Nanoseconds,
                    };
                    TemporalArray::new_timestamp(i64_values, unit, None).into_array()
                }
                DataType::Time32(arrow_schema::TimeUnit::Second) => {
                    TemporalArray::new_time(i32_values, TimeUnit::Seconds).into_array()
                }
                DataType::Time32(arrow_schema::TimeUnit::Millisecond) => {
                    TemporalArray::new_time(i32_values, TimeUnit::Milliseconds).into_array()
                }
                DataType::Time64(arrow_schema::TimeUnit::Microsecond) => {
                    TemporalArray::new_time(i64_values, TimeUnit::Microseconds).into_array()
                }
                DataType::Time64(arrow_schema::TimeUnit::Nanosecond) => {
                    TemporalArray::new_time(i64_values, TimeUnit::Nanoseconds).into_array()
                }
                DataType::Date32 => TemporalArray::new_date(
                    if nullable {
                        PrimitiveArray::from_option_iter([
                            Some(18000i32),
                            None,
                            Some(18002),
                            Some(18003),
                        ])
                    } else {
                        PrimitiveArray::from_iter([18000i32, 18001, 18002, 18003])
                    }
                    .into_array(),
                    TimeUnit::Days,
                )
                .into_array(),
                DataType::Date64 => TemporalArray::new_date(
                    if nullable {
                        PrimitiveArray::from_option_iter([
                            Some(1555200000000i64),
                            None,
                            Some(1555286400000),
                            Some(1555372800000),
                        ])
                    } else {
                        PrimitiveArray::from_iter([
                            1555200000000i64,
                            1555213600000,
                            1555286400000,
                            1555372800000,
                        ])
                    }
                    .into_array(),
                    TimeUnit::Milliseconds,
                )
                .into_array(),
                _ => unreachable!(),
            }
        };
        assert_arrays_eq!(
            v_null,
            expected(true),
            &mut array_session().create_execution_ctx()
        );
        assert_arrays_eq!(
            v_non_null,
            expected(false),
            &mut array_session().create_execution_ctx()
        );
    }

    #[test]
    fn test_timestamp_timezone_microsecond_array_conversion() {
        let arrow_array =
            TimestampMicrosecondArray::from(vec![Some(1000), None, Some(3000), Some(4000)])
                .with_timezone("UTC");
        let vortex_array = ArrayRef::from_arrow(&arrow_array, true).unwrap();

        let arrow_array_non_null =
            TimestampMicrosecondArray::from(vec![1000_i64, 2000, 3000, 4000]).with_timezone("UTC");
        let vortex_array_non_null = ArrayRef::from_arrow(&arrow_array_non_null, false).unwrap();

        assert_eq!(vortex_array.len(), 4);
        assert_eq!(
            vortex_array.dtype(),
            &DType::Extension(
                Timestamp::new_with_tz(
                    TimeUnit::Microseconds,
                    Some("UTC".into()),
                    Nullability::Nullable
                )
                .erased()
            ),
        );
        assert_eq!(vortex_array_non_null.len(), 4);
        assert_eq!(
            vortex_array_non_null.dtype(),
            &DType::Extension(
                Timestamp::new_with_tz(
                    TimeUnit::Microseconds,
                    Some("UTC".into()),
                    Nullability::NonNullable
                )
                .erased()
            )
        );
        assert_arrays_eq!(
            vortex_array,
            TemporalArray::new_timestamp(
                PrimitiveArray::from_option_iter([Some(1000i64), None, Some(3000), Some(4000)])
                    .into_array(),
                TimeUnit::Microseconds,
                Some("UTC".into())
            ),
            &mut array_session().create_execution_ctx()
        );
        assert_arrays_eq!(
            vortex_array_non_null,
            TemporalArray::new_timestamp(
                buffer![1000i64, 2000, 3000, 4000].into_array(),
                TimeUnit::Microseconds,
                Some("UTC".into())
            ),
            &mut array_session().create_execution_ctx()
        );
    }

    // Test string/binary array conversions
    #[rstest]
    #[case::utf8(
        Arc::new(StringArray::from(vec![Some("hello"), None, Some("world"), Some("test")])),
        Arc::new(StringArray::from(vec!["hello", "world", "test", "vortex"])),
        DType::Utf8(Nullability::NonNullable),
    )]
    #[case::large_utf8(
        Arc::new(LargeStringArray::from(vec![Some("hello"), None, Some("world"), Some("test")])),
        Arc::new(LargeStringArray::from(vec!["hello", "world", "test", "vortex"])),
        DType::Utf8(Nullability::NonNullable),
    )]
    #[case::binary(
        Arc::new(BinaryArray::from(vec![
            Some("hello".as_bytes()), None, Some("world".as_bytes()), Some("test".as_bytes()),
        ])),
        Arc::new(BinaryArray::from(vec![
            "hello".as_bytes(), "world".as_bytes(), "test".as_bytes(), "vortex".as_bytes(),
        ])),
        DType::Binary(Nullability::NonNullable),
    )]
    #[case::large_binary(
        Arc::new(LargeBinaryArray::from(vec![
            Some("hello".as_bytes()), None, Some("world".as_bytes()), Some("test".as_bytes()),
        ])),
        Arc::new(LargeBinaryArray::from(vec![
            "hello".as_bytes(), "world".as_bytes(), "test".as_bytes(), "vortex".as_bytes(),
        ])),
        DType::Binary(Nullability::NonNullable),
    )]
    fn test_string_binary_array_conversion(
        #[case] nullable: Arc<dyn ArrowArray>,
        #[case] non_nullable: Arc<dyn ArrowArray>,
        #[case] expected_non_nullable_dtype: DType,
    ) {
        let v_null = ArrayRef::from_arrow(nullable.as_ref(), true).unwrap();
        let v_non_null = ArrayRef::from_arrow(non_nullable.as_ref(), false).unwrap();
        assert_eq!(v_null.len(), 4);
        assert_eq!(v_non_null.len(), 4);
        assert_eq!(v_null.dtype(), &expected_non_nullable_dtype.as_nullable());
        assert_eq!(v_non_null.dtype(), &expected_non_nullable_dtype);
        assert_arrays_eq!(
            v_null,
            VarBinArray::from_iter(
                [Some("hello"), None, Some("world"), Some("test")],
                expected_non_nullable_dtype.as_nullable()
            ),
            &mut array_session().create_execution_ctx()
        );
        assert_arrays_eq!(
            v_non_null,
            VarBinArray::from_iter_nonnull(
                ["hello", "world", "test", "vortex"],
                expected_non_nullable_dtype
            ),
            &mut array_session().create_execution_ctx()
        );
    }

    #[test]
    fn test_utf8_view_array_conversion() {
        let mut builder = StringViewBuilder::new();
        builder.append_value("hello");
        builder.append_null();
        builder.append_value("world");
        builder.append_value("test");
        let arrow_array = builder.finish();
        let vortex_array = ArrayRef::from_arrow(&arrow_array, true).unwrap();

        let mut builder_non_null = StringViewBuilder::new();
        builder_non_null.append_value("hello");
        builder_non_null.append_value("world");
        builder_non_null.append_value("test");
        builder_non_null.append_value("vortex");
        let arrow_array_non_null = builder_non_null.finish();
        let vortex_array_non_null = ArrayRef::from_arrow(&arrow_array_non_null, false).unwrap();

        assert_eq!(vortex_array.len(), 4);
        assert_eq!(vortex_array_non_null.len(), 4);

        // Verify metadata - should be VarBinViewArray with correct buffer count and dtype
        let varbin_view_array = vortex_array.as_::<VarBinView>();
        assert_eq!(
            varbin_view_array.data_buffers().len(),
            arrow_array.data_buffers().len()
        );
        assert_eq!(varbin_view_array.dtype(), &DType::Utf8(true.into()));

        let varbin_view_array_non_null = vortex_array_non_null.as_::<VarBinView>();
        assert_eq!(
            varbin_view_array_non_null.data_buffers().len(),
            arrow_array_non_null.data_buffers().len()
        );
        assert_eq!(
            varbin_view_array_non_null.dtype(),
            &DType::Utf8(false.into())
        );
        assert_arrays_eq!(
            vortex_array,
            VarBinArray::from_iter(
                [Some("hello"), None, Some("world"), Some("test")],
                DType::Utf8(Nullability::NonNullable).as_nullable()
            ),
            &mut array_session().create_execution_ctx()
        );
        assert_arrays_eq!(
            vortex_array_non_null,
            VarBinArray::from_iter_nonnull(
                ["hello", "world", "test", "vortex"],
                DType::Utf8(Nullability::NonNullable)
            ),
            &mut array_session().create_execution_ctx()
        );
    }

    #[test]
    fn test_binary_view_array_conversion() {
        let mut builder = BinaryViewBuilder::new();
        builder.append_value(b"hello");
        builder.append_null();
        builder.append_value(b"world");
        builder.append_value(b"test");
        let arrow_array = builder.finish();
        let vortex_array = ArrayRef::from_arrow(&arrow_array, true).unwrap();

        let mut builder_non_null = BinaryViewBuilder::new();
        builder_non_null.append_value(b"hello");
        builder_non_null.append_value(b"world");
        builder_non_null.append_value(b"test");
        builder_non_null.append_value(b"vortex");
        let arrow_array_non_null = builder_non_null.finish();
        let vortex_array_non_null = ArrayRef::from_arrow(&arrow_array_non_null, false).unwrap();

        assert_eq!(vortex_array.len(), 4);
        assert_eq!(vortex_array_non_null.len(), 4);

        // Verify metadata - should be VarBinViewArray with correct buffer count and dtype
        let varbin_view_array = vortex_array.as_::<VarBinView>();
        assert_eq!(
            varbin_view_array.data_buffers().len(),
            arrow_array.data_buffers().len()
        );
        assert_eq!(varbin_view_array.dtype(), &DType::Binary(true.into()));

        let varbin_view_array_non_null = vortex_array_non_null.as_::<VarBinView>();
        assert_eq!(
            varbin_view_array_non_null.data_buffers().len(),
            arrow_array_non_null.data_buffers().len()
        );
        assert_eq!(
            varbin_view_array_non_null.dtype(),
            &DType::Binary(false.into())
        );
        assert_arrays_eq!(
            vortex_array,
            VarBinArray::from_iter(
                [Some("hello"), None, Some("world"), Some("test")],
                DType::Binary(Nullability::NonNullable).as_nullable()
            ),
            &mut array_session().create_execution_ctx()
        );
        assert_arrays_eq!(
            vortex_array_non_null,
            VarBinArray::from_iter_nonnull(
                ["hello", "world", "test", "vortex"],
                DType::Binary(Nullability::NonNullable)
            ),
            &mut array_session().create_execution_ctx()
        );
    }

    // Test boolean array conversions
    #[test]
    fn test_boolean_array_conversion() {
        let arrow_array = BooleanArray::from(vec![Some(true), None, Some(false), Some(true)]);
        let vortex_array = ArrayRef::from_arrow(&arrow_array, true).unwrap();

        let arrow_array_non_null = BooleanArray::from(vec![true, false, true, false]);
        let vortex_array_non_null = ArrayRef::from_arrow(&arrow_array_non_null, false).unwrap();

        assert_eq!(vortex_array.len(), 4);
        assert_eq!(vortex_array_non_null.len(), 4);
        assert_arrays_eq!(
            vortex_array,
            BoolArray::from_iter([Some(true), None, Some(false), Some(true)]),
            &mut array_session().create_execution_ctx()
        );
        assert_arrays_eq!(
            vortex_array_non_null,
            BoolArray::from_iter([true, false, true, false]),
            &mut array_session().create_execution_ctx()
        );
    }

    // Test struct array conversions
    #[test]
    fn test_struct_array_conversion() {
        let fields = vec![
            Field::new("field1", DataType::Int32, true),
            Field::new("field2", DataType::Utf8, false),
        ];
        let schema = Fields::from(fields);

        let field1_data = Int32Array::from(vec![Some(1), None, Some(3)]);
        let field2_data = StringArray::from(vec!["a", "b", "c"]);

        let arrow_array = StructArray::new(
            schema.clone(),
            vec![Arc::new(field1_data), Arc::new(field2_data)],
            None,
        );

        let vortex_array = ArrayRef::from_arrow(&arrow_array, false).unwrap();
        assert_eq!(vortex_array.len(), 3);

        // Verify metadata - should be StructArray with correct field names
        let struct_vortex_array = vortex_array.as_::<Struct>();
        assert_eq!(struct_vortex_array.names().len(), 2);
        assert_eq!(struct_vortex_array.names()[0], "field1");
        assert_eq!(struct_vortex_array.names()[1], "field2");

        // Test nullable struct
        let nullable_array = StructArray::new(
            schema,
            vec![
                Arc::new(Int32Array::from(vec![Some(1), None, Some(3)])),
                Arc::new(StringArray::from(vec!["a", "b", "c"])),
            ],
            Some(arrow_buffer::NullBuffer::new(BooleanBuffer::from(vec![
                true, false, true,
            ]))),
        );

        let vortex_nullable_array = ArrayRef::from_arrow(&nullable_array, true).unwrap();
        assert_eq!(vortex_nullable_array.len(), 3);

        // Verify metadata for nullable struct
        let struct_vortex_nullable_array = vortex_nullable_array.as_::<Struct>();
        assert_eq!(struct_vortex_nullable_array.names().len(), 2);
        assert_eq!(struct_vortex_nullable_array.names()[0], "field1");
        assert_eq!(struct_vortex_nullable_array.names()[1], "field2");
        let expected = |validity| {
            VortexStructArray::new(
                ["field1", "field2"].into(),
                vec![
                    PrimitiveArray::from_option_iter([Some(1i32), None, Some(3)]).into_array(),
                    VarBinArray::from_iter_nonnull(
                        ["a", "b", "c"],
                        DType::Utf8(Nullability::NonNullable),
                    )
                    .into_array(),
                ],
                3,
                validity,
            )
        };
        assert_arrays_eq!(
            vortex_array,
            expected(Validity::NonNullable),
            &mut array_session().create_execution_ctx()
        );
        assert_arrays_eq!(
            vortex_nullable_array,
            expected(Validity::from_iter([true, false, true])),
            &mut array_session().create_execution_ctx()
        );
    }

    // Test list array conversions
    #[test]
    fn test_list_array_conversion() {
        let mut builder = ListBuilder::new(Int32Builder::new());
        builder.append_value([Some(1), None, Some(3)]);
        builder.append_null();
        builder.append_value([Some(4), Some(5)]);
        let arrow_array = builder.finish();

        let vortex_array = ArrayRef::from_arrow(&arrow_array, true).unwrap();
        assert_eq!(vortex_array.len(), 3);

        // Verify metadata - should be ListArray with correct offsets
        let list_vortex_array = vortex_array.as_::<List>();
        let offsets_array = list_vortex_array.offsets().as_::<Primitive>();
        assert_eq!(offsets_array.len(), 4); // n+1 offsets for n lists
        assert_eq!(offsets_array.ptype(), PType::I32);

        // Test non-nullable list
        let mut builder_non_null = ListBuilder::new(Int32Builder::new());
        builder_non_null.append_value([Some(1), None, Some(3)]);
        builder_non_null.append_value([Some(4), Some(5)]);
        let arrow_array_non_null = builder_non_null.finish();

        let vortex_array_non_null = ArrayRef::from_arrow(&arrow_array_non_null, false).unwrap();
        assert_eq!(vortex_array_non_null.len(), 2);

        // Verify metadata for non-nullable list
        let list_vortex_array_non_null = vortex_array_non_null.as_::<List>();
        let offsets_array_non_null = list_vortex_array_non_null.offsets().as_::<Primitive>();
        assert_eq!(offsets_array_non_null.len(), 3); // n+1 offsets for n lists
        assert_eq!(offsets_array_non_null.ptype(), PType::I32);

        let expected_elements =
            PrimitiveArray::from_option_iter([Some(1i32), None, Some(3), Some(4), Some(5)])
                .into_array();
        let expected = ListArray::new(
            expected_elements.clone(),
            buffer![0u32, 3, 3, 5].into_array(),
            Validity::from_iter([true, false, true]),
        );
        assert_arrays_eq!(
            vortex_array,
            expected,
            &mut array_session().create_execution_ctx()
        );
        let expected = ListArray::new(
            expected_elements,
            buffer![0u32, 3, 5].into_array(),
            Validity::NonNullable,
        );
        assert_arrays_eq!(
            vortex_array_non_null,
            expected,
            &mut array_session().create_execution_ctx()
        );
    }

    #[test]
    fn test_sliced_offset_arrays_import_only_their_rows() -> VortexResult<()> {
        let strings = |rows: Range<i64>| {
            StringArray::from_iter_values(rows.map(|row| format!("value-{row}")))
        };
        let lists = |rows: Range<i64>| {
            let mut builder = ListBuilder::new(Int64Builder::new());
            for row in rows {
                builder.append_value([Some(row), Some(row + 1)]);
            }
            builder.finish()
        };
        let mut ctx = array_session().create_execution_ctx();

        // A slice keeps its parent's whole child; the import must not.
        let sliced = ArrayRef::from_arrow(&strings(0..1000).slice(500, 2), false)?;
        let fresh = ArrayRef::from_arrow(&strings(500..502), false)?;
        assert_eq!(sliced.nbytes(), fresh.nbytes());
        let expected = VarBinArray::from_iter_nonnull(
            ["value-500", "value-501"],
            DType::Utf8(Nullability::NonNullable),
        );
        assert_arrays_eq!(sliced, expected, &mut ctx);
        assert_arrays_eq!(fresh, expected, &mut ctx);

        let sliced = ArrayRef::from_arrow(&lists(0..1000).slice(500, 2), false)?;
        let fresh = ArrayRef::from_arrow(&lists(500..502), false)?;
        assert_eq!(sliced.nbytes(), fresh.nbytes());
        let expected = ListArray::new(
            PrimitiveArray::from_option_iter([Some(500i64), Some(501), Some(501), Some(502)])
                .into_array(),
            buffer![0u32, 2, 4].into_array(),
            Validity::NonNullable,
        );
        assert_arrays_eq!(sliced, expected, &mut ctx);
        assert_arrays_eq!(fresh, expected, &mut ctx);
        Ok(())
    }

    #[test]
    fn test_large_list_array_conversion() {
        let mut builder = LargeListBuilder::new(Int32Builder::new());
        builder.append_value([Some(1), None, Some(3)]);
        builder.append_null();
        builder.append_value([Some(4), Some(5)]);
        let arrow_array = builder.finish();

        let vortex_array = ArrayRef::from_arrow(&arrow_array, true).unwrap();
        assert_eq!(vortex_array.len(), 3);

        // Verify metadata - should be ListArray with correct offsets (I64 for large lists)
        let list_vortex_array = vortex_array.as_::<List>();
        let offsets_array = list_vortex_array.offsets().as_::<Primitive>();
        assert_eq!(offsets_array.len(), 4); // n+1 offsets for n lists
        assert_eq!(offsets_array.ptype(), PType::I64); // Large lists use I64 offsets

        // Test non-nullable large list
        let mut builder_non_null = LargeListBuilder::new(Int32Builder::new());
        builder_non_null.append_value([Some(1), None, Some(3)]);
        builder_non_null.append_value([Some(4), Some(5)]);
        let arrow_array_non_null = builder_non_null.finish();

        let vortex_array_non_null = ArrayRef::from_arrow(&arrow_array_non_null, false).unwrap();
        assert_eq!(vortex_array_non_null.len(), 2);

        // Verify metadata for non-nullable large list
        let list_vortex_array_non_null = vortex_array_non_null.as_::<List>();
        let offsets_array_non_null = list_vortex_array_non_null.offsets().as_::<Primitive>();
        assert_eq!(offsets_array_non_null.len(), 3); // n+1 offsets for n lists
        assert_eq!(offsets_array_non_null.ptype(), PType::I64); // Large lists use I64 offsets

        let expected_elements =
            PrimitiveArray::from_option_iter([Some(1i32), None, Some(3), Some(4), Some(5)])
                .into_array();
        let expected = ListArray::new(
            expected_elements.clone(),
            buffer![0u32, 3, 3, 5].into_array(),
            Validity::from_iter([true, false, true]),
        );
        assert_arrays_eq!(
            vortex_array,
            expected,
            &mut array_session().create_execution_ctx()
        );
        let expected = ListArray::new(
            expected_elements,
            buffer![0u32, 3, 5].into_array(),
            Validity::NonNullable,
        );
        assert_arrays_eq!(
            vortex_array_non_null,
            expected,
            &mut array_session().create_execution_ctx()
        );
    }

    #[test]
    fn test_fixed_size_list_array_conversion() {
        // Create elements for the fixed-size lists
        let values = Int32Array::from(vec![
            Some(1),
            Some(2),
            Some(3), // First list
            Some(4),
            None,
            Some(6), // Second list (with null element)
            Some(7),
            Some(8),
            Some(9), // Third list
            Some(10),
            Some(11),
            Some(12), // Fourth list
        ]);

        // Create a FixedSizeListArray with list_size=3
        let field = Arc::new(Field::new("item", DataType::Int32, true));
        let arrow_array =
            ArrowFixedSizeListArray::try_new(Arc::clone(&field), 3, Arc::new(values), None)
                .unwrap();
        let vortex_array = ArrayRef::from_arrow(&arrow_array, false).unwrap();

        assert_eq!(vortex_array.len(), 4);

        // Verify metadata - should be FixedSizeListArray with correct list size
        let fsl_vortex_array = vortex_array.as_::<FixedSizeList>();
        assert_eq!(fsl_vortex_array.list_size(), 3);
        assert_eq!(fsl_vortex_array.elements().len(), 12); // 4 lists * 3 elements

        // Test nullable fixed-size list
        let values_nullable = Int32Array::from(vec![
            Some(1),
            Some(2),
            Some(3), // First list
            Some(4),
            None,
            Some(6), // Second list (will be null)
            Some(7),
            Some(8),
            Some(9), // Third list
        ]);

        // Create nulls buffer - second list is null
        let null_buffer =
            arrow_buffer::NullBuffer::new(BooleanBuffer::from(vec![true, false, true]));

        let arrow_array_nullable = ArrowFixedSizeListArray::try_new(
            field,
            3,
            Arc::new(values_nullable),
            Some(null_buffer),
        )
        .unwrap();
        let vortex_array_nullable = ArrayRef::from_arrow(&arrow_array_nullable, true).unwrap();

        assert_eq!(vortex_array_nullable.len(), 3);

        // Verify metadata for nullable array
        let fsl_vortex_array_nullable = vortex_array_nullable.as_::<FixedSizeList>();
        assert_eq!(fsl_vortex_array_nullable.list_size(), 3);
        assert_eq!(fsl_vortex_array_nullable.elements().len(), 9); // 3 lists * 3 elements
        assert_arrays_eq!(
            vortex_array,
            VortexFixedSizeListArray::new(
                PrimitiveArray::from_option_iter([
                    Some(1i32),
                    Some(2),
                    Some(3),
                    Some(4),
                    None,
                    Some(6),
                    Some(7),
                    Some(8),
                    Some(9),
                    Some(10),
                    Some(11),
                    Some(12)
                ])
                .into_array(),
                3,
                Validity::NonNullable,
                4
            ),
            &mut array_session().create_execution_ctx()
        );
        assert_arrays_eq!(
            vortex_array_nullable,
            VortexFixedSizeListArray::new(
                PrimitiveArray::from_option_iter([
                    Some(1i32),
                    Some(2),
                    Some(3),
                    Some(4),
                    None,
                    Some(6),
                    Some(7),
                    Some(8),
                    Some(9)
                ])
                .into_array(),
                3,
                Validity::from_iter([true, false, true]),
                3
            ),
            &mut array_session().create_execution_ctx()
        );
    }

    #[test]
    fn test_list_view_array_conversion() {
        // Create values array for the lists
        let values = Int32Array::from(vec![
            Some(1),
            Some(2),
            Some(3), // First list [1, 2, 3]
            Some(4),
            Some(5), // Second list [4, 5]
            Some(6), // Third list [6]
            Some(7),
            Some(8),
            Some(9),
            Some(10), // Fourth list [7, 8, 9, 10]
        ]);

        // Create offsets and sizes for ListView
        let offsets = ScalarBuffer::from(vec![0i32, 3, 5, 6]);
        let sizes = ScalarBuffer::from(vec![3i32, 2, 1, 4]);

        let field = Arc::new(Field::new("item", DataType::Int32, true));
        let arrow_array = GenericListViewArray::try_new(
            Arc::clone(&field),
            offsets.clone(),
            sizes.clone(),
            Arc::new(values.clone()),
            None,
        )
        .unwrap();

        let vortex_array = ArrayRef::from_arrow(&arrow_array, false).unwrap();
        assert_eq!(vortex_array.len(), 4);

        // Verify metadata - should be ListViewArray with correct offsets and sizes
        let list_view_vortex_array = vortex_array.as_::<ListView>();
        let offsets_array = list_view_vortex_array.offsets().as_::<Primitive>();
        let sizes_array = list_view_vortex_array.sizes().as_::<Primitive>();

        assert_eq!(offsets_array.len(), 4);
        assert_eq!(offsets_array.ptype(), PType::I32);
        assert_eq!(sizes_array.len(), 4);
        assert_eq!(sizes_array.ptype(), PType::I32);

        // Test nullable ListView
        let null_buffer =
            arrow_buffer::NullBuffer::new(BooleanBuffer::from(vec![true, false, true, true]));

        let arrow_array_nullable = GenericListViewArray::try_new(
            Arc::clone(&field),
            offsets,
            sizes,
            Arc::new(values.clone()),
            Some(null_buffer),
        )
        .unwrap();

        let vortex_array_nullable = ArrayRef::from_arrow(&arrow_array_nullable, true).unwrap();
        assert_eq!(vortex_array_nullable.len(), 4);

        // Test LargeListView (i64 offsets and sizes)
        let large_offsets = ScalarBuffer::from(vec![0i64, 3, 5, 6]);
        let large_sizes = ScalarBuffer::from(vec![3i64, 2, 1, 4]);

        let large_arrow_array = GenericListViewArray::try_new(
            field,
            large_offsets,
            large_sizes,
            Arc::new(values),
            None,
        )
        .unwrap();

        let large_vortex_array = ArrayRef::from_arrow(&large_arrow_array, false).unwrap();
        assert_eq!(large_vortex_array.len(), 4);

        // Verify metadata for large ListView
        let large_list_view_vortex_array = large_vortex_array.as_::<ListView>();
        let large_offsets_array = large_list_view_vortex_array.offsets().as_::<Primitive>();
        let large_sizes_array = large_list_view_vortex_array.sizes().as_::<Primitive>();

        assert_eq!(large_offsets_array.len(), 4);
        assert_eq!(large_offsets_array.ptype(), PType::I64); // Large ListView uses I64 offsets
        assert_eq!(large_sizes_array.len(), 4);
        assert_eq!(large_sizes_array.ptype(), PType::I64); // Large ListView uses I64 sizes
        let expected = |validity| {
            ListViewArray::new(
                PrimitiveArray::from_option_iter((1i32..=10).map(Some)).into_array(),
                buffer![0u32, 3, 5, 6].into_array(),
                buffer![3u32, 2, 1, 4].into_array(),
                validity,
            )
        };
        assert_arrays_eq!(
            vortex_array,
            expected(Validity::NonNullable),
            &mut array_session().create_execution_ctx()
        );
        assert_arrays_eq!(
            vortex_array_nullable,
            expected(Validity::from_iter([true, false, true, true])),
            &mut array_session().create_execution_ctx()
        );
        assert_arrays_eq!(
            large_vortex_array,
            expected(Validity::NonNullable),
            &mut array_session().create_execution_ctx()
        );
    }

    // Test null array conversions
    #[test]
    fn test_null_array_conversion() {
        let arrow_array = NullArray::new(5);
        let vortex_array = ArrayRef::from_arrow(&arrow_array, true).unwrap();
        assert_eq!(vortex_array.len(), 5);
        assert_arrays_eq!(
            vortex_array,
            VortexNullArray::new(5),
            &mut array_session().create_execution_ctx()
        );
    }

    // Test buffer conversions
    #[test]
    fn test_arrow_buffer_conversion() {
        let data = vec![1u8, 2, 3, 4, 5];
        let arrow_buffer = ArrowBuffer::from_vec(data);
        let vortex_array = arrow_buffer.into_array();
        assert_eq!(vortex_array.len(), 5);
        assert_arrays_eq!(
            vortex_array,
            buffer![1u8, 2, 3, 4, 5].into_array(),
            &mut array_session().create_execution_ctx()
        );
    }

    #[test]
    fn test_boolean_buffer_conversion() {
        let data = vec![true, false, true, false, true];
        let boolean_buffer = BooleanBuffer::from(data);
        let vortex_array = boolean_buffer.into_array();
        assert_eq!(vortex_array.len(), 5);
        assert_arrays_eq!(
            vortex_array,
            BoolArray::from_iter([true, false, true, false, true]),
            &mut array_session().create_execution_ctx()
        );
    }

    #[test]
    fn test_scalar_buffer_conversion() {
        let data = vec![1i32, 2, 3, 4, 5];
        let scalar_buffer = ScalarBuffer::from(data);
        let vortex_array = scalar_buffer.into_array();
        assert_eq!(vortex_array.len(), 5);
        assert_arrays_eq!(
            vortex_array,
            buffer![1i32, 2, 3, 4, 5].into_array(),
            &mut array_session().create_execution_ctx()
        );
    }

    #[test]
    fn test_offset_buffer_conversion() {
        let data = vec![0i32, 2, 5, 8, 10];
        let offset_buffer = OffsetBuffer::new(ScalarBuffer::from(data));
        let vortex_array = offset_buffer.into_array();
        assert_eq!(vortex_array.len(), 5);
        assert_arrays_eq!(
            vortex_array,
            buffer![0i32, 2, 5, 8, 10].into_array(),
            &mut array_session().create_execution_ctx()
        );
    }

    // Test RecordBatch conversions
    #[test]
    fn test_record_batch_conversion() {
        let schema = Arc::new(Schema::new(vec![
            Field::new("field1", DataType::Int32, false),
            Field::new("field2", DataType::Utf8, false),
        ]));

        let field1_data = Arc::new(Int32Array::from(vec![1, 2, 3, 4]));
        let field2_data = Arc::new(StringArray::from(vec!["a", "b", "c", "d"]));

        let record_batch = RecordBatch::try_new(schema, vec![field1_data, field2_data]).unwrap();

        let vortex_array = ArrayRef::from_arrow(record_batch, false).unwrap();
        let expected = VortexStructArray::new(
            ["field1", "field2"].into(),
            vec![
                buffer![1i32, 2, 3, 4].into_array(),
                VarBinArray::from_iter_nonnull(
                    ["a", "b", "c", "d"],
                    DType::Utf8(Nullability::NonNullable),
                )
                .into_array(),
            ],
            4,
            Validity::NonNullable,
        );
        assert_arrays_eq!(
            vortex_array,
            expected,
            &mut array_session().create_execution_ctx()
        );

        // Test with reference
        let schema = Arc::new(Schema::new(vec![
            Field::new("field1", DataType::Int32, false),
            Field::new("field2", DataType::Utf8, false),
        ]));

        let field1_data = Arc::new(Int32Array::from(vec![1, 2, 3, 4]));
        let field2_data = Arc::new(StringArray::from(vec!["a", "b", "c", "d"]));

        let record_batch = RecordBatch::try_new(schema, vec![field1_data, field2_data]).unwrap();

        let vortex_array = ArrayRef::from_arrow(&record_batch, false).unwrap();
        let expected = VortexStructArray::new(
            ["field1", "field2"].into(),
            vec![
                buffer![1i32, 2, 3, 4].into_array(),
                VarBinArray::from_iter_nonnull(
                    ["a", "b", "c", "d"],
                    DType::Utf8(Nullability::NonNullable),
                )
                .into_array(),
            ],
            4,
            Validity::NonNullable,
        );
        assert_arrays_eq!(
            vortex_array,
            expected,
            &mut array_session().create_execution_ctx()
        );
    }

    // Test dynamic dispatch conversion
    #[test]
    fn test_dyn_array_conversion() {
        let int_array = Int32Array::from(vec![1, 2, 3, 4]);
        let dyn_array: &dyn ArrowArray = &int_array;
        let vortex_array = ArrayRef::from_arrow(dyn_array, false).unwrap();
        assert_arrays_eq!(
            vortex_array,
            buffer![1i32, 2, 3, 4].into_array(),
            &mut array_session().create_execution_ctx()
        );

        let string_array = StringArray::from(vec!["a", "b", "c"]);
        let dyn_array: &dyn ArrowArray = &string_array;
        let vortex_array = ArrayRef::from_arrow(dyn_array, false).unwrap();
        assert_arrays_eq!(
            vortex_array,
            VarBinArray::from_iter_nonnull(["a", "b", "c"], DType::Utf8(Nullability::NonNullable)),
            &mut array_session().create_execution_ctx()
        );

        let bool_array = BooleanArray::from(vec![true, false, true]);
        let dyn_array: &dyn ArrowArray = &bool_array;
        let vortex_array = ArrayRef::from_arrow(dyn_array, false).unwrap();
        assert_arrays_eq!(
            vortex_array,
            BoolArray::from_iter([true, false, true]),
            &mut array_session().create_execution_ctx()
        );
    }

    // Existing tests
    #[test]
    pub fn nullable_may_contain_non_nullable() {
        let null_struct_array_with_non_nullable_field = new_null_array(
            &DataType::Struct(Fields::from(vec![Field::new(
                "non_nullable_inner",
                DataType::Int32,
                false,
            )])),
            1,
        );
        let actual =
            ArrayRef::from_arrow(null_struct_array_with_non_nullable_field.as_ref(), true).unwrap();
        assert_arrays_eq!(
            actual,
            VortexStructArray::new(
                ["non_nullable_inner"].into(),
                vec![buffer![0i32].into_array()],
                1,
                Validity::AllInvalid
            ),
            &mut array_session().create_execution_ctx()
        );
    }

    #[test]
    pub fn nullable_may_contain_deeply_nested_non_nullable() {
        let null_struct_array_with_non_nullable_field = new_null_array(
            &DataType::Struct(Fields::from(vec![Field::new(
                "non_nullable_inner",
                DataType::Struct(Fields::from(vec![Field::new(
                    "non_nullable_deeper_inner",
                    DataType::Int32,
                    false,
                )])),
                false,
            )])),
            1,
        );
        let actual =
            ArrayRef::from_arrow(null_struct_array_with_non_nullable_field.as_ref(), true).unwrap();
        assert_arrays_eq!(
            actual,
            VortexStructArray::new(
                ["non_nullable_inner"].into(),
                vec![
                    VortexStructArray::new(
                        ["non_nullable_deeper_inner"].into(),
                        vec![buffer![0i32].into_array()],
                        1,
                        Validity::NonNullable
                    )
                    .into_array()
                ],
                1,
                Validity::AllInvalid
            ),
            &mut array_session().create_execution_ctx()
        );
    }

    #[test]
    fn dyn_dispatch_imports_map() -> VortexResult<()> {
        let mut builder = ArrowMapBuilder::new(None, StringBuilder::new(), Int32Builder::new());
        builder.keys().append_value("joe");
        builder.values().append_value(1);
        builder.append(true)?;
        builder.append(false)?;
        let arrow = builder.finish();

        let expected = DType::map(
            DType::Utf8(Nullability::NonNullable),
            DType::Primitive(PType::I32, Nullability::Nullable),
            false,
            Nullability::Nullable,
        )?;

        // Dispatching on the Arrow data type must cover maps...
        let direct = from_arrow_dyn(&arrow, true)?;
        assert_eq!(direct.dtype(), &expected);

        // ...including maps reached by recursing into a container.
        let batch = RecordBatch::try_from_iter_with_nullable([(
            "maps",
            Arc::new(arrow) as Arc<dyn ArrowArray>,
            true,
        )])?;
        let imported = from_arrow_batch(&batch, false)?;
        assert_eq!(
            imported.as_::<Struct>().unmasked_field(0).dtype(),
            &expected
        );

        let row = Scalar::try_map(
            expected.clone(),
            [(
                Scalar::utf8("joe", Nullability::NonNullable),
                Scalar::primitive(1i32, Nullability::Nullable),
            )],
        )?;
        let mut ctx = array_session().create_execution_ctx();
        for actual in [&direct, imported.as_::<Struct>().unmasked_field(0)] {
            assert_eq!(actual.len(), 2);
            assert_eq!(actual.execute_scalar(0, &mut ctx)?, row);
            assert_eq!(
                actual.execute_scalar(1, &mut ctx)?,
                Scalar::null(expected.clone())
            );
        }
        Ok(())
    }

    #[test]
    fn non_nullable_request_rejects_nulls() {
        // Requesting `nullable = false` on an Arrow array that physically contains nulls is a
        // contradiction and must surface as an error, not a panic.
        let arrow_array = Int32Array::from(vec![Some(1), None, Some(3)]);
        assert!(ArrayRef::from_arrow(&arrow_array, false).is_err());
    }

    #[test]
    fn non_nullable_request_rejects_null_array() {
        // An Arrow NullArray is entirely null, so it cannot be converted to a non-nullable
        // Vortex array.
        let arrow_array = NullArray::new(5);
        assert!(ArrayRef::from_arrow(&arrow_array, false).is_err());
    }

    #[test]
    fn non_nullable_struct_with_nulls_errors() {
        // A struct array carrying top-level nulls cannot be converted to a non-nullable Vortex
        // struct; the struct-level validity reconciliation must error rather than panic.
        let struct_array = new_null_array(
            &DataType::Struct(Fields::from(vec![Field::new("a", DataType::Int32, true)])),
            3,
        );
        assert!(ArrayRef::from_arrow(struct_array.as_ref(), false).is_err());
    }

    #[test]
    fn non_nullable_list_with_nulls_errors() {
        // Likewise for a list array with a null entry: requesting a non-nullable list must error
        // rather than panic.
        let mut builder = ListBuilder::new(Int32Builder::new());
        builder.append_value([Some(1), Some(2)]);
        builder.append_null();
        let list = builder.finish();
        assert!(ArrayRef::from_arrow(&list, false).is_err());
    }

    #[test]
    pub fn nullable_struct_containing_non_nullable_dictionary_with_nulls_errors() {
        // `remove_nulls` cannot strip pushed-down nulls out of a non-nullable dictionary field,
        // so the values end up converted with `nullable = false` while still containing nulls.
        // This must surface as an error rather than panicking.
        let null_struct_array_with_non_nullable_field = new_null_array(
            &DataType::Struct(Fields::from(vec![Field::new(
                "non_nullable_deeper_inner",
                DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)),
                false,
            )])),
            1,
        );

        assert!(
            ArrayRef::from_arrow(null_struct_array_with_non_nullable_field.as_ref(), true).is_err()
        );
    }
}