From 9172cd8b17139a09ba48c0ce3529bcd8fbdd96e5 Mon Sep 17 00:00:00 2001 From: jackylee-ch Date: Mon, 21 Sep 2026 20:54:14 +0800 Subject: [PATCH] fix(runend): clip min/max and is_constant to the array's logical window `RunEndData::validate_parts` only requires the runs to *cover* `offset..offset + length`, so a run-end array may legally carry runs that lie entirely outside its logical window. `RunEndMinMaxKernel` and `RunEndIsConstantKernel` aggregated over every entry of `values()`, so they saw values no row of the array holds. Measured against the decoded array: for `try_new_offset_length(ends=[7, 10], values=[2, 3], offset=2, length=3)` -- logical rows `[2, 2, 2]` -- `statistics().compute_stat(Stat::Max)` returned 3 where the decoded array returns 2. That is the path that fills a file's zone map, and DataFusion and DuckDB read those statistics as column statistics. The error is always outward, min too low and max too high, so pruning stays sound and no rows go missing; what is wrong is the statistic itself. The is_constant kernel is one-directional in the same way, reporting a constant window as non-constant. The sibling aggregate in this crate already clips: `RunEndSumKernel` takes `array.offset()..array.offset() + batch.len()` and `sum/runs.rs` trims the boundary runs to it. Share that treatment through a `windowed_values` helper, which returns `values()` untouched when the window spans every run -- the common case, since the slice kernel trims both children. Signed-off-by: jackylee-ch --- encodings/runend/src/compute/is_constant.rs | 8 +- encodings/runend/src/compute/min_max.rs | 101 +++++++++++++++++++- encodings/runend/src/compute/mod.rs | 36 +++++++ 3 files changed, 139 insertions(+), 6 deletions(-) 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;