Simplify RowFn execution contracts

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

@@ -4,16 +4,12 @@
 //! Row-loop execution for owned outputs and output sinks.
 //!
 //! [`owned`] stores one independent value per row and reduces compact failure evidence. [`sink`]
-//! drives output builders whose row handles may share batch state. Both return [`RowExecution`],
-//! which distinguishes a completed array from a deferred error that batch validity may suppress.
+//! drives output builders whose row handles may share batch state.
 
 mod owned;
 pub(super) use owned::execute_owned;
 pub(super) use owned::execute_owned_infallible;
 
-mod outcome;
-pub use outcome::RowExecution;
-
 mod sink;
 pub(super) use sink::execute_sink;
 pub(super) use sink::execute_sink_valid_rows;

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

@@ -13,13 +13,12 @@ use vortex_compute::lane_kernels::IndexedSourceExt;
 use vortex_error::VortexResult;
 use vortex_error::vortex_ensure;
 
-use super::RowExecution;
+use crate::ArrayRef;
 use crate::ExecutionCtx;
 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::ViewLen;
 use crate::scalar_fn::unstable::row::visitor::assert_owned_output_needs_no_drop;
 
 /// Zero-sized failure accumulator for infallible owned visits.
@@ -36,7 +35,7 @@ pub(crate) fn execute_owned_infallible<Args, Out, Prepared>(
     ctx: &mut ExecutionCtx,
     prepare: impl FnOnce(Args::ConstElems<'_>) -> Prepared,
     apply: impl Fn(&Prepared, Args::Elems<'_>) -> Out,
-) -> VortexResult<RowExecution>
+) -> VortexResult<ArrayRef>
 where
     Args: IndexedElementTuple,
     Out: OutputElement,
@@ -57,7 +56,7 @@ pub(crate) fn execute_owned<Args, Out, Prepared, Fail>(
     prepare: impl FnOnce(Args::ConstElems<'_>) -> Prepared,
     apply: impl Fn(&Prepared, Args::Elems<'_>) -> (Out, Fail),
     finish_failure: impl FnOnce(Fail) -> VortexResult<()>,
-) -> VortexResult<RowExecution>
+) -> VortexResult<ArrayRef>
 where
     Args: IndexedElementTuple,
     Out: OutputElement,
@@ -78,12 +77,12 @@ where
     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::ARITY == 0 || views.len() == row_count,
+            Args::view_lens_match(&views, row_count),
             "a decoded row input does not address exactly {row_count} rows",
         );
 
-        // SAFETY: the tuple length check proved every non-nullary view addresses exactly
-        // `row_count` rows immediately above. Nullary tuples do not access an input view.
+        // 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))
@@ -115,9 +114,8 @@ where
     // once, and `values` was allocated with at least `row_count` capacity.
     unsafe { values.set_len(row_count) };
 
-    // Defer failures so batch execution can retry with only valid rows.
-    match finish_failure(failure) {
-        Ok(()) => Ok(RowExecution::Output(Out::build(values))),
-        Err(error) => Ok(RowExecution::DeferredError(error)),
-    }
+    // Defer rich error construction until after the row loop.
+    finish_failure(failure)?;
+
+    Ok(Out::build(values))
 }

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

@@ -14,7 +14,7 @@ use vortex_error::vortex_ensure_eq;
 use vortex_mask::AllOr;
 use vortex_mask::Mask;
 
-use super::RowExecution;
+use crate::ArrayRef;
 use crate::ExecutionCtx;
 use crate::scalar_fn::ExecutionArgs;
 use crate::scalar_fn::unstable::row::ElementTuple;
@@ -32,7 +32,7 @@ pub(crate) fn execute_sink<Args, Prepared, Sink, ApplyResult, Options>(
     ctx: &mut ExecutionCtx,
     prepare: impl FnOnce(Args::ConstElems<'_>) -> Prepared,
     apply: impl Fn(&Prepared, Args::Elems<'_>, <Sink as OutputSink<Options>>::Row<'_>) -> ApplyResult,
-) -> VortexResult<RowExecution>
+) -> VortexResult<ArrayRef>
 where
     Args: ElementTuple,
     Sink: OutputSink<Options>,
@@ -60,13 +60,13 @@ where
 
         let views = Args::views_if_no_consts(&columns);
         if let Some(views) = views {
-            if Args::ARITY != 0 && views.len() != row_count {
+            if !Args::view_lens_match(&views, row_count) {
                 decoded_length_error(row_count)?;
             }
 
             for index in 0..row_count {
-                // SAFETY: the tuple length check proved every non-nullary view has `row_count`
-                // rows before the loop. Nullary tuples do not access an input view.
+                // SAFETY: `view_lens_match` checked that these exact retained views address
+                // `row_count` rows before the loop.
                 let elements = unsafe { Args::get_from_views_unchecked(&views, index) };
                 // SAFETY: the sink row-count check above proved every loop index is in bounds.
                 let output =
@@ -92,21 +92,21 @@ where
     }
 
     // SAFETY: every row callback completed successfully, so each returned the required write token.
-    unsafe { <Sink as OutputSink<Options>>::finish(sink) }.map(RowExecution::Output)
+    unsafe { <Sink as OutputSink<Options>>::finish(sink) }
 }
 
 /// Write only the rows set in `valid`, or decline when the inputs or sink cannot support
 /// skip-invalid execution.
 ///
-/// `Ok(None)` signals batch execution to filter every input to the valid rows, run the dense
-/// kernel, and scatter the results back into a null-padded array.
+/// `Ok(None)` signals that direct skip-invalid execution is unavailable. Batch execution decides
+/// how to handle the decline.
 pub(crate) fn execute_sink_valid_rows<Args, Prepared, Sink, ApplyResult, Options>(
     args: &dyn ExecutionArgs,
     valid: &Mask,
     ctx: &mut ExecutionCtx,
     prepare: impl FnOnce(Args::ConstElems<'_>) -> Prepared,
     apply: impl Fn(&Prepared, Args::Elems<'_>, <Sink as OutputSink<Options>>::Row<'_>) -> ApplyResult,
-) -> VortexResult<Option<RowExecution>>
+) -> VortexResult<Option<ArrayRef>>
 where
     Args: ElementTuple,
     Sink: OutputSink<Options>,
@@ -144,7 +144,7 @@ where
         );
 
         if let Some(views) = views {
-            if Args::ARITY != 0 && views.len() != row_count {
+            if !Args::view_lens_match(&views, row_count) {
                 decoded_length_error(row_count)?;
             }
 
@@ -154,9 +154,8 @@ where
                 let output =
                     unsafe { <Sink as OutputSink<Options>>::row_unchecked(&mut rows, index) };
 
-                // SAFETY: the tuple length check proved every non-nullary view has `row_count`
-                // rows, and mask indices are below `row_count`. Nullary tuples do not access an
-                // input view.
+                // SAFETY: `view_lens_match` checked that these exact retained views address
+                // `row_count` rows, and mask indices are below `row_count`.
                 let elements = unsafe { Args::get_from_views_unchecked(&views, index) };
 
                 apply(&prepared, elements, output).into_result()
@@ -179,9 +178,7 @@ where
 
     // SAFETY: the initializer completed before traversal, and every visited callback completed
     // successfully and returned the required write token.
-    unsafe { <Sink as OutputSink<Options>>::finish(sink) }
-        .map(RowExecution::Output)
-        .map(Some)
+    unsafe { <Sink as OutputSink<Options>>::finish(sink) }.map(Some)
 }
 
 /// Construct a decoded-length error outside the traversal branches.
@@ -267,21 +264,17 @@ mod tests {
     use vortex_error::vortex_err;
     use vortex_mask::Mask;
 
-    use super::RowExecution;
     use super::execute_sink_valid_rows;
     use crate::ArrayRef;
     use crate::IntoArray;
     use crate::VortexSessionExecute;
     use crate::array_session;
     use crate::arrays::PrimitiveArray;
-    use crate::assert_arrays_eq;
     use crate::dtype::DType;
     use crate::dtype::NativePType;
     use crate::scalar_fn::EmptyOptions;
     use crate::scalar_fn::VecExecutionArgs;
-    use crate::scalar_fn::unstable::row::InitializedElement;
     use crate::scalar_fn::unstable::row::OutputSink;
-    use crate::scalar_fn::unstable::row::UninitElementSink;
     use crate::validity::Validity;
 
     struct NonSkippingSink;
@@ -370,39 +363,6 @@ mod tests {
         Ok(())
     }
 
-    #[test]
-    fn test_skip_invalid_sink_initializes_and_writes_addressed_rows() -> VortexResult<()> {
-        let input = PrimitiveArray::from_iter([10_i64, 20, 30]).into_array();
-        let args = VecExecutionArgs::new(vec![input], 3);
-        let valid = Mask::from_iter([true, false, true]);
-        let mut ctx = array_session().create_execution_ctx();
-
-        let execution = execute_sink_valid_rows::<
-            (i64,),
-            (),
-            UninitElementSink<i64>,
-            InitializedElement,
-            EmptyOptions,
-        >(
-            &args,
-            &valid,
-            &mut ctx,
-            |_| (),
-            |_, (value,), output| {
-                // SAFETY: `output` is the row supplied to this callback.
-                unsafe { InitializedElement::write(output, value * 2) }
-            },
-        )?;
-        let Some(RowExecution::Output(actual)) = execution else {
-            vortex_bail!("the skip-invalid sink must produce an output");
-        };
-        let expected = PrimitiveArray::from_iter([20_i64, 0, 60]);
-
-        assert_arrays_eq!(&actual, expected.as_ref(), &mut ctx);
-
-        Ok(())
-    }
-
     #[test]
     fn test_skip_invalid_sink_rechecks_rows_after_initialization() -> VortexResult<()> {
         let input = PrimitiveArray::from_iter([10_i64, 20]).into_array();

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

@@ -14,13 +14,11 @@
 //! Unlike a general strict function, a [`RowFn`] cannot produce null from valid inputs.
 //!
 //! Prepared visits move work derived from constant operands outside the hot loop. Deferred visits
-//! reduce compact failure evidence in that loop and retry only valid rows when null payloads may
-//! have caused the failure.
+//! reduce compact failure evidence without constructing errors in that loop.
 
 // TODO(connor)[RowFn]: Remove this expectation when #9450 connects the batch executor.
 #[expect(dead_code)]
 mod execute;
-pub use execute::RowExecution;
 
 mod row_fn;
 pub use row_fn::RowFn;

vortex-array/src/scalar_fn/unstable/row/row_fn.rs

@@ -40,13 +40,13 @@ pub trait RowFn: 'static + Sized + Clone + Send + Sync {
     /// The arguments in display order. Its length is the function's exact arity.
     const ARG_NAMES: &'static [&'static str];
 
-    /// Whether any dispatch can raise a semantic error.
+    /// Whether every dispatch is infallible.
     ///
     /// See [`ScalarFnVTable::is_fallible`](crate::scalar_fn::ScalarFnVTable::is_fallible) for a
     /// more detailed explanation of semantic errors.
     ///
-    /// The framework checks dispatched element and result types. A conservative `true` is allowed.
-    const FALLIBLE: bool;
+    /// The framework checks dispatched element and result types. A conservative `false` is allowed.
+    const INFALLIBLE: bool;
 
     /// Returns the ID of the scalar function.
     fn id(&self) -> ScalarFnId;

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

@@ -54,8 +54,8 @@ pub unsafe trait InputElement: 'static {
 
     /// Decode `array` into its column representation.
     ///
-    /// Called once per row-kernel invocation, including deferred-error retries. Hoist dtype checks,
-    /// downcasts, and other invocation-invariant work into this method.
+    /// Called once per row-kernel invocation. Hoist dtype checks, downcasts, and other
+    /// invocation-invariant work into this method.
     fn decode(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self::Column>;
 
     /// Whether [`decode_null_tolerant`](Self::decode_null_tolerant) can decode this array.

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

@@ -175,8 +175,6 @@ pub trait ElementTuple: 'static + private::Sealed {
     fn validate(dtypes: &[DType]) -> VortexResult<()>;
 
     /// Decode every input column once for one row-kernel invocation.
-    ///
-    /// A dense deferred-error retry starts another invocation over filtered valid rows.
     fn decode(args: &dyn ExecutionArgs, ctx: &mut ExecutionCtx) -> VortexResult<Self::Columns>;
 
     /// Whether every input can be decoded without assuming that all rows are valid.
@@ -206,15 +204,19 @@ pub trait ElementTuple: 'static + private::Sealed {
     /// returned tuple removes batch-constant checks from the row loop.
     fn views_if_no_consts(columns: &Self::Columns) -> Option<Self::Views<'_>>;
 
-    /// Whether every argument decoded at full batch length contains exactly `row_count` rows.
+    /// Whether every borrowed view contains exactly `row_count` rows.
     ///
-    /// NB: `Columns` cannot implement [`ViewLen`] because a batch-constant [`ArgColumn`] physically
-    /// contains one decoded row while logically addressing every row in the batch. By contrast,
-    /// [`views_if_no_consts`](Self::views_if_no_consts) constructs `Views` only when every argument
-    /// is non-constant, so those views have one common physical length.
+    /// Executors must validate each retained view before unchecked traversal. [`ViewLen::len`] on
+    /// a tuple asserts that component lengths are equal and panics on a mismatch. Rebuilding views
+    /// from [`Columns`](Self::Columns) does not prove the lengths of the views passed to
+    /// [`get_from_views_unchecked`](Self::get_from_views_unchecked).
+    fn view_lens_match(views: &Self::Views<'_>, row_count: usize) -> bool;
+
+    /// Whether every argument decoded at full batch length contains exactly `row_count` rows.
     ///
-    /// This check runs once before the hot loop. A batch constant is exempt because its
-    /// [`ArgColumn`] constructor already validated the one-row representation produced by decoding.
+    /// `Columns` cannot implement [`ViewLen`] because a batch-constant `ArgColumn` stores one
+    /// decoded row while logically addressing the full batch. The batch-constant constructor
+    /// validates that one-row representation, so this method checks only non-constant columns.
     fn decoded_lens_match(columns: &Self::Columns, row_count: usize) -> bool;
 
     /// Read one row from borrowed views.
@@ -281,6 +283,10 @@ impl ElementTuple for () {
         Some(())
     }
 
+    fn view_lens_match(_views: &Self::Views<'_>, _row_count: usize) -> bool {
+        true
+    }
+
     fn decoded_lens_match(_columns: &Self::Columns, _row_count: usize) -> bool {
         true
     }
@@ -361,6 +367,10 @@ macro_rules! element_tuple {
                 Some(($($t::view(columns.$idx.as_column()?),)+))
             }
 
+            fn view_lens_match(views: &Self::Views<'_>, row_count: usize) -> bool {
+                $(views.$idx.len() == row_count &&)+ true
+            }
+
             fn decoded_lens_match(columns: &Self::Columns, row_count: usize) -> bool {
                 $(columns.$idx.addresses_rows(row_count) &&)+ true
             }

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

@@ -90,6 +90,14 @@ i64_test_element!(
 );
 i64_test_element!(DenseFallible, false);
 
+#[test]
+fn test_view_lens_match_checks_each_view() {
+    let first: &[i64] = &[1, 2];
+    let second: &[i64] = &[3];
+
+    assert!(!<(i64, i64)>::view_lens_match(&(first, second), 2));
+}
+
 #[test]
 fn test_null_tolerant_decline_precedes_decoding() -> VortexResult<()> {
     DECODE_CALLS.store(0, Ordering::Relaxed);