diff --git a/vortex-array/benches/row_fn_output.rs b/vortex-array/benches/row_fn_output.rs index 6ba02b6df8c..1bf15764ff0 100644 --- a/vortex-array/benches/row_fn_output.rs +++ b/vortex-array/benches/row_fn_output.rs @@ -92,6 +92,7 @@ impl RowFn for InfallibleBool { 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"); diff --git a/vortex-array/src/scalar_fn/fns/binary/numeric/row.rs b/vortex-array/src/scalar_fn/fns/binary/numeric/row.rs index ff7fbe7ab81..9baaa987fc2 100644 --- a/vortex-array/src/scalar_fn/fns/binary/numeric/row.rs +++ b/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 diff --git a/vortex-array/src/scalar_fn/unstable/row/batch/tests.rs b/vortex-array/src/scalar_fn/unstable/row/batch/tests.rs index a9f298de52a..0c651be8feb 100644 --- a/vortex-array/src/scalar_fn/unstable/row/batch/tests.rs +++ b/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; + type View<'a> = &'a [i64]; + type Elem<'a> = i64; + + const DENSE_SAFE: bool = true; + const DECODE_INFALLIBLE: bool = false; + + fn validate(dtype: &DType) -> VortexResult<()> { + ::validate(dtype) + } + + fn decode(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult { + let valid = array.validity()?.execute_mask(array.len(), ctx)?; + let values = ::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 { + ::get(column, index) + } + + fn view(column: &Self::Column) -> Self::View<'_> { + ::view(column) + } + + fn get_from_view<'a>(view: &Self::View<'a>, index: usize) -> i64 + where + Self: 'a, + { + ::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 { ::get_from_view_unchecked(view, index) } + } +} + +#[derive(Clone, Default)] +struct DecodeFallibleIdentity { + prepare_count: Arc, + apply_count: Arc, +} + +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( + &self, + _options: &Self::Options, + _args: &[DType], + visitor: V, + ) -> VortexResult { + 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"); diff --git a/vortex-array/src/scalar_fn/unstable/row/row_fn.rs b/vortex-array/src/scalar_fn/unstable/row/row_fn.rs index edf2f9371d0..e279a5f0c3c 100644 --- a/vortex-array/src/scalar_fn/unstable/row/row_fn.rs +++ b/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; diff --git a/vortex-array/src/scalar_fn/unstable/row/types/element/input.rs b/vortex-array/src/scalar_fn/unstable/row/types/element/input.rs index a9e1c09ed24..b54fe4c516e 100644 --- a/vortex-array/src/scalar_fn/unstable/row/types/element/input.rs +++ b/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. diff --git a/vortex-array/src/scalar_fn/unstable/row/visitor/check.rs b/vortex-array/src/scalar_fn/unstable/row/visitor/check.rs index f73325f285d..f619d00ed9f 100644 --- a/vortex-array/src/scalar_fn/unstable/row/visitor/check.rs +++ b/vortex-array/src/scalar_fn/unstable/row/visitor/check.rs @@ -33,6 +33,10 @@ const fn assert_input_visit_contract() { 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() diff --git a/vortex-array/src/scalar_fn/unstable/row/visitor/plan.rs b/vortex-array/src/scalar_fn/unstable/row/visitor/plan.rs index d7c72ea4561..c30644603e9 100644 --- a/vortex-array/src/scalar_fn/unstable/row/visitor/plan.rs +++ b/vortex-array/src/scalar_fn/unstable/row/visitor/plan.rs @@ -292,7 +292,7 @@ pub(crate) enum RowPolicy { impl RowPolicy { /// The policy for an infallible owned output. pub(crate) const fn for_owned_output() -> Self { - if Args::DENSE_SAFE && Args::DECODE_INFALLIBLE { + if Args::DENSE_SAFE { Self::Dense } else { Self::ValidOnly @@ -301,7 +301,7 @@ impl RowPolicy { /// The policy for an owned output carrying batch-deferred failure evidence. pub(crate) const fn for_deferred_output() -> Self { - if Args::DENSE_SAFE && Args::DECODE_INFALLIBLE { + if Args::DENSE_SAFE { Self::DenseWithRetry } else { Self::ValidOnly @@ -310,7 +310,7 @@ impl RowPolicy { /// The policy for a sink-writing output. pub(crate) const fn for_sink() -> Self { - if Args::DENSE_SAFE && Args::DECODE_INFALLIBLE && ApplyResult::INFALLIBLE { + if Args::DENSE_SAFE && ApplyResult::INFALLIBLE { Self::Dense } else { Self::ValidOnly diff --git a/vortex-array/src/scalar_fn/unstable/row/vtable.rs b/vortex-array/src/scalar_fn/unstable/row/vtable.rs index 2b483d6109b..c900b261c24 100644 --- a/vortex-array/src/scalar_fn/unstable/row/vtable.rs +++ b/vortex-array/src/scalar_fn/unstable/row/vtable.rs @@ -84,7 +84,7 @@ impl ScalarFnVTable for F { } fn is_infallible(&self, _options: &Self::Options) -> bool { - F::INFALLIBLE + F::INFALLIBLE && F::DECODE_INFALLIBLE } } @@ -231,6 +231,7 @@ mod tests { use crate::dtype::DType; 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::RowFn; use crate::scalar_fn::unstable::row::RowVisitor; @@ -259,6 +260,7 @@ mod tests { 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_seven"); @@ -281,6 +283,7 @@ mod tests { 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.indexing_row_fn"); @@ -304,6 +307,7 @@ mod tests { 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.changing_dispatch_row_fn"); @@ -328,6 +332,11 @@ mod tests { } } + #[test] + fn test_infallible_row_fn_advertises_both_guarantees() { + assert!(ScalarFnVTable::is_infallible(&IndexingRowFn, &EmptyOptions)); + } + #[test] fn test_return_dtype_rejects_wrong_arity_before_dispatch() { let error = row_fn_return_dtype(&IndexingRowFn, &EmptyOptions, &[]) diff --git a/vortex-spatial/src/scalar_fn/distance.rs b/vortex-spatial/src/scalar_fn/distance.rs index c78c8f89262..8b8a59f970a 100644 --- a/vortex-spatial/src/scalar_fn/distance.rs +++ b/vortex-spatial/src/scalar_fn/distance.rs @@ -41,6 +41,9 @@ impl RowFn for SpatialDistance { const ARG_NAMES: &'static [&'static str] = &["a", "b"]; const INFALLIBLE: bool = true; + // GeometryRow decoding can return a data-dependent error. + const DECODE_INFALLIBLE: bool = false; + fn id(&self) -> ScalarFnId { static ID: CachedId = CachedId::new("vortex.st.distance"); *ID @@ -202,6 +205,11 @@ mod tests { Ok(()) } + #[test] + fn distance_does_not_advertise_speculative_infallibility() { + assert!(!SpatialDistance.is_infallible(&EmptyOptions)); + } + /// A null row in a geometry operand yields a null result; valid rows are unaffected. #[test] fn distance_propagates_null_rows() -> VortexResult<()> {