Skip to main content

vortex_arrow/
session.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4//! Plugin layer for moving Arrow extension types in and out of Vortex.
5//!
6//! Vortex's canonical Arrow conversion (see [`crate::dtype`] and the executor in
7//! [`crate::executor`]) handles every non-extension Arrow type and the builtin temporal
8//! extensions. The plugins registered here cover the remaining case: **Arrow extension types**.
9//!
10//! * An [`ArrowExportVTable`] is dispatched purely by the **target Arrow extension Id** —
11//!   the plugin is selected when the caller asks for an Arrow [`Field`] carrying matching
12//!   `ARROW:extension:name` metadata. The Vortex source dtype/encoding is irrelevant to
13//!   dispatch.
14//! * An [`ArrowImportVTable`] is dispatched by the **source Arrow extension name** carried
15//!   on the incoming [`Field`]. The plugin is responsible for both preserving extension
16//!   identity and re-encoding storage if needed (e.g. Arrow `FixedSizeBinary[16]` for UUID
17//!   becomes Vortex `FixedSizeList<u8; 16>`).
18//!
19//! Multiple plugins may register against the same key. They are tried in registration order;
20//! each may return [`ArrowExport::Unsupported`] / [`ArrowImport::Unsupported`] to defer to
21//! the next.
22
23use 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 tracing::debug;
44use tracing::trace;
45use vortex_array::ArrayRef;
46use vortex_array::ExecutionCtx;
47use vortex_array::IntoArray;
48use vortex_array::arrays::DictArray;
49use vortex_array::arrays::FixedSizeListArray;
50use vortex_array::arrays::ListArray;
51use vortex_array::arrays::ListViewArray;
52use vortex_array::arrays::StructArray;
53use vortex_array::dtype::DType;
54use vortex_array::dtype::FieldName;
55use vortex_array::dtype::FieldNames;
56use vortex_array::dtype::NativePType;
57use vortex_array::dtype::Nullability;
58use vortex_array::dtype::StructFields;
59use vortex_array::dtype::extension::ExtId;
60use vortex_array::extension::datetime::AnyTemporal;
61use vortex_array::validity::Validity;
62use vortex_error::VortexResult;
63use vortex_error::vortex_bail;
64use vortex_error::vortex_ensure;
65use vortex_error::vortex_err;
66use vortex_session::ArcSwapMap;
67use vortex_session::SessionExt;
68use vortex_session::SessionGuard;
69use vortex_session::SessionVar;
70use vortex_session::registry::Id;
71
72use crate::IntoVortexArray;
73use crate::convert::from_arrow_dyn;
74use crate::convert::map_from_arrow_parts;
75use crate::convert::nulls;
76use crate::convert::remove_nulls;
77use crate::dtype::from_arrow_data_type;
78use crate::dtype::to_data_type_naive;
79use crate::executor::execute_arrow_naive;
80use crate::run_end_import::run_end_from_arrow;
81
82/// Outcome of a successful call to [`ArrowExportVTable::execute_arrow`].
83///
84/// Plugins that don't handle the supplied array return [`Unsupported`][Self::Unsupported]
85/// with ownership of the input so the session can probe the next plugin or fall back to the
86/// canonical path. Errors are propagated through [`VortexResult`].
87pub enum ArrowExport {
88    /// The plugin does not handle this input; the session may try another plugin.
89    Unsupported(ArrayRef),
90    /// A successful export.
91    Exported(ArrowArrayRef),
92}
93
94/// Outcome of a successful call to [`ArrowImportVTable::from_arrow_array`].
95///
96/// Plugins that don't handle the supplied array return [`Unsupported`][Self::Unsupported]
97/// with ownership of the input so the session can probe the next plugin or fall back to the
98/// canonical path. Errors are propagated through [`VortexResult`].
99pub enum ArrowImport {
100    /// The plugin does not handle this input; the session may try another plugin.
101    Unsupported(ArrowArrayRef),
102    /// A successful import.
103    Imported(ArrayRef),
104}
105
106/// The Arrow type description of an array being imported by [`ArrowSession::from_arrow_array`].
107///
108/// Callers holding an Arrow [`Field`] (or [`FieldRef`]) should pass it: its
109/// `ARROW:extension:name` metadata is what dispatches the array to a registered
110/// [`ArrowImportVTable`].
111///
112/// An Arrow array can carry a validity (null) buffer regardless of whether its schema declares
113/// the field nullable, so when no [`Field`] is in hand the caller passes the desired nullability
114/// instead, as a [`bool`] or a [`Nullability`]. An anonymous field is then synthesized from the
115/// array's own data type, which means no extension plugin is dispatched for the array itself;
116/// fields nested inside container data types still carry their metadata and are routed through
117/// their importers.
118pub trait IntoArrowField<'a> {
119    /// Resolve to the Arrow [`Field`] describing an array of `data_type`.
120    fn into_arrow_field(self, data_type: &DataType) -> Cow<'a, Field>;
121}
122
123impl<'a> IntoArrowField<'a> for &'a Field {
124    fn into_arrow_field(self, _data_type: &DataType) -> Cow<'a, Field> {
125        Cow::Borrowed(self)
126    }
127}
128
129impl<'a> IntoArrowField<'a> for &'a FieldRef {
130    fn into_arrow_field(self, _data_type: &DataType) -> Cow<'a, Field> {
131        Cow::Borrowed(self.as_ref())
132    }
133}
134
135impl<'a> IntoArrowField<'a> for bool {
136    fn into_arrow_field(self, data_type: &DataType) -> Cow<'a, Field> {
137        Cow::Owned(Field::new("", data_type.clone(), self))
138    }
139}
140
141impl<'a> IntoArrowField<'a> for Nullability {
142    fn into_arrow_field(self, data_type: &DataType) -> Cow<'a, Field> {
143        self.is_nullable().into_arrow_field(data_type)
144    }
145}
146
147/// Plugin layer for exporting a Vortex array to an Arrow extension type.
148///
149/// This is purely an implementation trait, its methods should not be called directly. Instead,
150/// use the methods on [`ArrowSession`].
151pub trait ArrowExportVTable: 'static + Send + Sync + Debug {
152    /// The Arrow extension ID this plugin produces.
153    fn arrow_ext_id(&self) -> Id;
154
155    /// The Vortex array or extension ID this plugin maps from. Used only for inference by
156    /// [`ArrowSession::to_arrow_field`] / [`ArrowSession::to_arrow_schema`]; never as a
157    /// dispatch key for [`execute_arrow`][Self::execute_arrow].
158    fn vortex_id(&self) -> Id;
159
160    /// Build the Arrow [`Field`] this plugin produces for the given Vortex extension
161    /// `dtype`. Used during schema inference.
162    fn to_arrow_field(
163        &self,
164        name: &str,
165        dtype: &DType,
166        session: &ArrowSession,
167    ) -> VortexResult<Option<Field>>;
168
169    /// Convert a Vortex array into an Arrow array shaped to `target`.
170    ///
171    /// Returns ownership of `array` via [`ArrowExport::Unsupported`] when the plugin cannot
172    /// handle the input.
173    fn execute_arrow(
174        &self,
175        array: ArrayRef,
176        target: &Field,
177        ctx: &mut ExecutionCtx,
178    ) -> VortexResult<ArrowExport>;
179}
180
181/// Plugin layer for importing an Arrow extension-typed array into a Vortex array.
182///
183/// Plugins are dispatched by `arrow_ext_id`.
184///
185/// This is purely an implementation trait, its methods should not be called directly. Instead,
186/// use the methods on [`ArrowSession`].
187pub trait ArrowImportVTable: 'static + Send + Sync + Debug {
188    /// The Arrow extension name this plugin handles.
189    fn arrow_ext_id(&self) -> Id;
190
191    /// Build the Vortex [`DType`] that corresponds to `field` (which carries this plugin's
192    /// Arrow extension metadata).
193    ///
194    /// `session` is provided so plugins can resolve nested or storage fields through the
195    /// plugin-aware conversion (e.g. [`ArrowSession::from_arrow_datatype`]) instead of the
196    /// naive Arrow → Vortex mapping.
197    #[allow(clippy::wrong_self_convention)]
198    fn from_arrow_field(
199        &self,
200        field: &Field,
201        session: &ArrowSession,
202    ) -> VortexResult<Option<DType>>;
203
204    /// Convert an Arrow array into a Vortex array of `dtype`.
205    ///
206    /// Returns ownership of `array` via [`ArrowImport::Unsupported`] when the plugin cannot
207    /// handle the input.
208    ///
209    /// `session` is provided so plugins can convert storage or nested arrays through the
210    /// session (e.g. [`ArrowSession::from_arrow_array`], passing a nullability rather than a
211    /// [`Field`] so the plugin is not dispatched again) instead of the deprecated
212    /// `FromArrowArray` trait.
213    #[allow(clippy::wrong_self_convention)]
214    fn from_arrow_array(
215        &self,
216        array: ArrowArrayRef,
217        field: &Field,
218        dtype: &DType,
219        session: &ArrowSession,
220    ) -> VortexResult<ArrowImport>;
221}
222
223pub type ArrowExportVTableRef = Arc<dyn ArrowExportVTable>;
224pub type ArrowImportVTableRef = Arc<dyn ArrowImportVTable>;
225
226/// Registry of Arrow exporters, keyed by target Arrow extension [`Id`].
227type ArrowExporterRegistry = ArcSwapMap<Id, Arc<[ArrowExportVTableRef]>>;
228/// Registry of Arrow exporters, keyed by source Vortex extension [`ExtId`].
229type VortexExporterRegistry = ArcSwapMap<ExtId, Arc<[ArrowExportVTableRef]>>;
230/// Registry of Arrow importers, keyed by source Arrow extension [`Id`].
231type ArrowImporterRegistry = ArcSwapMap<Id, Arc<[ArrowImportVTableRef]>>;
232
233/// Session-scoped registry of Arrow extension plugins.
234///
235/// Exporters are stored in two indices: one keyed by Arrow extension Id (used for
236/// `execute_arrow` dispatch) and one keyed by Vortex extension Id (used **only** by
237/// `to_arrow_field` / `to_arrow_schema` inference, when callers need to translate a Vortex
238/// extension `DType` into an Arrow `Field` with no target schema in hand). Importers are
239/// keyed by Arrow extension name. Extension plugins are registered by the crate that owns the
240/// extension type (for example `vortex-uuid`); temporal extensions are handled by the canonical
241/// Arrow ↔ Vortex path and do not need plugins.
242#[derive(Clone, Debug, Default)]
243pub struct ArrowSession {
244    exporters: ArrowExporterRegistry,
245    exporters_by_vortex: VortexExporterRegistry,
246    importers: ArrowImporterRegistry,
247}
248
249impl ArrowSession {
250    /// Register an [`ArrowExportVTable`] under its target Arrow extension Id (for dispatch)
251    /// and its source Vortex extension Id (for schema inference).
252    pub fn register_exporter(&self, exporter: ArrowExportVTableRef) {
253        self.exporters.push(
254            exporter.arrow_ext_id(),
255            ArrowExportVTableRef::clone(&exporter),
256        );
257        self.exporters_by_vortex
258            .push(exporter.vortex_id(), exporter);
259    }
260
261    /// Register an [`ArrowImportVTable`] under its source Arrow extension name.
262    pub fn register_importer(&self, importer: ArrowImportVTableRef) {
263        self.importers.push(importer.arrow_ext_id(), importer);
264    }
265
266    fn exporters(&self, id: &Id) -> Arc<[ArrowExportVTableRef]> {
267        self.exporters.get(id).unwrap_or_else(|| Arc::from([]))
268    }
269
270    fn exporters_by_vortex(&self, id: &Id) -> Arc<[ArrowExportVTableRef]> {
271        self.exporters_by_vortex
272            .get(id)
273            .unwrap_or_else(|| Arc::from([]))
274    }
275
276    fn importers(&self, id: &Id) -> Arc<[ArrowImportVTableRef]> {
277        self.importers.get(id).unwrap_or_else(|| Arc::from([]))
278    }
279
280    /// Build the Arrow [`Field`] for a Vortex [`DType`].
281    ///
282    /// For [`DType::Extension`]s, plugins registered against the extension's `Id`
283    /// are tried in registration order; the first plugin to return `Some(field)` wins.
284    pub fn to_arrow_field(&self, name: &str, dtype: &DType) -> VortexResult<Field> {
285        // Handle the structural encodings, which may have recursive types
286        match dtype {
287            DType::List(elem_dtype, nullability) => {
288                let elem_field = self.to_arrow_field(Field::LIST_FIELD_DEFAULT_NAME, elem_dtype)?;
289                Ok(Field::new_list(name, elem_field, nullability.is_nullable()))
290            }
291            DType::FixedSizeList(elem_dtype, elem_size, nullability) => {
292                let elem_field = self.to_arrow_field(Field::LIST_FIELD_DEFAULT_NAME, elem_dtype)?;
293                Ok(Field::new_fixed_size_list(
294                    name,
295                    elem_field,
296                    (*elem_size).try_into()?,
297                    nullability.is_nullable(),
298                ))
299            }
300            DType::Map(map_dtype, nullability) => {
301                let key = self.to_arrow_field("key", &map_dtype.key_dtype())?;
302                let value = self.to_arrow_field("value", &map_dtype.value_dtype())?;
303                let entries = Field::new_struct("entries", Fields::from(vec![key, value]), false);
304                Ok(Field::new(
305                    name,
306                    DataType::Map(Arc::new(entries), map_dtype.keys_sorted()),
307                    nullability.is_nullable(),
308                ))
309            }
310            DType::Struct(fields, nullability) => {
311                let arrow_fields = Fields::from_iter(
312                    fields
313                        .fields()
314                        .zip(fields.names().iter())
315                        .map(|(field, name)| self.to_arrow_field(name.as_ref(), &field))
316                        .collect::<VortexResult<Vec<_>>>()?,
317                );
318                Ok(Field::new_struct(
319                    name,
320                    arrow_fields,
321                    nullability.is_nullable(),
322                ))
323            }
324            DType::Extension(ext) if !ext.is::<AnyTemporal>() => {
325                for plugin in self.exporters_by_vortex(&ext.id()).iter() {
326                    if let Some(field) =
327                        plugin.to_arrow_field(name, &DType::Extension(ext.clone()), self)?
328                    {
329                        return Ok(field);
330                    }
331                }
332                vortex_bail!("extension type cannot be converted to Arrow without a plugin: {ext}");
333            }
334            DType::Variant(_) => {
335                // TODO(Adam): This currently encodes information about parquet-variant
336                // at this level. Variant's complexity with being an essentially logical type
337                // with multiple physical layout complicates handling this correctly.
338                Ok(Field::new(
339                    name,
340                    DataType::Struct(
341                        vec![
342                            Field::new("metadata", DataType::BinaryView, dtype.is_nullable()),
343                            Field::new("value", DataType::BinaryView, dtype.is_nullable()),
344                        ]
345                        .into(),
346                    ),
347                    dtype.is_nullable(),
348                )
349                .with_metadata(
350                    [(
351                        EXTENSION_TYPE_NAME_KEY.to_string(),
352                        "arrow.parquet.variant".to_string(),
353                    )]
354                    .into(),
355                ))
356            }
357            _ => Ok(Field::new(
358                name,
359                to_data_type_naive(dtype)?,
360                dtype.is_nullable(),
361            )),
362        }
363    }
364
365    /// Build the Arrow [`Schema`] for a Vortex top-level [`DType::Struct`], dispatching
366    /// extension fields through registered export plugins for inference. Nested
367    /// extensions are preserved via [`Self::to_arrow_field`].
368    pub fn to_arrow_schema(&self, dtype: &DType) -> VortexResult<Schema> {
369        let DType::Struct(struct_dtype, _) = dtype else {
370            vortex_bail!("to_arrow_schema requires a top-level struct dtype, got {dtype}");
371        };
372        let mut fields = Vec::with_capacity(struct_dtype.names().len());
373        for (name, field_dtype) in struct_dtype.names().iter().zip(struct_dtype.fields()) {
374            fields.push(self.to_arrow_field(name.as_ref(), &field_dtype)?);
375        }
376        Ok(Schema::new(fields))
377    }
378
379    /// Returns the Arrow [`DataType`] that best corresponds to the given Vortex [`DType`],
380    /// dispatching [`DType::Extension`]s through registered export plugins.
381    ///
382    /// Note that a bare [`DataType`] cannot carry `ARROW:extension:name` metadata; use
383    /// [`Self::to_arrow_field`] when extension identity must survive the roundtrip.
384    pub fn to_arrow_datatype(&self, dtype: &DType) -> VortexResult<DataType> {
385        Ok(self.to_arrow_field("", dtype)?.data_type().clone())
386    }
387
388    /// Build the Vortex [`DType`] for an Arrow [`Field`].
389    ///
390    /// Plugins registered against the field's Arrow extension name are tried in
391    /// registration order; the first plugin to return `Some(dtype)` wins. If none
392    /// match (or all return `None`), the builtin `arrow.parquet.variant` extension maps
393    /// to [`DType::Variant`], and any other field converts through
394    /// [`Self::from_arrow_datatype`] so extension metadata on nested element/struct
395    /// fields is preserved.
396    #[expect(clippy::disallowed_methods, reason = "interning a dynamic id")]
397    pub fn from_arrow_field(&self, field: &Field) -> VortexResult<DType> {
398        if let Some(name) = field.metadata().get(EXTENSION_TYPE_NAME_KEY) {
399            for plugin in self.importers(&Id::new(name)).iter() {
400                if let Some(dtype) = plugin.from_arrow_field(field, self)? {
401                    return Ok(dtype);
402                }
403            }
404            // Parquet Variant is understood even without a registered importer plugin.
405            if name == "arrow.parquet.variant" {
406                return Ok(DType::Variant(field.is_nullable().into()));
407            }
408        }
409        self.from_arrow_datatype(field.data_type(), field.is_nullable().into())
410    }
411
412    /// Build the Vortex [`DType`] for an Arrow [`DataType`].
413    ///
414    /// Recurses into container types ([`DataType::List`] family, [`DataType::FixedSizeList`],
415    /// [`DataType::Struct`], [`DataType::Map`], [`DataType::RunEndEncoded`]) via
416    /// [`Self::from_arrow_field`] so extension metadata on nested fields dispatches through
417    /// registered import plugins. Leaf types use the canonical Arrow → Vortex mapping.
418    ///
419    /// [`DataType::Dictionary`] is a partial exception: Arrow models its values as a bare
420    /// [`DataType`] rather than a [`Field`], so the values themselves cannot carry an extension
421    /// name. Fields nested *inside* that data type still do, and are dispatched normally.
422    pub fn from_arrow_datatype(
423        &self,
424        data_type: &DataType,
425        nullability: Nullability,
426    ) -> VortexResult<DType> {
427        Ok(match data_type {
428            DataType::List(elem)
429            | DataType::LargeList(elem)
430            | DataType::ListView(elem)
431            | DataType::LargeListView(elem) => {
432                DType::List(Arc::new(self.from_arrow_field(elem.as_ref())?), nullability)
433            }
434            DataType::FixedSizeList(elem, size) => DType::FixedSizeList(
435                Arc::new(self.from_arrow_field(elem.as_ref())?),
436                *size as u32,
437                nullability,
438            ),
439            DataType::Map(entries, keys_sorted) => {
440                vortex_ensure!(
441                    !entries.is_nullable(),
442                    "Arrow map entries field must be non-nullable"
443                );
444                let DataType::Struct(fields) = entries.data_type() else {
445                    vortex_bail!(
446                        "Arrow map entries field must have Struct type, got {:?}",
447                        entries.data_type()
448                    );
449                };
450                vortex_ensure!(
451                    fields.len() == 2,
452                    "Arrow map entries struct must contain exactly two fields"
453                );
454                vortex_ensure!(
455                    !fields[0].is_nullable(),
456                    "Arrow map key field must be non-nullable"
457                );
458                DType::map(
459                    self.from_arrow_field(fields[0].as_ref())?,
460                    self.from_arrow_field(fields[1].as_ref())?,
461                    *keys_sorted,
462                    nullability,
463                )?
464            }
465            DataType::Struct(fields) => DType::Struct(self.from_arrow_fields(fields)?, nullability),
466            DataType::Dictionary(_, value_type) => {
467                self.from_arrow_datatype(value_type.as_ref(), nullability)?
468            }
469            DataType::RunEndEncoded(_, value_field) => {
470                self.from_arrow_field(&run_end_values_field(value_field, nullability))?
471            }
472            _ => from_arrow_data_type(data_type, nullability)?,
473        })
474    }
475
476    /// Build Vortex [`StructFields`] for Arrow [`Fields`], dispatching each field through
477    /// [`Self::from_arrow_field`].
478    pub fn from_arrow_fields(&self, fields: &Fields) -> VortexResult<StructFields> {
479        fields
480            .iter()
481            .map(|f| {
482                self.from_arrow_field(f)
483                    .map(|dt| (FieldName::from(f.name().as_str()), dt))
484            })
485            .collect::<VortexResult<StructFields>>()
486    }
487
488    /// Build the Vortex [`DType`] for an Arrow [`Schema`], dispatching extension fields
489    /// through registered import plugins. The result is a top-level non-nullable struct
490    /// matching the schema's fields.
491    pub fn from_arrow_schema(&self, schema: &Schema) -> VortexResult<DType> {
492        Ok(DType::Struct(
493            self.from_arrow_fields(schema.fields())?,
494            Nullability::NonNullable,
495        ))
496    }
497
498    /// Decode an Arrow [`RecordBatch`] into a Vortex struct array, dispatching each
499    /// extension column through its registered import plugin.
500    ///
501    /// `schema` is the authoritative Arrow schema used for dispatch — the columns are
502    /// consumed positionally. Pass an external schema (rather than relying on
503    /// `batch.schema()`) when upstream DataFusion plumbing may have stripped Field-level
504    /// extension metadata from the runtime RecordBatch.
505    pub fn from_arrow_record_batch(
506        &self,
507        batch: RecordBatch,
508        schema: &Schema,
509    ) -> VortexResult<ArrayRef> {
510        vortex_ensure!(
511            batch.num_columns() == schema.fields().len(),
512            "RecordBatch has {} columns but schema has {} fields",
513            batch.num_columns(),
514            schema.fields().len()
515        );
516        let length = batch.num_rows();
517        let names = FieldNames::from_iter(
518            schema
519                .fields()
520                .iter()
521                .map(|f| FieldName::from(f.name().as_str())),
522        );
523        let mut columns = Vec::with_capacity(schema.fields().len());
524        for (col, field) in batch.columns().iter().zip(schema.fields().iter()) {
525            columns.push(self.from_arrow_array_inner(ArrowArrayRef::clone(col), field)?);
526        }
527        Ok(StructArray::try_new(names, columns, length, Validity::NonNullable)?.into_array())
528    }
529
530    /// Execute a Vortex array into an Arrow array.
531    ///
532    /// If `target` carries an `ARROW:extension:name`, the plugin registry is probed for one that
533    /// can support executing to the target extension type.
534    ///
535    /// With `target = None` the fallback path picks the array's preferred Arrow physical type
536    /// and executes directly into that, ignoring extension types.
537    #[expect(clippy::disallowed_methods, reason = "interning a dynamic id")]
538    pub fn execute_arrow(
539        &self,
540        array: ArrayRef,
541        target: Option<&Field>,
542        ctx: &mut ExecutionCtx,
543    ) -> VortexResult<ArrowArrayRef> {
544        // NOTE(aduffy): this looks strange, but we do this to keep target_field as &Field so
545        //  we can avoid cloning target when it is provided. It contains a HashMap internally that
546        //  can be expensive to copy.
547        let arrow_field;
548        let target_field = match target {
549            Some(field) => field,
550            None => {
551                let session = ctx.session().clone();
552                arrow_field = session.arrow().to_arrow_field("", array.dtype())?;
553                &arrow_field
554            }
555        };
556
557        if let Some(arrow_ext_name) = target_field.metadata().get(EXTENSION_TYPE_NAME_KEY) {
558            // There can be multiple plugins that report support for a particular extension type.
559            // We try them in order until one of them reports a successful conversion.
560            let len = array.len();
561            let mut current = array;
562
563            for plugin in self.exporters(&Id::new(arrow_ext_name)).iter() {
564                trace!(
565                    plugin = ?plugin,
566                    extension_name = arrow_ext_name,
567                    "probing plugin for converting Arrow array"
568                );
569
570                match plugin.execute_arrow(current, target_field, ctx)? {
571                    ArrowExport::Exported(arrow) => {
572                        vortex_ensure!(
573                            arrow.len() == len,
574                            "Arrow array length does not match Vortex array length after conversion to {:?}",
575                            arrow
576                        );
577                        return Ok(arrow);
578                    }
579                    ArrowExport::Unsupported(array) => current = array,
580                }
581            }
582
583            debug!(
584                extension_id = arrow_ext_name,
585                data_type = ?target_field.data_type(),
586                "unsupported Arrow extension type encountered, falling back to naive execution"
587            );
588
589            return execute_arrow_naive(current, Some(target_field.data_type()), ctx);
590        }
591
592        execute_arrow_naive(array, target.map(|field| field.data_type()), ctx)
593    }
594
595    /// Decode an Arrow array into a Vortex array.
596    ///
597    /// `field` describes the Arrow type the array is imported from: pass an Arrow [`Field`] when
598    /// one is available, or just the desired nullability (`true` / `false` /
599    /// [`Nullability`]) to synthesize an anonymous field from the array's own data type. See
600    /// [`IntoArrowField`] for the trade-off between the two.
601    ///
602    /// Routes through the registered import plugin if `field` carries an Arrow extension
603    /// name we recognize, probing each plugin in registration order until one handles the
604    /// input or all return [`ArrowImport::Unsupported`]. Otherwise recurses into container
605    /// arrays ([`arrow_array::StructArray`], [`arrow_array::GenericListArray`],
606    /// [`arrow_array::FixedSizeListArray`], [`arrow_array::GenericListViewArray`]) so
607    /// extension fields nested inside containers reach their importers; leaf types fall
608    /// through to the canonical Arrow → Vortex array conversion.
609    ///
610    /// # Errors
611    ///
612    /// Returns an error if the field (or requested nullability) is non-nullable but the array
613    /// physically contains nulls, or if the Arrow data type is unsupported.
614    pub fn from_arrow_array<'a>(
615        &self,
616        array: ArrowArrayRef,
617        field: impl IntoArrowField<'a>,
618    ) -> VortexResult<ArrayRef> {
619        let field = field.into_arrow_field(array.data_type());
620        self.from_arrow_array_inner(array, field.as_ref())
621    }
622
623    /// [`Self::from_arrow_array`] with the Arrow [`Field`] already resolved: probe the import
624    /// plugins registered for the field's extension name, then fall back to the canonical
625    /// conversion. Also the recursion point for nested fields, which already have a [`Field`].
626    #[allow(clippy::wrong_self_convention)]
627    fn from_arrow_array_inner(
628        &self,
629        array: ArrowArrayRef,
630        field: &Field,
631    ) -> VortexResult<ArrayRef> {
632        if let Some(extension_name) = field.metadata().get(EXTENSION_TYPE_NAME_KEY) {
633            #[expect(clippy::disallowed_methods, reason = "interning a dynamic id")]
634            let importers = self.importers(&Id::new(extension_name));
635            if !importers.is_empty() {
636                let dtype = self.from_arrow_field(field)?;
637                let mut current = array;
638                for plugin in importers.iter() {
639                    match plugin.from_arrow_array(current, field, &dtype, self)? {
640                        ArrowImport::Imported(arr) => return Ok(arr),
641                        ArrowImport::Unsupported(arr) => current = arr,
642                    }
643                }
644                return self.from_arrow_array_canonical(current.as_ref(), field);
645            }
646        }
647        self.from_arrow_array_canonical(array.as_ref(), field)
648    }
649
650    /// Recurse into Arrow container arrays so nested fields with extension metadata reach
651    /// their importers, falling through to the canonical conversion for leaf types.
652    #[allow(clippy::wrong_self_convention)]
653    fn from_arrow_array_canonical(
654        &self,
655        array: &dyn ArrowArray,
656        field: &Field,
657    ) -> VortexResult<ArrayRef> {
658        use arrow_array::cast::AsArray;
659
660        match field.data_type() {
661            DataType::Struct(fields) => {
662                let arrow_struct = array.as_struct();
663                let names = FieldNames::from_iter(
664                    fields.iter().map(|f| FieldName::from(f.name().as_str())),
665                );
666                let columns = arrow_struct
667                    .columns()
668                    .iter()
669                    .zip(fields.iter())
670                    .map(|(col, child_field)| {
671                        // Arrow pushes nulls into non-nullable fields; strip before recursing
672                        // so Vortex's stricter validity invariants are upheld.
673                        let inner = if col.null_count() > 0 && !child_field.is_nullable() {
674                            make_array(remove_nulls(col.to_data())?)
675                        } else {
676                            ArrowArrayRef::clone(col)
677                        };
678                        self.from_arrow_array_inner(inner, child_field.as_ref())
679                    })
680                    .collect::<VortexResult<Vec<_>>>()?;
681                let validity = nulls(arrow_struct.nulls(), field.is_nullable())?;
682                Ok(
683                    StructArray::try_new(names, columns, arrow_struct.len(), validity)?
684                        .into_array(),
685                )
686            }
687            DataType::List(elem_field) => {
688                let list = array.as_list::<i32>();
689                let elements = self
690                    .from_arrow_array(ArrowArrayRef::clone(list.values()), elem_field.as_ref())?;
691                let offsets = list.offsets().clone().into_array();
692                let validity = nulls(list.nulls(), field.is_nullable())?;
693                Ok(ListArray::try_new(elements, offsets, validity)?.into_array())
694            }
695            DataType::LargeList(elem_field) => {
696                let list = array.as_list::<i64>();
697                let elements = self
698                    .from_arrow_array(ArrowArrayRef::clone(list.values()), elem_field.as_ref())?;
699                let offsets = list.offsets().clone().into_array();
700                let validity = nulls(list.nulls(), field.is_nullable())?;
701                Ok(ListArray::try_new(elements, offsets, validity)?.into_array())
702            }
703            DataType::FixedSizeList(elem_field, list_size) => {
704                let fsl = array.as_fixed_size_list();
705                let elements = self.from_arrow_array_inner(
706                    ArrowArrayRef::clone(fsl.values()),
707                    elem_field.as_ref(),
708                )?;
709                let validity = nulls(fsl.nulls(), field.is_nullable())?;
710                Ok(
711                    FixedSizeListArray::try_new(elements, *list_size as u32, validity, fsl.len())?
712                        .into_array(),
713                )
714            }
715            DataType::ListView(elem_field) => {
716                let list = array.as_list_view::<i32>();
717                let elements = self
718                    .from_arrow_array(ArrowArrayRef::clone(list.values()), elem_field.as_ref())?;
719                let offsets = list.offsets().clone().into_array();
720                let sizes = list.sizes().clone().into_array();
721                let validity = nulls(list.nulls(), field.is_nullable())?;
722                Ok(ListViewArray::try_new(elements, offsets, sizes, validity)?.into_array())
723            }
724            DataType::LargeListView(elem_field) => {
725                let list = array.as_list_view::<i64>();
726                let elements = self
727                    .from_arrow_array(ArrowArrayRef::clone(list.values()), elem_field.as_ref())?;
728                let offsets = list.offsets().clone().into_array();
729                let sizes = list.sizes().clone().into_array();
730                let validity = nulls(list.nulls(), field.is_nullable())?;
731                Ok(ListViewArray::try_new(elements, offsets, sizes, validity)?.into_array())
732            }
733            DataType::Map(entries_field, keys_sorted) => {
734                let map = array.as_map();
735                let entries_array: ArrowArrayRef = Arc::new(map.entries().clone());
736                let entries = self.from_arrow_array_inner(entries_array, entries_field.as_ref())?;
737                map_from_arrow_parts(
738                    entries,
739                    map.offsets(),
740                    map.nulls(),
741                    *keys_sorted,
742                    field.is_nullable(),
743                )
744            }
745            DataType::RunEndEncoded(ends_field, values_field) => {
746                let values_field = run_end_values_field(values_field, field.is_nullable().into());
747                match ends_field.data_type() {
748                    DataType::Int16 => self.run_end_from_arrow::<Int16Type>(array, &values_field),
749                    DataType::Int32 => self.run_end_from_arrow::<Int32Type>(array, &values_field),
750                    DataType::Int64 => self.run_end_from_arrow::<Int64Type>(array, &values_field),
751                    ends_dt => vortex_bail!(
752                        "Arrow run-end array run ends must be Int16, Int32 or Int64, got {ends_dt}"
753                    ),
754                }
755            }
756            DataType::Dictionary(..) => {
757                let dict = array.as_any_dictionary();
758                // Arrow models dictionary values as a bare `DataType`, so there is no field
759                // metadata to carry an extension name for the values themselves. Fields *nested
760                // inside* that data type (list elements, struct fields, map entries) do keep
761                // their metadata, so importing the values by nullability alone still routes them
762                // back through the plugin-aware conversion.
763                let values = self
764                    .from_arrow_array(ArrowArrayRef::clone(dict.values()), field.is_nullable())?;
765                let codes = dict.keys();
766                let codes = from_arrow_dyn(codes, codes.is_nullable())?;
767                // SAFETY: arrow-rs enforces the dictionary invariants on construction, so the
768                // codes are in-bounds for the values.
769                Ok(unsafe { DictArray::new_unchecked(codes, values) }.into_array())
770            }
771            _ => from_arrow_dyn(array, field.is_nullable()),
772        }
773    }
774
775    /// Decode an Arrow run-end array, recursing into its values so extension metadata on the
776    /// values field reaches its importer.
777    #[allow(clippy::wrong_self_convention)]
778    fn run_end_from_arrow<R: RunEndIndexType>(
779        &self,
780        array: &dyn ArrowArray,
781        values_field: &Field,
782    ) -> VortexResult<ArrayRef>
783    where
784        R::Native: NativePType,
785    {
786        let run_array = array
787            .as_any()
788            .downcast_ref::<RunArray<R>>()
789            .ok_or_else(|| vortex_err!("expected an Arrow RunArray, got {}", array.data_type()))?;
790        let values =
791            self.from_arrow_array_inner(ArrowArrayRef::clone(run_array.values()), values_field)?;
792        run_end_from_arrow(run_array, values)
793    }
794}
795
796/// The values field of an Arrow [`DataType::RunEndEncoded`], re-stamped with the run-end array's
797/// own nullability.
798fn run_end_values_field(values_field: &FieldRef, nullability: Nullability) -> Field {
799    values_field
800        .as_ref()
801        .clone()
802        .with_nullable(nullability.into())
803}
804
805impl SessionVar for ArrowSession {
806    fn as_any(&self) -> &dyn Any {
807        self
808    }
809
810    fn as_any_mut(&mut self) -> &mut dyn Any {
811        self
812    }
813}
814
815/// Extension trait for accessing the [`ArrowSession`] on a Vortex session.
816pub trait ArrowSessionExt: SessionExt {
817    /// Get the Arrow session.
818    fn arrow(&self) -> SessionGuard<'_, ArrowSession>;
819}
820
821impl<S: SessionExt> ArrowSessionExt for S {
822    fn arrow(&self) -> SessionGuard<'_, ArrowSession> {
823        self.get::<ArrowSession>()
824    }
825}
826
827#[cfg(test)]
828mod tests {
829    use std::sync::Arc;
830
831    use arrow_array::DictionaryArray;
832    use arrow_array::Int32Array;
833    use arrow_array::StringArray;
834    use arrow_array::types::Int32Type;
835    use arrow_schema::DataType;
836    use arrow_schema::Field;
837    use vortex_array::arrays::Dict;
838    use vortex_array::dtype::DType;
839    use vortex_array::dtype::Nullability;
840    use vortex_error::VortexResult;
841
842    use super::*;
843
844    #[test]
845    fn from_arrow_fields_matches_schema_conversion() -> VortexResult<()> {
846        let session = ArrowSession::default();
847        let fields = Fields::from(vec![
848            Field::new("a", DataType::Int32, false),
849            Field::new("b", DataType::Utf8View, true),
850        ]);
851        let struct_fields = session.from_arrow_fields(&fields)?;
852        let schema_dtype = session.from_arrow_schema(&Schema::new(fields))?;
853        assert_eq!(
854            schema_dtype,
855            DType::Struct(struct_fields, Nullability::NonNullable)
856        );
857        Ok(())
858    }
859
860    #[test]
861    fn from_arrow_field_maps_variant_without_importer() -> VortexResult<()> {
862        let session = ArrowSession::default();
863        let storage = DataType::Struct(
864            vec![
865                Field::new("metadata", DataType::BinaryView, false),
866                Field::new("value", DataType::BinaryView, true),
867            ]
868            .into(),
869        );
870        let field = Field::new("v", storage, true).with_metadata(
871            [(
872                "ARROW:extension:name".to_string(),
873                "arrow.parquet.variant".to_string(),
874            )]
875            .into(),
876        );
877        assert_eq!(
878            session.from_arrow_field(&field)?,
879            DType::Variant(Nullability::Nullable)
880        );
881        Ok(())
882    }
883
884    /// A plain Arrow dictionary imports as a Vortex `Dict` array over the dictionary values,
885    /// matching the dtype the schema conversion reports.
886    #[test]
887    fn dictionary_imports_as_dict_encoding() -> VortexResult<()> {
888        let session = ArrowSession::default();
889        let dict: ArrowArrayRef = Arc::new(DictionaryArray::<Int32Type>::try_new(
890            Int32Array::from(vec![0, 1, 0, 1]),
891            Arc::new(StringArray::from(vec!["a", "b"])),
892        )?);
893        let field = Field::new("s", dict.data_type().clone(), false);
894
895        let dtype = session.from_arrow_field(&field)?;
896        assert_eq!(dtype, DType::Utf8(Nullability::NonNullable));
897
898        let array = session.from_arrow_array(dict, &field)?;
899        assert!(array.is::<Dict>(), "expected a Dict encoding, got {array}");
900        assert_eq!(array.dtype(), &dtype);
901        assert_eq!(array.len(), 4);
902        Ok(())
903    }
904}