Implement RowFn row execution

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

@@ -0,0 +1,19 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! 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.
+
+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/outcome.rs

@@ -0,0 +1,43 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! The result of a completed row loop before batch-level null handling.
+//!
+//! [`RowExecution`] preserves deferred failure evidence until batch execution can determine whether
+//! the failing payload belonged to a valid row.
+
+use vortex_error::VortexError;
+use vortex_error::VortexResult;
+
+use crate::ArrayRef;
+
+/// The outcome of a row loop before batch execution decides whether an error is observable.
+///
+/// Together with the surrounding [`VortexResult`], this represents three outcomes:
+///
+/// - `Err(error)` is a non-retryable execution or immediate row error.
+/// - [`Output`](Self::Output) is a successful row loop.
+/// - [`DeferredError`](Self::DeferredError) is failure evidence from a completed row loop.
+///
+/// A dense loop can evaluate null payloads, so its deferred error is not always observable. Batch
+/// execution can retry only valid rows to discard errors caused by null payloads. A plain
+/// `VortexResult<ArrayRef>` cannot distinguish these errors from failures that a retry cannot fix.
+///
+/// Once execution is known to contain only valid rows, converting this outcome into a
+/// `VortexResult<ArrayRef>` turns [`DeferredError`](Self::DeferredError) into an ordinary error.
+pub enum RowExecution {
+    /// The successfully built, full-length output column.
+    Output(ArrayRef),
+
+    /// An error constructed from failure evidence reduced across a completed row loop.
+    DeferredError(VortexError),
+}
+
+impl From<RowExecution> for VortexResult<ArrayRef> {
+    fn from(execution: RowExecution) -> Self {
+        match execution {
+            RowExecution::Output(output) => Ok(output),
+            RowExecution::DeferredError(error) => Err(error),
+        }
+    }
+}

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

@@ -0,0 +1,123 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Executes row kernels that return one independent owned value per row.
+//!
+//! [`execute_owned`] decodes inputs once, prepares constant state, writes into spare vector
+//! capacity, and reduces compact failure evidence without putting error construction in the hot
+//! loop. [`execute_owned_infallible`] removes that failure path for infallible kernels.
+
+use std::ops::BitOrAssign;
+
+use vortex_compute::lane_kernels::IndexedSourceExt;
+use vortex_error::VortexResult;
+use vortex_error::vortex_ensure;
+
+use super::RowExecution;
+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.
+#[derive(Clone, Copy, Default)]
+struct NoFailure;
+
+impl BitOrAssign for NoFailure {
+    fn bitor_assign(&mut self, _rhs: Self) {}
+}
+
+/// Decode every input column, then store one output per row from an infallible kernel.
+pub(crate) fn execute_owned_infallible<Args, Out, Prepared>(
+    args: &dyn ExecutionArgs,
+    ctx: &mut ExecutionCtx,
+    prepare: impl FnOnce(Args::ConstElems<'_>) -> Prepared,
+    apply: impl Fn(&Prepared, Args::Elems<'_>) -> Out,
+) -> VortexResult<RowExecution>
+where
+    Args: IndexedElementTuple,
+    Out: OutputElement,
+{
+    execute_owned::<Args, Out, Prepared, NoFailure>(
+        args,
+        ctx,
+        prepare,
+        move |prepared, args| (apply(prepared, args), NoFailure),
+        |_| Ok(()),
+    )
+}
+
+/// Decode every input column, then store outputs and combine per-row failure evidence.
+pub(crate) fn execute_owned<Args, Out, Prepared, Fail>(
+    args: &dyn ExecutionArgs,
+    ctx: &mut ExecutionCtx,
+    prepare: impl FnOnce(Args::ConstElems<'_>) -> Prepared,
+    apply: impl Fn(&Prepared, Args::Elems<'_>) -> (Out, Fail),
+    finish_failure: impl FnOnce(Fail) -> VortexResult<()>,
+) -> VortexResult<RowExecution>
+where
+    Args: IndexedElementTuple,
+    Out: OutputElement,
+    Fail: FailureEvidence,
+{
+    // The output vector stays at length zero until every slot is initialized so that an unwind
+    // abandons partially initialized spare capacity. This no-drop assertion proves that no
+    // initialized value requires a destructor to run.
+    const { assert_owned_output_needs_no_drop::<Out>() };
+
+    let columns = Args::decode(args, ctx)?;
+    let prepared = prepare(Args::const_values(&columns));
+
+    let row_count = args.row_count();
+    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::ARITY == 0 || views.len() == 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.
+        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
+    };
+
+    // SAFETY: normal completion of either execution path initializes `0..row_count` exactly
+    // 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)),
+    }
+}