Arrow Executor (#5647)

We eventually need to port all of our execution logic over to executing
using Vectors + VortexSession.
We can release this new API prior to deprecating the existing one.
We should figure out what this API should be.
Once the full operators change has finished, we should be able to pull
this module out into vortex-arrow.

---------

Signed-off-by: Nicholas Gates <[email protected]>

Author: gatesn

vortex-array/Cargo.toml

@@ -25,6 +25,7 @@ arcref = { workspace = true }
 arrow-arith = { workspace = true }
 arrow-array = { workspace = true, features = ["ffi"] }
 arrow-buffer = { workspace = true }
+arrow-cast = { workspace = true }
 arrow-data = { workspace = true }
 arrow-ord = { workspace = true }
 arrow-schema = { workspace = true }

vortex-array/src/array/mod.rs

@@ -8,6 +8,7 @@ use std::fmt::Debug;
 use std::fmt::Formatter;
 use std::hash::Hash;
 use std::hash::Hasher;
+use std::ops::Deref;
 use std::ops::Range;
 use std::sync::Arc;
 
@@ -18,6 +19,7 @@ use vortex_dtype::Nullability;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_ensure;
+use vortex_error::vortex_err;
 use vortex_error::vortex_panic;
 use vortex_mask::Mask;
 use vortex_scalar::Scalar;
@@ -73,6 +75,9 @@ pub trait Array:
     /// Returns the array as a reference to a generic [`Any`] trait object.
     fn as_any(&self) -> &dyn Any;
 
+    /// Returns the array as an `Arc<dyn Any + Send + Sync>`.
+    fn as_any_arc(self: Arc<Self>) -> Arc<dyn Any + Send + Sync>;
+
     /// Returns the array as an [`ArrayRef`].
     fn to_array(&self) -> ArrayRef;
 
@@ -211,6 +216,10 @@ impl Array for Arc<dyn Array> {
         self.as_ref().as_any()
     }
 
+    fn as_any_arc(self: Arc<Self>) -> Arc<dyn Any + Send + Sync> {
+        self
+    }
+
     #[inline]
     fn to_array(&self) -> ArrayRef {
         self.clone()
@@ -350,6 +359,24 @@ impl dyn Array + '_ {
             .map(|array_adapter| &array_adapter.0)
     }
 
+    /// Returns the array downcast to the given `A` as an owned object.
+    pub fn try_into<V: VTable>(self: Arc<Self>) -> Result<V::Array, Arc<Self>> {
+        match self.is::<V>() {
+            true => {
+                let arc = self
+                    .as_any_arc()
+                    .downcast::<ArrayAdapter<V>>()
+                    .map_err(|_| vortex_err!("failed to downcast"))
+                    .vortex_expect("Failed to downcast");
+                Ok(match Arc::try_unwrap(arc) {
+                    Ok(array) => array.0,
+                    Err(arc) => arc.deref().0.clone(),
+                })
+            }
+            false => Err(self),
+        }
+    }
+
     /// Is self an array with encoding from vtable `V`.
     pub fn is<V: VTable>(&self) -> bool {
         self.as_opt::<V>().is_some()
@@ -443,6 +470,10 @@ impl<V: VTable> Array for ArrayAdapter<V> {
         self
     }
 
+    fn as_any_arc(self: Arc<Self>) -> Arc<dyn Any + Send + Sync> {
+        self
+    }
+
     fn to_array(&self) -> ArrayRef {
         Arc::new(ArrayAdapter::<V>(self.0.clone()))
     }
@@ -691,13 +722,6 @@ impl<V: VTable> Array for ArrayAdapter<V> {
     }
 
     fn reduce_parent(&self, parent: &ArrayRef, child_idx: usize) -> VortexResult<Option<ArrayRef>> {
-        #[cfg(debug_assertions)]
-        vortex_ensure!(
-            Arc::as_ptr(&parent.children()[child_idx]) == self,
-            "Parent array's child at index {} does not match self",
-            child_idx
-        );
-
         let Some(reduced) = V::reduce_parent(&self.0, parent, child_idx)? else {
             return Ok(None);
         };

vortex-array/src/arrays/dict/array.rs

@@ -114,6 +114,10 @@ impl DictArray {
         Ok(unsafe { Self::new_unchecked(codes, values) })
     }
 
+    pub fn into_parts(self) -> (ArrayRef, ArrayRef) {
+        (self.codes, self.values)
+    }
+
     #[inline]
     pub fn codes(&self) -> &ArrayRef {
         &self.codes

vortex-array/src/arrays/filter/array.rs

@@ -30,4 +30,8 @@ impl FilterArray {
             stats: ArrayStats::default(),
         }
     }
+
+    pub fn mask(&self) -> &Mask {
+        &self.mask
+    }
 }

vortex-array/src/arrays/filter/vtable.rs

@@ -22,8 +22,8 @@ use crate::ArrayHash;
 use crate::ArrayRef;
 use crate::Canonical;
 use crate::IntoArray;
+use crate::LEGACY_SESSION;
 use crate::Precision;
-use crate::arrays::LEGACY_SESSION;
 use crate::arrays::filter::array::FilterArray;
 use crate::arrays::filter::kernel::FilterKernel;
 use crate::kernel::BindCtx;

vortex-array/src/arrays/listview/array.rs

@@ -319,6 +319,10 @@ impl ListViewArray {
         .is_ok()
     }
 
+    pub fn into_parts(self) -> (ArrayRef, ArrayRef, ArrayRef, Validity) {
+        (self.elements, self.offsets, self.sizes, self.validity)
+    }
+
     /// Returns the offset at the given index.
     ///
     /// Note that it is possible the corresponding list view is null (which is only defined by the

vortex-array/src/arrays/mod.rs

@@ -6,8 +6,6 @@
 #[cfg(any(test, feature = "test-harness"))]
 mod assertions;
 
-use std::sync::LazyLock;
-
 #[cfg(any(test, feature = "test-harness"))]
 pub use assertions::format_indices;
 
@@ -59,12 +57,3 @@ pub use scalar_fn::*;
 pub use struct_::*;
 pub use varbin::*;
 pub use varbinview::*;
-use vortex_session::VortexSession;
-
-use crate::session::ArraySession;
-
-// TODO(ngates): canonicalize doesn't currently take a session, therefore we cannot invoke execute
-//  from the new array encodings to support back-compat for legacy encodings. So we hold a session
-//  here...
-static LEGACY_SESSION: LazyLock<VortexSession> =
-    LazyLock::new(|| VortexSession::empty().with::<ArraySession>());

vortex-array/src/arrays/scalar_fn/rules.rs

@@ -27,11 +27,14 @@ use crate::expr::ReduceNode;
 use crate::expr::ReduceNodeRef;
 use crate::expr::ScalarFn;
 use crate::optimizer::rules::ArrayReduceRule;
+use crate::optimizer::rules::ParentRuleSet;
 use crate::optimizer::rules::ReduceRuleSet;
 
 pub(super) const RULES: ReduceRuleSet<ScalarFnVTable> =
     ReduceRuleSet::new(&[&ScalarFnConstantRule, &ScalarFnAbstractReduceRule]);
 
+pub(super) const PARENT_RULES: ParentRuleSet<ScalarFnVTable> = ParentRuleSet::new(&[]);
+
 #[derive(Debug)]
 struct ScalarFnConstantRule;
 impl ArrayReduceRule<ScalarFnVTable> for ScalarFnConstantRule {

vortex-array/src/arrays/scalar_fn/vtable/canonical.rs

@@ -5,7 +5,7 @@ use vortex_error::VortexExpect;
 
 use crate::Array;
 use crate::Canonical;
-use crate::arrays::LEGACY_SESSION;
+use crate::LEGACY_SESSION;
 use crate::arrays::scalar_fn::array::ScalarFnArray;
 use crate::arrays::scalar_fn::vtable::ScalarFnVTable;
 use crate::executor::VectorExecutor;

vortex-array/src/arrays/scalar_fn/vtable/mod.rs

@@ -25,6 +25,7 @@ use crate::arrays::scalar_fn::array::ScalarFnArray;
 use crate::arrays::scalar_fn::kernel::KernelInput;
 use crate::arrays::scalar_fn::kernel::ScalarFnKernel;
 use crate::arrays::scalar_fn::metadata::ScalarFnMetadata;
+use crate::arrays::scalar_fn::rules::PARENT_RULES;
 use crate::arrays::scalar_fn::rules::RULES;
 use crate::expr;
 use crate::expr::ExprVTable;
@@ -167,6 +168,14 @@ impl VTable for ScalarFnVTable {
     fn reduce(array: &Self::Array) -> VortexResult<Option<ArrayRef>> {
         RULES.evaluate(array)
     }
+
+    fn reduce_parent(
+        array: &Self::Array,
+        parent: &ArrayRef,
+        child_idx: usize,
+    ) -> VortexResult<Option<ArrayRef>> {
+        PARENT_RULES.evaluate(array, parent, child_idx)
+    }
 }
 
 /// Array factory functions for scalar functions.