[Experiment] Self-defined arrow-export

encodings/runend/Cargo.toml

@@ -15,7 +15,9 @@ version = { workspace = true }
 
 [dependencies]
 arbitrary = { workspace = true, optional = true }
-arrow-array = { workspace = true, optional = true }
+arrow-array = { workspace = true }
+arrow-buffer = { workspace = true }
+arrow-schema = { workspace = true }
 itertools = { workspace = true }
 num-traits = { workspace = true }
 prost = { workspace = true }
@@ -39,7 +41,6 @@ vortex-array = { workspace = true, features = ["_test-harness"] }
 
 [features]
 arbitrary = ["dep:arbitrary", "vortex-array/arbitrary"]
-arrow = ["dep:arrow-array"]
 
 [[bench]]
 name = "run_end_null_count"

encodings/runend/src/array.rs

@@ -3,7 +3,14 @@
 
 use std::fmt::Debug;
 use std::hash::Hash;
-
+use std::sync::Arc;
+
+use arrow_array::ArrayRef as ArrowArrayRef;
+use arrow_array::RunArray;
+use arrow_array::cast::AsArray;
+use arrow_array::types::*;
+use arrow_buffer::ArrowNativeType;
+use arrow_schema::DataType;
 use vortex_array::ArrayEq;
 use vortex_array::ArrayHash;
 use vortex_array::ArrayRef;
@@ -503,6 +510,59 @@ pub(super) fn run_end_canonicalize(
     })
 }
 
+pub(crate) fn build_run_array(
+    ends: &ArrowArrayRef,
+    values: &ArrowArrayRef,
+    ends_type: &DataType,
+    offset: usize,
+    length: usize,
+) -> VortexResult<ArrowArrayRef> {
+    match ends_type {
+        DataType::Int16 => build_run_array_typed::<Int16Type>(ends, values, offset, length),
+        DataType::Int32 => build_run_array_typed::<Int32Type>(ends, values, offset, length),
+        DataType::Int64 => build_run_array_typed::<Int64Type>(ends, values, offset, length),
+        _ => vortex_bail!("Unsupported run-end index type: {:?}", ends_type),
+    }
+}
+
+fn build_run_array_typed<R: RunEndIndexType>(
+    ends: &ArrowArrayRef,
+    values: &ArrowArrayRef,
+    offset: usize,
+    length: usize,
+) -> VortexResult<ArrowArrayRef>
+where
+    R::Native: std::ops::Sub<Output = R::Native> + Ord,
+{
+    let offset_native = R::Native::from_usize(offset).ok_or_else(|| {
+        vortex_error::vortex_err!("Offset {offset} exceeds run-end index capacity")
+    })?;
+    let length_native = R::Native::from_usize(length).ok_or_else(|| {
+        vortex_error::vortex_err!("Length {length} exceeds run-end index capacity")
+    })?;
+
+    let ends_prim = ends.as_primitive::<R>();
+    if offset == 0 && ends_prim.values().last() == Some(&length_native) {
+        return Ok(Arc::new(RunArray::<R>::try_new(ends_prim, values)?) as ArrowArrayRef);
+    }
+
+    // Trim to only include runs covering the [offset, offset+length) range.
+    let num_runs = (ends_prim
+        .values()
+        .partition_point(|&e| e - offset_native < length_native)
+        + 1)
+    .min(ends_prim.len());
+
+    let trimmed_ends = ends.slice(0, num_runs);
+    let trimmed_values = values.slice(0, num_runs);
+
+    let adjusted = trimmed_ends
+        .as_primitive::<R>()
+        .unary(|end| (end - offset_native).min(length_native));
+
+    Ok(Arc::new(RunArray::<R>::try_new(&adjusted, &trimmed_values)?) as ArrowArrayRef)
+}
+
 #[cfg(test)]
 mod tests {
     use vortex_array::IntoArray;

encodings/runend/src/kernel.rs

@@ -3,21 +3,27 @@
 
 use std::ops::Range;
 
+use arrow_schema::DataType;
 use vortex_array::ArrayRef;
 use vortex_array::ExecutionCtx;
 use vortex_array::IntoArray;
+use vortex_array::arrays::ArrowExport;
+use vortex_array::arrays::ArrowExportArray;
 use vortex_array::arrays::ConstantArray;
+use vortex_array::arrays::NativeArrowArray;
 use vortex_array::arrays::Slice;
 use vortex_array::arrays::SliceArray;
 use vortex_array::arrays::dict::TakeExecuteAdaptor;
 use vortex_array::arrays::filter::FilterExecuteAdaptor;
+use vortex_array::arrow::ArrowArrayExecutor;
 use vortex_array::kernel::ExecuteParentKernel;
 use vortex_array::kernel::ParentKernelSet;
 use vortex_array::scalar_fn::fns::binary::CompareExecuteAdaptor;
 use vortex_error::VortexResult;
 
 use crate::RunEnd;
 use crate::RunEndArray;
+use crate::array::build_run_array;
 use crate::compute::take_from::RunEndTakeFrom;
 
 pub(super) const PARENT_KERNELS: ParentKernelSet<RunEnd> = ParentKernelSet::new(&[
@@ -26,8 +32,46 @@ pub(super) const PARENT_KERNELS: ParentKernelSet<RunEnd> = ParentKernelSet::new(
     ParentKernelSet::lift(&FilterExecuteAdaptor(RunEnd)),
     ParentKernelSet::lift(&TakeExecuteAdaptor(RunEnd)),
     ParentKernelSet::lift(&RunEndTakeFrom),
+    ParentKernelSet::lift(&RunEndArrowExportKernel),
 ]);
 
+#[derive(Debug)]
+struct RunEndArrowExportKernel;
+
+impl ExecuteParentKernel<RunEnd> for RunEndArrowExportKernel {
+    type Parent = ArrowExport;
+
+    fn execute_parent(
+        &self,
+        array: &RunEndArray,
+        parent: &ArrowExportArray,
+        _child_idx: usize,
+        ctx: &mut ExecutionCtx,
+    ) -> VortexResult<Option<ArrayRef>> {
+        let DataType::RunEndEncoded(ends_field, values_field) = parent.target() else {
+            return Ok(None);
+        };
+        let arrow_ends = array
+            .ends()
+            .clone()
+            .execute_arrow(Some(ends_field.data_type()), ctx)?;
+        let arrow_values = array
+            .values()
+            .clone()
+            .execute_arrow(Some(values_field.data_type()), ctx)?;
+        let arrow = build_run_array(
+            &arrow_ends,
+            &arrow_values,
+            ends_field.data_type(),
+            array.offset(),
+            array.len(),
+        )?;
+        Ok(Some(
+            NativeArrowArray::new(arrow, parent.dtype().clone()).into_array(),
+        ))
+    }
+}
+
 /// Kernel to execute slicing on a RunEnd array.
 ///
 /// This directly slices the RunEnd array using OperationsVTable::slice and then

encodings/runend/src/lib.rs

@@ -9,7 +9,6 @@ pub use array::*;
 pub use iter::trimmed_ends_iter;
 
 mod array;
-#[cfg(feature = "arrow")]
 mod arrow;
 pub mod compress;
 mod compute;