Pass Columnar arguments to scalar functions

vortex-array/public-api.lock

@@ -13716,7 +13716,7 @@ pub fn vortex_array::scalar_fn::fns::get_item::GetItem::child_name(&self, _insta
 
 pub fn vortex_array::scalar_fn::fns::get_item::GetItem::deserialize(&self, _metadata: &[u8], _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<Self::Options>
 
-pub fn vortex_array::scalar_fn::fns::get_item::GetItem::execute(&self, field_name: &vortex_array::dtype::FieldName, args: &dyn vortex_array::scalar_fn::ExecutionArgs, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ArrayRef>
+pub fn vortex_array::scalar_fn::fns::get_item::GetItem::execute(&self, field_name: &vortex_array::dtype::FieldName, args: &dyn vortex_array::scalar_fn::ExecutionArgs, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ArrayRef>
 
 pub fn vortex_array::scalar_fn::fns::get_item::GetItem::fmt_sql(&self, field_name: &vortex_array::dtype::FieldName, expr: &vortex_array::expr::Expression, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 
@@ -14232,7 +14232,7 @@ pub fn vortex_array::scalar_fn::fns::merge::Merge::child_name(&self, _instance:
 
 pub fn vortex_array::scalar_fn::fns::merge::Merge::deserialize(&self, _metadata: &[u8], _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<Self::Options>
 
-pub fn vortex_array::scalar_fn::fns::merge::Merge::execute(&self, options: &Self::Options, args: &dyn vortex_array::scalar_fn::ExecutionArgs, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ArrayRef>
+pub fn vortex_array::scalar_fn::fns::merge::Merge::execute(&self, options: &Self::Options, args: &dyn vortex_array::scalar_fn::ExecutionArgs, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ArrayRef>
 
 pub fn vortex_array::scalar_fn::fns::merge::Merge::fmt_sql(&self, _options: &Self::Options, expr: &vortex_array::expr::Expression, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 
@@ -14276,7 +14276,7 @@ pub fn vortex_array::scalar_fn::fns::not::Not::child_name(&self, _options: &Self
 
 pub fn vortex_array::scalar_fn::fns::not::Not::deserialize(&self, _metadata: &[u8], _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<Self::Options>
 
-pub fn vortex_array::scalar_fn::fns::not::Not::execute(&self, _data: &Self::Options, args: &dyn vortex_array::scalar_fn::ExecutionArgs, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ArrayRef>
+pub fn vortex_array::scalar_fn::fns::not::Not::execute(&self, _data: &Self::Options, args: &dyn vortex_array::scalar_fn::ExecutionArgs, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ArrayRef>
 
 pub fn vortex_array::scalar_fn::fns::not::Not::fmt_sql(&self, _options: &Self::Options, expr: &vortex_array::expr::Expression, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 
@@ -15004,27 +15004,27 @@ pub struct vortex_array::scalar_fn::VecExecutionArgs
 
 impl vortex_array::scalar_fn::VecExecutionArgs
 
-pub fn vortex_array::scalar_fn::VecExecutionArgs::new(inputs: alloc::vec::Vec<vortex_array::ArrayRef>, row_count: usize) -> Self
+pub fn vortex_array::scalar_fn::VecExecutionArgs::new(inputs: alloc::vec::Vec<vortex_array::Columnar>, row_count: usize) -> Self
 
 impl vortex_array::scalar_fn::ExecutionArgs for vortex_array::scalar_fn::VecExecutionArgs
 
-pub fn vortex_array::scalar_fn::VecExecutionArgs::get(&self, index: usize) -> vortex_error::VortexResult<vortex_array::ArrayRef>
+pub fn vortex_array::scalar_fn::VecExecutionArgs::get(&self, index: usize) -> vortex_error::VortexResult<vortex_array::Columnar>
 
 pub fn vortex_array::scalar_fn::VecExecutionArgs::num_inputs(&self) -> usize
 
 pub fn vortex_array::scalar_fn::VecExecutionArgs::row_count(&self) -> usize
 
 pub trait vortex_array::scalar_fn::ExecutionArgs
 
-pub fn vortex_array::scalar_fn::ExecutionArgs::get(&self, index: usize) -> vortex_error::VortexResult<vortex_array::ArrayRef>
+pub fn vortex_array::scalar_fn::ExecutionArgs::get(&self, index: usize) -> vortex_error::VortexResult<vortex_array::Columnar>
 
 pub fn vortex_array::scalar_fn::ExecutionArgs::num_inputs(&self) -> usize
 
 pub fn vortex_array::scalar_fn::ExecutionArgs::row_count(&self) -> usize
 
 impl vortex_array::scalar_fn::ExecutionArgs for vortex_array::scalar_fn::VecExecutionArgs
 
-pub fn vortex_array::scalar_fn::VecExecutionArgs::get(&self, index: usize) -> vortex_error::VortexResult<vortex_array::ArrayRef>
+pub fn vortex_array::scalar_fn::VecExecutionArgs::get(&self, index: usize) -> vortex_error::VortexResult<vortex_array::Columnar>
 
 pub fn vortex_array::scalar_fn::VecExecutionArgs::num_inputs(&self) -> usize
 
@@ -15320,7 +15320,7 @@ pub fn vortex_array::scalar_fn::fns::get_item::GetItem::child_name(&self, _insta
 
 pub fn vortex_array::scalar_fn::fns::get_item::GetItem::deserialize(&self, _metadata: &[u8], _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<Self::Options>
 
-pub fn vortex_array::scalar_fn::fns::get_item::GetItem::execute(&self, field_name: &vortex_array::dtype::FieldName, args: &dyn vortex_array::scalar_fn::ExecutionArgs, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ArrayRef>
+pub fn vortex_array::scalar_fn::fns::get_item::GetItem::execute(&self, field_name: &vortex_array::dtype::FieldName, args: &dyn vortex_array::scalar_fn::ExecutionArgs, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ArrayRef>
 
 pub fn vortex_array::scalar_fn::fns::get_item::GetItem::fmt_sql(&self, field_name: &vortex_array::dtype::FieldName, expr: &vortex_array::expr::Expression, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 
@@ -15536,7 +15536,7 @@ pub fn vortex_array::scalar_fn::fns::merge::Merge::child_name(&self, _instance:
 
 pub fn vortex_array::scalar_fn::fns::merge::Merge::deserialize(&self, _metadata: &[u8], _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<Self::Options>
 
-pub fn vortex_array::scalar_fn::fns::merge::Merge::execute(&self, options: &Self::Options, args: &dyn vortex_array::scalar_fn::ExecutionArgs, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ArrayRef>
+pub fn vortex_array::scalar_fn::fns::merge::Merge::execute(&self, options: &Self::Options, args: &dyn vortex_array::scalar_fn::ExecutionArgs, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ArrayRef>
 
 pub fn vortex_array::scalar_fn::fns::merge::Merge::fmt_sql(&self, _options: &Self::Options, expr: &vortex_array::expr::Expression, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 
@@ -15572,7 +15572,7 @@ pub fn vortex_array::scalar_fn::fns::not::Not::child_name(&self, _options: &Self
 
 pub fn vortex_array::scalar_fn::fns::not::Not::deserialize(&self, _metadata: &[u8], _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<Self::Options>
 
-pub fn vortex_array::scalar_fn::fns::not::Not::execute(&self, _data: &Self::Options, args: &dyn vortex_array::scalar_fn::ExecutionArgs, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ArrayRef>
+pub fn vortex_array::scalar_fn::fns::not::Not::execute(&self, _data: &Self::Options, args: &dyn vortex_array::scalar_fn::ExecutionArgs, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ArrayRef>
 
 pub fn vortex_array::scalar_fn::fns::not::Not::fmt_sql(&self, _options: &Self::Options, expr: &vortex_array::expr::Expression, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 
@@ -18034,6 +18034,10 @@ impl vortex_array::IntoArray for vortex_array::Canonical
 
 pub fn vortex_array::Canonical::into_array(self) -> vortex_array::ArrayRef
 
+impl<'a> core::convert::From<&'a vortex_array::Canonical> for vortex_array::CanonicalView<'a>
+
+pub fn vortex_array::CanonicalView<'a>::from(value: &'a vortex_array::Canonical) -> Self
+
 pub enum vortex_array::CanonicalView<'a>
 
 pub vortex_array::CanonicalView::Bool(&'a vortex_array::arrays::BoolArray)
@@ -18066,6 +18070,10 @@ impl<'a> core::clone::Clone for vortex_array::CanonicalView<'a>
 
 pub fn vortex_array::CanonicalView<'a>::clone(&self) -> vortex_array::CanonicalView<'a>
 
+impl<'a> core::convert::From<&'a vortex_array::Canonical> for vortex_array::CanonicalView<'a>
+
+pub fn vortex_array::CanonicalView<'a>::from(value: &'a vortex_array::Canonical) -> Self
+
 impl<'a> core::fmt::Debug for vortex_array::CanonicalView<'a>
 
 pub fn vortex_array::CanonicalView<'a>::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
@@ -18086,6 +18094,14 @@ pub fn vortex_array::Columnar::is_empty(&self) -> bool
 
 pub fn vortex_array::Columnar::len(&self) -> usize
 
+impl core::clone::Clone for vortex_array::Columnar
+
+pub fn vortex_array::Columnar::clone(&self) -> vortex_array::Columnar
+
+impl core::convert::AsRef<dyn vortex_array::Array> for vortex_array::Columnar
+
+pub fn vortex_array::Columnar::as_ref(&self) -> &dyn vortex_array::Array
+
 impl vortex_array::Executable for vortex_array::Columnar
 
 pub fn vortex_array::Columnar::execute(array: vortex_array::ArrayRef, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<Self>

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

@@ -21,6 +21,8 @@ use crate::Array;
 use crate::ArrayEq;
 use crate::ArrayHash;
 use crate::ArrayRef;
+use crate::Columnar;
+use crate::Executable;
 use crate::IntoArray;
 use crate::Precision;
 use crate::arrays::scalar_fn::array::ScalarFnArray;
@@ -196,7 +198,12 @@ impl VTable for ScalarFnVTable {
 
     fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         ctx.log(format_args!("scalar_fn({}): executing", array.scalar_fn));
-        let args = VecExecutionArgs::new(array.children.clone(), array.len);
+        let inputs: Vec<Columnar> = array
+            .children
+            .iter()
+            .map(|child| Columnar::execute(child.clone(), ctx))
+            .collect::<VortexResult<_>>()?;
+        let args = VecExecutionArgs::new(inputs, array.len);
         array.scalar_fn.execute(&args, ctx)
     }
 
@@ -361,7 +368,7 @@ impl scalar_fn::ScalarFnVTable for ArrayExpr {
         _args: &dyn ExecutionArgs,
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<ArrayRef> {
-        crate::Executable::execute(options.0.clone(), ctx)
+        Executable::execute(options.0.clone(), ctx)
     }
 
     fn validity(

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

@@ -4,7 +4,6 @@
 use vortex_error::VortexResult;
 
 use crate::Array;
-use crate::IntoArray;
 use crate::LEGACY_SESSION;
 use crate::VortexSessionExecute;
 use crate::arrays::ConstantArray;
@@ -20,7 +19,12 @@ impl OperationsVTable<ScalarFnVTable> for ScalarFnVTable {
         let inputs: Vec<_> = array
             .children
             .iter()
-            .map(|child| Ok(ConstantArray::new(child.scalar_at(index)?, 1).into_array()))
+            .map(|child| {
+                Ok(Columnar::Constant(ConstantArray::new(
+                    child.scalar_at(index)?,
+                    1,
+                )))
+            })
             .collect::<VortexResult<_>>()?;
 
         let mut ctx = LEGACY_SESSION.create_execution_ctx();

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

@@ -4,9 +4,12 @@
 use vortex_error::VortexResult;
 
 use crate::ArrayRef;
+use crate::Columnar;
+use crate::Executable;
 use crate::IntoArray;
 use crate::LEGACY_SESSION;
 use crate::VortexSessionExecute;
+use crate::arrays::ConstantArray;
 use crate::arrays::scalar_fn::array::ScalarFnArray;
 use crate::arrays::scalar_fn::vtable::ArrayExpr;
 use crate::arrays::scalar_fn::vtable::FakeEq;
@@ -34,14 +37,17 @@ fn execute_expr(expr: &Expression, row_count: usize) -> VortexResult<ArrayRef> {
     // Handle Literal expression - create a constant array
     if expr.is::<Literal>() {
         let scalar = expr.as_::<Literal>();
-        return Ok(crate::arrays::ConstantArray::new(scalar.clone(), row_count).into_array());
+        return Ok(ConstantArray::new(scalar.clone(), row_count).into_array());
     }
 
-    // Recursively execute child expressions to get input arrays
-    let inputs: Vec<ArrayRef> = expr
+    // Recursively execute child expressions to get input arrays, then categorize as Columnar
+    let inputs: Vec<Columnar> = expr
         .children()
         .iter()
-        .map(|child| execute_expr(child, row_count))
+        .map(|child| {
+            let arr = execute_expr(child, row_count)?;
+            Columnar::execute(arr, &mut ctx)
+        })
         .collect::<VortexResult<_>>()?;
 
     let args = VecExecutionArgs::new(inputs, row_count);

vortex-array/src/canonical.rs

@@ -837,6 +837,22 @@ pub enum CanonicalView<'a> {
     Extension(&'a ExtensionArray),
 }
 
+impl<'a> From<&'a Canonical> for CanonicalView<'a> {
+    fn from(value: &'a Canonical) -> Self {
+        match value {
+            Canonical::Null(a) => CanonicalView::Null(a),
+            Canonical::Bool(a) => CanonicalView::Bool(a),
+            Canonical::Primitive(a) => CanonicalView::Primitive(a),
+            Canonical::Decimal(a) => CanonicalView::Decimal(a),
+            Canonical::VarBinView(a) => CanonicalView::VarBinView(a),
+            Canonical::List(a) => CanonicalView::List(a),
+            Canonical::FixedSizeList(a) => CanonicalView::FixedSizeList(a),
+            Canonical::Struct(a) => CanonicalView::Struct(a),
+            Canonical::Extension(a) => CanonicalView::Extension(a),
+        }
+    }
+}
+
 impl From<CanonicalView<'_>> for Canonical {
     fn from(value: CanonicalView<'_>) -> Self {
         match value {

vortex-array/src/columnar.rs

@@ -27,6 +27,7 @@ use crate::scalar::Scalar;
 /// Since the [`Canonical`] enum has one variant per logical data type, it is inefficient for
 /// representing constant arrays. The [`Columnar`] enum allows holding an array of data in either
 /// canonical or constant form enabling efficient handling of constants during execution.
+#[derive(Clone)]
 pub enum Columnar {
     /// A columnar array in canonical form.
     Canonical(Canonical),
@@ -62,6 +63,15 @@ impl Columnar {
     }
 }
 
+impl AsRef<dyn Array> for Columnar {
+    fn as_ref(&self) -> &dyn Array {
+        match self {
+            Columnar::Canonical(c) => c.as_ref(),
+            Columnar::Constant(c) => c.as_ref(),
+        }
+    }
+}
+
 impl IntoArray for Columnar {
     fn into_array(self) -> ArrayRef {
         match self {

vortex-array/src/scalar_fn/fns/between/mod.rs

@@ -299,18 +299,9 @@ impl ScalarFnVTable for Between {
         args: &dyn ExecutionArgs,
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<ArrayRef> {
-        let arr = args.get(0)?;
-        let lower = args.get(1)?;
-        let upper = args.get(2)?;
-
-        // canonicalize the arr and we might be able to run a between kernels over that.
-        if !arr.is_canonical() {
-            return arr.execute::<Canonical>(ctx)?.into_array().between(
-                lower,
-                upper,
-                options.clone(),
-            );
-        }
+        let arr = args.get(0)?.into_array();
+        let lower = args.get(1)?.into_array();
+        let upper = args.get(2)?.into_array();
 
         between_canonical(arr.as_ref(), lower.as_ref(), upper.as_ref(), options, ctx)
     }

vortex-array/src/scalar_fn/fns/binary/mod.rs

@@ -15,6 +15,7 @@ use vortex_session::VortexSession;
 
 use crate::ArrayRef;
 use crate::ExecutionCtx;
+use crate::IntoArray;
 use crate::dtype::DType;
 use crate::expr::StatsCatalog;
 use crate::expr::and;
@@ -128,8 +129,8 @@ impl ScalarFnVTable for Binary {
         args: &dyn ExecutionArgs,
         _ctx: &mut ExecutionCtx,
     ) -> VortexResult<ArrayRef> {
-        let lhs = args.get(0)?;
-        let rhs = args.get(1)?;
+        let lhs = args.get(0)?.into_array();
+        let rhs = args.get(1)?.into_array();
 
         match op {
             Operator::Eq => execute_compare(&lhs, &rhs, CompareOperator::Eq),

vortex-array/src/scalar_fn/fns/cast/mod.rs

@@ -13,10 +13,9 @@ use vortex_error::vortex_err;
 use vortex_proto::expr as pb;
 use vortex_session::VortexSession;
 
-use crate::AnyColumnar;
 use crate::ArrayRef;
 use crate::CanonicalView;
-use crate::ColumnarView;
+use crate::Columnar;
 use crate::ExecutionCtx;
 use crate::arrays::BoolVTable;
 use crate::arrays::ConstantArray;
@@ -29,7 +28,6 @@ use crate::arrays::NullVTable;
 use crate::arrays::PrimitiveVTable;
 use crate::arrays::StructVTable;
 use crate::arrays::VarBinViewVTable;
-use crate::builtins::ArrayBuiltins;
 use crate::dtype::DType;
 use crate::expr::StatsCatalog;
 use crate::expr::cast;
@@ -109,23 +107,20 @@ impl ScalarFnVTable for Cast {
     ) -> VortexResult<ArrayRef> {
         let input = args.get(0)?;
 
-        let Some(columnar) = input.as_opt::<AnyColumnar>() else {
-            return input.execute::<ArrayRef>(ctx)?.cast(target_dtype.clone());
-        };
-
-        match columnar {
-            ColumnarView::Canonical(canonical) => {
-                match cast_canonical(canonical.clone(), target_dtype, ctx)? {
+        match input {
+            Columnar::Canonical(ref canonical) => {
+                let view = CanonicalView::from(canonical);
+                match cast_canonical(view, target_dtype, ctx)? {
                     Some(result) => Ok(result),
                     None => vortex_bail!(
                         "No CastKernel to cast canonical array {} from {} to {}",
                         canonical.as_ref().encoding_id(),
-                        canonical.as_ref().dtype(),
+                        canonical.dtype(),
                         target_dtype,
                     ),
                 }
             }
-            ColumnarView::Constant(constant) => match cast_constant(constant, target_dtype)? {
+            Columnar::Constant(ref constant) => match cast_constant(constant, target_dtype)? {
                 Some(result) => Ok(result),
                 None => vortex_bail!(
                     "No CastReduce to cast constant array from {} to {}",

vortex-array/src/scalar_fn/fns/dynamic.rs

@@ -13,8 +13,8 @@ use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 
-use crate::Array;
 use crate::ArrayRef;
+use crate::Columnar;
 use crate::ExecutionCtx;
 use crate::IntoArray;
 use crate::arrays::ConstantArray;
@@ -102,7 +102,7 @@ impl ScalarFnVTable for DynamicComparison {
     ) -> VortexResult<ArrayRef> {
         if let Some(scalar) = data.rhs.scalar() {
             let lhs = args.get(0)?;
-            let rhs = ConstantArray::new(scalar, args.row_count()).into_array();
+            let rhs = Columnar::Constant(ConstantArray::new(scalar, args.row_count()));
 
             let delegate_args = VecExecutionArgs::new(vec![lhs, rhs], args.row_count());
             return Binary