Skip to content

Commit 37ea28f

Browse files
committed
Make OutputSink skipped-row initialization a required method
Skipped-row initialization was an optional capability because the old filter-and-scatter fallback could serve a sink without one by running it densely at the compact length and scattering afterwards. With that fallback gone, a sink without an initializer failed partially valid batches at runtime. A sink allocates its own storage, so it can always write placeholder values into it, and the values are unobservable: callbacks overwrite valid rows and batch execution masks skipped rows. Replace the Option-returning skipped_rows_initializer with a required initialize_skipped_rows method, removing the sink decline in direct skip-invalid execution and the runtime error in filtered execution. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0161JW9aU6W4zmzRQsnjbE3Q
1 parent 469658d commit 37ea28f

6 files changed

Lines changed: 34 additions & 210 deletions

File tree

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

Lines changed: 1 addition & 94 deletions
Original file line numberDiff line numberDiff line change
@@ -220,42 +220,7 @@ unsafe impl OutputSink for I64Sink {
220220
type Row<'a> = &'a mut i64;
221221
type WriteToken = ();
222222

223-
fn skipped_rows_initializer() -> Option<for<'a> fn(&mut Self::Rows<'a>)> {
224-
Some(|_| {})
225-
}
226-
227-
fn storage_dtype(_params: &Self::Params) -> DType {
228-
DType::from(i64::PTYPE)
229-
}
230-
231-
fn with_capacity(rows: usize, _params: &Self::Params) -> VortexResult<Self> {
232-
Ok(Self(BufferMut::zeroed(rows)))
233-
}
234-
235-
fn rows(&mut self) -> Self::Rows<'_> {
236-
self.0.as_mut_slice()
237-
}
238-
239-
unsafe fn row_unchecked<'a>(rows: &'a mut Self::Rows<'_>, index: usize) -> Self::Row<'a> {
240-
// SAFETY: required by this method's contract.
241-
unsafe { rows.get_unchecked_mut(index) }
242-
}
243-
244-
unsafe fn finish(self) -> VortexResult<ArrayRef> {
245-
Ok(PrimitiveArray::new(self.0.freeze(), Validity::NonNullable).into_array())
246-
}
247-
}
248-
249-
/// An [`I64Sink`] without a skipped-row initializer, so it cannot skip invalid rows.
250-
struct NoSkipI64Sink(BufferMut<i64>);
251-
252-
// SAFETY: every row is initialized by `BufferMut::zeroed`, and the sink exposes exactly that
253-
// initialized slice. The `()` write token therefore proves no additional invariant.
254-
unsafe impl OutputSink for NoSkipI64Sink {
255-
type Params = ();
256-
type Rows<'a> = &'a mut [i64];
257-
type Row<'a> = &'a mut i64;
258-
type WriteToken = ();
223+
fn initialize_skipped_rows(_rows: &mut Self::Rows<'_>) {}
259224

260225
fn storage_dtype(_params: &Self::Params) -> DType {
261226
DType::from(i64::PTYPE)
@@ -279,33 +244,6 @@ unsafe impl OutputSink for NoSkipI64Sink {
279244
}
280245
}
281246

282-
#[derive(Clone)]
283-
struct NoSkipIdentity;
284-
285-
impl RowFn for NoSkipIdentity {
286-
type Options = EmptyOptions;
287-
288-
const ARG_NAMES: &'static [&'static str] = &["value"];
289-
const INFALLIBLE: bool = false;
290-
291-
fn id(&self) -> ScalarFnId {
292-
static ID: CachedId = CachedId::new("test.no_skip_identity");
293-
*ID
294-
}
295-
296-
fn dispatch<V: RowVisitor>(
297-
&self,
298-
_options: &Self::Options,
299-
_args: &[DType],
300-
visitor: V,
301-
) -> VortexResult<V::VisitResult> {
302-
visitor.visit_into::<(i64,), NoSkipI64Sink, VortexResult<()>>((), |(value,), output| {
303-
*output = value;
304-
Ok(())
305-
})
306-
}
307-
}
308-
309247
#[derive(Clone)]
310248
struct RepeatValue;
311249

@@ -839,37 +777,6 @@ fn test_deferred_owned_execution_handles_constant_lhs() -> VortexResult<()> {
839777
Ok(())
840778
}
841779

842-
#[test]
843-
fn test_no_skip_sink_executes_all_valid_batch() -> VortexResult<()> {
844-
let values = vec![1_i64, 2];
845-
let input = PrimitiveArray::from_iter(values.clone()).into_array();
846-
let args = VecExecutionArgs::new(vec![input], 2);
847-
let mut ctx = array_session().create_execution_ctx();
848-
849-
let actual = execute_rows(&NoSkipIdentity, &EmptyOptions, &args, &mut ctx)?;
850-
let expected = PrimitiveArray::from_iter(values).into_array();
851-
852-
assert_arrays_eq!(&actual, &expected, &mut ctx);
853-
Ok(())
854-
}
855-
856-
#[test]
857-
fn test_partially_valid_batch_requires_sink_skipped_row_initializer() -> VortexResult<()> {
858-
let input =
859-
PrimitiveArray::new(vec![1_i64, 2], Validity::from_iter([true, false])).into_array();
860-
let args = VecExecutionArgs::new(vec![input], 2);
861-
let mut ctx = array_session().create_execution_ctx();
862-
863-
let error = execute_rows(&NoSkipIdentity, &EmptyOptions, &args, &mut ctx)
864-
.expect_err("a sink without a skipped-row initializer must reject a partially valid batch");
865-
866-
assert!(
867-
error.to_string().contains("cannot initialize skipped rows"),
868-
"unexpected error: {error}",
869-
);
870-
Ok(())
871-
}
872-
873780
#[test]
874781
fn test_deferred_owned_execution_retries_null_row_failure() -> VortexResult<()> {
875782
let function = DeferredAdd::default();

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

Lines changed: 9 additions & 83 deletions
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,7 @@
66
//! Dense execution visits every row. Skip-invalid execution initializes skipped output rows and
77
//! visits only rows that are valid in every input. Direct skip-invalid execution declines when the
88
//! input representation cannot decode null payloads; filtered execution then reads inputs filtered
9-
//! to the valid rows while still writing into the original row domain. Both skip-invalid paths
10-
//! require the sink's skipped-row initializer.
9+
//! to the valid rows while still writing into the original row domain.
1110
1211
use vortex_buffer::BitBuffer;
1312
use vortex_error::VortexResult;
@@ -95,7 +94,7 @@ where
9594
unsafe { Sink::finish(sink) }
9695
}
9796

98-
/// Write only the rows set in `valid`, or decline when the inputs or sink cannot support
97+
/// Write only the rows set in `valid`, or decline when the inputs cannot support direct
9998
/// skip-invalid execution.
10099
///
101100
/// `Ok(None)` signals that direct skip-invalid execution is unavailable. Batch execution decides
@@ -114,7 +113,6 @@ where
114113
ApplyResult: SinkResult<WriteToken = Sink::WriteToken>,
115114
{
116115
let Some(ValidRowsSetup {
117-
initialize_skipped_rows,
118116
columns,
119117
valid_rows,
120118
row_count,
@@ -133,7 +131,7 @@ where
133131
{
134132
// Initialize every slot before visiting only valid rows.
135133
let mut rows = Sink::rows(&mut sink);
136-
initialize_skipped_rows(&mut rows);
134+
Sink::initialize_skipped_rows(&mut rows);
137135

138136
// The initializer can change addressability. Recheck it so LLVM can prove every mask
139137
// index is in bounds.
@@ -184,9 +182,8 @@ where
184182
///
185183
/// `args` addresses only the valid rows of the original batch, in order. The sink covers the
186184
/// original row domain: each set position of `valid` receives the output of the next filtered
187-
/// row, and the skipped-row initializer makes unset positions safe to finish before batch
188-
/// execution masks them. A sink without an initializer cannot represent skipped rows, and there
189-
/// is no compact execution to scatter, so this execution fails.
185+
/// row, and [`OutputSink::initialize_skipped_rows`] makes unset positions safe to finish before
186+
/// batch execution masks them.
190187
pub(crate) fn execute_sink_filtered<Args, Prepared, Sink, ApplyResult>(
191188
args: &dyn ExecutionArgs,
192189
valid: &MaskValuesRef,
@@ -200,13 +197,6 @@ where
200197
Sink: OutputSink,
201198
ApplyResult: SinkResult<WriteToken = Sink::WriteToken>,
202199
{
203-
let Some(initialize_skipped_rows) = Sink::skipped_rows_initializer() else {
204-
vortex_bail!(
205-
"the output sink cannot initialize skipped rows, which a partially valid batch \
206-
requires",
207-
);
208-
};
209-
210200
let columns = Args::decode(args, ctx)?;
211201

212202
let filtered_len = args.row_count();
@@ -229,7 +219,7 @@ where
229219
{
230220
// Initialize every slot before visiting only valid rows.
231221
let mut rows = Sink::rows(&mut sink);
232-
initialize_skipped_rows(&mut rows);
222+
Sink::initialize_skipped_rows(&mut rows);
233223

234224
// The initializer can change addressability. Recheck it so LLVM can prove every mask
235225
// index is in bounds.
@@ -299,14 +289,13 @@ where
299289
Args: ElementTuple,
300290
Sink: OutputSink,
301291
{
302-
initialize_skipped_rows: for<'rows> fn(&mut Sink::Rows<'rows>),
303292
columns: Args::Columns,
304293
valid_rows: &'valid BitBuffer,
305294
row_count: usize,
306295
sink: Sink,
307296
}
308297

309-
/// Resolve the capabilities, inputs, sink, and validity mask for skip-invalid execution.
298+
/// Resolve the inputs, sink, and validity mask for direct skip-invalid execution.
310299
fn setup_sink_valid_rows<'valid, Args, Sink>(
311300
args: &dyn ExecutionArgs,
312301
valid: &'valid MaskValuesRef,
@@ -317,11 +306,6 @@ where
317306
Args: ElementTuple,
318307
Sink: OutputSink,
319308
{
320-
// The initializer both declares support for skipping rows and initializes those rows.
321-
let Some(initialize_skipped_rows) = Sink::skipped_rows_initializer() else {
322-
return Ok(None);
323-
};
324-
325309
// Null-tolerant decoding exposes values behind nulls without filtering. Decline when any input
326310
// cannot provide those values safely.
327311
let Some(columns) = Args::decode_null_tolerant(args, ctx)? else {
@@ -344,7 +328,6 @@ where
344328
);
345329

346330
Ok(Some(ValidRowsSetup {
347-
initialize_skipped_rows,
348331
columns,
349332
valid_rows,
350333
row_count,
@@ -356,7 +339,6 @@ where
356339
mod tests {
357340
use vortex_error::VortexResult;
358341
use vortex_error::vortex_bail;
359-
use vortex_error::vortex_err;
360342
use vortex_mask::Mask;
361343

362344
use super::execute_sink_valid_rows;
@@ -369,40 +351,9 @@ mod tests {
369351
use crate::dtype::NativePType;
370352
use crate::scalar_fn::VecExecutionArgs;
371353
use crate::scalar_fn::unstable::row::OutputSink;
372-
use crate::validity::Validity;
373-
374-
struct NonSkippingSink;
375354

376355
struct ShrinkingSink(Vec<i64>);
377356

378-
// SAFETY: `with_capacity` always returns an error, so no sink value can reach `rows`, `row`, or
379-
// `finish` through the executor. The row-initialization requirements are therefore vacuous.
380-
unsafe impl OutputSink for NonSkippingSink {
381-
type Params = ();
382-
type Rows<'a> = ();
383-
type Row<'a> = ();
384-
type WriteToken = ();
385-
386-
fn storage_dtype(_params: &Self::Params) -> DType {
387-
DType::from(i64::PTYPE)
388-
}
389-
390-
fn with_capacity(_rows: usize, _params: &Self::Params) -> VortexResult<Self> {
391-
Err(vortex_err!(
392-
"a non-skipping sink must decline before allocation"
393-
))
394-
}
395-
396-
fn rows(&mut self) -> Self::Rows<'_> {}
397-
398-
unsafe fn row_unchecked<'a>(_rows: &'a mut Self::Rows<'_>, _index: usize) -> Self::Row<'a> {
399-
}
400-
401-
unsafe fn finish(self) -> VortexResult<ArrayRef> {
402-
Err(vortex_err!("a non-skipping sink must not finish"))
403-
}
404-
}
405-
406357
// SAFETY: the initializer deliberately shrinks the row collection to exercise the executor's
407358
// post-initialization length check. If execution incorrectly continues, safe indexing in
408359
// `row_unchecked` panics instead of accessing invalid memory.
@@ -412,10 +363,8 @@ mod tests {
412363
type Row<'a> = &'a mut i64;
413364
type WriteToken = ();
414365

415-
fn skipped_rows_initializer() -> Option<for<'a> fn(&mut Self::Rows<'a>)> {
416-
Some(|rows| {
417-
rows.pop();
418-
})
366+
fn initialize_skipped_rows(rows: &mut Self::Rows<'_>) {
367+
rows.pop();
419368
}
420369

421370
fn storage_dtype(_params: &Self::Params) -> DType {
@@ -439,29 +388,6 @@ mod tests {
439388
}
440389
}
441390

442-
#[test]
443-
fn test_non_skipping_sink_declines_before_allocation() -> VortexResult<()> {
444-
let input = PrimitiveArray::new(vec![1_i64, 2], Validity::NonNullable).into_array();
445-
let args = VecExecutionArgs::new(vec![input], 2);
446-
let Mask::Values(valid) = Mask::from_iter([true, false]) else {
447-
vortex_bail!("the test validity must be partially valid");
448-
};
449-
let mut ctx = array_session().create_execution_ctx();
450-
451-
let execution = execute_sink_valid_rows::<(i64,), (), NonSkippingSink, ()>(
452-
&args,
453-
&valid,
454-
&(),
455-
&mut ctx,
456-
|_| (),
457-
|_, _, _| (),
458-
)?;
459-
460-
assert!(execution.is_none());
461-
462-
Ok(())
463-
}
464-
465391
#[test]
466392
fn test_skip_invalid_sink_rechecks_rows_after_initialization() -> VortexResult<()> {
467393
let input = PrimitiveArray::from_iter([10_i64, 20]).into_array();

‎vortex-array/src/scalar_fn/unstable/row/types/sink/fixed_size_list.rs‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -92,12 +92,10 @@ unsafe impl<T: OutputElement + Copy + Default> OutputSink for FixedSizeListSink<
9292
type Row<'a> = &'a mut [MaybeUninit<T>];
9393
type WriteToken = InitializedRow;
9494

95-
fn skipped_rows_initializer() -> Option<for<'a> fn(&mut Self::Rows<'a>)> {
96-
Some(|rows| {
97-
for element in rows.elements.iter_mut() {
98-
element.write(T::default());
99-
}
100-
})
95+
fn initialize_skipped_rows(rows: &mut Self::Rows<'_>) {
96+
for element in rows.elements.iter_mut() {
97+
element.write(T::default());
98+
}
10199
}
102100

103101
fn storage_dtype(params: &Self::Params) -> DType {

0 commit comments

Comments
 (0)