Skip to content

Commit d49a7be

Browse files
connortsui20connortsui20
committed
Separate RowFn decode and semantic infallibility
Signed-off-by: Connor Tsui <[email protected]>
1 parent 42a5d3b commit d49a7be

9 files changed

Lines changed: 230 additions & 14 deletions

File tree

‎vortex-array/benches/row_fn_output.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,7 @@ impl<T: BenchPrimitive> RowFn for InfallibleBool<T> {
9292

9393
const ARG_NAMES: &'static [&'static str] = &["lhs", "rhs"];
9494
const INFALLIBLE: bool = true;
95+
const DECODE_INFALLIBLE: bool = true;
9596

9697
fn id(&self) -> ScalarFnId {
9798
static ID: CachedId = CachedId::new("bench.row_fn_output.infallible_bool");

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@ impl RowFn for NumericBinary {
5555

5656
// Fallibility is queried without input dtypes, so this conservatively covers integer widths.
5757
const INFALLIBLE: bool = false;
58+
const DECODE_INFALLIBLE: bool = true;
5859

5960
fn id(&self) -> ScalarFnId {
6061
// `NumericBinary` is a private implementation detail of `Binary`: it is never registered or

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

Lines changed: 172 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@ use crate::extension::datetime::Timestamp;
3535
use crate::scalar::Scalar;
3636
use crate::scalar_fn::EmptyOptions;
3737
use crate::scalar_fn::ScalarFnId;
38+
use crate::scalar_fn::ScalarFnVTable;
3839
use crate::scalar_fn::VecExecutionArgs;
3940
use crate::scalar_fn::unstable::row::FixedSizeListSink;
4041
use crate::scalar_fn::unstable::row::InitializedRow;
@@ -192,6 +193,113 @@ unsafe impl InputElement for DenseRetryI64 {
192193
}
193194
}
194195

196+
const REJECTED_DECODE_VALUE: i64 = i64::MIN;
197+
198+
struct DecodeFallibleI64;
199+
200+
// SAFETY: the view and unchecked access delegate to the `i64` implementation. Decoding inspects
201+
// only valid rows, so null-row payloads cannot cause its data-dependent error.
202+
unsafe impl InputElement for DecodeFallibleI64 {
203+
type Column = Buffer<i64>;
204+
type View<'a> = &'a [i64];
205+
type Elem<'a> = i64;
206+
207+
const DENSE_SAFE: bool = true;
208+
const DECODE_INFALLIBLE: bool = false;
209+
210+
fn validate(dtype: &DType) -> VortexResult<()> {
211+
<i64 as InputElement>::validate(dtype)
212+
}
213+
214+
fn decode(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self::Column> {
215+
let valid = array.validity()?.execute_mask(array.len(), ctx)?;
216+
let values = <i64 as InputElement>::decode(array, ctx)?;
217+
218+
if valid
219+
.to_bit_buffer()
220+
.iter()
221+
.zip(values.iter())
222+
.any(|(is_valid, value)| is_valid && *value == REJECTED_DECODE_VALUE)
223+
{
224+
vortex_bail!(InvalidArgument: "test decoder rejected a valid value");
225+
}
226+
227+
Ok(values)
228+
}
229+
230+
fn get(column: &Self::Column, index: usize) -> i64 {
231+
<i64 as InputElement>::get(column, index)
232+
}
233+
234+
fn view(column: &Self::Column) -> Self::View<'_> {
235+
<i64 as InputElement>::view(column)
236+
}
237+
238+
fn get_from_view<'a>(view: &Self::View<'a>, index: usize) -> i64
239+
where
240+
Self: 'a,
241+
{
242+
<i64 as InputElement>::get_from_view(view, index)
243+
}
244+
245+
unsafe fn get_from_view_unchecked<'a>(view: &Self::View<'a>, index: usize) -> i64
246+
where
247+
Self: 'a,
248+
{
249+
// SAFETY: forwarded from this method's contract.
250+
unsafe { <i64 as InputElement>::get_from_view_unchecked(view, index) }
251+
}
252+
}
253+
254+
#[derive(Clone, Default)]
255+
struct DecodeFallibleIdentity {
256+
prepare_count: Arc<AtomicUsize>,
257+
apply_count: Arc<AtomicUsize>,
258+
}
259+
260+
impl DecodeFallibleIdentity {
261+
fn prepare_count(&self) -> usize {
262+
self.prepare_count.load(Ordering::Relaxed)
263+
}
264+
265+
fn apply_count(&self) -> usize {
266+
self.apply_count.load(Ordering::Relaxed)
267+
}
268+
}
269+
270+
impl RowFn for DecodeFallibleIdentity {
271+
type Options = EmptyOptions;
272+
273+
const ARG_NAMES: &'static [&'static str] = &["value"];
274+
const INFALLIBLE: bool = true;
275+
const DECODE_INFALLIBLE: bool = false;
276+
277+
fn id(&self) -> ScalarFnId {
278+
static ID: CachedId = CachedId::new("test.decode_fallible_identity");
279+
*ID
280+
}
281+
282+
fn dispatch<V: RowVisitor>(
283+
&self,
284+
_options: &Self::Options,
285+
_args: &[DType],
286+
visitor: V,
287+
) -> VortexResult<V::VisitResult> {
288+
let prepare_count = Arc::clone(&self.prepare_count);
289+
let apply_count = Arc::clone(&self.apply_count);
290+
291+
visitor.visit_prepared::<(DecodeFallibleI64,), i64, _>(
292+
move |_| {
293+
prepare_count.fetch_add(1, Ordering::Relaxed);
294+
},
295+
move |&(), (value,)| {
296+
apply_count.fetch_add(1, Ordering::Relaxed);
297+
value
298+
},
299+
)
300+
}
301+
}
302+
195303
/// Produces a null row to exercise output validation at the row-function boundary.
196304
#[derive(Default)]
197305
struct NullProducingI64(i64);
@@ -249,6 +357,7 @@ impl RowFn for RepeatValue {
249357

250358
const ARG_NAMES: &'static [&'static str] = &["value"];
251359
const INFALLIBLE: bool = true;
360+
const DECODE_INFALLIBLE: bool = true;
252361

253362
fn id(&self) -> ScalarFnId {
254363
static ID: CachedId = CachedId::new("test.repeat_value");
@@ -278,6 +387,7 @@ impl RowFn for DeferredAdd {
278387

279388
const ARG_NAMES: &'static [&'static str] = &["lhs", "rhs"];
280389
const INFALLIBLE: bool = false;
390+
const DECODE_INFALLIBLE: bool = true;
281391

282392
fn id(&self) -> ScalarFnId {
283393
static ID: CachedId = CachedId::new("test.deferred_add");
@@ -313,6 +423,7 @@ impl RowFn for ValidOnlyIdentity {
313423

314424
const ARG_NAMES: &'static [&'static str] = &["value"];
315425
const INFALLIBLE: bool = false;
426+
const DECODE_INFALLIBLE: bool = true;
316427

317428
fn id(&self) -> ScalarFnId {
318429
static ID: CachedId = CachedId::new("test.valid_only_identity");
@@ -337,6 +448,7 @@ impl RowFn for FilterAndScatterIdentity {
337448

338449
const ARG_NAMES: &'static [&'static str] = &["value"];
339450
const INFALLIBLE: bool = false;
451+
const DECODE_INFALLIBLE: bool = false;
340452

341453
fn id(&self) -> ScalarFnId {
342454
static ID: CachedId = CachedId::new("test.filter_and_scatter_identity");
@@ -358,6 +470,7 @@ impl RowFn for DenseRetryIncrement {
358470

359471
const ARG_NAMES: &'static [&'static str] = &["value"];
360472
const INFALLIBLE: bool = false;
473+
const DECODE_INFALLIBLE: bool = true;
361474

362475
fn id(&self) -> ScalarFnId {
363476
static ID: CachedId = CachedId::new("test.dense_retry_increment");
@@ -388,6 +501,7 @@ impl RowFn for FilterAndScatterRepeat {
388501

389502
const ARG_NAMES: &'static [&'static str] = &["value"];
390503
const INFALLIBLE: bool = true;
504+
const DECODE_INFALLIBLE: bool = false;
391505

392506
fn id(&self) -> ScalarFnId {
393507
static ID: CachedId = CachedId::new("test.filter_and_scatter_repeat");
@@ -418,6 +532,7 @@ impl RowFn for InvalidKernelOutput {
418532

419533
const ARG_NAMES: &'static [&'static str] = &["value"];
420534
const INFALLIBLE: bool = true;
535+
const DECODE_INFALLIBLE: bool = true;
421536

422537
fn id(&self) -> ScalarFnId {
423538
static ID: CachedId = CachedId::new("test.invalid_kernel_output");
@@ -434,6 +549,55 @@ impl RowFn for InvalidKernelOutput {
434549
}
435550
}
436551

552+
#[test]
553+
fn test_decode_fallibility_disables_scalar_fn_infallibility_not_dense_execution() -> VortexResult<()>
554+
{
555+
let function = DecodeFallibleIdentity::default();
556+
let input = PrimitiveArray::new(
557+
vec![
558+
1_i64, // valid
559+
REJECTED_DECODE_VALUE, // null
560+
3, // valid
561+
],
562+
Validity::from_iter([true, false, true]),
563+
)
564+
.into_array();
565+
let expected = input.clone();
566+
let args = VecExecutionArgs::new(vec![input], 3);
567+
let mut ctx = array_session().create_execution_ctx();
568+
569+
let actual = execute_rows(&function, &EmptyOptions, &args, &mut ctx)?;
570+
571+
assert_arrays_eq!(actual, expected, &mut ctx);
572+
assert_eq!(function.prepare_count(), 1);
573+
assert_eq!(function.apply_count(), 3);
574+
assert!(!ScalarFnVTable::is_infallible(&function, &EmptyOptions));
575+
Ok(())
576+
}
577+
578+
#[test]
579+
fn test_decode_error_precedes_prepare_and_apply() -> VortexResult<()> {
580+
let function = DecodeFallibleIdentity::default();
581+
let input = PrimitiveArray::from_iter([REJECTED_DECODE_VALUE]).into_array();
582+
let args = VecExecutionArgs::new(vec![input], 1);
583+
let mut ctx = array_session().create_execution_ctx();
584+
585+
let error = match execute_rows(&function, &EmptyOptions, &args, &mut ctx) {
586+
Err(error) => error,
587+
Ok(_) => vortex_bail!("a rejected decoded value must fail execution"),
588+
};
589+
590+
assert!(
591+
error
592+
.to_string()
593+
.contains("test decoder rejected a valid value"),
594+
"unexpected error: {error}",
595+
);
596+
assert_eq!(function.prepare_count(), 0);
597+
assert_eq!(function.apply_count(), 0);
598+
Ok(())
599+
}
600+
437601
#[rstest]
438602
#[case::dense_width_two(
439603
vec![1_i64, 2],
@@ -491,6 +655,7 @@ impl RowFn for PackedPositive {
491655

492656
const ARG_NAMES: &'static [&'static str] = &["value"];
493657
const INFALLIBLE: bool = true;
658+
const DECODE_INFALLIBLE: bool = true;
494659

495660
fn id(&self) -> ScalarFnId {
496661
static ID: CachedId = CachedId::new("test.packed_positive");
@@ -512,6 +677,7 @@ impl RowFn for PackedGreaterThan {
512677

513678
const ARG_NAMES: &'static [&'static str] = &["lhs", "rhs"];
514679
const INFALLIBLE: bool = true;
680+
const DECODE_INFALLIBLE: bool = true;
515681

516682
fn id(&self) -> ScalarFnId {
517683
static ID: CachedId = CachedId::new("test.packed_greater_than");
@@ -533,6 +699,7 @@ impl RowFn for ValidOnlyPositive {
533699

534700
const ARG_NAMES: &'static [&'static str] = &["value"];
535701
const INFALLIBLE: bool = true;
702+
const DECODE_INFALLIBLE: bool = true;
536703

537704
fn id(&self) -> ScalarFnId {
538705
static ID: CachedId = CachedId::new("test.valid_only_positive");
@@ -557,6 +724,7 @@ impl RowFn for NullaryTrue {
557724

558725
const ARG_NAMES: &'static [&'static str] = &[];
559726
const INFALLIBLE: bool = true;
727+
const DECODE_INFALLIBLE: bool = true;
560728

561729
fn id(&self) -> ScalarFnId {
562730
static ID: CachedId = CachedId::new("test.nullary_true");
@@ -903,6 +1071,7 @@ impl RowFn for DeclaredOutput {
9031071

9041072
const ARG_NAMES: &'static [&'static str] = &["value"];
9051073
const INFALLIBLE: bool = true;
1074+
const DECODE_INFALLIBLE: bool = true;
9061075

9071076
fn id(&self) -> ScalarFnId {
9081077
static ID: CachedId = CachedId::new("test.declared_output");
@@ -933,6 +1102,7 @@ impl RowFn for DeclaredSinkOutput {
9331102

9341103
const ARG_NAMES: &'static [&'static str] = &["value"];
9351104
const INFALLIBLE: bool = false;
1105+
const DECODE_INFALLIBLE: bool = true;
9361106

9371107
fn id(&self) -> ScalarFnId {
9381108
static ID: CachedId = CachedId::new("test.declared_sink_output");
@@ -965,6 +1135,7 @@ impl RowFn for ChangingOutputDType {
9651135

9661136
const ARG_NAMES: &'static [&'static str] = &["value"];
9671137
const INFALLIBLE: bool = true;
1138+
const DECODE_INFALLIBLE: bool = true;
9681139

9691140
fn id(&self) -> ScalarFnId {
9701141
static ID: CachedId = CachedId::new("test.changing_output_dtype");
@@ -1125,6 +1296,7 @@ impl RowFn for MismatchedStorage {
11251296

11261297
const ARG_NAMES: &'static [&'static str] = &["value"];
11271298
const INFALLIBLE: bool = true;
1299+
const DECODE_INFALLIBLE: bool = true;
11281300

11291301
fn id(&self) -> ScalarFnId {
11301302
static ID: CachedId = CachedId::new("test.mismatched_storage");

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

Lines changed: 27 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,9 @@
33

44
//! The [`RowFn`] contract for scalar functions whose natural kernel computes one row at a time.
55
//!
6-
//! Implementations declare their arity and fallibility, then use [`RowFn::dispatch`] to select the
7-
//! typed row signature for each supported dtype combination. Optional methods provide
8-
//! serialization without putting persistence plumbing in the row kernel.
6+
//! Implementations declare their arity, semantic fallibility, and decode fallibility, then use
7+
//! [`RowFn::dispatch`] to select the typed row signature for each supported dtype combination.
8+
//! Optional methods provide serialization without putting persistence plumbing in the row kernel.
99
1010
use std::fmt::Debug;
1111
use std::fmt::Display;
@@ -51,15 +51,36 @@ pub trait RowFn: 'static + Sized + Clone + Send + Sync {
5151
/// infallible.
5252
///
5353
/// Input decoding declares its fallibility independently through
54-
/// [`InputElement::DECODE_INFALLIBLE`] and does not affect this flag. The framework checks
55-
/// row-result fallibility. A conservative `false` is allowed.
54+
/// [`DECODE_INFALLIBLE`](Self::DECODE_INFALLIBLE) and does not affect this flag. The framework
55+
/// checks row-result fallibility. A conservative `false` is allowed.
5656
///
5757
/// See [`ScalarFnVTable::is_infallible`] for the definition of semantic errors.
5858
///
59-
/// [`InputElement::DECODE_INFALLIBLE`]: crate::scalar_fn::unstable::row::InputElement::DECODE_INFALLIBLE
6059
/// [`ScalarFnVTable::is_infallible`]: crate::scalar_fn::ScalarFnVTable::is_infallible
6160
const INFALLIBLE: bool;
6261

62+
/// Whether every input element selected by [`dispatch`](Self::dispatch) declares
63+
/// [`InputElement::DECODE_INFALLIBLE`].
64+
///
65+
/// This describes conversion into the Rust-native column representation. It is independent of
66+
/// [`INFALLIBLE`](Self::INFALLIBLE), which describes the row operation, and
67+
/// [`InputElement::DENSE_SAFE`], which describes whether dense execution can tolerate null-row
68+
/// payloads.
69+
///
70+
/// This function-wide summary is necessary because [`ScalarFnVTable::is_infallible`] receives
71+
/// no argument dtypes and cannot inspect the tuple selected by [`dispatch`](Self::dispatch).
72+
/// The framework checks every selected tuple against a positive declaration. A conservative
73+
/// `false` is allowed.
74+
///
75+
/// The blanket [`ScalarFnVTable`] advertises infallibility only when both this constant and
76+
/// [`INFALLIBLE`](Self::INFALLIBLE) are `true`.
77+
///
78+
/// [`InputElement::DECODE_INFALLIBLE`]: crate::scalar_fn::unstable::row::InputElement::DECODE_INFALLIBLE
79+
/// [`InputElement::DENSE_SAFE`]: crate::scalar_fn::unstable::row::InputElement::DENSE_SAFE
80+
/// [`ScalarFnVTable`]: crate::scalar_fn::ScalarFnVTable
81+
/// [`ScalarFnVTable::is_infallible`]: crate::scalar_fn::ScalarFnVTable::is_infallible
82+
const DECODE_INFALLIBLE: bool;
83+
6384
/// Returns the ID of the scalar function.
6485
fn id(&self) -> ScalarFnId;
6586

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

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -44,12 +44,12 @@ pub unsafe trait InputElement: 'static {
4444
/// null rows to the row closure.
4545
const DENSE_SAFE: bool;
4646

47-
/// Whether [`decode`](Self::decode) is infallible for _legal_ input data.
47+
/// Whether [`decode`](Self::decode) is guaranteed not to reject legal, well-typed values while
48+
/// constructing the Rust-native [`Column`](Self::Column) representation.
4849
///
49-
/// This excludes infrastructural failures such as IO or allocation.
50-
/// It is independent of
50+
/// This excludes infrastructural failures such as IO or allocation. It is independent of
5151
/// [`RowFn::INFALLIBLE`](crate::scalar_fn::unstable::row::RowFn::INFALLIBLE), which describes
52-
/// the row operation rather than input decoding.
52+
/// the row operation, and [`DENSE_SAFE`](Self::DENSE_SAFE), which describes null-row payloads.
5353
const DECODE_INFALLIBLE: bool;
5454

5555
/// Validate that `dtype` is an acceptable input column dtype for this element type.

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

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,10 @@ const fn assert_input_visit_contract<F: RowFn, Args: ElementTuple>() {
3333
Args::ARITY == F::ARG_NAMES.len(),
3434
"the visited argument tuple must have the arity declared by RowFn::ARG_NAMES",
3535
);
36+
assert!(
37+
!F::DECODE_INFALLIBLE || Args::DECODE_INFALLIBLE,
38+
"RowFn::DECODE_INFALLIBLE must be false when an input decoder can fail",
39+
);
3640
}
3741

3842
pub(super) const fn assert_owned_visit_contract<Function, Args, Out>()

0 commit comments

Comments
 (0)