pluggable registry for input/export arrow kernels (#7824)

## Summary

Adds a pluggable `ArrowSession` registry on `VortexSession` for
round-tripping Vortex extension types in and out of Arrow extension
types. Unblocks Arrow round-trip for `arrow.uuid` today, with
`arrow.parquet.variant`, GeoArrow, and tensor types as the next
consumers.

  Part of #7686.

  ## API changes

  The session exposes two trait-driven plugin slots:

- `ArrowExportVTable` — dispatched by **target Arrow extension name**
(`ARROW:extension:name`). Implementations turn a Vortex `ArrayRef` into
an Arrow `ArrayRef` shaped to the requested `Field`. Also provides
`to_arrow_field` for schema inference when only a Vortex `DType` is in
  hand.
- `ArrowImportVTable` — dispatched by **source Arrow extension name**
carried on the incoming `Field`. Implementations turn an Arrow
`ArrayRef` back into a Vortex `ArrayRef`, including any storage
re-encoding (e.g. `FixedSizeBinary[16]` → `FixedSizeList<u8; 16>` for
UUID).

Both traits return `Unsupported(input)` to defer to the next plugin or
to the canonical fallback, so multiple plugins can register against the
same key and probe in order.

  New session entry points (`vortex-array/src/arrow/session.rs`):

- `ArrowSession::to_arrow_field` / `to_arrow_schema` — Vortex `DType` →
Arrow `Field`/`Schema`, recursing into containers so nested extension
fields go through the registered plugin.
- `ArrowSession::from_arrow_field` / `from_arrow_schema` — inverse
direction, plugin-aware.
- `ArrowSession::from_arrow_record_batch` / `execute_record_batch` —
`RecordBatch` round-trip.
- `ArrowSessionExt` extension trait so any `SessionExt` can call
`session.arrow().…`.

The default session pre-registers the builtin UUID plugin
(`vortex-array/src/extension/uuid/arrow.rs`).

  ## What's *not* in the plugin layer

`Date`, `Time`, and `Timestamp` are Vortex builtin extensions that map
directly to native Arrow temporal types, so they continue to go through
the canonical executor (`vortex-array/src/arrow/executor/temporal.rs`)
rather than the plugin registry. The plugin layer is reserved for
**Arrow extension types** that the canonical path can't express.

  ## DataFusion wiring

`vortex-datafusion` now goes through the session for schema/array
conversion:

- `convert/schema.rs::calculate_physical_schema` uses
`ArrowSession::to_arrow_field` so extension metadata survives
projection.
- `persistent/format.rs` and `persistent/opener.rs` route schema
inference through the session.
- `persistent/sink.rs` uses `from_arrow_record_batch`, passing the
original schema separately from `RecordBatch::schema()` to preserve
`ARROW:extension:name` metadata that DataFusion strips at runtime.

  ## Tests

Two new end-to-end tests in `vortex-datafusion/src/persistent/tests.rs`:

- `arrow_uuid_extension_roundtrip` — write Arrow UUID column to a Vortex
file via the session, `SELECT *` it back, assert the field still carries
the `Uuid` extension type and the values match.
- `arrow_uuid_extension_roundtrip_nested_struct` — same flow with the
UUID nested in a top-level `Struct`, exercising recursive session-aware
schema inference.

---------

Signed-off-by: Andrew Duffy <andrew@a10y.dev>
Signed-off-by: Baris Palaska <barispalaska@gmail.com>
Co-authored-by: Baris Palaska <barispalaska@gmail.com>

Author: a10y

Cargo.lock

@@ -382,7 +382,7 @@ debug = "full"
 inherits = "release"
 
 [profile.bench]
-codegen-units = 16
+codegen-units = 1
 debug = "full"
 lto = false
 

Cargo.toml

@@ -22,6 +22,10 @@ use vortex_array::arrays::StructArray;
 use vortex_array::arrays::VariantArray;
 use vortex_array::arrays::list::ListArrayExt;
 use vortex_array::arrays::struct_::StructArrayExt;
+#[expect(
+    deprecated,
+    reason = "TODO(aduffy): figure out what to do with Parquet Variant"
+)]
 use vortex_array::arrow::ArrowArrayExecutor;
 use vortex_array::arrow::FromArrowArray;
 use vortex_array::arrow::to_arrow_null_buffer;
@@ -331,6 +335,10 @@ pub trait ParquetVariantArrayExt: TypedArrayRef<ParquetVariant> {
     }
 
     /// Converts this storage array to Arrow's canonical Parquet Variant extension storage.
+    #[expect(
+        deprecated,
+        reason = "TODO(aduffy): figure out what to do with Parquet Variant"
+    )]
     fn to_arrow(&self, ctx: &mut ExecutionCtx) -> VortexResult<ArrowVariantArray> {
         let metadata = self.metadata_array();
         let len = metadata.len();

encodings/parquet-variant/src/array.rs

@@ -104,6 +104,10 @@ fn to_parquet_variant_path(path: &VariantPath) -> VortexResult<PqVariantPath<'st
         .map(PqVariantPath::new)
 }
 
+#[expect(
+    deprecated,
+    reason = "TODO(aduffy): figure out what to do with Parquet Variant"
+)]
 fn to_arrow_as_type(dtype: Option<&DType>) -> VortexResult<Option<FieldRef>> {
     match dtype {
         Some(dtype) if !dtype.is_variant() => Ok(Some(Arc::new(Field::new(

encodings/parquet-variant/src/kernel.rs

@@ -10,7 +10,7 @@ use vortex_array::LEGACY_SESSION;
 use vortex_array::VortexSessionExecute;
 use vortex_array::arrays::BoolArray;
 use vortex_array::arrays::PrimitiveArray;
-use vortex_array::arrow::ArrowArrayExecutor;
+use vortex_array::arrow::ArrowSessionExt;
 use vortex_array::assert_arrays_eq;
 use vortex_array::assert_nth_scalar;
 use vortex_array::dtype::DType;
@@ -213,9 +213,14 @@ fn test_serde() -> VortexResult<()> {
         &ReadContext::new(context.to_ids()),
         &SESSION,
     )?;
-    let data_type = data.dtype().to_arrow_dtype()?;
-    let pco_arrow = pco.execute_arrow(Some(&data_type), &mut ctx)?;
-    let decoded_arrow = decoded.execute_arrow(Some(&data_type), &mut ctx)?;
+    let data_type = LEGACY_SESSION.arrow().to_arrow_field("", data.dtype())?;
+    let pco_arrow = LEGACY_SESSION
+        .arrow()
+        .execute_arrow(pco, Some(&data_type), &mut ctx)?;
+    let decoded_arrow =
+        LEGACY_SESSION
+            .arrow()
+            .execute_arrow(decoded, Some(&data_type), &mut ctx)?;
     assert!(pco_arrow == decoded_arrow);
     Ok(())
 }

encodings/pco/src/tests.rs

@@ -85,7 +85,7 @@ mod tests {
     use vortex_array::VortexSessionExecute as _;
     use vortex_array::arrays::PrimitiveArray;
     use vortex_array::arrays::primitive::PrimitiveArrayExt;
-    use vortex_array::arrow::ArrowArrayExecutor;
+    use vortex_array::arrow::ArrowSessionExt;
     use vortex_array::arrow::FromArrowArray;
     use vortex_array::assert_arrays_eq;
     use vortex_array::dtype::DType;
@@ -301,7 +301,10 @@ mod tests {
     }
 
     fn execute(array: ArrayRef, dt: &DataType) -> VortexResult<arrow_array::ArrayRef> {
-        array.execute_arrow(Some(dt), &mut SESSION.create_execution_ctx())
+        let field = Field::new("", dt.clone(), true);
+        SESSION
+            .arrow()
+            .execute_arrow(array, Some(&field), &mut SESSION.create_execution_ctx())
     }
 
     #[test]

encodings/runend/src/arrow.rs

@@ -599,7 +599,7 @@ mod test {
     use vortex_array::arrays::VarBinArray;
     use vortex_array::arrays::VarBinViewArray;
     use vortex_array::arrays::listview::ListViewArrayExt;
-    use vortex_array::arrow::ArrowArrayExecutor;
+    use vortex_array::arrow::ArrowSessionExt;
     use vortex_array::assert_arrays_eq;
     use vortex_array::dtype::DType;
     use vortex_array::dtype::DecimalDType;
@@ -845,23 +845,33 @@ mod test {
         let fill_scalar = Scalar::decimal(DecimalValue::I32(123), decimal_dtype, Nullable);
         let sparse_struct = Sparse::try_new(indices, patch_values, len, fill_scalar).unwrap();
 
-        let expected = DecimalArray::new(
-            buffer![100i128, 200, 123, 123, 123, 123, 123, 300, 4000, 123],
-            decimal_dtype,
-            // NB: patch indices: [0, 1, 7, 8]; patch validity: [Valid, Valid, Valid, Invalid]; ergo 0, 1, 7 are valid.
-            Validity::from_mask(Mask::from_excluded_indices(10, vec![8]), Nullable),
-        )
-        .into_array()
-        .execute_arrow(None, &mut ctx)
-        .unwrap();
+        let expected = LEGACY_SESSION
+            .arrow()
+            .execute_arrow(
+                DecimalArray::new(
+                    buffer![100i128, 200, 123, 123, 123, 123, 123, 300, 4000, 123],
+                    decimal_dtype,
+                    // NB: patch indices: [0, 1, 7, 8]; patch validity: [Valid, Valid, Valid, Invalid]; ergo 0, 1, 7 are valid.
+                    Validity::from_mask(Mask::from_excluded_indices(10, vec![8]), Nullable),
+                )
+                .into_array(),
+                None,
+                &mut ctx,
+            )
+            .unwrap();
 
-        let actual = sparse_struct
-            .as_array()
-            .clone()
-            .execute::<DecimalArray>(&mut ctx)
-            .unwrap()
-            .into_array()
-            .execute_arrow(None, &mut ctx)
+        let actual = LEGACY_SESSION
+            .arrow()
+            .execute_arrow(
+                sparse_struct
+                    .as_array()
+                    .clone()
+                    .execute::<DecimalArray>(&mut ctx)
+                    .unwrap()
+                    .into_array(),
+                None,
+                &mut ctx,
+            )
             .unwrap();
 
         assert_eq!(expected.data_type(), actual.data_type());
@@ -1544,9 +1554,16 @@ mod test {
         assert_arrays_eq!(&actual, &expected);
 
         // Note that the preferred arrow list representation is `List` (not `ListView`).
-        let arrow_dtype = expected.dtype().to_arrow_dtype()?;
-        let actual = actual.execute_arrow(Some(&arrow_dtype), &mut ctx)?;
-        let expected = expected.execute_arrow(Some(&arrow_dtype), &mut ctx)?;
+        let arrow_dtype = LEGACY_SESSION
+            .arrow()
+            .to_arrow_field("", expected.dtype())?;
+        let actual = LEGACY_SESSION
+            .arrow()
+            .execute_arrow(actual, Some(&arrow_dtype), &mut ctx)?;
+        let expected =
+            LEGACY_SESSION
+                .arrow()
+                .execute_arrow(expected, Some(&arrow_dtype), &mut ctx)?;
 
         assert_eq!(actual.data_type(), expected.data_type());
         Ok(())

encodings/sparse/src/canonical.rs

@@ -29,7 +29,7 @@ arrow-buffer = { workspace = true }
 arrow-cast = { workspace = true }
 arrow-data = { workspace = true }
 arrow-ord = { workspace = true }
-arrow-schema = { workspace = true }
+arrow-schema = { workspace = true, features = ["canonical_extension_types"] }
 arrow-select = { workspace = true }
 arrow-string = { workspace = true }
 async-lock = { workspace = true }
@@ -206,3 +206,7 @@ harness = false
 [[bench]]
 name = "slice_dict_primitive"
 harness = false
+
+[[bench]]
+name = "to_arrow"
+harness = false

vortex-array/Cargo.toml

@@ -0,0 +1,130 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+#![expect(clippy::unwrap_used)]
+
+use std::sync::Arc;
+
+use divan::Bencher;
+use vortex_array::ArrayRef;
+use vortex_array::IntoArray;
+use vortex_array::LEGACY_SESSION;
+use vortex_array::VortexSessionExecute;
+use vortex_array::arrays::DecimalArray;
+use vortex_array::arrays::ListArray;
+use vortex_array::arrays::PrimitiveArray;
+use vortex_array::arrays::StructArray;
+#[expect(
+    deprecated,
+    reason = "benchmark comparing deprecated method with new one"
+)]
+use vortex_array::arrow::ArrowArrayExecutor;
+use vortex_array::arrow::ArrowSessionExt;
+use vortex_array::dtype::DType;
+use vortex_array::dtype::DecimalDType;
+use vortex_array::dtype::Nullability;
+use vortex_array::dtype::PType;
+use vortex_array::dtype::StructFields;
+
+fn main() {
+    divan::main();
+}
+
+fn schema() -> DType {
+    let fields = StructFields::from_iter([
+        (
+            "primitive",
+            DType::Primitive(PType::F32, Nullability::Nullable),
+        ),
+        (
+            "list",
+            DType::List(
+                Arc::new(DType::Binary(Nullability::NonNullable)),
+                Nullability::Nullable,
+            ),
+        ),
+        (
+            "decimal",
+            DType::Decimal(DecimalDType::new(19, 10), Nullability::Nullable),
+        ),
+    ]);
+    DType::Struct(fields, Nullability::NonNullable)
+}
+
+fn array() -> ArrayRef {
+    StructArray::from_fields(&[
+        (
+            "primitive",
+            PrimitiveArray::from_iter(0i16..1024).into_array(),
+        ),
+        (
+            "list",
+            ListArray::from_iter_slow::<u32, _>(
+                (0..1024).map(|_| vec!["a", "b", "c"]).collect::<Vec<_>>(),
+                Arc::new(DType::Utf8(Nullability::NonNullable)),
+            )
+            .unwrap()
+            .into_array(),
+        ),
+        (
+            "decimal",
+            DecimalArray::from_iter(0i64..1024, DecimalDType::new(19, 2)).into_array(),
+        ),
+    ])
+    .unwrap()
+    .into_array()
+}
+
+#[divan::bench]
+fn to_arrow_dtype(bencher: Bencher) {
+    bencher.with_inputs(schema).bench_values(|dtype| {
+        #[expect(deprecated, reason = "benchmarking deprecated code path")]
+        dtype.to_arrow_dtype().unwrap()
+    });
+}
+
+#[allow(non_snake_case)]
+#[divan::bench]
+fn ArrowExportVTable_to_arrow_field(bencher: Bencher) {
+    // Warm the ArrowSession
+    drop(
+        LEGACY_SESSION
+            .arrow()
+            .to_arrow_field("", &schema())
+            .unwrap(),
+    );
+
+    bencher
+        .with_inputs(schema)
+        .bench_values(|dtype| LEGACY_SESSION.arrow().to_arrow_field("", &dtype).unwrap())
+}
+
+#[divan::bench]
+fn to_arrow_array(bencher: Bencher) {
+    bencher
+        .with_inputs(|| (array(), LEGACY_SESSION.create_execution_ctx()))
+        .bench_values(|(array, mut ctx)| {
+            #[expect(deprecated, reason = "benchmarking deprecated code path")]
+            array.execute_arrow(None, &mut ctx).unwrap()
+        });
+}
+
+#[allow(non_snake_case)]
+#[divan::bench]
+fn ArrowExportVTable_execute_arrow(bencher: Bencher) {
+    // Warm the ArrowSession
+    drop(LEGACY_SESSION.arrow().execute_arrow(
+        array(),
+        None,
+        &mut LEGACY_SESSION.create_execution_ctx(),
+    ));
+
+    bencher
+        .with_inputs(|| (array(), LEGACY_SESSION.create_execution_ctx()))
+        .bench_values(|(array, mut ctx)| {
+            LEGACY_SESSION
+                .arrow()
+                .execute_arrow(array, None, &mut ctx)
+                .unwrap()
+        })
+}