diff --git a/encodings/runend/src/compute/is_constant.rs b/encodings/runend/src/compute/is_constant.rs index 710fb7a3712..0be30a1e1f5 100644 --- a/encodings/runend/src/compute/is_constant.rs +++ b/encodings/runend/src/compute/is_constant.rs @@ -11,11 +11,12 @@ use vortex_array::scalar::Scalar; use vortex_error::VortexResult; use crate::RunEnd; -use crate::array::RunEndArraySlotsExt; +use crate::compute::windowed_values; /// RunEnd-specific is_constant kernel. /// -/// If the values array of a run-end array is constant, the entire array is constant. +/// If every run value the array's logical window covers is the same, the whole array is constant. +/// Runs outside that window hold no rows, so they must not take part — see [`windowed_values`]. #[derive(Debug)] pub(crate) struct RunEndIsConstantKernel; @@ -34,7 +35,8 @@ impl DynAggregateKernel for RunEndIsConstantKernel { return Ok(None); }; - let result = is_constant(array.values(), ctx)?; + let values = windowed_values(&array, batch.len(), ctx)?; + let result = is_constant(&values, ctx)?; Ok(Some(IsConstant::make_partial(batch, result, ctx)?)) } } diff --git a/encodings/runend/src/compute/min_max.rs b/encodings/runend/src/compute/min_max.rs index d17c5b3c65c..1be103c8da2 100644 --- a/encodings/runend/src/compute/min_max.rs +++ b/encodings/runend/src/compute/min_max.rs @@ -12,12 +12,13 @@ use vortex_array::scalar::Scalar; use vortex_error::VortexResult; use crate::RunEnd; -use crate::array::RunEndArraySlotsExt; +use crate::compute::windowed_values; /// RunEnd-specific min/max kernel. /// /// Run-end encoded arrays store each unique run value once, so min/max can be computed directly -/// on the values array without decoding. +/// on the run values without decoding — but only over the runs the array's logical window covers, +/// since runs outside it hold no rows. See [`windowed_values`]. #[derive(Debug)] pub(crate) struct RunEndMinMaxKernel; @@ -37,7 +38,8 @@ impl DynAggregateKernel for RunEndMinMaxKernel { }; let struct_dtype = make_minmax_dtype(batch.dtype()); - match min_max(run_end.values(), ctx, *options)? { + let values = windowed_values(&run_end, batch.len(), ctx)?; + match min_max(&values, ctx, *options)? { Some(result) => Ok(Some(Scalar::struct_( struct_dtype, vec![result.min, result.max], @@ -46,3 +48,96 @@ impl DynAggregateKernel for RunEndMinMaxKernel { } } } + +#[cfg(test)] +mod tests { + use rstest::rstest; + use vortex_array::ArrayRef; + use vortex_array::IntoArray; + use vortex_array::VortexSessionExecute; + use vortex_array::aggregate_fn::AggregateFnVTableExt; + use vortex_array::aggregate_fn::NumericalAggregateOpts; + use vortex_array::aggregate_fn::fns::min_max::MinMax; + use vortex_array::aggregate_fn::kernels::DynAggregateKernel; + use vortex_array::arrays::PrimitiveArray; + use vortex_array::expr::stats::Stat; + use vortex_buffer::buffer; + use vortex_error::VortexExpect; + use vortex_error::VortexResult; + + use super::RunEndMinMaxKernel; + use crate::RunEnd; + use crate::tests::SESSION; + + /// `validate_parts` only requires the runs to *cover* `offset..offset + length`, so runs may + /// legally sit outside the window. Their values hold no rows and must not reach min/max. + #[rstest] + // Trailing run is entirely past the window: logical rows are [2, 2, 2]. + #[case::trailing_run_outside(buffer![7u64, 10].into_array(), buffer![2i64, 3].into_array(), 2, 3)] + // Leading run is entirely before the window: logical rows are [3, 3]. + #[case::leading_run_outside(buffer![4u64, 9].into_array(), buffer![2i64, 3].into_array(), 4, 2)] + // Both ends outside: logical rows are [5, 5]. `validate_parts` requires + // `first_run_end >= offset`, so a leading run can only sit outside when it ends exactly at the + // offset — hence 3 here rather than 2. + #[case::both_ends_outside( + buffer![3u64, 6, 9].into_array(), + buffer![1i64, 5, 9].into_array(), + 3, + 2 + )] + // The whole window, so the fast path must still be taken. + #[case::window_spans_every_run( + buffer![3u64, 5].into_array(), + buffer![2i64, 3].into_array(), + 0, + 5 + )] + fn min_max_only_sees_the_logical_window( + #[case] ends: ArrayRef, + #[case] values: ArrayRef, + #[case] offset: usize, + #[case] length: usize, + ) -> VortexResult<()> { + let mut ctx = SESSION.create_execution_ctx(); + let array = + RunEnd::try_new_offset_length(ends, values, offset, length, &mut ctx)?.into_array(); + let decoded = array + .clone() + .execute::(&mut ctx)? + .into_array(); + + // The statistics path is what fills a file's zone map, so check it first and on its own: + // it is the route by which a wrong extremum becomes wrong pruning. + let stat_min = array.statistics().compute_stat(Stat::Min, &mut ctx)?; + let decoded_min = decoded.statistics().compute_stat(Stat::Min, &mut ctx)?; + assert_eq!( + stat_min, decoded_min, + "Stat::Min disagrees with the decoded array" + ); + let stat_max = array.statistics().compute_stat(Stat::Max, &mut ctx)?; + let decoded_max = decoded.statistics().compute_stat(Stat::Max, &mut ctx)?; + assert_eq!( + stat_max, decoded_max, + "Stat::Max disagrees with the decoded array" + ); + + // Also drive the kernel directly, so the test cannot pass merely because some dispatch + // path decided to decode. + let aggregate = MinMax.bind(NumericalAggregateOpts::default()); + let partial = RunEndMinMaxKernel + .aggregate(&aggregate, &array, &mut ctx)? + .vortex_expect("the run-end min/max kernel handles a primitive run-end array"); + let mut direct = aggregate.accumulator(array.dtype())?; + direct.combine_partials(partial)?; + + let mut reference = aggregate.accumulator(array.dtype())?; + reference.accumulate(&decoded, &mut ctx)?; + + assert_eq!( + direct.finish()?, + reference.finish()?, + "the kernel disagrees with the decoded array" + ); + Ok(()) + } +} diff --git a/encodings/runend/src/compute/mod.rs b/encodings/runend/src/compute/mod.rs index 296ab525952..ffe5bd47216 100644 --- a/encodings/runend/src/compute/mod.rs +++ b/encodings/runend/src/compute/mod.rs @@ -12,6 +12,42 @@ pub(crate) mod sum; pub(crate) mod take; pub(crate) mod take_from; +use vortex_array::ArrayRef; +use vortex_array::ExecutionCtx; +use vortex_array::TypedArrayRef; +use vortex_error::VortexResult; + +use crate::RunEnd; +use crate::array::RunEndArrayExt; +use crate::array::RunEndArraySlotsExt; + +/// The run values that the array's logical window actually covers. +/// +/// `offset` and `len` describe a window into the runs, and +/// [`RunEndData::validate_parts`](crate::RunEndData::validate_parts) only requires the runs to +/// *cover* that window, so an array may legally carry runs that lie entirely outside it. A kernel +/// that reads `values()` directly therefore sees values no row of the array holds; slice first. +/// +/// Returns `values()` untouched when the window already spans every run, which is what a trimming +/// slice produces and so the common case. +fn windowed_values( + array: &impl TypedArrayRef, + len: usize, + ctx: &mut ExecutionCtx, +) -> VortexResult { + let values = array.values(); + if len == 0 { + return values.slice(0..0); + } + + let begin = array.find_physical_index(0, ctx)?; + let end = array.find_slice_end_index(len, ctx)?; + if begin == 0 && end == values.len() { + return Ok(values.clone()); + } + values.slice(begin..end) +} + #[cfg(test)] mod tests { use std::sync::LazyLock;