Map constant RowFn inputs directly

vortex-array/src/scalar_fn/unstable/row/batch/tests.rs

@@ -68,6 +68,9 @@ struct InvalidKernelOutput;
 #[derive(Clone)]
 struct PackedPositive;
 
+#[derive(Clone)]
+struct PackedGreaterThan;
+
 #[derive(Clone)]
 struct ValidOnlyPositive;
 
@@ -344,6 +347,27 @@ impl RowFn for PackedPositive {
     }
 }
 
+impl RowFn for PackedGreaterThan {
+    type Options = EmptyOptions;
+
+    const ARG_NAMES: &'static [&'static str] = &["lhs", "rhs"];
+    const INFALLIBLE: bool = true;
+
+    fn id(&self) -> ScalarFnId {
+        static ID: CachedId = CachedId::new("test.packed_greater_than");
+        *ID
+    }
+
+    fn dispatch<V: RowVisitor>(
+        &self,
+        _options: &Self::Options,
+        _args: &[DType],
+        visitor: V,
+    ) -> VortexResult<V::VisitResult> {
+        visitor.visit::<(i64, i64), bool>(|(lhs, rhs)| lhs > rhs)
+    }
+}
+
 impl RowFn for ValidOnlyPositive {
     type Options = EmptyOptions;
 
@@ -473,6 +497,39 @@ fn test_bool_output_word_boundaries(#[case] len: usize) -> VortexResult<()> {
     Ok(())
 }
 
+#[test]
+fn test_bool_output_handles_partial_constants() -> VortexResult<()> {
+    let values: Vec<_> = (0_i64..65).collect();
+    let varying = PrimitiveArray::from_iter(values.iter().copied()).into_array();
+    let constant = ConstantArray::new(32_i64, values.len()).into_array();
+    let mut ctx = array_session().create_execution_ctx();
+
+    let lhs_varying = VecExecutionArgs::new(vec![varying.clone(), constant.clone()], values.len());
+    let actual = execute_rows(&PackedGreaterThan, &EmptyOptions, &lhs_varying, &mut ctx)?;
+    let expected = BoolArray::from_iter(values.iter().map(|value| *value > 32)).into_array();
+    assert_arrays_eq!(&actual, &expected, &mut ctx);
+
+    let rhs_varying = VecExecutionArgs::new(vec![constant, varying], values.len());
+    let actual = execute_rows(&PackedGreaterThan, &EmptyOptions, &rhs_varying, &mut ctx)?;
+    let expected = BoolArray::from_iter(values.iter().map(|value| 32 > *value)).into_array();
+    assert_arrays_eq!(&actual, &expected, &mut ctx);
+
+    Ok(())
+}
+
+#[test]
+fn test_bool_output_handles_all_constant_input() -> VortexResult<()> {
+    let input = ConstantArray::new(7_i64, 65).into_array();
+    let args = VecExecutionArgs::new(vec![input], 65);
+    let mut ctx = array_session().create_execution_ctx();
+
+    let actual = execute_rows(&PackedPositive, &EmptyOptions, &args, &mut ctx)?;
+    let expected = BoolArray::from_iter(std::iter::repeat_n(true, 65)).into_array();
+
+    assert_arrays_eq!(&actual, &expected, &mut ctx);
+    Ok(())
+}
+
 #[test]
 fn test_bool_output_handles_nullary_rows() -> VortexResult<()> {
     let args = VecExecutionArgs::new(vec![], 65);

vortex-array/src/scalar_fn/unstable/row/execute/owned.rs

@@ -5,12 +5,13 @@
 //!
 //! [`execute_owned`] writes fallible row results into spare vector capacity and reduces compact
 //! failure evidence outside the hot loop. [`execute_owned_infallible`] lets the output type map a
-//! validated non-constant row source directly into its physical representation.
+//! validated row source directly into its physical representation.
 
 use std::ops::BitOrAssign;
 
 use vortex_compute::lane_kernels::IndexedSourceExt;
 use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
 use vortex_error::vortex_ensure;
 use vortex_error::vortex_ensure_eq;
 use vortex_mask::MaskValuesRef;
@@ -21,6 +22,7 @@ use crate::scalar_fn::ExecutionArgs;
 use crate::scalar_fn::unstable::row::FailureEvidence;
 use crate::scalar_fn::unstable::row::IndexedElementTuple;
 use crate::scalar_fn::unstable::row::OutputElement;
+use crate::scalar_fn::unstable::row::types::decoded_source;
 use crate::scalar_fn::unstable::row::visitor::assert_owned_output_needs_no_drop;
 
 /// Zero-sized failure accumulator for infallible owned visits.
@@ -48,38 +50,13 @@ where
     let prepared = prepare(Args::const_values(&columns));
     let row_count = args.row_count();
 
-    if let Some(views) = Args::views_if_no_consts(&columns) {
-        vortex_ensure!(
-            Args::view_lens_match(&views, row_count),
-            "a decoded row input does not address exactly {row_count} rows",
-        );
-
-        // SAFETY: `view_lens_match` checked that these exact retained views address `row_count`
-        // rows.
-        let source = unsafe { Args::indexed_source(views, row_count) };
-
-        return Ok(Out::build_from(source, |elements| {
-            apply(&prepared, elements)
-        }));
-    }
-
-    vortex_ensure!(
-        Args::decoded_lens_match(&columns, row_count),
-        "a decoded row input does not address exactly {row_count} rows",
-    );
-
-    let mut values = Vec::<Out>::with_capacity(row_count);
-    let output = &mut values.spare_capacity_mut()[..row_count];
-
-    for (index, slot) in output.iter_mut().enumerate() {
-        slot.write(apply(&prepared, Args::get(&columns, index)));
-    }
-
-    // SAFETY: normal completion initializes every output slot exactly once, and `values` was
-    // allocated with at least `row_count` capacity.
-    unsafe { values.set_len(row_count) };
+    let Some(source) = decoded_source::<Args>(&columns, row_count) else {
+        vortex_bail!("a decoded row input does not address exactly {row_count} rows");
+    };
 
-    Ok(Out::build(values))
+    Ok(Out::build_from(source, |elements| {
+        apply(&prepared, elements)
+    }))
 }
 
 /// Decode nullable inputs, then store one output for each valid row from an infallible kernel.

vortex-array/src/scalar_fn/unstable/row/types/element/input.rs

@@ -93,7 +93,9 @@ pub unsafe trait InputElement: 'static {
     /// Borrow the representation used when this argument varies within the batch.
     ///
     /// Called once before the hot loop. Constants do not use this view because the tuple adapter
-    /// keeps their one-row decoded representation separate.
+    /// keeps their one-row decoded representation separate. For every index below the returned
+    /// view's length, [`get_from_view`](Self::get_from_view) must produce the same element as
+    /// [`get`](Self::get) on `column`.
     fn view(column: &Self::Column) -> Self::View<'_>;
 
     /// Read one row from a [`View`](Self::View).

vortex-array/src/scalar_fn/unstable/row/types/element/mod.rs

@@ -21,3 +21,4 @@ mod tuple;
 pub use tuple::ElementTuple;
 pub use tuple::IndexedElementTuple;
 pub use tuple::batch_const;
+pub(in crate::scalar_fn::unstable::row) use tuple::decoded_source;

vortex-array/src/scalar_fn/unstable/row/types/element/tuple/element_tuple.rs

@@ -24,10 +24,10 @@ use crate::scalar_fn::unstable::row::ViewLen;
 /// One decoded input, collapsed to a single row when it is constant for the batch.
 pub struct ArgColumn<T: InputElement>(
     /// The decoded argument, classified by how the row loop addresses it.
-    ArgColumnKind<T>,
+    pub(super) ArgColumnKind<T>,
 );
 
-enum ArgColumnKind<T: InputElement> {
+pub(super) enum ArgColumnKind<T: InputElement> {
     /// A decoded column covering the full batch; executors validate its length before traversal.
     Column(T::Column),