use std::any::Any;
use std::borrow::Cow;
use std::fmt::Debug;
use std::sync::Arc;
use arrow_array::Array as ArrowArray;
use arrow_array::ArrayRef as ArrowArrayRef;
use arrow_array::RecordBatch;
use arrow_array::RunArray;
use arrow_array::make_array;
use arrow_array::types::Int16Type;
use arrow_array::types::Int32Type;
use arrow_array::types::Int64Type;
use arrow_array::types::RunEndIndexType;
use arrow_schema::DataType;
use arrow_schema::Field;
use arrow_schema::FieldRef;
use arrow_schema::Fields;
use arrow_schema::Schema;
use arrow_schema::extension::EXTENSION_TYPE_NAME_KEY;
use arrow_schema::extension::ExtensionType;
use tracing::debug;
use tracing::trace;
use vortex_array::ArrayRef;
use vortex_array::ExecutionCtx;
use vortex_array::IntoArray;
use vortex_array::arrays::DictArray;
use vortex_array::arrays::FixedSizeListArray;
use vortex_array::arrays::ListArray;
use vortex_array::arrays::ListViewArray;
use vortex_array::arrays::StructArray;
use vortex_array::dtype::DType;
use vortex_array::dtype::FieldName;
use vortex_array::dtype::FieldNames;
use vortex_array::dtype::NativePType;
use vortex_array::dtype::Nullability;
use vortex_array::dtype::StructFields;
use vortex_array::dtype::extension::ExtId;
use vortex_array::extension::datetime::AnyTemporal;
use vortex_array::extension::uuid::Uuid;
use vortex_array::validity::Validity;
use vortex_error::VortexResult;
use vortex_error::vortex_bail;
use vortex_error::vortex_ensure;
use vortex_error::vortex_err;
use vortex_session::ArcSwapMap;
use vortex_session::SessionExt;
use vortex_session::SessionGuard;
use vortex_session::SessionVar;
use vortex_session::registry::Id;
use crate::IntoVortexArray;
use crate::convert::from_arrow_dyn;
use crate::convert::map_from_arrow_parts;
use crate::convert::nulls;
use crate::convert::remove_nulls;
use crate::dtype::from_arrow_data_type;
use crate::dtype::to_data_type_naive;
use crate::executor::execute_arrow_naive;
use crate::run_end_import::run_end_from_arrow;
pub enum ArrowExport {
Unsupported(ArrayRef),
Exported(ArrowArrayRef),
}
pub enum ArrowImport {
Unsupported(ArrowArrayRef),
Imported(ArrayRef),
}
pub trait IntoArrowField<'a> {
fn into_arrow_field(self, data_type: &DataType) -> Cow<'a, Field>;
}
impl<'a> IntoArrowField<'a> for &'a Field {
fn into_arrow_field(self, _data_type: &DataType) -> Cow<'a, Field> {
Cow::Borrowed(self)
}
}
impl<'a> IntoArrowField<'a> for &'a FieldRef {
fn into_arrow_field(self, _data_type: &DataType) -> Cow<'a, Field> {
Cow::Borrowed(self.as_ref())
}
}
impl<'a> IntoArrowField<'a> for bool {
fn into_arrow_field(self, data_type: &DataType) -> Cow<'a, Field> {
Cow::Owned(Field::new("", data_type.clone(), self))
}
}
impl<'a> IntoArrowField<'a> for Nullability {
fn into_arrow_field(self, data_type: &DataType) -> Cow<'a, Field> {
self.is_nullable().into_arrow_field(data_type)
}
}
pub trait ArrowExportVTable: 'static + Send + Sync + Debug {
fn arrow_ext_id(&self) -> Id;
fn vortex_id(&self) -> Id;
fn to_arrow_field(
&self,
name: &str,
dtype: &DType,
session: &ArrowSession,
) -> VortexResult<Option<Field>>;
fn execute_arrow(
&self,
array: ArrayRef,
target: &Field,
ctx: &mut ExecutionCtx,
) -> VortexResult<ArrowExport>;
}
pub trait ArrowImportVTable: 'static + Send + Sync + Debug {
fn arrow_ext_id(&self) -> Id;
#[allow(clippy::wrong_self_convention)]
fn from_arrow_field(
&self,
field: &Field,
session: &ArrowSession,
) -> VortexResult<Option<DType>>;
#[allow(clippy::wrong_self_convention)]
fn from_arrow_array(
&self,
array: ArrowArrayRef,
field: &Field,
dtype: &DType,
session: &ArrowSession,
) -> VortexResult<ArrowImport>;
}
pub type ArrowExportVTableRef = Arc<dyn ArrowExportVTable>;
pub type ArrowImportVTableRef = Arc<dyn ArrowImportVTable>;
type ArrowExporterRegistry = ArcSwapMap<Id, Arc<[ArrowExportVTableRef]>>;
type VortexExporterRegistry = ArcSwapMap<ExtId, Arc<[ArrowExportVTableRef]>>;
type ArrowImporterRegistry = ArcSwapMap<Id, Arc<[ArrowImportVTableRef]>>;
#[derive(Clone, Debug)]
pub struct ArrowSession {
exporters: ArrowExporterRegistry,
exporters_by_vortex: VortexExporterRegistry,
importers: ArrowImporterRegistry,
}
impl Default for ArrowSession {
fn default() -> Self {
let session = Self {
exporters: ArrowExporterRegistry::default(),
exporters_by_vortex: VortexExporterRegistry::default(),
importers: ArrowImporterRegistry::default(),
};
session.register_exporter(Arc::new(Uuid));
session.register_importer(Arc::new(Uuid));
session
}
}
impl ArrowSession {
pub fn register_exporter(&self, exporter: ArrowExportVTableRef) {
self.exporters.push(
exporter.arrow_ext_id(),
ArrowExportVTableRef::clone(&exporter),
);
self.exporters_by_vortex
.push(exporter.vortex_id(), exporter);
}
pub fn register_importer(&self, importer: ArrowImportVTableRef) {
self.importers.push(importer.arrow_ext_id(), importer);
}
fn exporters(&self, id: &Id) -> Arc<[ArrowExportVTableRef]> {
self.exporters.get(id).unwrap_or_else(|| Arc::from([]))
}
fn exporters_by_vortex(&self, id: &Id) -> Arc<[ArrowExportVTableRef]> {
self.exporters_by_vortex
.get(id)
.unwrap_or_else(|| Arc::from([]))
}
fn importers(&self, id: &Id) -> Arc<[ArrowImportVTableRef]> {
self.importers.get(id).unwrap_or_else(|| Arc::from([]))
}
pub fn to_arrow_field(&self, name: &str, dtype: &DType) -> VortexResult<Field> {
match dtype {
DType::List(elem_dtype, nullability) => {
let elem_field = self.to_arrow_field(Field::LIST_FIELD_DEFAULT_NAME, elem_dtype)?;
Ok(Field::new_list(name, elem_field, nullability.is_nullable()))
}
DType::FixedSizeList(elem_dtype, elem_size, nullability) => {
let elem_field = self.to_arrow_field(Field::LIST_FIELD_DEFAULT_NAME, elem_dtype)?;
Ok(Field::new_fixed_size_list(
name,
elem_field,
(*elem_size).try_into()?,
nullability.is_nullable(),
))
}
DType::Map(map_dtype, nullability) => {
let key = self.to_arrow_field("key", &map_dtype.key_dtype())?;
let value = self.to_arrow_field("value", &map_dtype.value_dtype())?;
let entries = Field::new_struct("entries", Fields::from(vec![key, value]), false);
Ok(Field::new(
name,
DataType::Map(Arc::new(entries), map_dtype.keys_sorted()),
nullability.is_nullable(),
))
}
DType::Struct(fields, nullability) => {
let arrow_fields = Fields::from_iter(
fields
.fields()
.zip(fields.names().iter())
.map(|(field, name)| self.to_arrow_field(name.as_ref(), &field))
.collect::<VortexResult<Vec<_>>>()?,
);
Ok(Field::new_struct(
name,
arrow_fields,
nullability.is_nullable(),
))
}
DType::Extension(ext) if !ext.is::<AnyTemporal>() => {
for plugin in self.exporters_by_vortex(&ext.id()).iter() {
if let Some(field) =
plugin.to_arrow_field(name, &DType::Extension(ext.clone()), self)?
{
return Ok(field);
}
}
vortex_bail!("extension type cannot be converted to Arrow without a plugin: {ext}");
}
DType::Variant(_) => {
Ok(Field::new(
name,
DataType::Struct(
vec![
Field::new("metadata", DataType::BinaryView, dtype.is_nullable()),
Field::new("value", DataType::BinaryView, dtype.is_nullable()),
]
.into(),
),
dtype.is_nullable(),
)
.with_metadata(
[(
EXTENSION_TYPE_NAME_KEY.to_string(),
"arrow.parquet.variant".to_string(),
)]
.into(),
))
}
_ => Ok(Field::new(
name,
to_data_type_naive(dtype)?,
dtype.is_nullable(),
)),
}
}
pub fn to_arrow_schema(&self, dtype: &DType) -> VortexResult<Schema> {
let DType::Struct(struct_dtype, _) = dtype else {
vortex_bail!("to_arrow_schema requires a top-level struct dtype, got {dtype}");
};
let mut fields = Vec::with_capacity(struct_dtype.names().len());
for (name, field_dtype) in struct_dtype.names().iter().zip(struct_dtype.fields()) {
fields.push(self.to_arrow_field(name.as_ref(), &field_dtype)?);
}
Ok(Schema::new(fields))
}
pub fn to_arrow_datatype(&self, dtype: &DType) -> VortexResult<DataType> {
Ok(self.to_arrow_field("", dtype)?.data_type().clone())
}
#[expect(clippy::disallowed_methods, reason = "interning a dynamic id")]
pub fn from_arrow_field(&self, field: &Field) -> VortexResult<DType> {
if let Some(name) = field.metadata().get(EXTENSION_TYPE_NAME_KEY) {
for plugin in self.importers(&Id::new(name)).iter() {
if let Some(dtype) = plugin.from_arrow_field(field, self)? {
return Ok(dtype);
}
}
if name == "arrow.parquet.variant" {
return Ok(DType::Variant(field.is_nullable().into()));
}
}
self.from_arrow_datatype(field.data_type(), field.is_nullable().into())
}
pub fn from_arrow_datatype(
&self,
data_type: &DataType,
nullability: Nullability,
) -> VortexResult<DType> {
Ok(match data_type {
DataType::List(elem)
| DataType::LargeList(elem)
| DataType::ListView(elem)
| DataType::LargeListView(elem) => {
DType::List(Arc::new(self.from_arrow_field(elem.as_ref())?), nullability)
}
DataType::FixedSizeList(elem, size) => DType::FixedSizeList(
Arc::new(self.from_arrow_field(elem.as_ref())?),
*size as u32,
nullability,
),
DataType::Map(entries, keys_sorted) => {
vortex_ensure!(
!entries.is_nullable(),
"Arrow map entries field must be non-nullable"
);
let DataType::Struct(fields) = entries.data_type() else {
vortex_bail!(
"Arrow map entries field must have Struct type, got {:?}",
entries.data_type()
);
};
vortex_ensure!(
fields.len() == 2,
"Arrow map entries struct must contain exactly two fields"
);
vortex_ensure!(
!fields[0].is_nullable(),
"Arrow map key field must be non-nullable"
);
DType::map(
self.from_arrow_field(fields[0].as_ref())?,
self.from_arrow_field(fields[1].as_ref())?,
*keys_sorted,
nullability,
)?
}
DataType::Struct(fields) => DType::Struct(self.from_arrow_fields(fields)?, nullability),
DataType::Dictionary(_, value_type) => {
self.from_arrow_datatype(value_type.as_ref(), nullability)?
}
DataType::RunEndEncoded(_, value_field) => {
self.from_arrow_field(&run_end_values_field(value_field, nullability))?
}
_ => from_arrow_data_type(data_type, nullability)?,
})
}
pub fn from_arrow_fields(&self, fields: &Fields) -> VortexResult<StructFields> {
fields
.iter()
.map(|f| {
self.from_arrow_field(f)
.map(|dt| (FieldName::from(f.name().as_str()), dt))
})
.collect::<VortexResult<StructFields>>()
}
pub fn from_arrow_schema(&self, schema: &Schema) -> VortexResult<DType> {
Ok(DType::Struct(
self.from_arrow_fields(schema.fields())?,
Nullability::NonNullable,
))
}
pub fn from_arrow_record_batch(
&self,
batch: RecordBatch,
schema: &Schema,
) -> VortexResult<ArrayRef> {
vortex_ensure!(
batch.num_columns() == schema.fields().len(),
"RecordBatch has {} columns but schema has {} fields",
batch.num_columns(),
schema.fields().len()
);
let length = batch.num_rows();
let names = FieldNames::from_iter(
schema
.fields()
.iter()
.map(|f| FieldName::from(f.name().as_str())),
);
let mut columns = Vec::with_capacity(schema.fields().len());
for (col, field) in batch.columns().iter().zip(schema.fields().iter()) {
columns.push(self.from_arrow_array_inner(ArrowArrayRef::clone(col), field)?);
}
Ok(StructArray::try_new(names, columns, length, Validity::NonNullable)?.into_array())
}
#[expect(clippy::disallowed_methods, reason = "interning a dynamic id")]
pub fn execute_arrow(
&self,
array: ArrayRef,
target: Option<&Field>,
ctx: &mut ExecutionCtx,
) -> VortexResult<ArrowArrayRef> {
let arrow_field;
let target_field = match target {
Some(field) => field,
None => {
let session = ctx.session().clone();
arrow_field = session.arrow().to_arrow_field("", array.dtype())?;
&arrow_field
}
};
if let Some(arrow_ext_name) = target_field.metadata().get(EXTENSION_TYPE_NAME_KEY) {
let len = array.len();
let mut current = array;
for plugin in self.exporters(&Id::new(arrow_ext_name)).iter() {
trace!(
plugin = ?plugin,
extension_name = arrow_ext_name,
"probing plugin for converting Arrow array"
);
match plugin.execute_arrow(current, target_field, ctx)? {
ArrowExport::Exported(arrow) => {
vortex_ensure!(
arrow.len() == len,
"Arrow array length does not match Vortex array length after conversion to {:?}",
arrow
);
return Ok(arrow);
}
ArrowExport::Unsupported(array) => current = array,
}
}
debug!(
extension_id = arrow_ext_name,
data_type = ?target_field.data_type(),
"unsupported Arrow extension type encountered, falling back to naive execution"
);
return execute_arrow_naive(current, Some(target_field.data_type()), ctx);
}
execute_arrow_naive(array, target.map(|field| field.data_type()), ctx)
}
pub fn from_arrow_array<'a>(
&self,
array: ArrowArrayRef,
field: impl IntoArrowField<'a>,
) -> VortexResult<ArrayRef> {
let field = field.into_arrow_field(array.data_type());
self.from_arrow_array_inner(array, field.as_ref())
}
#[allow(clippy::wrong_self_convention)]
fn from_arrow_array_inner(
&self,
array: ArrowArrayRef,
field: &Field,
) -> VortexResult<ArrayRef> {
if let Some(extension_name) = field.metadata().get(EXTENSION_TYPE_NAME_KEY) {
#[expect(clippy::disallowed_methods, reason = "interning a dynamic id")]
let importers = self.importers(&Id::new(extension_name));
if !importers.is_empty() {
let dtype = self.from_arrow_field(field)?;
let mut current = array;
for plugin in importers.iter() {
match plugin.from_arrow_array(current, field, &dtype, self)? {
ArrowImport::Imported(arr) => return Ok(arr),
ArrowImport::Unsupported(arr) => current = arr,
}
}
return self.from_arrow_array_canonical(current.as_ref(), field);
}
}
self.from_arrow_array_canonical(array.as_ref(), field)
}
#[allow(clippy::wrong_self_convention)]
fn from_arrow_array_canonical(
&self,
array: &dyn ArrowArray,
field: &Field,
) -> VortexResult<ArrayRef> {
use arrow_array::cast::AsArray;
match field.data_type() {
DataType::Struct(fields) => {
let arrow_struct = array.as_struct();
let names = FieldNames::from_iter(
fields.iter().map(|f| FieldName::from(f.name().as_str())),
);
let columns = arrow_struct
.columns()
.iter()
.zip(fields.iter())
.map(|(col, child_field)| {
let inner = if col.null_count() > 0 && !child_field.is_nullable() {
make_array(remove_nulls(col.to_data())?)
} else {
ArrowArrayRef::clone(col)
};
self.from_arrow_array_inner(inner, child_field.as_ref())
})
.collect::<VortexResult<Vec<_>>>()?;
let validity = nulls(arrow_struct.nulls(), field.is_nullable())?;
Ok(
StructArray::try_new(names, columns, arrow_struct.len(), validity)?
.into_array(),
)
}
DataType::List(elem_field) => {
let list = array.as_list::<i32>();
let elements = self
.from_arrow_array(ArrowArrayRef::clone(list.values()), elem_field.as_ref())?;
let offsets = list.offsets().clone().into_array();
let validity = nulls(list.nulls(), field.is_nullable())?;
Ok(ListArray::try_new(elements, offsets, validity)?.into_array())
}
DataType::LargeList(elem_field) => {
let list = array.as_list::<i64>();
let elements = self
.from_arrow_array(ArrowArrayRef::clone(list.values()), elem_field.as_ref())?;
let offsets = list.offsets().clone().into_array();
let validity = nulls(list.nulls(), field.is_nullable())?;
Ok(ListArray::try_new(elements, offsets, validity)?.into_array())
}
DataType::FixedSizeList(elem_field, list_size) => {
let fsl = array.as_fixed_size_list();
let elements = self.from_arrow_array_inner(
ArrowArrayRef::clone(fsl.values()),
elem_field.as_ref(),
)?;
let validity = nulls(fsl.nulls(), field.is_nullable())?;
Ok(
FixedSizeListArray::try_new(elements, *list_size as u32, validity, fsl.len())?
.into_array(),
)
}
DataType::ListView(elem_field) => {
let list = array.as_list_view::<i32>();
let elements = self
.from_arrow_array(ArrowArrayRef::clone(list.values()), elem_field.as_ref())?;
let offsets = list.offsets().clone().into_array();
let sizes = list.sizes().clone().into_array();
let validity = nulls(list.nulls(), field.is_nullable())?;
Ok(ListViewArray::try_new(elements, offsets, sizes, validity)?.into_array())
}
DataType::LargeListView(elem_field) => {
let list = array.as_list_view::<i64>();
let elements = self
.from_arrow_array(ArrowArrayRef::clone(list.values()), elem_field.as_ref())?;
let offsets = list.offsets().clone().into_array();
let sizes = list.sizes().clone().into_array();
let validity = nulls(list.nulls(), field.is_nullable())?;
Ok(ListViewArray::try_new(elements, offsets, sizes, validity)?.into_array())
}
DataType::Map(entries_field, keys_sorted) => {
let map = array.as_map();
let entries_array: ArrowArrayRef = Arc::new(map.entries().clone());
let entries = self.from_arrow_array_inner(entries_array, entries_field.as_ref())?;
map_from_arrow_parts(
entries,
map.offsets(),
map.nulls(),
*keys_sorted,
field.is_nullable(),
)
}
DataType::RunEndEncoded(ends_field, values_field) => {
let values_field = run_end_values_field(values_field, field.is_nullable().into());
match ends_field.data_type() {
DataType::Int16 => self.run_end_from_arrow::<Int16Type>(array, &values_field),
DataType::Int32 => self.run_end_from_arrow::<Int32Type>(array, &values_field),
DataType::Int64 => self.run_end_from_arrow::<Int64Type>(array, &values_field),
ends_dt => vortex_bail!(
"Arrow run-end array run ends must be Int16, Int32 or Int64, got {ends_dt}"
),
}
}
DataType::Dictionary(..) => {
let dict = array.as_any_dictionary();
let values = self
.from_arrow_array(ArrowArrayRef::clone(dict.values()), field.is_nullable())?;
let codes = dict.keys();
let codes = from_arrow_dyn(codes, codes.is_nullable())?;
Ok(unsafe { DictArray::new_unchecked(codes, values) }.into_array())
}
_ => from_arrow_dyn(array, field.is_nullable()),
}
}
#[allow(clippy::wrong_self_convention)]
fn run_end_from_arrow<R: RunEndIndexType>(
&self,
array: &dyn ArrowArray,
values_field: &Field,
) -> VortexResult<ArrayRef>
where
R::Native: NativePType,
{
let run_array = array
.as_any()
.downcast_ref::<RunArray<R>>()
.ok_or_else(|| vortex_err!("expected an Arrow RunArray, got {}", array.data_type()))?;
let values =
self.from_arrow_array_inner(ArrowArrayRef::clone(run_array.values()), values_field)?;
run_end_from_arrow(run_array, values)
}
}
fn run_end_values_field(values_field: &FieldRef, nullability: Nullability) -> Field {
values_field
.as_ref()
.clone()
.with_nullable(nullability.into())
}
pub(crate) fn has_valid_extension_type<E: ExtensionType>(field: &Field) -> bool {
if field.extension_type_name() != Some(E::NAME) {
return false;
}
E::try_new_from_field_metadata(field.data_type(), field.metadata()).is_ok()
}
impl SessionVar for ArrowSession {
fn as_any(&self) -> &dyn Any {
self
}
fn as_any_mut(&mut self) -> &mut dyn Any {
self
}
}
pub trait ArrowSessionExt: SessionExt {
fn arrow(&self) -> SessionGuard<'_, ArrowSession>;
}
impl<S: SessionExt> ArrowSessionExt for S {
fn arrow(&self) -> SessionGuard<'_, ArrowSession> {
self.get::<ArrowSession>()
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use arrow_array::DictionaryArray;
use arrow_array::FixedSizeBinaryArray;
use arrow_array::Int32Array;
use arrow_array::ListArray as ArrowListArray;
use arrow_array::StringArray;
use arrow_array::StructArray as ArrowStructArray;
use arrow_array::cast::AsArray;
use arrow_buffer::OffsetBuffer;
use arrow_schema::DataType;
use arrow_schema::Field;
use arrow_schema::extension::Uuid as ArrowUuid;
use rstest::rstest;
use vortex_array::IntoArray;
use vortex_array::VortexSessionExecute;
use vortex_array::array_session;
use vortex_array::arrays::Dict;
use vortex_array::arrays::ListArray;
use vortex_array::arrays::PrimitiveArray;
use vortex_array::dtype::DType;
use vortex_array::dtype::FieldName;
use vortex_array::dtype::Nullability;
use vortex_array::dtype::PType;
use vortex_array::dtype::StructFields;
use vortex_array::dtype::extension::ExtDType;
use vortex_array::dtype::extension::ExtVTable;
use vortex_array::extension::uuid::Uuid;
use vortex_array::extension::uuid::UuidMetadata;
use vortex_error::VortexExpect;
use vortex_error::VortexResult;
use super::*;
fn uuid_dtype(nullable: bool) -> DType {
let storage = DType::FixedSizeList(
Arc::new(DType::Primitive(PType::U8, Nullability::NonNullable)),
16,
nullable.into(),
);
DType::Extension(
ExtDType::try_with_vtable(Uuid, UuidMetadata::default(), storage)
.expect("uuid ext dtype")
.erased(),
)
}
#[test]
fn to_arrow_field_top_level_uuid_carries_extension_metadata() -> VortexResult<()> {
let session = ArrowSession::default();
let field = session.to_arrow_field("id", &uuid_dtype(false))?;
assert!(has_valid_extension_type::<ArrowUuid>(&field));
Ok(())
}
#[test]
fn to_arrow_field_struct_with_nested_uuid_preserves_metadata() -> VortexResult<()> {
let session = ArrowSession::default();
let dtype = DType::Struct(
StructFields::from_iter([(FieldName::from("id"), uuid_dtype(false))]),
Nullability::NonNullable,
);
let field = session.to_arrow_field("row", &dtype)?;
let DataType::Struct(inner) = field.data_type() else {
panic!("expected Struct, got {:?}", field.data_type());
};
assert_eq!(inner.len(), 1);
assert_eq!(inner[0].data_type(), &DataType::FixedSizeBinary(16));
assert!(has_valid_extension_type::<ArrowUuid>(&inner[0]));
Ok(())
}
#[test]
fn to_arrow_field_list_of_uuid_preserves_metadata() -> VortexResult<()> {
let session = ArrowSession::default();
let dtype = DType::List(Arc::new(uuid_dtype(true)), Nullability::NonNullable);
let field = session.to_arrow_field("ids", &dtype)?;
let DataType::List(elem) = field.data_type() else {
panic!("expected List, got {:?}", field.data_type());
};
assert!(has_valid_extension_type::<ArrowUuid>(elem));
Ok(())
}
#[test]
fn to_arrow_field_fixed_size_list_of_uuid_preserves_metadata() -> VortexResult<()> {
let session = ArrowSession::default();
let dtype = DType::FixedSizeList(Arc::new(uuid_dtype(false)), 3, Nullability::NonNullable);
let field = session.to_arrow_field("triple", &dtype)?;
let DataType::FixedSizeList(elem, size) = field.data_type() else {
panic!("expected FixedSizeList, got {:?}", field.data_type());
};
assert_eq!(*size, 3);
assert!(has_valid_extension_type::<ArrowUuid>(elem));
Ok(())
}
#[test]
fn schema_roundtrip_preserves_map_uuid_fields() -> VortexResult<()> {
let session = ArrowSession::default();
let map = DType::map(
uuid_dtype(false),
uuid_dtype(true),
true,
Nullability::Nullable,
)?;
let dtype = DType::Struct(
StructFields::from_iter([(FieldName::from("ids"), map)]),
Nullability::NonNullable,
);
let schema = session.to_arrow_schema(&dtype)?;
let field = schema.field(0);
let DataType::Map(entries, keys_sorted) = field.data_type() else {
panic!("expected Map, got {:?}", field.data_type());
};
assert!(*keys_sorted);
assert_eq!(entries.name(), "entries");
assert!(!entries.is_nullable());
let DataType::Struct(fields) = entries.data_type() else {
panic!("expected map entries struct, got {:?}", entries.data_type());
};
assert!(has_valid_extension_type::<ArrowUuid>(&fields[0]));
assert!(has_valid_extension_type::<ArrowUuid>(&fields[1]));
assert!(!fields[0].is_nullable());
assert!(fields[1].is_nullable());
assert_eq!(session.from_arrow_schema(&schema)?, dtype);
Ok(())
}
#[test]
fn to_arrow_schema_struct_of_struct_uuid() -> VortexResult<()> {
let session = ArrowSession::default();
let inner = DType::Struct(
StructFields::from_iter([(FieldName::from("id"), uuid_dtype(true))]),
Nullability::NonNullable,
);
let outer = DType::Struct(
StructFields::from_iter([(FieldName::from("payload"), inner)]),
Nullability::NonNullable,
);
let schema = session.to_arrow_schema(&outer)?;
let payload = schema.field(0);
let DataType::Struct(inner_fields) = payload.data_type() else {
panic!("expected Struct, got {:?}", payload.data_type());
};
assert!(has_valid_extension_type::<ArrowUuid>(&inner_fields[0]));
Ok(())
}
#[test]
fn from_arrow_field_recurses_into_nested_uuid() -> VortexResult<()> {
let session = ArrowSession::default();
let mut elem = Field::new("item", DataType::FixedSizeBinary(16), false);
elem.try_with_extension_type(ArrowUuid)?;
let outer = Field::new("ids", DataType::List(Arc::new(elem)), false);
let dtype = session.from_arrow_field(&outer)?;
let DType::List(inner_dt, _) = dtype else {
panic!("expected List dtype, got {dtype}");
};
assert!(
matches!(inner_dt.as_ref(), DType::Extension(ext) if ext.id() == Uuid.id()),
"expected Uuid extension element, got {inner_dt}",
);
Ok(())
}
#[test]
fn schema_roundtrip_preserves_nested_uuid() -> VortexResult<()> {
let session = ArrowSession::default();
let dtype = DType::Struct(
StructFields::from_iter([
(FieldName::from("id"), uuid_dtype(false)),
(
FieldName::from("ids"),
DType::List(Arc::new(uuid_dtype(true)), Nullability::NonNullable),
),
]),
Nullability::NonNullable,
);
let schema = session.to_arrow_schema(&dtype)?;
let roundtripped = session.from_arrow_schema(&schema)?;
assert_eq!(roundtripped, dtype);
Ok(())
}
#[test]
fn to_arrow_datatype_dispatches_plugins() -> VortexResult<()> {
let session = ArrowSession::default();
assert_eq!(
session.to_arrow_datatype(&uuid_dtype(false))?,
DataType::FixedSizeBinary(16)
);
assert_eq!(
session.to_arrow_datatype(&DType::Utf8(Nullability::Nullable))?,
DataType::Utf8View
);
Ok(())
}
#[test]
fn from_arrow_datatype_recurses_into_nested_extension_fields() -> VortexResult<()> {
let session = ArrowSession::default();
let mut elem = Field::new("item", DataType::FixedSizeBinary(16), false);
elem.try_with_extension_type(ArrowUuid)?;
let data_type = DataType::List(Arc::new(elem));
let dtype = session.from_arrow_datatype(&data_type, Nullability::Nullable)?;
let DType::List(inner_dt, Nullability::Nullable) = dtype else {
panic!("expected nullable List dtype, got {dtype}");
};
assert!(
matches!(inner_dt.as_ref(), DType::Extension(ext) if ext.id() == Uuid.id()),
"expected Uuid extension element, got {inner_dt}",
);
Ok(())
}
#[test]
fn from_arrow_fields_matches_schema_conversion() -> VortexResult<()> {
let session = ArrowSession::default();
let fields = Fields::from(vec![
Field::new("a", DataType::Int32, false),
Field::new("b", DataType::Utf8View, true),
]);
let struct_fields = session.from_arrow_fields(&fields)?;
let schema_dtype = session.from_arrow_schema(&Schema::new(fields))?;
assert_eq!(
schema_dtype,
DType::Struct(struct_fields, Nullability::NonNullable)
);
Ok(())
}
#[test]
fn from_arrow_field_maps_variant_without_importer() -> VortexResult<()> {
let session = ArrowSession::default();
let storage = DataType::Struct(
vec![
Field::new("metadata", DataType::BinaryView, false),
Field::new("value", DataType::BinaryView, true),
]
.into(),
);
let field = Field::new("v", storage, true).with_metadata(
[(
"ARROW:extension:name".to_string(),
"arrow.parquet.variant".to_string(),
)]
.into(),
);
assert_eq!(
session.from_arrow_field(&field)?,
DType::Variant(Nullability::Nullable)
);
Ok(())
}
#[test]
fn execute_arrow_target_none_preserves_top_level_uuid_metadata() -> VortexResult<()> {
let vortex_session = array_session();
let mut ctx = vortex_session.create_execution_ctx();
let session = vortex_session.arrow();
let mut field = Field::new("id", DataType::FixedSizeBinary(16), false);
field.try_with_extension_type(ArrowUuid)?;
let arrow_array: ArrowArrayRef = Arc::new(FixedSizeBinaryArray::try_from_iter(
[*b"0123456789abcdef", *b"fedcba9876543210"].into_iter(),
)?);
let vortex_array = session.from_arrow_array(arrow_array, &field)?;
let vortex_ext = vortex_array.dtype().as_extension();
assert!(vortex_ext.is::<Uuid>());
let exported = session.execute_arrow(vortex_array, None, &mut ctx)?;
assert_eq!(exported.data_type(), &DataType::FixedSizeBinary(16));
let fsb = exported.as_fixed_size_binary();
assert_eq!(fsb.len(), 2);
assert_eq!(fsb.value(0), b"0123456789abcdef");
assert_eq!(fsb.value(1), b"fedcba9876543210");
Ok(())
}
fn uuid_array(session: &ArrowSession) -> VortexResult<ArrayRef> {
let mut field = Field::new("id", DataType::FixedSizeBinary(16), false);
field.try_with_extension_type(ArrowUuid)?;
let arrow_array: ArrowArrayRef = Arc::new(FixedSizeBinaryArray::try_from_iter(
[*b"0123456789abcdef", *b"fedcba9876543210"].into_iter(),
)?);
session.from_arrow_array(arrow_array, &field)
}
#[test]
fn execute_arrow_target_none_preserves_nested_uuid_metadata() -> VortexResult<()> {
let vortex_session = array_session();
let mut ctx = vortex_session.create_execution_ctx();
let session = vortex_session.arrow();
let uuids = uuid_array(&session)?;
let struct_array = StructArray::try_new(
FieldNames::from(["id"]),
vec![uuids],
2,
Validity::NonNullable,
)?
.into_array();
let exported = session.execute_arrow(struct_array, None, &mut ctx)?;
let DataType::Struct(fields) = exported.data_type() else {
panic!("expected Struct, got {:?}", exported.data_type());
};
assert_eq!(fields[0].data_type(), &DataType::FixedSizeBinary(16));
assert!(has_valid_extension_type::<ArrowUuid>(&fields[0]));
let uuids = exported.as_struct().column(0).as_fixed_size_binary();
assert_eq!(uuids.value(0), b"0123456789abcdef");
assert_eq!(uuids.value(1), b"fedcba9876543210");
Ok(())
}
#[test]
fn execute_arrow_target_none_preserves_list_element_uuid_metadata() -> VortexResult<()> {
let vortex_session = array_session();
let mut ctx = vortex_session.create_execution_ctx();
let session = vortex_session.arrow();
let list = ListArray::try_new(
uuid_array(&session)?,
PrimitiveArray::from_iter([0i32, 1, 2]).into_array(),
Validity::NonNullable,
)?
.into_array();
let exported = session.execute_arrow(list, None, &mut ctx)?;
let DataType::List(elem) = exported.data_type() else {
panic!("expected List, got {:?}", exported.data_type());
};
assert_eq!(elem.data_type(), &DataType::FixedSizeBinary(16));
assert!(has_valid_extension_type::<ArrowUuid>(elem));
let uuids = exported.as_list::<i32>().values().as_fixed_size_binary();
assert_eq!(uuids.value(0), b"0123456789abcdef");
assert_eq!(uuids.value(1), b"fedcba9876543210");
Ok(())
}
#[test]
fn run_end_recurses_into_extension_values() -> VortexResult<()> {
let vortex_session = array_session();
let mut ctx = vortex_session.create_execution_ctx();
let session = vortex_session.arrow();
let mut values_field = Field::new("values", DataType::FixedSizeBinary(16), false);
values_field.try_with_extension_type(ArrowUuid)?;
let field = Field::new(
"id",
DataType::RunEndEncoded(
Arc::new(Field::new("run_ends", DataType::Int32, false)),
Arc::new(values_field),
),
false,
);
let dtype = session.from_arrow_field(&field)?;
assert!(
dtype.as_extension().is::<Uuid>(),
"expected a Uuid extension dtype, got {dtype}"
);
let values = FixedSizeBinaryArray::try_from_iter(
[*b"0123456789abcdef", *b"fedcba9876543210"].into_iter(),
)?;
let run_array: ArrowArrayRef = Arc::new(RunArray::<Int32Type>::try_new(
&Int32Array::from(vec![2i32, 5]),
&values,
)?);
let vortex_array = session.from_arrow_array(run_array, &field)?;
assert_eq!(vortex_array.len(), 5);
assert_eq!(vortex_array.dtype(), &dtype);
let exported = session.execute_arrow(vortex_array, Some(&field), &mut ctx)?;
assert_eq!(exported.len(), 5);
let ree = exported
.as_any()
.downcast_ref::<RunArray<Int32Type>>()
.ok_or_else(|| {
vortex_err!(
"expected an Int32 run-end array, got {}",
exported.data_type()
)
})?;
let values = ree.values().as_fixed_size_binary();
assert_eq!(values.value(0), b"0123456789abcdef");
assert_eq!(values.value(1), b"fedcba9876543210");
Ok(())
}
#[test]
fn dictionary_recurses_into_nested_extension_values() -> VortexResult<()> {
let session = ArrowSession::default();
let mut elem = Field::new("item", DataType::FixedSizeBinary(16), false);
elem.try_with_extension_type(ArrowUuid)?;
let uuids = FixedSizeBinaryArray::try_from_iter(
[*b"0123456789abcdef", *b"fedcba9876543210"].into_iter(),
)?;
let values = ArrowListArray::try_new(
Arc::new(elem),
OffsetBuffer::new(vec![0, 1, 2].into()),
Arc::new(uuids),
None,
)?;
let dict: ArrowArrayRef = Arc::new(DictionaryArray::<Int32Type>::try_new(
Int32Array::from(vec![0, 1, 0]),
Arc::new(values),
)?);
let field = Field::new("ids", dict.data_type().clone(), false);
let dtype = session.from_arrow_field(&field)?;
let DType::List(elem_dt, _) = &dtype else {
panic!("expected a List dtype, got {dtype}");
};
assert!(
elem_dt.as_extension().is::<Uuid>(),
"expected a Uuid extension element, got {elem_dt}"
);
let array = session.from_arrow_array(dict, &field)?;
assert_eq!(array.len(), 3);
assert_eq!(array.dtype(), &dtype);
let mut ctx = array_session().create_execution_ctx();
let exported = session.execute_arrow(array, Some(&field), &mut ctx)?;
assert_eq!(exported.data_type(), field.data_type());
let values = exported.as_any_dictionary().values().as_list::<i32>();
let uuids = values.values().as_fixed_size_binary();
assert_eq!(uuids.value(0), b"0123456789abcdef");
assert_eq!(uuids.value(1), b"fedcba9876543210");
Ok(())
}
#[rstest]
#[case(true)]
#[case(false)]
fn from_arrow_array_by_nullability(#[case] nullable: bool) -> VortexResult<()> {
let session = ArrowSession::default();
let mut uuid_field = Field::new("id", DataType::FixedSizeBinary(16), false);
uuid_field.try_with_extension_type(ArrowUuid)?;
let uuids: ArrowArrayRef = Arc::new(FixedSizeBinaryArray::try_from_iter(
[*b"0123456789abcdef", *b"fedcba9876543210"].into_iter(),
)?);
let arrow_struct: ArrowArrayRef = Arc::new(ArrowStructArray::try_new(
Fields::from(vec![uuid_field]),
vec![uuids],
None,
)?);
let array = session.from_arrow_array(ArrowArrayRef::clone(&arrow_struct), nullable)?;
assert_eq!(array.dtype().nullability(), nullable.into());
let DType::Struct(fields, _) = array.dtype() else {
panic!("expected a Struct dtype, got {}", array.dtype());
};
assert!(
fields
.field_by_index(0)
.vortex_expect("struct dtype has one field")
.as_extension()
.is::<Uuid>(),
"expected the nested Uuid extension to survive, got {}",
array.dtype()
);
let by_nullability = session.from_arrow_array(arrow_struct, Nullability::from(nullable))?;
assert_eq!(by_nullability.dtype(), array.dtype());
Ok(())
}
#[test]
fn dictionary_imports_as_dict_encoding() -> VortexResult<()> {
let session = ArrowSession::default();
let dict: ArrowArrayRef = Arc::new(DictionaryArray::<Int32Type>::try_new(
Int32Array::from(vec![0, 1, 0, 1]),
Arc::new(StringArray::from(vec!["a", "b"])),
)?);
let field = Field::new("s", dict.data_type().clone(), false);
let dtype = session.from_arrow_field(&field)?;
assert_eq!(dtype, DType::Utf8(Nullability::NonNullable));
let array = session.from_arrow_array(dict, &field)?;
assert!(array.is::<Dict>(), "expected a Dict encoding, got {array}");
assert_eq!(array.dtype(), &dtype);
assert_eq!(array.len(), 4);
Ok(())
}
}