Skip to content

Commit 5541df9

Browse files
robert3005connortsui20
authored andcommitted
Separate RowFn batch inspection from execution planning
Inspect borrowed inputs once for lengths, dtypes, constant classification, and strict validity. Pair collected inputs with a separate execution plan while retaining the existing dispatch and retry paths. Extracted from #9979. Signed-off-by: Robert Kruszewski <github@robertk.io> Signed-off-by: Connor Tsui <connor.tsui20@gmail.com>
1 parent 80cbea7 commit 5541df9

15 files changed

Lines changed: 320 additions & 94 deletions

File tree

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -10,21 +10,21 @@ use crate::ExecutionCtx;
1010
use crate::IntoArray;
1111
use crate::arrays::ConstantArray;
1212

13-
impl RowFnExecutionArgs {
13+
impl RowFnExecutionArgs<'_> {
1414
/// Execute all-constant inputs by evaluating one row and broadcasting the validated result.
1515
pub(super) fn execute_all_constant(
1616
&self,
1717
kernel: impl Fn(BorrowedRowFnArgs<'_>, &mut ExecutionCtx) -> VortexResult<ArrayRef>,
1818
ctx: &mut ExecutionCtx,
1919
) -> VortexResult<ArrayRef> {
2020
let result = self.validate_kernel_output(
21-
kernel(self.execution_args(&self.inputs, 1), ctx)?,
21+
kernel(self.execution_args(&self.batch.inputs, 1), ctx)?,
2222
1,
2323
ctx,
2424
)?;
2525
let result = self.finalize_output(result, 1)?;
2626
let scalar = result.execute_scalar(0, ctx)?;
2727

28-
Ok(ConstantArray::new(scalar, self.row_count).into_array())
28+
Ok(ConstantArray::new(scalar, self.batch.row_count).into_array())
2929
}
3030
}

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

Lines changed: 14 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -14,14 +14,17 @@ use crate::builtins::ArrayBuiltins;
1414
use crate::scalar_fn::unstable::row::execute::DenseAttempt;
1515
use crate::validity::Validity;
1616

17-
impl RowFnExecutionArgs {
17+
impl RowFnExecutionArgs<'_> {
1818
/// Run every stored payload, then attach the input validity without materializing its mask.
1919
pub(super) fn execute_dense(
2020
&self,
2121
kernel: impl Fn(BorrowedRowFnArgs<'_>, &mut ExecutionCtx) -> VortexResult<ArrayRef>,
2222
ctx: &mut ExecutionCtx,
2323
) -> VortexResult<ArrayRef> {
24-
let values = kernel(self.execution_args(&self.inputs, self.row_count), ctx)?;
24+
let values = kernel(
25+
self.execution_args(&self.batch.inputs, self.batch.row_count),
26+
ctx,
27+
)?;
2528

2629
self.finalize_dense_output(values, ctx)
2730
}
@@ -49,8 +52,10 @@ impl RowFnExecutionArgs {
4952
) -> VortexResult<ArrayRef>,
5053
ctx: &mut ExecutionCtx,
5154
) -> VortexResult<ArrayRef> {
52-
let attempt =
53-
execute_dense_attempt(self.execution_args(&self.inputs, self.row_count), ctx)?;
55+
let attempt = execute_dense_attempt(
56+
self.execution_args(&self.batch.inputs, self.batch.row_count),
57+
ctx,
58+
)?;
5459

5560
match attempt {
5661
DenseAttempt::Values(values) => self.finalize_dense_output(values, ctx),
@@ -75,7 +80,7 @@ impl RowFnExecutionArgs {
7580
) -> VortexResult<ArrayRef>,
7681
ctx: &mut ExecutionCtx,
7782
) -> VortexResult<ArrayRef> {
78-
let valid_rows = match self.validity.execute_mask(self.row_count, ctx)? {
83+
let valid_rows = match self.batch.validity.execute_mask(self.batch.row_count, ctx)? {
7984
// An array-backed validity can materialize to all valid even though the cheap checks
8085
// in `RowFnExecutionArgs::execute` could not prove that. The deferred error therefore
8186
// came from an observable row and remains terminal. An empty mask is both all-valid
@@ -101,13 +106,13 @@ impl RowFnExecutionArgs {
101106
values: ArrayRef,
102107
ctx: &mut ExecutionCtx,
103108
) -> VortexResult<ArrayRef> {
104-
let values = self.validate_kernel_output(values, self.row_count, ctx)?;
109+
let values = self.validate_kernel_output(values, self.batch.row_count, ctx)?;
105110

106-
match self.validity.clone() {
111+
match self.batch.validity.clone() {
107112
Validity::NonNullable | Validity::AllValid => {
108-
self.finalize_output(values, self.row_count)
113+
self.finalize_output(values, self.batch.row_count)
109114
}
110-
Validity::Array(valid) => self.finalize_output(values.mask(valid)?, self.row_count),
115+
Validity::Array(valid) => self.finalize_output(values.mask(valid)?, self.batch.row_count),
111116
Validity::AllInvalid => {
112117
unreachable!("all-invalid validity is handled before dense row execution")
113118
}

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

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ use crate::ExecutionCtx;
3131
use crate::IntoArray;
3232
use crate::builtins::ArrayBuiltins;
3333

34-
impl RowFnExecutionArgs {
34+
impl RowFnExecutionArgs<'_> {
3535
/// Filter the batch to valid rows, then execute the kernel into the original row domain.
3636
pub(super) fn execute_filtered(
3737
&self,
@@ -47,16 +47,17 @@ impl RowFnExecutionArgs {
4747
let filter_mask = Mask::Values(MaskValuesRef::clone(valid));
4848

4949
let filtered_inputs: SmallVec<[ArrayRef; 4]> = self
50+
.batch
5051
.inputs
5152
.iter()
5253
.map(|input| input.filter(filter_mask.clone()))
5354
.collect::<VortexResult<_>>()?;
5455

5556
let filtered_args = self.execution_args(&filtered_inputs, filtered_len);
5657
let values = execute_filtered_rows(filtered_args, MaskValuesRef::clone(valid), ctx)?;
57-
let values = self.validate_kernel_output(values, self.row_count, ctx)?;
58+
let values = self.validate_kernel_output(values, self.batch.row_count, ctx)?;
5859

5960
let mask = valid.as_ref().into_array();
60-
self.finalize_output(values.mask(mask)?, self.row_count)
61+
self.finalize_output(values.mask(mask)?, self.batch.row_count)
6162
}
6263
}

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

Lines changed: 6 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -14,9 +14,7 @@ use super::RowPolicy;
1414
use super::args::BorrowedRowFnArgs;
1515
use crate::ArrayRef;
1616
use crate::ExecutionCtx;
17-
use crate::arrays::Constant;
1817
use crate::scalar_fn::unstable::row::execute::DenseAttempt;
19-
use crate::scalar_fn::unstable::row::types::batch_const;
2018
use crate::validity::Validity;
2119

2220
mod constant;
@@ -27,7 +25,7 @@ mod valid_only;
2725
mod output;
2826
pub(crate) use output::finalize_kernel_output;
2927

30-
impl RowFnExecutionArgs {
28+
impl RowFnExecutionArgs<'_> {
3129
/// Apply constant folding and null handling around `kernel`.
3230
///
3331
/// For a partially valid batch, `try_valid_rows` can avoid filtering. `Ok(None)` filters the
@@ -55,29 +53,23 @@ impl RowFnExecutionArgs {
5553
) -> VortexResult<ArrayRef> {
5654
// Strictness: an all-null batch has no observable row work. Keep the literal-constant
5755
// check explicit alongside the conjoined validity invariant.
58-
if matches!(self.validity, Validity::AllInvalid)
59-
|| self.inputs.iter().any(|input| {
60-
input
61-
.as_opt::<Constant>()
62-
.is_some_and(|constant| constant.scalar().is_null())
63-
})
64-
{
56+
if matches!(self.batch.validity, Validity::AllInvalid) || self.batch.has_null_const {
6557
return Ok(self.all_null());
6658
}
6759

6860
// All inputs are constant, and their conjoined validity proves that every row is non-null.
6961
// The constant check sees through extension and masked wrappers, just like argument
7062
// decoding.
71-
if self.row_count > 0
72-
&& self.validity.definitely_no_nulls()
73-
&& self.inputs.iter().all(|input| batch_const(input).is_some())
63+
if self.batch.row_count > 0
64+
&& self.batch.validity.definitely_no_nulls()
65+
&& self.batch.all_batch_const()
7466
{
7567
return self.execute_all_constant(kernel, ctx);
7668
}
7769

7870
// Do not resolve array-backed validity for the uncommon all-valid or all-null cases here.
7971
// That can execute and scan the full mask; each policy resolves it only when necessary.
80-
if self.validity.definitely_no_nulls() {
72+
if self.batch.validity.definitely_no_nulls() {
8173
return self.execute_dense(kernel, ctx);
8274
}
8375

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

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -15,9 +15,13 @@ use crate::dtype::DType;
1515
use crate::scalar::Scalar;
1616
use crate::scalar_fn::ScalarFnId;
1717

18-
impl RowFnExecutionArgs {
18+
impl RowFnExecutionArgs<'_> {
1919
pub(super) fn all_null(&self) -> ArrayRef {
20-
ConstantArray::new(Scalar::null(self.result_dtype.clone()), self.row_count).into_array()
20+
ConstantArray::new(
21+
Scalar::null(self.result_dtype.clone()),
22+
self.batch.row_count,
23+
)
24+
.into_array()
2125
}
2226

2327
/// Label the finished column, validate it, and apply the row function's logical outer
@@ -32,7 +36,7 @@ impl RowFnExecutionArgs {
3236
// partially valid batch carries the same metadata as a dense one.
3337
let values = self.plan.relabel_output(values)?;
3438

35-
validate_output(self.id, &self.result_dtype, expected_len, &values)?;
39+
validate_output(self.batch.id, &self.result_dtype, expected_len, &values)?;
3640

3741
cast_output_nullability(&self.result_dtype, values)
3842
}
@@ -45,7 +49,7 @@ impl RowFnExecutionArgs {
4549
ctx: &mut ExecutionCtx,
4650
) -> VortexResult<ArrayRef> {
4751
finalize_kernel_output(
48-
self.id,
52+
self.batch.id,
4953
self.plan.storage_dtype(),
5054
expected_len,
5155
values,

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

Lines changed: 13 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ use crate::ExecutionCtx;
1212
use crate::IntoArray;
1313
use crate::builtins::ArrayBuiltins;
1414

15-
impl RowFnExecutionArgs {
15+
impl RowFnExecutionArgs<'_> {
1616
/// Resolve validity and try direct valid-row execution before filtered execution.
1717
pub(super) fn execute_valid_only(
1818
&self,
@@ -29,15 +29,22 @@ impl RowFnExecutionArgs {
2929
) -> VortexResult<ArrayRef>,
3030
ctx: &mut ExecutionCtx,
3131
) -> VortexResult<ArrayRef> {
32-
let validity = self.validity.clone().execute_mask(self.row_count, ctx)?;
32+
let validity = self
33+
.batch
34+
.validity
35+
.clone()
36+
.execute_mask(self.batch.row_count, ctx)?;
3337

3438
let valid_rows = match validity {
3539
// An empty mask is both all-valid and all-null. Preserve the all-valid behavior.
3640
Mask::AllTrue(_) | Mask::AllFalse(0) => {
37-
let values = kernel(self.execution_args(&self.inputs, self.row_count), ctx)?;
38-
let values = self.validate_kernel_output(values, self.row_count, ctx)?;
41+
let values = kernel(
42+
self.execution_args(&self.batch.inputs, self.batch.row_count),
43+
ctx,
44+
)?;
45+
let values = self.validate_kernel_output(values, self.batch.row_count, ctx)?;
3946

40-
return self.finalize_output(values, self.row_count);
47+
return self.finalize_output(values, self.batch.row_count);
4148
}
4249
Mask::AllFalse(_) => return Ok(self.all_null()),
4350
Mask::Values(valid_rows) => valid_rows,
@@ -62,7 +69,7 @@ impl RowFnExecutionArgs {
6269
ctx: &mut ExecutionCtx,
6370
) -> VortexResult<Option<ArrayRef>> {
6471
let Some(values) = try_valid_rows(
65-
self.execution_args(&self.inputs, self.row_count),
72+
self.execution_args(&self.batch.inputs, self.batch.row_count),
6673
MaskValuesRef::clone(valid),
6774
ctx,
6875
)?
Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
// SPDX-License-Identifier: Apache-2.0
2+
// SPDX-FileCopyrightText: Copyright the Vortex contributors
3+
4+
//! Borrowed access to the input columns of one row-function execution.
5+
//!
6+
//! [`RowInputs`] lets batch execution inspect and decode its inputs without cloning them. An
7+
//! internal caller that holds its operands implements it directly. [`ArrayInputs`] adapts the
8+
//! columns collected from a general [`ExecutionArgs`](crate::scalar_fn::ExecutionArgs).
9+
10+
use crate::ArrayRef;
11+
12+
/// The same-length input columns of one row-function execution, borrowed by index.
13+
///
14+
/// Batch inspection and execution share this borrowed view of the inputs.
15+
pub(crate) trait RowInputs {
16+
/// Return the number of input columns.
17+
fn num_inputs(&self) -> usize;
18+
19+
/// Return the number of rows in the execution scope.
20+
fn row_count(&self) -> usize;
21+
22+
/// Return input `index`. `index` **must** be less than [`num_inputs`](Self::num_inputs).
23+
fn input(&self, index: usize) -> &ArrayRef;
24+
}
25+
26+
/// [`RowInputs`] over a slice of collected columns.
27+
pub(crate) struct ArrayInputs<'a> {
28+
/// The input columns.
29+
arrays: &'a [ArrayRef],
30+
31+
/// The number of rows in the execution scope.
32+
row_count: usize,
33+
}
34+
35+
impl<'a> ArrayInputs<'a> {
36+
pub(crate) fn new(arrays: &'a [ArrayRef], row_count: usize) -> Self {
37+
Self { arrays, row_count }
38+
}
39+
}
40+
41+
impl RowInputs for ArrayInputs<'_> {
42+
fn num_inputs(&self) -> usize {
43+
self.arrays.len()
44+
}
45+
46+
fn row_count(&self) -> usize {
47+
self.row_count
48+
}
49+
50+
fn input(&self, index: usize) -> &ArrayRef {
51+
&self.arrays[index]
52+
}
53+
}

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

Lines changed: 47 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -9,9 +9,10 @@
99
//! propagating strict validity, selecting an execution strategy, and validating the finished
1010
//! output.
1111
//!
12-
//! [`BatchPlan`] carries the nullable execution strategy selected by a concrete dispatch.
13-
//! [`RowFnExecutionArgs`] applies that strategy, and [`BorrowedRowFnArgs`] pairs each kernel
14-
//! invocation with its planning metadata.
12+
//! [`RowInputs`] borrows the input columns, and [`BatchProbe`] classifies them before dispatch.
13+
//! [`BatchInputs`] retains the columns and their classification. [`BatchPlan`] carries the nullable
14+
//! execution strategy selected by a concrete dispatch. [`RowFnExecutionArgs`] applies that strategy,
15+
//! and [`BorrowedRowFnArgs`] pairs each kernel invocation with its planning metadata.
1516
1617
use smallvec::SmallVec;
1718

@@ -23,6 +24,10 @@ use crate::validity::Validity;
2324
mod args;
2425
pub(super) use args::BorrowedRowFnArgs;
2526

27+
mod inputs;
28+
pub(crate) use inputs::ArrayInputs;
29+
pub(crate) use inputs::RowInputs;
30+
2631
mod execute;
2732
pub(super) use execute::finalize_kernel_output;
2833

@@ -31,8 +36,29 @@ mod planning;
3136
pub(super) use super::visitor::BatchPlan;
3237
pub(super) use super::visitor::RowPolicy;
3338

34-
/// The same-length input columns and metadata for one row-function execution.
35-
pub(crate) struct RowFnExecutionArgs {
39+
/// What batch execution derives from its borrowed inputs before a dispatch plans them.
40+
///
41+
/// A probe holds no input column. [`BatchInputs`] retains the columns and their probe when a batch
42+
/// needs owned inputs for execution.
43+
pub(crate) struct BatchProbe {
44+
/// The number of rows in the original execution scope.
45+
row_count: usize,
46+
47+
/// The conjoined input validity. An output row is valid exactly when it is valid in every
48+
/// input. Conjoining is lazy, and null handling materializes the mask only when required.
49+
validity: Validity,
50+
51+
/// Bit `i` is set when input `i` is a batch constant, seen through extension and masked
52+
/// wrappers.
53+
const_mask: u32,
54+
55+
/// Whether any input is a null literal constant.
56+
has_null_const: bool,
57+
}
58+
59+
/// The same-length input columns collected for one row-function execution, with the facts of
60+
/// their [`BatchProbe`].
61+
pub(crate) struct BatchInputs {
3662
/// The function being executed, named in the errors this raises.
3763
id: ScalarFnId,
3864

@@ -42,13 +68,26 @@ pub(crate) struct RowFnExecutionArgs {
4268
/// The input columns, collected once for validity, constant folding, filtering, and execution.
4369
inputs: SmallVec<[ArrayRef; 4]>,
4470

45-
/// The input dtypes, collected with the columns and reused by both planning and execution.
46-
arg_dtypes: SmallVec<[DType; 4]>,
47-
4871
/// The conjoined input validity. An output row is valid exactly when it is valid in every
4972
/// input. Conjoining is lazy, and null handling materializes the mask only when required.
5073
validity: Validity,
5174

75+
/// Bit `i` is set when input `i` is a batch constant, seen through extension and masked
76+
/// wrappers.
77+
const_mask: u32,
78+
79+
/// Whether any input is a null literal constant.
80+
has_null_const: bool,
81+
}
82+
83+
/// The same-length input columns and metadata for one row-function execution.
84+
pub(crate) struct RowFnExecutionArgs<'a> {
85+
/// The collected input columns and their conjoined validity.
86+
batch: &'a BatchInputs,
87+
88+
/// The input dtypes, collected with the columns and reused by both planning and execution.
89+
arg_dtypes: &'a [DType],
90+
5291
/// The output dtype, widened to nullable when any input is nullable. The finished column is
5392
/// reconciled against this dtype.
5493
result_dtype: DType,

0 commit comments

Comments
 (0)