1use std::any::Any;
24use std::borrow::Cow;
25use std::fmt::Debug;
26use std::sync::Arc;
27
28use arrow_array::Array as ArrowArray;
29use arrow_array::ArrayRef as ArrowArrayRef;
30use arrow_array::RecordBatch;
31use arrow_array::RunArray;
32use arrow_array::make_array;
33use arrow_array::types::Int16Type;
34use arrow_array::types::Int32Type;
35use arrow_array::types::Int64Type;
36use arrow_array::types::RunEndIndexType;
37use arrow_schema::DataType;
38use arrow_schema::Field;
39use arrow_schema::FieldRef;
40use arrow_schema::Fields;
41use arrow_schema::Schema;
42use arrow_schema::extension::EXTENSION_TYPE_NAME_KEY;
43use arrow_schema::extension::ExtensionType;
44use tracing::debug;
45use tracing::trace;
46use vortex_array::ArrayRef;
47use vortex_array::ExecutionCtx;
48use vortex_array::IntoArray;
49use vortex_array::arrays::DictArray;
50use vortex_array::arrays::FixedSizeListArray;
51use vortex_array::arrays::ListArray;
52use vortex_array::arrays::ListViewArray;
53use vortex_array::arrays::StructArray;
54use vortex_array::dtype::DType;
55use vortex_array::dtype::FieldName;
56use vortex_array::dtype::FieldNames;
57use vortex_array::dtype::NativePType;
58use vortex_array::dtype::Nullability;
59use vortex_array::dtype::StructFields;
60use vortex_array::dtype::extension::ExtId;
61use vortex_array::extension::datetime::AnyTemporal;
62use vortex_array::extension::uuid::Uuid;
63use vortex_array::validity::Validity;
64use vortex_error::VortexResult;
65use vortex_error::vortex_bail;
66use vortex_error::vortex_ensure;
67use vortex_error::vortex_err;
68use vortex_session::ArcSwapMap;
69use vortex_session::SessionExt;
70use vortex_session::SessionGuard;
71use vortex_session::SessionVar;
72use vortex_session::registry::Id;
73
74use crate::IntoVortexArray;
75use crate::convert::from_arrow_dyn;
76use crate::convert::map_from_arrow_parts;
77use crate::convert::nulls;
78use crate::convert::remove_nulls;
79use crate::dtype::from_arrow_data_type;
80use crate::dtype::to_data_type_naive;
81use crate::executor::execute_arrow_naive;
82use crate::run_end_import::run_end_from_arrow;
83
84pub enum ArrowExport {
90 Unsupported(ArrayRef),
92 Exported(ArrowArrayRef),
94}
95
96pub enum ArrowImport {
102 Unsupported(ArrowArrayRef),
104 Imported(ArrayRef),
106}
107
108pub trait IntoArrowField<'a> {
121 fn into_arrow_field(self, data_type: &DataType) -> Cow<'a, Field>;
123}
124
125impl<'a> IntoArrowField<'a> for &'a Field {
126 fn into_arrow_field(self, _data_type: &DataType) -> Cow<'a, Field> {
127 Cow::Borrowed(self)
128 }
129}
130
131impl<'a> IntoArrowField<'a> for &'a FieldRef {
132 fn into_arrow_field(self, _data_type: &DataType) -> Cow<'a, Field> {
133 Cow::Borrowed(self.as_ref())
134 }
135}
136
137impl<'a> IntoArrowField<'a> for bool {
138 fn into_arrow_field(self, data_type: &DataType) -> Cow<'a, Field> {
139 Cow::Owned(Field::new("", data_type.clone(), self))
140 }
141}
142
143impl<'a> IntoArrowField<'a> for Nullability {
144 fn into_arrow_field(self, data_type: &DataType) -> Cow<'a, Field> {
145 self.is_nullable().into_arrow_field(data_type)
146 }
147}
148
149pub trait ArrowExportVTable: 'static + Send + Sync + Debug {
154 fn arrow_ext_id(&self) -> Id;
156
157 fn vortex_id(&self) -> Id;
161
162 fn to_arrow_field(
165 &self,
166 name: &str,
167 dtype: &DType,
168 session: &ArrowSession,
169 ) -> VortexResult<Option<Field>>;
170
171 fn execute_arrow(
176 &self,
177 array: ArrayRef,
178 target: &Field,
179 ctx: &mut ExecutionCtx,
180 ) -> VortexResult<ArrowExport>;
181}
182
183pub trait ArrowImportVTable: 'static + Send + Sync + Debug {
190 fn arrow_ext_id(&self) -> Id;
192
193 #[allow(clippy::wrong_self_convention)]
200 fn from_arrow_field(
201 &self,
202 field: &Field,
203 session: &ArrowSession,
204 ) -> VortexResult<Option<DType>>;
205
206 #[allow(clippy::wrong_self_convention)]
216 fn from_arrow_array(
217 &self,
218 array: ArrowArrayRef,
219 field: &Field,
220 dtype: &DType,
221 session: &ArrowSession,
222 ) -> VortexResult<ArrowImport>;
223}
224
225pub type ArrowExportVTableRef = Arc<dyn ArrowExportVTable>;
226pub type ArrowImportVTableRef = Arc<dyn ArrowImportVTable>;
227
228type ArrowExporterRegistry = ArcSwapMap<Id, Arc<[ArrowExportVTableRef]>>;
230type VortexExporterRegistry = ArcSwapMap<ExtId, Arc<[ArrowExportVTableRef]>>;
232type ArrowImporterRegistry = ArcSwapMap<Id, Arc<[ArrowImportVTableRef]>>;
234
235#[derive(Clone, Debug)]
245pub struct ArrowSession {
246 exporters: ArrowExporterRegistry,
247 exporters_by_vortex: VortexExporterRegistry,
248 importers: ArrowImporterRegistry,
249}
250
251impl Default for ArrowSession {
252 fn default() -> Self {
253 let session = Self {
254 exporters: ArrowExporterRegistry::default(),
255 exporters_by_vortex: VortexExporterRegistry::default(),
256 importers: ArrowImporterRegistry::default(),
257 };
258
259 session.register_exporter(Arc::new(Uuid));
260 session.register_importer(Arc::new(Uuid));
261
262 session
263 }
264}
265
266impl ArrowSession {
267 pub fn register_exporter(&self, exporter: ArrowExportVTableRef) {
270 self.exporters.push(
271 exporter.arrow_ext_id(),
272 ArrowExportVTableRef::clone(&exporter),
273 );
274 self.exporters_by_vortex
275 .push(exporter.vortex_id(), exporter);
276 }
277
278 pub fn register_importer(&self, importer: ArrowImportVTableRef) {
280 self.importers.push(importer.arrow_ext_id(), importer);
281 }
282
283 fn exporters(&self, id: &Id) -> Arc<[ArrowExportVTableRef]> {
284 self.exporters.get(id).unwrap_or_else(|| Arc::from([]))
285 }
286
287 fn exporters_by_vortex(&self, id: &Id) -> Arc<[ArrowExportVTableRef]> {
288 self.exporters_by_vortex
289 .get(id)
290 .unwrap_or_else(|| Arc::from([]))
291 }
292
293 fn importers(&self, id: &Id) -> Arc<[ArrowImportVTableRef]> {
294 self.importers.get(id).unwrap_or_else(|| Arc::from([]))
295 }
296
297 pub fn to_arrow_field(&self, name: &str, dtype: &DType) -> VortexResult<Field> {
302 match dtype {
304 DType::List(elem_dtype, nullability) => {
305 let elem_field = self.to_arrow_field(Field::LIST_FIELD_DEFAULT_NAME, elem_dtype)?;
306 Ok(Field::new_list(name, elem_field, nullability.is_nullable()))
307 }
308 DType::FixedSizeList(elem_dtype, elem_size, nullability) => {
309 let elem_field = self.to_arrow_field(Field::LIST_FIELD_DEFAULT_NAME, elem_dtype)?;
310 Ok(Field::new_fixed_size_list(
311 name,
312 elem_field,
313 (*elem_size).try_into()?,
314 nullability.is_nullable(),
315 ))
316 }
317 DType::Map(map_dtype, nullability) => {
318 let key = self.to_arrow_field("key", &map_dtype.key_dtype())?;
319 let value = self.to_arrow_field("value", &map_dtype.value_dtype())?;
320 let entries = Field::new_struct("entries", Fields::from(vec![key, value]), false);
321 Ok(Field::new(
322 name,
323 DataType::Map(Arc::new(entries), map_dtype.keys_sorted()),
324 nullability.is_nullable(),
325 ))
326 }
327 DType::Struct(fields, nullability) => {
328 let arrow_fields = Fields::from_iter(
329 fields
330 .fields()
331 .zip(fields.names().iter())
332 .map(|(field, name)| self.to_arrow_field(name.as_ref(), &field))
333 .collect::<VortexResult<Vec<_>>>()?,
334 );
335 Ok(Field::new_struct(
336 name,
337 arrow_fields,
338 nullability.is_nullable(),
339 ))
340 }
341 DType::Extension(ext) if !ext.is::<AnyTemporal>() => {
342 for plugin in self.exporters_by_vortex(&ext.id()).iter() {
343 if let Some(field) =
344 plugin.to_arrow_field(name, &DType::Extension(ext.clone()), self)?
345 {
346 return Ok(field);
347 }
348 }
349 vortex_bail!("extension type cannot be converted to Arrow without a plugin: {ext}");
350 }
351 DType::Variant(_) => {
352 Ok(Field::new(
356 name,
357 DataType::Struct(
358 vec![
359 Field::new("metadata", DataType::BinaryView, dtype.is_nullable()),
360 Field::new("value", DataType::BinaryView, dtype.is_nullable()),
361 ]
362 .into(),
363 ),
364 dtype.is_nullable(),
365 )
366 .with_metadata(
367 [(
368 EXTENSION_TYPE_NAME_KEY.to_string(),
369 "arrow.parquet.variant".to_string(),
370 )]
371 .into(),
372 ))
373 }
374 _ => Ok(Field::new(
375 name,
376 to_data_type_naive(dtype)?,
377 dtype.is_nullable(),
378 )),
379 }
380 }
381
382 pub fn to_arrow_schema(&self, dtype: &DType) -> VortexResult<Schema> {
386 let DType::Struct(struct_dtype, _) = dtype else {
387 vortex_bail!("to_arrow_schema requires a top-level struct dtype, got {dtype}");
388 };
389 let mut fields = Vec::with_capacity(struct_dtype.names().len());
390 for (name, field_dtype) in struct_dtype.names().iter().zip(struct_dtype.fields()) {
391 fields.push(self.to_arrow_field(name.as_ref(), &field_dtype)?);
392 }
393 Ok(Schema::new(fields))
394 }
395
396 pub fn to_arrow_datatype(&self, dtype: &DType) -> VortexResult<DataType> {
402 Ok(self.to_arrow_field("", dtype)?.data_type().clone())
403 }
404
405 #[expect(clippy::disallowed_methods, reason = "interning a dynamic id")]
414 pub fn from_arrow_field(&self, field: &Field) -> VortexResult<DType> {
415 if let Some(name) = field.metadata().get(EXTENSION_TYPE_NAME_KEY) {
416 for plugin in self.importers(&Id::new(name)).iter() {
417 if let Some(dtype) = plugin.from_arrow_field(field, self)? {
418 return Ok(dtype);
419 }
420 }
421 if name == "arrow.parquet.variant" {
423 return Ok(DType::Variant(field.is_nullable().into()));
424 }
425 }
426 self.from_arrow_datatype(field.data_type(), field.is_nullable().into())
427 }
428
429 pub fn from_arrow_datatype(
440 &self,
441 data_type: &DataType,
442 nullability: Nullability,
443 ) -> VortexResult<DType> {
444 Ok(match data_type {
445 DataType::List(elem)
446 | DataType::LargeList(elem)
447 | DataType::ListView(elem)
448 | DataType::LargeListView(elem) => {
449 DType::List(Arc::new(self.from_arrow_field(elem.as_ref())?), nullability)
450 }
451 DataType::FixedSizeList(elem, size) => DType::FixedSizeList(
452 Arc::new(self.from_arrow_field(elem.as_ref())?),
453 *size as u32,
454 nullability,
455 ),
456 DataType::Map(entries, keys_sorted) => {
457 vortex_ensure!(
458 !entries.is_nullable(),
459 "Arrow map entries field must be non-nullable"
460 );
461 let DataType::Struct(fields) = entries.data_type() else {
462 vortex_bail!(
463 "Arrow map entries field must have Struct type, got {:?}",
464 entries.data_type()
465 );
466 };
467 vortex_ensure!(
468 fields.len() == 2,
469 "Arrow map entries struct must contain exactly two fields"
470 );
471 vortex_ensure!(
472 !fields[0].is_nullable(),
473 "Arrow map key field must be non-nullable"
474 );
475 DType::map(
476 self.from_arrow_field(fields[0].as_ref())?,
477 self.from_arrow_field(fields[1].as_ref())?,
478 *keys_sorted,
479 nullability,
480 )?
481 }
482 DataType::Struct(fields) => DType::Struct(self.from_arrow_fields(fields)?, nullability),
483 DataType::Dictionary(_, value_type) => {
484 self.from_arrow_datatype(value_type.as_ref(), nullability)?
485 }
486 DataType::RunEndEncoded(_, value_field) => {
487 self.from_arrow_field(&run_end_values_field(value_field, nullability))?
488 }
489 _ => from_arrow_data_type(data_type, nullability)?,
490 })
491 }
492
493 pub fn from_arrow_fields(&self, fields: &Fields) -> VortexResult<StructFields> {
496 fields
497 .iter()
498 .map(|f| {
499 self.from_arrow_field(f)
500 .map(|dt| (FieldName::from(f.name().as_str()), dt))
501 })
502 .collect::<VortexResult<StructFields>>()
503 }
504
505 pub fn from_arrow_schema(&self, schema: &Schema) -> VortexResult<DType> {
509 Ok(DType::Struct(
510 self.from_arrow_fields(schema.fields())?,
511 Nullability::NonNullable,
512 ))
513 }
514
515 pub fn from_arrow_record_batch(
523 &self,
524 batch: RecordBatch,
525 schema: &Schema,
526 ) -> VortexResult<ArrayRef> {
527 vortex_ensure!(
528 batch.num_columns() == schema.fields().len(),
529 "RecordBatch has {} columns but schema has {} fields",
530 batch.num_columns(),
531 schema.fields().len()
532 );
533 let length = batch.num_rows();
534 let names = FieldNames::from_iter(
535 schema
536 .fields()
537 .iter()
538 .map(|f| FieldName::from(f.name().as_str())),
539 );
540 let mut columns = Vec::with_capacity(schema.fields().len());
541 for (col, field) in batch.columns().iter().zip(schema.fields().iter()) {
542 columns.push(self.from_arrow_array_inner(ArrowArrayRef::clone(col), field)?);
543 }
544 Ok(StructArray::try_new(names, columns, length, Validity::NonNullable)?.into_array())
545 }
546
547 #[expect(clippy::disallowed_methods, reason = "interning a dynamic id")]
555 pub fn execute_arrow(
556 &self,
557 array: ArrayRef,
558 target: Option<&Field>,
559 ctx: &mut ExecutionCtx,
560 ) -> VortexResult<ArrowArrayRef> {
561 let arrow_field;
565 let target_field = match target {
566 Some(field) => field,
567 None => {
568 let session = ctx.session().clone();
569 arrow_field = session.arrow().to_arrow_field("", array.dtype())?;
570 &arrow_field
571 }
572 };
573
574 if let Some(arrow_ext_name) = target_field.metadata().get(EXTENSION_TYPE_NAME_KEY) {
575 let len = array.len();
578 let mut current = array;
579
580 for plugin in self.exporters(&Id::new(arrow_ext_name)).iter() {
581 trace!(
582 plugin = ?plugin,
583 extension_name = arrow_ext_name,
584 "probing plugin for converting Arrow array"
585 );
586
587 match plugin.execute_arrow(current, target_field, ctx)? {
588 ArrowExport::Exported(arrow) => {
589 vortex_ensure!(
590 arrow.len() == len,
591 "Arrow array length does not match Vortex array length after conversion to {:?}",
592 arrow
593 );
594 return Ok(arrow);
595 }
596 ArrowExport::Unsupported(array) => current = array,
597 }
598 }
599
600 debug!(
601 extension_id = arrow_ext_name,
602 data_type = ?target_field.data_type(),
603 "unsupported Arrow extension type encountered, falling back to naive execution"
604 );
605
606 return execute_arrow_naive(current, Some(target_field.data_type()), ctx);
607 }
608
609 execute_arrow_naive(array, target.map(|field| field.data_type()), ctx)
610 }
611
612 pub fn from_arrow_array<'a>(
632 &self,
633 array: ArrowArrayRef,
634 field: impl IntoArrowField<'a>,
635 ) -> VortexResult<ArrayRef> {
636 let field = field.into_arrow_field(array.data_type());
637 self.from_arrow_array_inner(array, field.as_ref())
638 }
639
640 #[allow(clippy::wrong_self_convention)]
644 fn from_arrow_array_inner(
645 &self,
646 array: ArrowArrayRef,
647 field: &Field,
648 ) -> VortexResult<ArrayRef> {
649 if let Some(extension_name) = field.metadata().get(EXTENSION_TYPE_NAME_KEY) {
650 #[expect(clippy::disallowed_methods, reason = "interning a dynamic id")]
651 let importers = self.importers(&Id::new(extension_name));
652 if !importers.is_empty() {
653 let dtype = self.from_arrow_field(field)?;
654 let mut current = array;
655 for plugin in importers.iter() {
656 match plugin.from_arrow_array(current, field, &dtype, self)? {
657 ArrowImport::Imported(arr) => return Ok(arr),
658 ArrowImport::Unsupported(arr) => current = arr,
659 }
660 }
661 return self.from_arrow_array_canonical(current.as_ref(), field);
662 }
663 }
664 self.from_arrow_array_canonical(array.as_ref(), field)
665 }
666
667 #[allow(clippy::wrong_self_convention)]
670 fn from_arrow_array_canonical(
671 &self,
672 array: &dyn ArrowArray,
673 field: &Field,
674 ) -> VortexResult<ArrayRef> {
675 use arrow_array::cast::AsArray;
676
677 match field.data_type() {
678 DataType::Struct(fields) => {
679 let arrow_struct = array.as_struct();
680 let names = FieldNames::from_iter(
681 fields.iter().map(|f| FieldName::from(f.name().as_str())),
682 );
683 let columns = arrow_struct
684 .columns()
685 .iter()
686 .zip(fields.iter())
687 .map(|(col, child_field)| {
688 let inner = if col.null_count() > 0 && !child_field.is_nullable() {
691 make_array(remove_nulls(col.to_data())?)
692 } else {
693 ArrowArrayRef::clone(col)
694 };
695 self.from_arrow_array_inner(inner, child_field.as_ref())
696 })
697 .collect::<VortexResult<Vec<_>>>()?;
698 let validity = nulls(arrow_struct.nulls(), field.is_nullable())?;
699 Ok(
700 StructArray::try_new(names, columns, arrow_struct.len(), validity)?
701 .into_array(),
702 )
703 }
704 DataType::List(elem_field) => {
705 let list = array.as_list::<i32>();
706 let elements = self
707 .from_arrow_array(ArrowArrayRef::clone(list.values()), elem_field.as_ref())?;
708 let offsets = list.offsets().clone().into_array();
709 let validity = nulls(list.nulls(), field.is_nullable())?;
710 Ok(ListArray::try_new(elements, offsets, validity)?.into_array())
711 }
712 DataType::LargeList(elem_field) => {
713 let list = array.as_list::<i64>();
714 let elements = self
715 .from_arrow_array(ArrowArrayRef::clone(list.values()), elem_field.as_ref())?;
716 let offsets = list.offsets().clone().into_array();
717 let validity = nulls(list.nulls(), field.is_nullable())?;
718 Ok(ListArray::try_new(elements, offsets, validity)?.into_array())
719 }
720 DataType::FixedSizeList(elem_field, list_size) => {
721 let fsl = array.as_fixed_size_list();
722 let elements = self.from_arrow_array_inner(
723 ArrowArrayRef::clone(fsl.values()),
724 elem_field.as_ref(),
725 )?;
726 let validity = nulls(fsl.nulls(), field.is_nullable())?;
727 Ok(
728 FixedSizeListArray::try_new(elements, *list_size as u32, validity, fsl.len())?
729 .into_array(),
730 )
731 }
732 DataType::ListView(elem_field) => {
733 let list = array.as_list_view::<i32>();
734 let elements = self
735 .from_arrow_array(ArrowArrayRef::clone(list.values()), elem_field.as_ref())?;
736 let offsets = list.offsets().clone().into_array();
737 let sizes = list.sizes().clone().into_array();
738 let validity = nulls(list.nulls(), field.is_nullable())?;
739 Ok(ListViewArray::try_new(elements, offsets, sizes, validity)?.into_array())
740 }
741 DataType::LargeListView(elem_field) => {
742 let list = array.as_list_view::<i64>();
743 let elements = self
744 .from_arrow_array(ArrowArrayRef::clone(list.values()), elem_field.as_ref())?;
745 let offsets = list.offsets().clone().into_array();
746 let sizes = list.sizes().clone().into_array();
747 let validity = nulls(list.nulls(), field.is_nullable())?;
748 Ok(ListViewArray::try_new(elements, offsets, sizes, validity)?.into_array())
749 }
750 DataType::Map(entries_field, keys_sorted) => {
751 let map = array.as_map();
752 let entries_array: ArrowArrayRef = Arc::new(map.entries().clone());
753 let entries = self.from_arrow_array_inner(entries_array, entries_field.as_ref())?;
754 map_from_arrow_parts(
755 entries,
756 map.offsets(),
757 map.nulls(),
758 *keys_sorted,
759 field.is_nullable(),
760 )
761 }
762 DataType::RunEndEncoded(ends_field, values_field) => {
763 let values_field = run_end_values_field(values_field, field.is_nullable().into());
764 match ends_field.data_type() {
765 DataType::Int16 => self.run_end_from_arrow::<Int16Type>(array, &values_field),
766 DataType::Int32 => self.run_end_from_arrow::<Int32Type>(array, &values_field),
767 DataType::Int64 => self.run_end_from_arrow::<Int64Type>(array, &values_field),
768 ends_dt => vortex_bail!(
769 "Arrow run-end array run ends must be Int16, Int32 or Int64, got {ends_dt}"
770 ),
771 }
772 }
773 DataType::Dictionary(..) => {
774 let dict = array.as_any_dictionary();
775 let values = self
781 .from_arrow_array(ArrowArrayRef::clone(dict.values()), field.is_nullable())?;
782 let codes = dict.keys();
783 let codes = from_arrow_dyn(codes, codes.is_nullable())?;
784 Ok(unsafe { DictArray::new_unchecked(codes, values) }.into_array())
787 }
788 _ => from_arrow_dyn(array, field.is_nullable()),
789 }
790 }
791
792 #[allow(clippy::wrong_self_convention)]
795 fn run_end_from_arrow<R: RunEndIndexType>(
796 &self,
797 array: &dyn ArrowArray,
798 values_field: &Field,
799 ) -> VortexResult<ArrayRef>
800 where
801 R::Native: NativePType,
802 {
803 let run_array = array
804 .as_any()
805 .downcast_ref::<RunArray<R>>()
806 .ok_or_else(|| vortex_err!("expected an Arrow RunArray, got {}", array.data_type()))?;
807 let values =
808 self.from_arrow_array_inner(ArrowArrayRef::clone(run_array.values()), values_field)?;
809 run_end_from_arrow(run_array, values)
810 }
811}
812
813fn run_end_values_field(values_field: &FieldRef, nullability: Nullability) -> Field {
816 values_field
817 .as_ref()
818 .clone()
819 .with_nullable(nullability.into())
820}
821
822pub(crate) fn has_valid_extension_type<E: ExtensionType>(field: &Field) -> bool {
826 if field.extension_type_name() != Some(E::NAME) {
827 return false;
828 }
829
830 E::try_new_from_field_metadata(field.data_type(), field.metadata()).is_ok()
831}
832
833impl SessionVar for ArrowSession {
834 fn as_any(&self) -> &dyn Any {
835 self
836 }
837
838 fn as_any_mut(&mut self) -> &mut dyn Any {
839 self
840 }
841}
842
843pub trait ArrowSessionExt: SessionExt {
845 fn arrow(&self) -> SessionGuard<'_, ArrowSession>;
847}
848
849impl<S: SessionExt> ArrowSessionExt for S {
850 fn arrow(&self) -> SessionGuard<'_, ArrowSession> {
851 self.get::<ArrowSession>()
852 }
853}
854
855#[cfg(test)]
856mod tests {
857 use std::sync::Arc;
858
859 use arrow_array::DictionaryArray;
860 use arrow_array::FixedSizeBinaryArray;
861 use arrow_array::Int32Array;
862 use arrow_array::ListArray as ArrowListArray;
863 use arrow_array::StringArray;
864 use arrow_array::StructArray as ArrowStructArray;
865 use arrow_array::cast::AsArray;
866 use arrow_buffer::OffsetBuffer;
867 use arrow_schema::DataType;
868 use arrow_schema::Field;
869 use arrow_schema::extension::Uuid as ArrowUuid;
870 use rstest::rstest;
871 use vortex_array::IntoArray;
872 use vortex_array::VortexSessionExecute;
873 use vortex_array::array_session;
874 use vortex_array::arrays::Dict;
875 use vortex_array::arrays::ListArray;
876 use vortex_array::arrays::PrimitiveArray;
877 use vortex_array::dtype::DType;
878 use vortex_array::dtype::FieldName;
879 use vortex_array::dtype::Nullability;
880 use vortex_array::dtype::PType;
881 use vortex_array::dtype::StructFields;
882 use vortex_array::dtype::extension::ExtDType;
883 use vortex_array::dtype::extension::ExtVTable;
884 use vortex_array::extension::uuid::Uuid;
885 use vortex_array::extension::uuid::UuidMetadata;
886 use vortex_error::VortexExpect;
887 use vortex_error::VortexResult;
888
889 use super::*;
890
891 fn uuid_dtype(nullable: bool) -> DType {
892 let storage = DType::FixedSizeList(
893 Arc::new(DType::Primitive(PType::U8, Nullability::NonNullable)),
894 16,
895 nullable.into(),
896 );
897 DType::Extension(
898 ExtDType::try_with_vtable(Uuid, UuidMetadata::default(), storage)
899 .expect("uuid ext dtype")
900 .erased(),
901 )
902 }
903
904 #[test]
905 fn to_arrow_field_top_level_uuid_carries_extension_metadata() -> VortexResult<()> {
906 let session = ArrowSession::default();
907 let field = session.to_arrow_field("id", &uuid_dtype(false))?;
908 assert!(has_valid_extension_type::<ArrowUuid>(&field));
909 Ok(())
910 }
911
912 #[test]
913 fn to_arrow_field_struct_with_nested_uuid_preserves_metadata() -> VortexResult<()> {
914 let session = ArrowSession::default();
915 let dtype = DType::Struct(
916 StructFields::from_iter([(FieldName::from("id"), uuid_dtype(false))]),
917 Nullability::NonNullable,
918 );
919 let field = session.to_arrow_field("row", &dtype)?;
920 let DataType::Struct(inner) = field.data_type() else {
921 panic!("expected Struct, got {:?}", field.data_type());
922 };
923 assert_eq!(inner.len(), 1);
924 assert_eq!(inner[0].data_type(), &DataType::FixedSizeBinary(16));
925 assert!(has_valid_extension_type::<ArrowUuid>(&inner[0]));
926 Ok(())
927 }
928
929 #[test]
930 fn to_arrow_field_list_of_uuid_preserves_metadata() -> VortexResult<()> {
931 let session = ArrowSession::default();
932 let dtype = DType::List(Arc::new(uuid_dtype(true)), Nullability::NonNullable);
933 let field = session.to_arrow_field("ids", &dtype)?;
934 let DataType::List(elem) = field.data_type() else {
935 panic!("expected List, got {:?}", field.data_type());
936 };
937 assert!(has_valid_extension_type::<ArrowUuid>(elem));
938 Ok(())
939 }
940
941 #[test]
942 fn to_arrow_field_fixed_size_list_of_uuid_preserves_metadata() -> VortexResult<()> {
943 let session = ArrowSession::default();
944 let dtype = DType::FixedSizeList(Arc::new(uuid_dtype(false)), 3, Nullability::NonNullable);
945 let field = session.to_arrow_field("triple", &dtype)?;
946 let DataType::FixedSizeList(elem, size) = field.data_type() else {
947 panic!("expected FixedSizeList, got {:?}", field.data_type());
948 };
949 assert_eq!(*size, 3);
950 assert!(has_valid_extension_type::<ArrowUuid>(elem));
951 Ok(())
952 }
953
954 #[test]
955 fn schema_roundtrip_preserves_map_uuid_fields() -> VortexResult<()> {
956 let session = ArrowSession::default();
957 let map = DType::map(
958 uuid_dtype(false),
959 uuid_dtype(true),
960 true,
961 Nullability::Nullable,
962 )?;
963 let dtype = DType::Struct(
964 StructFields::from_iter([(FieldName::from("ids"), map)]),
965 Nullability::NonNullable,
966 );
967
968 let schema = session.to_arrow_schema(&dtype)?;
969 let field = schema.field(0);
970 let DataType::Map(entries, keys_sorted) = field.data_type() else {
971 panic!("expected Map, got {:?}", field.data_type());
972 };
973 assert!(*keys_sorted);
974 assert_eq!(entries.name(), "entries");
975 assert!(!entries.is_nullable());
976 let DataType::Struct(fields) = entries.data_type() else {
977 panic!("expected map entries struct, got {:?}", entries.data_type());
978 };
979 assert!(has_valid_extension_type::<ArrowUuid>(&fields[0]));
980 assert!(has_valid_extension_type::<ArrowUuid>(&fields[1]));
981 assert!(!fields[0].is_nullable());
982 assert!(fields[1].is_nullable());
983
984 assert_eq!(session.from_arrow_schema(&schema)?, dtype);
985 Ok(())
986 }
987
988 #[test]
989 fn to_arrow_schema_struct_of_struct_uuid() -> VortexResult<()> {
990 let session = ArrowSession::default();
991 let inner = DType::Struct(
992 StructFields::from_iter([(FieldName::from("id"), uuid_dtype(true))]),
993 Nullability::NonNullable,
994 );
995 let outer = DType::Struct(
996 StructFields::from_iter([(FieldName::from("payload"), inner)]),
997 Nullability::NonNullable,
998 );
999 let schema = session.to_arrow_schema(&outer)?;
1000 let payload = schema.field(0);
1001 let DataType::Struct(inner_fields) = payload.data_type() else {
1002 panic!("expected Struct, got {:?}", payload.data_type());
1003 };
1004 assert!(has_valid_extension_type::<ArrowUuid>(&inner_fields[0]));
1005 Ok(())
1006 }
1007
1008 #[test]
1009 fn from_arrow_field_recurses_into_nested_uuid() -> VortexResult<()> {
1010 let session = ArrowSession::default();
1011 let mut elem = Field::new("item", DataType::FixedSizeBinary(16), false);
1012 elem.try_with_extension_type(ArrowUuid)?;
1013 let outer = Field::new("ids", DataType::List(Arc::new(elem)), false);
1014
1015 let dtype = session.from_arrow_field(&outer)?;
1016 let DType::List(inner_dt, _) = dtype else {
1017 panic!("expected List dtype, got {dtype}");
1018 };
1019 assert!(
1020 matches!(inner_dt.as_ref(), DType::Extension(ext) if ext.id() == Uuid.id()),
1021 "expected Uuid extension element, got {inner_dt}",
1022 );
1023 Ok(())
1024 }
1025
1026 #[test]
1027 fn schema_roundtrip_preserves_nested_uuid() -> VortexResult<()> {
1028 let session = ArrowSession::default();
1029 let dtype = DType::Struct(
1030 StructFields::from_iter([
1031 (FieldName::from("id"), uuid_dtype(false)),
1032 (
1033 FieldName::from("ids"),
1034 DType::List(Arc::new(uuid_dtype(true)), Nullability::NonNullable),
1035 ),
1036 ]),
1037 Nullability::NonNullable,
1038 );
1039 let schema = session.to_arrow_schema(&dtype)?;
1040 let roundtripped = session.from_arrow_schema(&schema)?;
1041 assert_eq!(roundtripped, dtype);
1042 Ok(())
1043 }
1044
1045 #[test]
1046 fn to_arrow_datatype_dispatches_plugins() -> VortexResult<()> {
1047 let session = ArrowSession::default();
1048 assert_eq!(
1049 session.to_arrow_datatype(&uuid_dtype(false))?,
1050 DataType::FixedSizeBinary(16)
1051 );
1052 assert_eq!(
1053 session.to_arrow_datatype(&DType::Utf8(Nullability::Nullable))?,
1054 DataType::Utf8View
1055 );
1056 Ok(())
1057 }
1058
1059 #[test]
1060 fn from_arrow_datatype_recurses_into_nested_extension_fields() -> VortexResult<()> {
1061 let session = ArrowSession::default();
1062 let mut elem = Field::new("item", DataType::FixedSizeBinary(16), false);
1063 elem.try_with_extension_type(ArrowUuid)?;
1064 let data_type = DataType::List(Arc::new(elem));
1065
1066 let dtype = session.from_arrow_datatype(&data_type, Nullability::Nullable)?;
1067 let DType::List(inner_dt, Nullability::Nullable) = dtype else {
1068 panic!("expected nullable List dtype, got {dtype}");
1069 };
1070 assert!(
1071 matches!(inner_dt.as_ref(), DType::Extension(ext) if ext.id() == Uuid.id()),
1072 "expected Uuid extension element, got {inner_dt}",
1073 );
1074 Ok(())
1075 }
1076
1077 #[test]
1078 fn from_arrow_fields_matches_schema_conversion() -> VortexResult<()> {
1079 let session = ArrowSession::default();
1080 let fields = Fields::from(vec![
1081 Field::new("a", DataType::Int32, false),
1082 Field::new("b", DataType::Utf8View, true),
1083 ]);
1084 let struct_fields = session.from_arrow_fields(&fields)?;
1085 let schema_dtype = session.from_arrow_schema(&Schema::new(fields))?;
1086 assert_eq!(
1087 schema_dtype,
1088 DType::Struct(struct_fields, Nullability::NonNullable)
1089 );
1090 Ok(())
1091 }
1092
1093 #[test]
1094 fn from_arrow_field_maps_variant_without_importer() -> VortexResult<()> {
1095 let session = ArrowSession::default();
1096 let storage = DataType::Struct(
1097 vec![
1098 Field::new("metadata", DataType::BinaryView, false),
1099 Field::new("value", DataType::BinaryView, true),
1100 ]
1101 .into(),
1102 );
1103 let field = Field::new("v", storage, true).with_metadata(
1104 [(
1105 "ARROW:extension:name".to_string(),
1106 "arrow.parquet.variant".to_string(),
1107 )]
1108 .into(),
1109 );
1110 assert_eq!(
1111 session.from_arrow_field(&field)?,
1112 DType::Variant(Nullability::Nullable)
1113 );
1114 Ok(())
1115 }
1116
1117 #[test]
1118 fn execute_arrow_target_none_preserves_top_level_uuid_metadata() -> VortexResult<()> {
1119 let vortex_session = array_session();
1120 let mut ctx = vortex_session.create_execution_ctx();
1121 let session = vortex_session.arrow();
1122
1123 let mut field = Field::new("id", DataType::FixedSizeBinary(16), false);
1124 field.try_with_extension_type(ArrowUuid)?;
1125 let arrow_array: ArrowArrayRef = Arc::new(FixedSizeBinaryArray::try_from_iter(
1126 [*b"0123456789abcdef", *b"fedcba9876543210"].into_iter(),
1127 )?);
1128
1129 let vortex_array = session.from_arrow_array(arrow_array, &field)?;
1130
1131 let vortex_ext = vortex_array.dtype().as_extension();
1132 assert!(vortex_ext.is::<Uuid>());
1133
1134 let exported = session.execute_arrow(vortex_array, None, &mut ctx)?;
1135 assert_eq!(exported.data_type(), &DataType::FixedSizeBinary(16));
1136 let fsb = exported.as_fixed_size_binary();
1137 assert_eq!(fsb.len(), 2);
1138 assert_eq!(fsb.value(0), b"0123456789abcdef");
1139 assert_eq!(fsb.value(1), b"fedcba9876543210");
1140 Ok(())
1141 }
1142
1143 fn uuid_array(session: &ArrowSession) -> VortexResult<ArrayRef> {
1145 let mut field = Field::new("id", DataType::FixedSizeBinary(16), false);
1146 field.try_with_extension_type(ArrowUuid)?;
1147 let arrow_array: ArrowArrayRef = Arc::new(FixedSizeBinaryArray::try_from_iter(
1148 [*b"0123456789abcdef", *b"fedcba9876543210"].into_iter(),
1149 )?);
1150 session.from_arrow_array(arrow_array, &field)
1151 }
1152
1153 #[test]
1157 fn execute_arrow_target_none_preserves_nested_uuid_metadata() -> VortexResult<()> {
1158 let vortex_session = array_session();
1159 let mut ctx = vortex_session.create_execution_ctx();
1160 let session = vortex_session.arrow();
1161
1162 let uuids = uuid_array(&session)?;
1163 let struct_array = StructArray::try_new(
1164 FieldNames::from(["id"]),
1165 vec![uuids],
1166 2,
1167 Validity::NonNullable,
1168 )?
1169 .into_array();
1170
1171 let exported = session.execute_arrow(struct_array, None, &mut ctx)?;
1172 let DataType::Struct(fields) = exported.data_type() else {
1173 panic!("expected Struct, got {:?}", exported.data_type());
1174 };
1175 assert_eq!(fields[0].data_type(), &DataType::FixedSizeBinary(16));
1176 assert!(has_valid_extension_type::<ArrowUuid>(&fields[0]));
1177
1178 let uuids = exported.as_struct().column(0).as_fixed_size_binary();
1179 assert_eq!(uuids.value(0), b"0123456789abcdef");
1180 assert_eq!(uuids.value(1), b"fedcba9876543210");
1181 Ok(())
1182 }
1183
1184 #[test]
1187 fn execute_arrow_target_none_preserves_list_element_uuid_metadata() -> VortexResult<()> {
1188 let vortex_session = array_session();
1189 let mut ctx = vortex_session.create_execution_ctx();
1190 let session = vortex_session.arrow();
1191
1192 let list = ListArray::try_new(
1193 uuid_array(&session)?,
1194 PrimitiveArray::from_iter([0i32, 1, 2]).into_array(),
1195 Validity::NonNullable,
1196 )?
1197 .into_array();
1198
1199 let exported = session.execute_arrow(list, None, &mut ctx)?;
1200 let DataType::List(elem) = exported.data_type() else {
1201 panic!("expected List, got {:?}", exported.data_type());
1202 };
1203 assert_eq!(elem.data_type(), &DataType::FixedSizeBinary(16));
1204 assert!(has_valid_extension_type::<ArrowUuid>(elem));
1205
1206 let uuids = exported.as_list::<i32>().values().as_fixed_size_binary();
1207 assert_eq!(uuids.value(0), b"0123456789abcdef");
1208 assert_eq!(uuids.value(1), b"fedcba9876543210");
1209 Ok(())
1210 }
1211
1212 #[test]
1215 fn run_end_recurses_into_extension_values() -> VortexResult<()> {
1216 let vortex_session = array_session();
1217 let mut ctx = vortex_session.create_execution_ctx();
1218 let session = vortex_session.arrow();
1219
1220 let mut values_field = Field::new("values", DataType::FixedSizeBinary(16), false);
1221 values_field.try_with_extension_type(ArrowUuid)?;
1222 let field = Field::new(
1223 "id",
1224 DataType::RunEndEncoded(
1225 Arc::new(Field::new("run_ends", DataType::Int32, false)),
1226 Arc::new(values_field),
1227 ),
1228 false,
1229 );
1230
1231 let dtype = session.from_arrow_field(&field)?;
1232 assert!(
1233 dtype.as_extension().is::<Uuid>(),
1234 "expected a Uuid extension dtype, got {dtype}"
1235 );
1236
1237 let values = FixedSizeBinaryArray::try_from_iter(
1238 [*b"0123456789abcdef", *b"fedcba9876543210"].into_iter(),
1239 )?;
1240 let run_array: ArrowArrayRef = Arc::new(RunArray::<Int32Type>::try_new(
1241 &Int32Array::from(vec![2i32, 5]),
1242 &values,
1243 )?);
1244
1245 let vortex_array = session.from_arrow_array(run_array, &field)?;
1246 assert_eq!(vortex_array.len(), 5);
1247 assert_eq!(vortex_array.dtype(), &dtype);
1249
1250 let exported = session.execute_arrow(vortex_array, Some(&field), &mut ctx)?;
1252 assert_eq!(exported.len(), 5);
1253 let ree = exported
1254 .as_any()
1255 .downcast_ref::<RunArray<Int32Type>>()
1256 .ok_or_else(|| {
1257 vortex_err!(
1258 "expected an Int32 run-end array, got {}",
1259 exported.data_type()
1260 )
1261 })?;
1262 let values = ree.values().as_fixed_size_binary();
1263 assert_eq!(values.value(0), b"0123456789abcdef");
1264 assert_eq!(values.value(1), b"fedcba9876543210");
1265 Ok(())
1266 }
1267
1268 #[test]
1272 fn dictionary_recurses_into_nested_extension_values() -> VortexResult<()> {
1273 let session = ArrowSession::default();
1274
1275 let mut elem = Field::new("item", DataType::FixedSizeBinary(16), false);
1276 elem.try_with_extension_type(ArrowUuid)?;
1277
1278 let uuids = FixedSizeBinaryArray::try_from_iter(
1279 [*b"0123456789abcdef", *b"fedcba9876543210"].into_iter(),
1280 )?;
1281 let values = ArrowListArray::try_new(
1282 Arc::new(elem),
1283 OffsetBuffer::new(vec![0, 1, 2].into()),
1284 Arc::new(uuids),
1285 None,
1286 )?;
1287 let dict: ArrowArrayRef = Arc::new(DictionaryArray::<Int32Type>::try_new(
1288 Int32Array::from(vec![0, 1, 0]),
1289 Arc::new(values),
1290 )?);
1291 let field = Field::new("ids", dict.data_type().clone(), false);
1292
1293 let dtype = session.from_arrow_field(&field)?;
1294 let DType::List(elem_dt, _) = &dtype else {
1295 panic!("expected a List dtype, got {dtype}");
1296 };
1297 assert!(
1298 elem_dt.as_extension().is::<Uuid>(),
1299 "expected a Uuid extension element, got {elem_dt}"
1300 );
1301
1302 let array = session.from_arrow_array(dict, &field)?;
1303 assert_eq!(array.len(), 3);
1304 assert_eq!(array.dtype(), &dtype);
1306
1307 let mut ctx = array_session().create_execution_ctx();
1310 let exported = session.execute_arrow(array, Some(&field), &mut ctx)?;
1311 assert_eq!(exported.data_type(), field.data_type());
1312 let values = exported.as_any_dictionary().values().as_list::<i32>();
1313 let uuids = values.values().as_fixed_size_binary();
1314 assert_eq!(uuids.value(0), b"0123456789abcdef");
1315 assert_eq!(uuids.value(1), b"fedcba9876543210");
1316 Ok(())
1317 }
1318
1319 #[rstest]
1323 #[case(true)]
1324 #[case(false)]
1325 fn from_arrow_array_by_nullability(#[case] nullable: bool) -> VortexResult<()> {
1326 let session = ArrowSession::default();
1327
1328 let mut uuid_field = Field::new("id", DataType::FixedSizeBinary(16), false);
1329 uuid_field.try_with_extension_type(ArrowUuid)?;
1330 let uuids: ArrowArrayRef = Arc::new(FixedSizeBinaryArray::try_from_iter(
1331 [*b"0123456789abcdef", *b"fedcba9876543210"].into_iter(),
1332 )?);
1333 let arrow_struct: ArrowArrayRef = Arc::new(ArrowStructArray::try_new(
1334 Fields::from(vec![uuid_field]),
1335 vec![uuids],
1336 None,
1337 )?);
1338
1339 let array = session.from_arrow_array(ArrowArrayRef::clone(&arrow_struct), nullable)?;
1340 assert_eq!(array.dtype().nullability(), nullable.into());
1341 let DType::Struct(fields, _) = array.dtype() else {
1342 panic!("expected a Struct dtype, got {}", array.dtype());
1343 };
1344 assert!(
1345 fields
1346 .field_by_index(0)
1347 .vortex_expect("struct dtype has one field")
1348 .as_extension()
1349 .is::<Uuid>(),
1350 "expected the nested Uuid extension to survive, got {}",
1351 array.dtype()
1352 );
1353
1354 let by_nullability = session.from_arrow_array(arrow_struct, Nullability::from(nullable))?;
1356 assert_eq!(by_nullability.dtype(), array.dtype());
1357 Ok(())
1358 }
1359
1360 #[test]
1363 fn dictionary_imports_as_dict_encoding() -> VortexResult<()> {
1364 let session = ArrowSession::default();
1365 let dict: ArrowArrayRef = Arc::new(DictionaryArray::<Int32Type>::try_new(
1366 Int32Array::from(vec![0, 1, 0, 1]),
1367 Arc::new(StringArray::from(vec!["a", "b"])),
1368 )?);
1369 let field = Field::new("s", dict.data_type().clone(), false);
1370
1371 let dtype = session.from_arrow_field(&field)?;
1372 assert_eq!(dtype, DType::Utf8(Nullability::NonNullable));
1373
1374 let array = session.from_arrow_array(dict, &field)?;
1375 assert!(array.is::<Dict>(), "expected a Dict encoding, got {array}");
1376 assert_eq!(array.dtype(), &dtype);
1377 assert_eq!(array.len(), 4);
1378 Ok(())
1379 }
1380}