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