Separate RowFn decode and semantic infallibility

vortex-array/benches/row_fn_output.rs

@@ -92,6 +92,7 @@ impl<T: BenchPrimitive> RowFn for InfallibleBool<T> {
 
     const ARG_NAMES: &'static [&'static str] = &["lhs", "rhs"];
     const INFALLIBLE: bool = true;
+    const DECODE_INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("bench.row_fn_output.infallible_bool");

vortex-array/src/scalar_fn/fns/binary/numeric/row.rs

@@ -55,6 +55,7 @@ impl RowFn for NumericBinary {
 
     // Fallibility is queried without input dtypes, so this conservatively covers integer widths.
     const INFALLIBLE: bool = false;
+    const DECODE_INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
         // `NumericBinary` is a private implementation detail of `Binary`: it is never registered or

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

@@ -35,6 +35,7 @@ use crate::extension::datetime::Timestamp;
 use crate::scalar::Scalar;
 use crate::scalar_fn::EmptyOptions;
 use crate::scalar_fn::ScalarFnId;
+use crate::scalar_fn::ScalarFnVTable;
 use crate::scalar_fn::VecExecutionArgs;
 use crate::scalar_fn::unstable::row::FixedSizeListSink;
 use crate::scalar_fn::unstable::row::InitializedRow;
@@ -192,6 +193,113 @@ unsafe impl InputElement for DenseRetryI64 {
     }
 }
 
+const REJECTED_DECODE_VALUE: i64 = i64::MIN;
+
+struct DecodeFallibleI64;
+
+// SAFETY: the view and unchecked access delegate to the `i64` implementation. Decoding inspects
+// only valid rows, so null-row payloads cannot cause its data-dependent error.
+unsafe impl InputElement for DecodeFallibleI64 {
+    type Column = Buffer<i64>;
+    type View<'a> = &'a [i64];
+    type Elem<'a> = i64;
+
+    const DENSE_SAFE: bool = true;
+    const DECODE_INFALLIBLE: bool = false;
+
+    fn validate(dtype: &DType) -> VortexResult<()> {
+        <i64 as InputElement>::validate(dtype)
+    }
+
+    fn decode(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self::Column> {
+        let valid = array.validity()?.execute_mask(array.len(), ctx)?;
+        let values = <i64 as InputElement>::decode(array, ctx)?;
+
+        if valid
+            .to_bit_buffer()
+            .iter()
+            .zip(values.iter())
+            .any(|(is_valid, value)| is_valid && *value == REJECTED_DECODE_VALUE)
+        {
+            vortex_bail!(InvalidArgument: "test decoder rejected a valid value");
+        }
+
+        Ok(values)
+    }
+
+    fn get(column: &Self::Column, index: usize) -> i64 {
+        <i64 as InputElement>::get(column, index)
+    }
+
+    fn view(column: &Self::Column) -> Self::View<'_> {
+        <i64 as InputElement>::view(column)
+    }
+
+    fn get_from_view<'a>(view: &Self::View<'a>, index: usize) -> i64
+    where
+        Self: 'a,
+    {
+        <i64 as InputElement>::get_from_view(view, index)
+    }
+
+    unsafe fn get_from_view_unchecked<'a>(view: &Self::View<'a>, index: usize) -> i64
+    where
+        Self: 'a,
+    {
+        // SAFETY: forwarded from this method's contract.
+        unsafe { <i64 as InputElement>::get_from_view_unchecked(view, index) }
+    }
+}
+
+#[derive(Clone, Default)]
+struct DecodeFallibleIdentity {
+    prepare_count: Arc<AtomicUsize>,
+    apply_count: Arc<AtomicUsize>,
+}
+
+impl DecodeFallibleIdentity {
+    fn prepare_count(&self) -> usize {
+        self.prepare_count.load(Ordering::Relaxed)
+    }
+
+    fn apply_count(&self) -> usize {
+        self.apply_count.load(Ordering::Relaxed)
+    }
+}
+
+impl RowFn for DecodeFallibleIdentity {
+    type Options = EmptyOptions;
+
+    const ARG_NAMES: &'static [&'static str] = &["value"];
+    const INFALLIBLE: bool = true;
+    const DECODE_INFALLIBLE: bool = false;
+
+    fn id(&self) -> ScalarFnId {
+        static ID: CachedId = CachedId::new("test.decode_fallible_identity");
+        *ID
+    }
+
+    fn dispatch<V: RowVisitor>(
+        &self,
+        _options: &Self::Options,
+        _args: &[DType],
+        visitor: V,
+    ) -> VortexResult<V::VisitResult> {
+        let prepare_count = Arc::clone(&self.prepare_count);
+        let apply_count = Arc::clone(&self.apply_count);
+
+        visitor.visit_prepared::<(DecodeFallibleI64,), i64, _>(
+            move |_| {
+                prepare_count.fetch_add(1, Ordering::Relaxed);
+            },
+            move |&(), (value,)| {
+                apply_count.fetch_add(1, Ordering::Relaxed);
+                value
+            },
+        )
+    }
+}
+
 /// Produces a null row to exercise output validation at the row-function boundary.
 #[derive(Default)]
 struct NullProducingI64(i64);
@@ -249,6 +357,7 @@ impl RowFn for RepeatValue {
 
     const ARG_NAMES: &'static [&'static str] = &["value"];
     const INFALLIBLE: bool = true;
+    const DECODE_INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("test.repeat_value");
@@ -278,6 +387,7 @@ impl RowFn for DeferredAdd {
 
     const ARG_NAMES: &'static [&'static str] = &["lhs", "rhs"];
     const INFALLIBLE: bool = false;
+    const DECODE_INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("test.deferred_add");
@@ -313,6 +423,7 @@ impl RowFn for ValidOnlyIdentity {
 
     const ARG_NAMES: &'static [&'static str] = &["value"];
     const INFALLIBLE: bool = false;
+    const DECODE_INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("test.valid_only_identity");
@@ -337,6 +448,7 @@ impl RowFn for FilterAndScatterIdentity {
 
     const ARG_NAMES: &'static [&'static str] = &["value"];
     const INFALLIBLE: bool = false;
+    const DECODE_INFALLIBLE: bool = false;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("test.filter_and_scatter_identity");
@@ -358,6 +470,7 @@ impl RowFn for DenseRetryIncrement {
 
     const ARG_NAMES: &'static [&'static str] = &["value"];
     const INFALLIBLE: bool = false;
+    const DECODE_INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("test.dense_retry_increment");
@@ -388,6 +501,7 @@ impl RowFn for FilterAndScatterRepeat {
 
     const ARG_NAMES: &'static [&'static str] = &["value"];
     const INFALLIBLE: bool = true;
+    const DECODE_INFALLIBLE: bool = false;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("test.filter_and_scatter_repeat");
@@ -418,6 +532,7 @@ impl RowFn for InvalidKernelOutput {
 
     const ARG_NAMES: &'static [&'static str] = &["value"];
     const INFALLIBLE: bool = true;
+    const DECODE_INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("test.invalid_kernel_output");
@@ -434,6 +549,55 @@ impl RowFn for InvalidKernelOutput {
     }
 }
 
+#[test]
+fn test_decode_fallibility_disables_scalar_fn_infallibility_not_dense_execution() -> VortexResult<()>
+{
+    let function = DecodeFallibleIdentity::default();
+    let input = PrimitiveArray::new(
+        vec![
+            1_i64,                 // valid
+            REJECTED_DECODE_VALUE, // null
+            3,                     // valid
+        ],
+        Validity::from_iter([true, false, true]),
+    )
+    .into_array();
+    let expected = input.clone();
+    let args = VecExecutionArgs::new(vec![input], 3);
+    let mut ctx = array_session().create_execution_ctx();
+
+    let actual = execute_rows(&function, &EmptyOptions, &args, &mut ctx)?;
+
+    assert_arrays_eq!(actual, expected, &mut ctx);
+    assert_eq!(function.prepare_count(), 1);
+    assert_eq!(function.apply_count(), 3);
+    assert!(!ScalarFnVTable::is_infallible(&function, &EmptyOptions));
+    Ok(())
+}
+
+#[test]
+fn test_decode_error_precedes_prepare_and_apply() -> VortexResult<()> {
+    let function = DecodeFallibleIdentity::default();
+    let input = PrimitiveArray::from_iter([REJECTED_DECODE_VALUE]).into_array();
+    let args = VecExecutionArgs::new(vec![input], 1);
+    let mut ctx = array_session().create_execution_ctx();
+
+    let error = match execute_rows(&function, &EmptyOptions, &args, &mut ctx) {
+        Err(error) => error,
+        Ok(_) => vortex_bail!("a rejected decoded value must fail execution"),
+    };
+
+    assert!(
+        error
+            .to_string()
+            .contains("test decoder rejected a valid value"),
+        "unexpected error: {error}",
+    );
+    assert_eq!(function.prepare_count(), 0);
+    assert_eq!(function.apply_count(), 0);
+    Ok(())
+}
+
 #[rstest]
 #[case::dense_width_two(
     vec![1_i64, 2],
@@ -491,6 +655,7 @@ impl RowFn for PackedPositive {
 
     const ARG_NAMES: &'static [&'static str] = &["value"];
     const INFALLIBLE: bool = true;
+    const DECODE_INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("test.packed_positive");
@@ -512,6 +677,7 @@ impl RowFn for PackedGreaterThan {
 
     const ARG_NAMES: &'static [&'static str] = &["lhs", "rhs"];
     const INFALLIBLE: bool = true;
+    const DECODE_INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("test.packed_greater_than");
@@ -533,6 +699,7 @@ impl RowFn for ValidOnlyPositive {
 
     const ARG_NAMES: &'static [&'static str] = &["value"];
     const INFALLIBLE: bool = true;
+    const DECODE_INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("test.valid_only_positive");
@@ -557,6 +724,7 @@ impl RowFn for NullaryTrue {
 
     const ARG_NAMES: &'static [&'static str] = &[];
     const INFALLIBLE: bool = true;
+    const DECODE_INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("test.nullary_true");
@@ -903,6 +1071,7 @@ impl RowFn for DeclaredOutput {
 
     const ARG_NAMES: &'static [&'static str] = &["value"];
     const INFALLIBLE: bool = true;
+    const DECODE_INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("test.declared_output");
@@ -933,6 +1102,7 @@ impl RowFn for DeclaredSinkOutput {
 
     const ARG_NAMES: &'static [&'static str] = &["value"];
     const INFALLIBLE: bool = false;
+    const DECODE_INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("test.declared_sink_output");
@@ -965,6 +1135,7 @@ impl RowFn for ChangingOutputDType {
 
     const ARG_NAMES: &'static [&'static str] = &["value"];
     const INFALLIBLE: bool = true;
+    const DECODE_INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("test.changing_output_dtype");
@@ -1125,6 +1296,7 @@ impl RowFn for MismatchedStorage {
 
     const ARG_NAMES: &'static [&'static str] = &["value"];
     const INFALLIBLE: bool = true;
+    const DECODE_INFALLIBLE: bool = true;
 
     fn id(&self) -> ScalarFnId {
         static ID: CachedId = CachedId::new("test.mismatched_storage");

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

@@ -3,9 +3,9 @@
 
 //! The [`RowFn`] contract for scalar functions whose natural kernel computes one row at a time.
 //!
-//! Implementations declare their arity and fallibility, then use [`RowFn::dispatch`] to select the
-//! typed row signature for each supported dtype combination. Optional methods provide
-//! serialization without putting persistence plumbing in the row kernel.
+//! Implementations declare their arity, semantic fallibility, and decode fallibility, then use
+//! [`RowFn::dispatch`] to select the typed row signature for each supported dtype combination.
+//! Optional methods provide serialization without putting persistence plumbing in the row kernel.
 
 use std::fmt::Debug;
 use std::fmt::Display;
@@ -51,15 +51,36 @@ pub trait RowFn: 'static + Sized + Clone + Send + Sync {
     /// infallible.
     ///
     /// Input decoding declares its fallibility independently through
-    /// [`InputElement::DECODE_INFALLIBLE`] and does not affect this flag. The framework checks
-    /// row-result fallibility. A conservative `false` is allowed.
+    /// [`DECODE_INFALLIBLE`](Self::DECODE_INFALLIBLE) and does not affect this flag. The framework
+    /// checks row-result fallibility. A conservative `false` is allowed.
     ///
     /// See [`ScalarFnVTable::is_infallible`] for the definition of semantic errors.
     ///
-    /// [`InputElement::DECODE_INFALLIBLE`]: crate::scalar_fn::unstable::row::InputElement::DECODE_INFALLIBLE
     /// [`ScalarFnVTable::is_infallible`]: crate::scalar_fn::ScalarFnVTable::is_infallible
     const INFALLIBLE: bool;
 
+    /// Whether every input element selected by [`dispatch`](Self::dispatch) declares
+    /// [`InputElement::DECODE_INFALLIBLE`].
+    ///
+    /// This describes conversion into the Rust-native column representation. It is independent of
+    /// [`INFALLIBLE`](Self::INFALLIBLE), which describes the row operation, and
+    /// [`InputElement::DENSE_SAFE`], which describes whether dense execution can tolerate null-row
+    /// payloads.
+    ///
+    /// This function-wide summary is necessary because [`ScalarFnVTable::is_infallible`] receives
+    /// no argument dtypes and cannot inspect the tuple selected by [`dispatch`](Self::dispatch).
+    /// The framework checks every selected tuple against a positive declaration. A conservative
+    /// `false` is allowed.
+    ///
+    /// The blanket [`ScalarFnVTable`] advertises infallibility only when both this constant and
+    /// [`INFALLIBLE`](Self::INFALLIBLE) are `true`.
+    ///
+    /// [`InputElement::DECODE_INFALLIBLE`]: crate::scalar_fn::unstable::row::InputElement::DECODE_INFALLIBLE
+    /// [`InputElement::DENSE_SAFE`]: crate::scalar_fn::unstable::row::InputElement::DENSE_SAFE
+    /// [`ScalarFnVTable`]: crate::scalar_fn::ScalarFnVTable
+    /// [`ScalarFnVTable::is_infallible`]: crate::scalar_fn::ScalarFnVTable::is_infallible
+    const DECODE_INFALLIBLE: bool;
+
     /// Returns the ID of the scalar function.
     fn id(&self) -> ScalarFnId;
 

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

@@ -44,12 +44,12 @@ pub unsafe trait InputElement: 'static {
     /// null rows to the row closure.
     const DENSE_SAFE: bool;
 
-    /// Whether [`decode`](Self::decode) is infallible for _legal_ input data.
+    /// Whether [`decode`](Self::decode) is guaranteed not to reject legal, well-typed values while
+    /// constructing the Rust-native [`Column`](Self::Column) representation.
     ///
-    /// This excludes infrastructural failures such as IO or allocation.
-    /// It is independent of
+    /// This excludes infrastructural failures such as IO or allocation. It is independent of
     /// [`RowFn::INFALLIBLE`](crate::scalar_fn::unstable::row::RowFn::INFALLIBLE), which describes
-    /// the row operation rather than input decoding.
+    /// the row operation, and [`DENSE_SAFE`](Self::DENSE_SAFE), which describes null-row payloads.
     const DECODE_INFALLIBLE: bool;
 
     /// Validate that `dtype` is an acceptable input column dtype for this element type.

vortex-array/src/scalar_fn/unstable/row/visitor/check.rs

@@ -33,6 +33,10 @@ const fn assert_input_visit_contract<F: RowFn, Args: ElementTuple>() {
         Args::ARITY == F::ARG_NAMES.len(),
         "the visited argument tuple must have the arity declared by RowFn::ARG_NAMES",
     );
+    assert!(
+        !F::DECODE_INFALLIBLE || Args::DECODE_INFALLIBLE,
+        "RowFn::DECODE_INFALLIBLE must be false when an input decoder can fail",
+    );
 }
 
 pub(super) const fn assert_owned_visit_contract<Function, Args, Out>()