Define experimental RowFn contracts

Signed-off-by: Connor Tsui <connor.tsui20@gmail.com>

Author: connortsui20

vortex-array/Cargo.toml

@@ -79,6 +79,8 @@ cudarc = ["dep:cudarc"]
 table-display = ["dep:tabled"]
 _test-harness = ["dep:goldenfile", "dep:rstest", "dep:rstest_reuse"]
 serde = ["dep:serde", "vortex-buffer/serde", "vortex-mask/serde"]
+# Exposes experimental row-function APIs without compatibility guarantees.
+unstable_row_fns = []
 
 [dev-dependencies]
 divan = { workspace = true }
@@ -90,7 +92,11 @@ rstest = { workspace = true }
 serde_json = { workspace = true }
 serde_test = { workspace = true }
 test-with = { workspace = true }
-vortex-array = { path = ".", features = ["_test-harness", "table-display"] }
+vortex-array = { path = ".", features = [
+    "_test-harness",
+    "table-display",
+    "unstable_row_fns",
+] }
 
 [[bench]]
 name = "aggregate_max"

vortex-array/src/scalar_fn/mod.rs

@@ -6,6 +6,10 @@
 //! This module contains the [`ScalarFnVTable`] trait and all built-in scalar function
 //! implementations. Expressions ([`crate::expr::Expression`]) reference scalar functions
 //! at each node.
+//!
+//! [`unstable::row`] defines experimental contracts for strict functions whose natural kernel
+//! computes one row at a time. This API requires the `unstable_row_fns` feature and has no
+//! compatibility guarantees.
 
 use vortex_session::registry::Id;
 
@@ -35,6 +39,12 @@ pub use options::*;
 mod signature;
 pub use signature::*;
 
+#[cfg(feature = "unstable_row_fns")]
+pub mod unstable;
+#[cfg(not(feature = "unstable_row_fns"))]
+#[allow(dead_code, unused_imports)]
+pub(crate) mod unstable;
+
 pub mod fns;
 pub mod internal;
 pub mod session;

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

@@ -0,0 +1,9 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Experimental scalar-function APIs without compatibility guarantees.
+//!
+//! These APIs can change or disappear without a deprecation period. External users must enable
+//! the corresponding `unstable_*` Cargo feature before importing them.
+
+pub mod row;

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

@@ -0,0 +1,7 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Outcomes returned by encoding-aware and row-oriented kernels.
+
+mod outcome;
+pub use outcome::RowExecution;

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

@@ -0,0 +1,39 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+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 may evaluate values behind nulls. Its deferred error is therefore not necessarily
+/// observable: batch execution can retry over only valid rows, suppressing an error that came from
+/// a null row while preserving one from a valid row. A plain `VortexResult<ArrayRef>` would lose
+/// the distinction between that retryable error and an error for which retrying cannot help.
+///
+/// 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/mod.rs

@@ -0,0 +1,33 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Scalar functions computed one row at a time.
+//!
+//! This module is experimental and has no compatibility guarantees. External users must enable
+//! the `unstable_row_fns` Cargo feature before importing it.
+//!
+//! Start with [`RowFn`] to define an operation and [`RowVisitor`] to select one of its output
+//! capabilities. [`InputElement`] and [`ElementTuple`] describe how columns become row values.
+//! [`OutputElement`] covers independent owned values, while [`OutputSink`] supports outputs that
+//! need row handles or shared batch state. [`SinkResult`] describes how a sink-writing closure
+//! reports errors. [`RowExecution`] represents the outcome of an encoding-aware or row-oriented
+//! kernel before an execution backend resolves deferred failures.
+
+mod execute;
+pub use execute::RowExecution;
+
+mod row_fn;
+pub use row_fn::RowFn;
+
+mod types;
+pub use types::ElementTuple;
+pub use types::IndexedElementTuple;
+pub use types::InitializedElement;
+pub use types::InputElement;
+pub use types::OutputElement;
+pub use types::OutputSink;
+pub use types::SinkResult;
+pub use types::UninitElementSink;
+
+mod visitor;
+pub use visitor::RowVisitor;

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

@@ -0,0 +1,105 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Scalar functions computed one row at a time.
+
+use std::fmt::Debug;
+use std::fmt::Display;
+use std::hash::Hash;
+
+use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
+use vortex_session::VortexSession;
+
+use super::visitor::RowVisitor;
+use crate::ArrayRef;
+use crate::ExecutionCtx;
+use crate::dtype::DType;
+use crate::scalar_fn::ScalarFnId;
+use crate::scalar_fn::unstable::row::RowExecution;
+
+/// A scalar function computed one row at a time.
+///
+/// Declare the argument names and use [`dispatch`](Self::dispatch) to choose concrete element and
+/// sink types for each accepted dtype combination.
+///
+/// An execution backend can implement the sealed [`RowVisitor`] contract to plan and execute the
+/// concrete signature selected by [`dispatch`](Self::dispatch).
+pub trait RowFn: 'static + Sized + Clone + Send + Sync {
+    /// Options for this function, or [`EmptyOptions`](crate::scalar_fn::EmptyOptions) for none.
+    type Options: 'static + Send + Sync + Clone + Debug + Display + PartialEq + Eq + Hash;
+
+    /// The arguments in display order. Its length is the function's exact arity.
+    const ARG_NAMES: &'static [&'static str];
+
+    /// Whether any legal dispatch or encoding-aware execution can raise a semantic error.
+    ///
+    /// The framework checks this at compile time for every fallible dispatched element or result.
+    /// A conservative `true` is allowed when only some dtype choices are fallible.
+    ///
+    /// Set this to `true` when [`reduce_encoded`](Self::reduce_encoded) can return a semantic
+    /// error or [`RowExecution::DeferredError`]. Compile-time checks cover dispatched element and
+    /// result types, but cannot inspect that hook.
+    ///
+    /// Semantic errors are defined by
+    /// [`ScalarFnVTable::is_fallible`](crate::scalar_fn::ScalarFnVTable::is_fallible).
+    const FALLIBLE: bool = false;
+
+    /// Returns the ID of the scalar function.
+    fn id(&self) -> ScalarFnId;
+
+    /// Serialize this function's options, or return `None` when the function is not serializable.
+    fn serialize(&self, options: &Self::Options) -> VortexResult<Option<Vec<u8>>> {
+        _ = options;
+        Ok(None)
+    }
+
+    /// Restore options written by [`serialize`](Self::serialize).
+    fn deserialize(
+        &self,
+        _metadata: &[u8],
+        _session: &VortexSession,
+    ) -> VortexResult<Self::Options> {
+        vortex_bail!("Expression {} is not deserializable", self.id())
+    }
+
+    /// Choose element types for these input dtypes and visit the framework with them.
+    ///
+    /// Plan time and run time both call this method, so the choice **must** be a pure function of
+    /// `options` and `args`. Cross-argument dtype validation belongs here.
+    fn dispatch<V: RowVisitor<Self::Options>>(
+        &self,
+        options: &Self::Options,
+        args: &[DType],
+        visitor: V,
+    ) -> VortexResult<V::VisitResult>;
+
+    /// Try an encoding-aware implementation before decoding the inputs into row elements.
+    ///
+    /// `None` continues to the dispatched row loop. [`Output`](RowExecution::Output) skips that
+    /// loop and may remain encoded or lazy. [`DeferredError`](RowExecution::DeferredError) reruns
+    /// only valid rows when null payloads may have caused the failure. For non-nullary functions,
+    /// batch execution calls this hook at most once with the original, unfiltered arrays; slices
+    /// and compacted retries do not reach it.
+    ///
+    /// Like a dense row closure, this hook must be total over every stored payload, including
+    /// payloads behind null rows. An `Err` is immediately user-visible and is never suppressed or
+    /// retried through the row layer.
+    ///
+    /// # Requirements
+    ///
+    /// - `output.len()` **must** equal `args[0].len()`.
+    /// - The output dtype **must** match the planned dtype when ignoring nullability.
+    /// - The output **must not** introduce a null where every input is valid.
+    ///
+    /// The framework skips this hook for nullary functions.
+    fn reduce_encoded(
+        &self,
+        options: &Self::Options,
+        args: &[ArrayRef],
+        ctx: &mut ExecutionCtx,
+    ) -> VortexResult<Option<RowExecution>> {
+        _ = (options, args, ctx);
+        Ok(None)
+    }
+}

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

@@ -0,0 +1,77 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_buffer::BitBuffer;
+use vortex_error::VortexResult;
+use vortex_error::vortex_ensure;
+
+use crate::ArrayRef;
+use crate::ExecutionCtx;
+use crate::IntoArray;
+use crate::arrays::BoolArray;
+use crate::dtype::DType;
+use crate::dtype::Nullability;
+use crate::scalar_fn::unstable::row::InputElement;
+use crate::scalar_fn::unstable::row::OutputElement;
+use crate::validity::Validity;
+
+// SAFETY: the per-row view is a bit buffer, and its reported length is the buffer length.
+unsafe impl InputElement for bool {
+    type Column = BitBuffer;
+    type View<'a> = &'a BitBuffer;
+    type Elem<'a> = bool;
+
+    // Every bit of the buffer is readable, valid or not.
+    const DENSE_SAFE: bool = true;
+    const DECODE_FALLIBLE: bool = false;
+
+    fn validate(dtype: &DType) -> VortexResult<()> {
+        vortex_ensure!(
+            matches!(dtype, DType::Bool(_)),
+            "expected a Bool column, got {dtype}",
+        );
+        Ok(())
+    }
+
+    fn decode(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self::Column> {
+        Ok(array.execute::<BoolArray>(ctx)?.into_bit_buffer())
+    }
+
+    fn get(column: &Self::Column, index: usize) -> bool {
+        column.value(index)
+    }
+
+    fn view(column: &Self::Column) -> Self::View<'_> {
+        column
+    }
+
+    fn view_len(view: &Self::View<'_>) -> usize {
+        view.len()
+    }
+
+    fn get_from_view<'a>(view: &Self::View<'a>, index: usize) -> bool
+    where
+        Self: 'a,
+    {
+        view.value(index)
+    }
+
+    unsafe fn get_from_view_unchecked<'a>(view: &Self::View<'a>, index: usize) -> bool
+    where
+        Self: 'a,
+    {
+        // SAFETY: forwarded from this method's contract.
+        unsafe { view.value_unchecked(index) }
+    }
+}
+
+impl OutputElement for bool {
+    fn element_dtype() -> DType {
+        DType::Bool(Nullability::NonNullable)
+    }
+
+    fn build(values: Vec<Self>) -> ArrayRef {
+        // `From<Vec<bool>>` uses the bulk bit-packing path.
+        BoolArray::new(BitBuffer::from(values), Validity::NonNullable).into_array()
+    }
+}