Use indexed sources for fallible constant rows

vortex-array/benches/row_fn_output.rs

@@ -28,6 +28,7 @@ use vortex_array::scalar_fn::unstable::row::RowVisitor;
 use vortex_array::scalar_fn::unstable::row::execute_rows;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
+use vortex_error::vortex_ensure;
 use vortex_session::VortexSession;
 use vortex_session::registry::CachedId;
 
@@ -87,6 +88,9 @@ impl BenchPrimitive for i64 {
 #[derive(Clone)]
 struct InfallibleBool<T>(PhantomData<T>);
 
+#[derive(Clone)]
+struct DeferredI64;
+
 impl<T: BenchPrimitive> RowFn for InfallibleBool<T> {
     type Options = EmptyOptions;
 
@@ -108,22 +112,69 @@ impl<T: BenchPrimitive> RowFn for InfallibleBool<T> {
     }
 }
 
+impl RowFn for DeferredI64 {
+    type Options = EmptyOptions;
+
+    const ARG_NAMES: &'static [&'static str] = &["lhs", "rhs"];
+    const INFALLIBLE: bool = false;
+
+    fn id(&self) -> ScalarFnId {
+        static ID: CachedId = CachedId::new("bench.row_fn_output.deferred_i64");
+        *ID
+    }
+
+    fn dispatch<V: RowVisitor>(
+        &self,
+        _options: &Self::Options,
+        _args: &[DType],
+        visitor: V,
+    ) -> VortexResult<V::VisitResult> {
+        visitor.visit_deferred::<(i64, i64), i64, bool>(
+            |(lhs, rhs)| lhs.overflowing_add(rhs),
+            |overflowed| {
+                vortex_ensure!(
+                    !overflowed,
+                    "deferred-i64 benchmark inputs must not overflow"
+                );
+                Ok(())
+            },
+        )
+    }
+}
+
 #[vortex_bench_support::cpu_features]
 #[divan::bench(types = [i32, i64], args = INPUT_SHAPES)]
 fn infallible_bool<T: BenchPrimitive>(bencher: Bencher, &shape: &InputShape) {
     let function = InfallibleBool::<T>(PhantomData);
+    bench_row_fn(bencher, &function, make_args::<T>(shape));
+}
+
+#[vortex_bench_support::cpu_features]
+#[divan::bench(args = INPUT_SHAPES)]
+fn deferred_i64(bencher: Bencher, &shape: &InputShape) {
+    bench_row_fn(bencher, &DeferredI64, make_args::<i64>(shape));
+}
+
+fn make_args<T: BenchPrimitive>(shape: InputShape) -> VecExecutionArgs {
     let args = match shape {
         InputShape::PerRowPerRow => vec![T::per_row(0), T::per_row(1)],
         InputShape::PerRowConstant => vec![T::per_row(0), T::constant()],
         InputShape::ConstantPerRow => vec![T::constant(), T::per_row(1)],
     };
-    let args = VecExecutionArgs::new(args, ROWS);
 
+    VecExecutionArgs::new(args, ROWS)
+}
+
+fn bench_row_fn<F: RowFn<Options = EmptyOptions>>(
+    bencher: Bencher,
+    function: &F,
+    args: VecExecutionArgs,
+) {
     bencher
         .counter(ItemsCount::new(ROWS))
         .with_inputs(|| (&args, SESSION.create_execution_ctx()))
         .bench_refs(|(args, ctx)| {
-            execute_rows(&function, &EmptyOptions, *args, ctx)
+            execute_rows(function, &EmptyOptions, *args, ctx)
                 .vortex_expect("row execution should succeed in benchmark")
         });
 }

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

@@ -559,6 +559,20 @@ fn test_valid_only_bool_output_skips_invalid_rows() -> VortexResult<()> {
     Ok(())
 }
 
+#[test]
+fn test_deferred_owned_execution_handles_constant_lhs() -> VortexResult<()> {
+    let lhs = ConstantArray::new(10_i64, 3).into_array();
+    let rhs = PrimitiveArray::from_iter([1_i64, 2, 3]).into_array();
+    let args = VecExecutionArgs::new(vec![lhs, rhs], 3);
+    let mut ctx = array_session().create_execution_ctx();
+
+    let actual = execute_rows(&DeferredAdd::default(), &EmptyOptions, &args, &mut ctx)?;
+    let expected = PrimitiveArray::from_iter([11_i64, 12, 13]).into_array();
+
+    assert_arrays_eq!(&actual, &expected, &mut ctx);
+    Ok(())
+}
+
 #[test]
 fn test_deferred_owned_execution_retries_null_row_failure() -> VortexResult<()> {
     let function = DeferredAdd::default();

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

@@ -177,44 +177,13 @@ where
     let mut values = Vec::<Out>::with_capacity(row_count);
     let output = &mut values.spare_capacity_mut()[..row_count];
 
-    let failure = if let Some(views) = Args::views_if_no_consts(&columns) {
-        // Keep this validation beside the views so LLVM sees their common length here.
-        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) };
-
-        source.map_checked_into(output, |elements| apply(&prepared, elements))
-    } else {
-        // Keep this proof branch-local. Shared validation prevents LLVM from specializing this
-        // loop for each batch-constant arrangement, leaving it scalar under multiple CGUs without
-        // LTO.
-        vortex_ensure!(
-            Args::decoded_lens_match(&columns, row_count),
-            "a decoded row input does not address exactly {row_count} rows",
-        );
-
-        let mut accumulated = Fail::default();
-
-        // Iterate over `output` directly. A `0..row_count` range reuses the address-taken value
-        // from the validation error formatter and retains an output bounds check.
-        for (index, slot) in output.iter_mut().enumerate() {
-            // LLVM unswitches the batch-constant checks in `Args::get` before vectorizing the loop.
-            let (value, row_failure) = apply(&prepared, Args::get(&columns, index));
-
-            slot.write(value);
-            accumulated |= row_failure;
-        }
-
-        accumulated
+    let Some(source) = decoded_source::<Args>(&columns, row_count) else {
+        vortex_bail!("a decoded row input does not address exactly {row_count} rows");
     };
+    let failure = source.map_checked_into(output, |elements| apply(&prepared, elements));
 
-    // SAFETY: normal completion of either execution path initializes `0..row_count` exactly
-    // once, and `values` was allocated with at least `row_count` capacity.
+    // SAFETY: normal completion initializes `0..row_count` exactly once, and `values` was
+    // allocated with at least `row_count` capacity.
     unsafe { values.set_len(row_count) };
 
     // Defer rich error construction until after the row loop.

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

@@ -90,12 +90,11 @@ pub unsafe trait InputElement: 'static {
     /// Read one row without repeating batch-constant work from [`decode`](Self::decode).
     fn get(column: &Self::Column, index: usize) -> Self::Elem<'_>;
 
-    /// Borrow the representation used when this argument varies within the batch.
+    /// Borrow the representation used inside the row loop.
     ///
-    /// Called once before the hot loop. Constants do not use this view because the tuple adapter
-    /// 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`.
+    /// Executors call this before the hot loop. 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/tuple/indexed.rs

@@ -51,8 +51,10 @@ pub(in crate::scalar_fn::unstable::row) fn decoded_source<'a, Args: IndexedEleme
 /// the row loop.
 enum ArgColumnSource<'a, T: InputElement> {
     Rows(T::View<'a>),
+
+    /// A validated one-row view that logically addresses `row_count` rows.
     Constant {
-        column: &'a T::Column,
+        view: T::View<'a>,
         row_count: usize,
     },
 }
@@ -64,7 +66,10 @@ impl<'a, T: InputElement> ArgColumnSource<'a, T> {
                 let view = T::view(column);
                 (view.len() == row_count).then_some(Self::Rows(view))
             }
-            ArgColumnKind::Const(column) => Some(Self::Constant { column, row_count }),
+            ArgColumnKind::Const(column) => {
+                let view = T::view(column);
+                (view.len() == 1).then_some(Self::Constant { view, row_count })
+            }
         }
     }
 }
@@ -86,8 +91,10 @@ impl<'a, T: InputElement> IndexedSource for ArgColumnSource<'a, T> {
                 // caller guarantees that `index` is below the source length.
                 unsafe { T::get_from_view_unchecked(view, index) }
             }
-            // `ArgColumn::try_from_const` validated row zero when it constructed this column.
-            Self::Constant { column, .. } => T::get(column, 0),
+            Self::Constant { view, .. } => {
+                // SAFETY: `try_new` checked that this exact retained view contains row zero.
+                unsafe { T::get_from_view_unchecked(view, 0) }
+            }
         }
     }
 }