vortex_arrow/

convert.rs

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

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::FromArrowType;

/// 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()
    }
}

macro_rules! impl_from_arrow_primitive {
    ($T:path) => {
        impl FromArrowArray<&ArrowPrimitiveArray<$T>> for ArrayRef {
            fn from_arrow(value: &ArrowPrimitiveArray<$T>, nullable: bool) -> VortexResult<Self> {
                let buffer = Buffer::from_arrow_scalar_buffer(value.values().clone());
                let validity = nulls(value.nulls(), nullable)?;
                Ok(PrimitiveArray::new(buffer, validity).into_array())
            }
        }
    };
}

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);

impl FromArrowArray<&ArrowPrimitiveArray<Decimal32Type>> for ArrayRef {
    fn from_arrow(
        array: &ArrowPrimitiveArray<Decimal32Type>,
        nullable: bool,
    ) -> VortexResult<Self> {
        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> {
        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> {
        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<Decimal256Type>> for ArrayRef {
    fn from_arrow(
        array: &ArrowPrimitiveArray<Decimal256Type>,
        nullable: bool,
    ) -> VortexResult<Self> {
        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())
    }
}

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> {
                temporal_array(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);

fn temporal_array<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, TimeUnit::from_arrow(time_unit), tz.clone()).into()
        }
        DataType::Time32(time_unit) => {
            TemporalArray::new_time(arr, TimeUnit::from_arrow(time_unit)).into()
        }
        DataType::Time64(time_unit) => {
            TemporalArray::new_time(arr, TimeUnit::from_arrow(time_unit)).into()
        }
        DataType::Date32 => TemporalArray::new_date(arr, TimeUnit::Days).into(),
        DataType::Date64 => TemporalArray::new_date(arr, TimeUnit::Milliseconds).into(),
        DataType::Duration(_) => unimplemented!(),
        DataType::Interval(_) => unimplemented!(),
        _ => vortex_panic!("Invalid temporal type: {}", value.data_type()),
    })
}

impl<T: ByteArrayType> FromArrowArray<&GenericByteArray<T>> for ArrayRef
where
    <T as ByteArrayType>::Offset: IntegerPType,
{
    fn from_arrow(value: &GenericByteArray<T>, nullable: bool) -> VortexResult<Self> {
        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}"),
        };
        // SAFETY: Arrow arrays are already validated (valid UTF-8, valid offsets, correct validity).
        Ok(unsafe {
            VarBinArray::new_unchecked(
                value.offsets().clone().into_array(),
                ByteBuffer::from_arrow_buffer(value.values().clone(), Alignment::of::<u8>()),
                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> {
        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 FromArrowArray<&ArrowBooleanArray> for ArrayRef {
    fn from_arrow(value: &ArrowBooleanArray, nullable: bool) -> VortexResult<Self> {
        Ok(BoolArray::new(
            value.values().clone().into(),
            nulls(value.nulls(), nullable)?,
        )
        .into_array())
    }
}

/// 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}"))
}

impl FromArrowArray<&ArrowStructArray> for ArrayRef {
    fn from_arrow(value: &ArrowStructArray, nullable: bool) -> VortexResult<Self> {
        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())?);
                        Self::from_arrow(stripped.as_ref(), false)
                    } else {
                        Self::from_arrow(c.as_ref(), field.is_nullable())
                    }
                })
                .collect::<VortexResult<Vec<_>>>()?,
            value.len(),
            nulls(value.nulls(), nullable)?,
        )?
        .into_array())
    }
}

impl<O: IntegerPType + OffsetSizeTrait> FromArrowArray<&GenericListArray<O>> for ArrayRef {
    fn from_arrow(value: &GenericListArray<O>, nullable: bool) -> VortexResult<Self> {
        // 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 elements = Self::from_arrow(value.values().as_ref(), elements_are_nullable)?;

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

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

impl<O: OffsetSizeTrait + NativePType> FromArrowArray<&GenericListViewArray<O>> for ArrayRef {
    fn from_arrow(array: &GenericListViewArray<O>, nullable: bool) -> VortexResult<Self> {
        // 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 = Self::from_arrow(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 FromArrowArray<&ArrowFixedSizeListArray> for ArrayRef {
    fn from_arrow(array: &ArrowFixedSizeListArray, nullable: bool) -> VortexResult<Self> {
        let DataType::FixedSizeList(field, list_size) = array.data_type() else {
            vortex_panic!("Invalid data type for ListArray: {}", array.data_type());
        };

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

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!(
        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())
}

impl FromArrowArray<&ArrowMapArray> for ArrayRef {
    fn from_arrow(array: &ArrowMapArray, nullable: bool) -> VortexResult<Self> {
        let DataType::Map(_, keys_sorted) = array.data_type() else {
            vortex_panic!("Invalid data type for MapArray: {}", array.data_type());
        };
        let entries = Self::from_arrow(array.entries(), false)?;
        map_from_arrow_parts(
            entries,
            array.offsets(),
            array.nulls(),
            *keys_sorted,
            nullable,
        )
    }
}

impl FromArrowArray<&ArrowNullArray> for ArrayRef {
    fn from_arrow(value: &ArrowNullArray, nullable: bool) -> VortexResult<Self> {
        vortex_ensure!(
            nullable,
            "Cannot convert an Arrow NullArray into a non-nullable Vortex array"
        );
        Ok(NullArray::new(value.len()).into_array())
    }
}

impl<K: ArrowDictionaryKeyType> FromArrowArray<&DictionaryArray<K>> for DictArray {
    fn from_arrow(array: &DictionaryArray<K>, nullable: bool) -> VortexResult<Self> {
        let keys = AnyDictionaryArray::keys(array);
        let keys = ArrayRef::from_arrow(keys, keys.is_nullable())?;
        let values = ArrayRef::from_arrow(array.values().as_ref(), nullable)?;
        // SAFETY: we assume that Arrow has checked the invariants on construction.
        Ok(unsafe { DictArray::new_unchecked(keys, values) })
    }
}

pub(crate) 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 {null_count} nulls into a non-nullable Vortex array"
        );
        Ok(Validity::NonNullable)
    }
}

impl FromArrowArray<&dyn ArrowArray> for ArrayRef {
    fn from_arrow(array: &dyn ArrowArray, nullable: bool) -> VortexResult<Self> {
        match array.data_type() {
            DataType::Boolean => Self::from_arrow(array.as_boolean(), nullable),
            DataType::UInt8 => Self::from_arrow(array.as_primitive::<UInt8Type>(), nullable),
            DataType::UInt16 => Self::from_arrow(array.as_primitive::<UInt16Type>(), nullable),
            DataType::UInt32 => Self::from_arrow(array.as_primitive::<UInt32Type>(), nullable),
            DataType::UInt64 => Self::from_arrow(array.as_primitive::<UInt64Type>(), nullable),
            DataType::Int8 => Self::from_arrow(array.as_primitive::<Int8Type>(), nullable),
            DataType::Int16 => Self::from_arrow(array.as_primitive::<Int16Type>(), nullable),
            DataType::Int32 => Self::from_arrow(array.as_primitive::<Int32Type>(), nullable),
            DataType::Int64 => Self::from_arrow(array.as_primitive::<Int64Type>(), nullable),
            DataType::Float16 => Self::from_arrow(array.as_primitive::<Float16Type>(), nullable),
            DataType::Float32 => Self::from_arrow(array.as_primitive::<Float32Type>(), nullable),
            DataType::Float64 => Self::from_arrow(array.as_primitive::<Float64Type>(), nullable),
            DataType::Utf8 => Self::from_arrow(array.as_string::<i32>(), nullable),
            DataType::LargeUtf8 => Self::from_arrow(array.as_string::<i64>(), nullable),
            DataType::Binary => Self::from_arrow(array.as_binary::<i32>(), nullable),
            DataType::LargeBinary => Self::from_arrow(array.as_binary::<i64>(), nullable),
            DataType::BinaryView => Self::from_arrow(array.as_binary_view(), nullable),
            DataType::Utf8View => Self::from_arrow(array.as_string_view(), nullable),
            DataType::Struct(_) => Self::from_arrow(array.as_struct(), nullable),
            DataType::List(_) => Self::from_arrow(array.as_list::<i32>(), nullable),
            DataType::LargeList(_) => Self::from_arrow(array.as_list::<i64>(), nullable),
            DataType::ListView(_) => Self::from_arrow(array.as_list_view::<i32>(), nullable),
            DataType::LargeListView(_) => Self::from_arrow(array.as_list_view::<i64>(), nullable),
            DataType::FixedSizeList(..) => Self::from_arrow(array.as_fixed_size_list(), nullable),
            DataType::Map(..) => Self::from_arrow(array.as_map(), nullable),
            DataType::Null => Self::from_arrow(as_null_array(array), nullable),
            DataType::Timestamp(u, _) => match u {
                ArrowTimeUnit::Second => {
                    Self::from_arrow(array.as_primitive::<TimestampSecondType>(), nullable)
                }
                ArrowTimeUnit::Millisecond => {
                    Self::from_arrow(array.as_primitive::<TimestampMillisecondType>(), nullable)
                }
                ArrowTimeUnit::Microsecond => {
                    Self::from_arrow(array.as_primitive::<TimestampMicrosecondType>(), nullable)
                }
                ArrowTimeUnit::Nanosecond => {
                    Self::from_arrow(array.as_primitive::<TimestampNanosecondType>(), nullable)
                }
            },
            DataType::Date32 => Self::from_arrow(array.as_primitive::<Date32Type>(), nullable),
            DataType::Date64 => Self::from_arrow(array.as_primitive::<Date64Type>(), nullable),
            DataType::Time32(u) => match u {
                ArrowTimeUnit::Second => {
                    Self::from_arrow(array.as_primitive::<Time32SecondType>(), nullable)
                }
                ArrowTimeUnit::Millisecond => {
                    Self::from_arrow(array.as_primitive::<Time32MillisecondType>(), nullable)
                }
                ArrowTimeUnit::Microsecond | ArrowTimeUnit::Nanosecond => unreachable!(),
            },
            DataType::Time64(u) => match u {
                ArrowTimeUnit::Microsecond => {
                    Self::from_arrow(array.as_primitive::<Time64MicrosecondType>(), nullable)
                }
                ArrowTimeUnit::Nanosecond => {
                    Self::from_arrow(array.as_primitive::<Time64NanosecondType>(), nullable)
                }
                ArrowTimeUnit::Second | ArrowTimeUnit::Millisecond => unreachable!(),
            },
            DataType::Decimal32(..) => {
                Self::from_arrow(array.as_primitive::<Decimal32Type>(), nullable)
            }
            DataType::Decimal64(..) => {
                Self::from_arrow(array.as_primitive::<Decimal64Type>(), nullable)
            }
            DataType::Decimal128(..) => {
                Self::from_arrow(array.as_primitive::<Decimal128Type>(), nullable)
            }
            DataType::Decimal256(..) => {
                Self::from_arrow(array.as_primitive::<Decimal256Type>(), nullable)
            }
            DataType::Dictionary(key_type, _) => match key_type.as_ref() {
                DataType::Int8 => Ok(DictArray::from_arrow(
                    array.as_dictionary::<Int8Type>(),
                    nullable,
                )?
                .into_array()),
                DataType::Int16 => Ok(DictArray::from_arrow(
                    array.as_dictionary::<Int16Type>(),
                    nullable,
                )?
                .into_array()),
                DataType::Int32 => Ok(DictArray::from_arrow(
                    array.as_dictionary::<Int32Type>(),
                    nullable,
                )?
                .into_array()),
                DataType::Int64 => Ok(DictArray::from_arrow(
                    array.as_dictionary::<Int64Type>(),
                    nullable,
                )?
                .into_array()),
                DataType::UInt8 => Ok(DictArray::from_arrow(
                    array.as_dictionary::<UInt8Type>(),
                    nullable,
                )?
                .into_array()),
                DataType::UInt16 => Ok(DictArray::from_arrow(
                    array.as_dictionary::<UInt16Type>(),
                    nullable,
                )?
                .into_array()),
                DataType::UInt32 => Ok(DictArray::from_arrow(
                    array.as_dictionary::<UInt32Type>(),
                    nullable,
                )?
                .into_array()),
                DataType::UInt64 => Ok(DictArray::from_arrow(
                    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<RecordBatch> for ArrayRef {
    fn from_arrow(array: RecordBatch, nullable: bool) -> VortexResult<Self> {
        ArrayRef::from_arrow(&arrow_array::StructArray::from(array), nullable)
    }
}

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

#[cfg(test)]
mod tests {
    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::LargeListBuilder;
    use arrow_array::builder::ListBuilder;
    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::arrays::Decimal;
    use vortex_array::arrays::FixedSizeList;
    use vortex_array::arrays::List;
    use vortex_array::arrays::ListView;
    use vortex_array::arrays::Primitive;
    use vortex_array::arrays::Struct;
    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::dtype::DType;
    use vortex_array::dtype::Nullability;
    use vortex_array::dtype::PType;
    use vortex_array::extension::datetime::TimeUnit;
    use vortex_array::extension::datetime::Timestamp;

    use crate::FromArrowArray as _;
    use crate::IntoVortexArray as _;

    #[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);
    }

    #[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);
    }

    // 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);
    }

    #[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);
    }

    // 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);
    }

    #[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()
            )
        );
    }

    // 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);
    }

    #[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())
        );
    }

    #[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())
        );
    }

    // 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);
    }

    // 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");
    }

    // 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);
    }

    #[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
    }

    #[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
    }

    #[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
    }

    // 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);
    }

    // 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);
    }

    #[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);
    }

    #[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);
    }

    #[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);
    }

    // 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();
        assert_eq!(vortex_array.len(), 4);

        // 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();
        assert_eq!(vortex_array.len(), 4);
    }

    // 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_eq!(vortex_array.len(), 4);

        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_eq!(vortex_array.len(), 3);

        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_eq!(vortex_array.len(), 3);
    }

    // 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,
        );
        ArrayRef::from_arrow(null_struct_array_with_non_nullable_field.as_ref(), true).unwrap();
    }

    #[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,
        );
        ArrayRef::from_arrow(null_struct_array_with_non_nullable_field.as_ref(), true).unwrap();
    }

    #[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()
        );
    }
}