From 65ea0704d30f658906d666b0278f8c49f43ce3fe Mon Sep 17 00:00:00 2001 From: Connor Tsui Date: Tue, 3 Nov 2026 13:48:03 -0500 Subject: [PATCH] Push decimal aggregates through Widen Decimal widening preserves order, equality, and validity, so ordered aggregates can use child partials without widening every input value. Reuse exact cached child extrema for MinMax. Compute decimal Sum in the logical sum dtype to avoid overflow at the child precision. Serve it only through the legacy Stat::Sum slot, whose independent batch totals preserve canonical overflow behavior. Keep the running-sum variants on the fallback path. Cover batch equivalence, logical precision, overflow, and child statistics without repeating partial-combination tests or adding benchmark cases. Signed-off-by: Connor Tsui --- .../src/arrays/widen/compute/aggregate.rs | 140 ++++++++- .../src/arrays/widen/tests/aggregates.rs | 92 +++++- .../arrays/widen/tests/decimal_aggregates.rs | 286 ++++++++++++++++++ vortex-array/src/arrays/widen/tests/mod.rs | 1 + 4 files changed, 488 insertions(+), 31 deletions(-) create mode 100644 vortex-array/src/arrays/widen/tests/decimal_aggregates.rs diff --git a/vortex-array/src/arrays/widen/compute/aggregate.rs b/vortex-array/src/arrays/widen/compute/aggregate.rs index aeca44dc042..731505109c7 100644 --- a/vortex-array/src/arrays/widen/compute/aggregate.rs +++ b/vortex-array/src/arrays/widen/compute/aggregate.rs @@ -1,12 +1,28 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright the Vortex contributors +//! Aggregates over the narrow child of a [`Widen`] array. +//! +//! Results must match aggregation over the widened values. Widening preserves order and equality, +//! but sums also depend on the logical dtype's precision and overflow rules. + use vortex_error::VortexExpect; use vortex_error::VortexResult; use crate::ArrayRef; +use crate::Columnar; use crate::ExecutionCtx; +use crate::aggregate_fn::AggregateDTypes; use crate::aggregate_fn::AggregateFnRef; +use crate::aggregate_fn::AggregateFnVTable; +use crate::aggregate_fn::NumericalAggregateOpts; +use crate::aggregate_fn::fns::is_constant::IsConstant; +use crate::aggregate_fn::fns::is_sorted::IsSorted; +use crate::aggregate_fn::fns::max::Max; +use crate::aggregate_fn::fns::min::Min; +use crate::aggregate_fn::fns::min_max::MinMax; +use crate::aggregate_fn::fns::min_max::MinMaxResult; +use crate::aggregate_fn::fns::min_max::min_max; use crate::aggregate_fn::fns::sum::Sum; use crate::aggregate_fn::kernels::DynAggregateKernel; use crate::arrays::Widen; @@ -15,11 +31,7 @@ use crate::dtype::DType; use crate::expr::stats::Stat; use crate::scalar::Scalar; -/// Runs an aggregate of a [`Widen`] on its values and widens the partial to the logical dtype. -/// -/// The values run the same accumulator code as the canonical array, so edge cases such as an empty -/// batch or a single null match by construction, and the values keep their own cached statistics -/// and kernels. [`Sum`] is served only when [`is_exact_sum`] holds. +/// Computes aggregate partials from a [`Widen`]'s child without materializing widened values. #[derive(Debug)] pub(crate) struct WidenArrayAggregate; @@ -34,27 +46,123 @@ impl DynAggregateKernel for WidenArrayAggregate { return Ok(None); }; - // NB: Exactness rests on integer decoding. A decimal `Widen` (CD4 in #10733) needs its own - // argument, because `Sum` checks a decimal for overflow against the precision of its input. - if !batch.dtype().is_int() { + let values = widen.values(); + let logical_dtype = batch.dtype(); + + if logical_dtype.is_decimal() { + return aggregate_decimal(aggregate_fn, values, logical_dtype, ctx); + } + + if !logical_dtype.is_int() { return Ok(None); } - let values = widen.values(); if aggregate_fn.is::() && !is_exact_sum(aggregate_fn, batch, values) { return Ok(None); } - let mut accumulator = aggregate_fn.accumulator(values.dtype())?; - accumulator.accumulate(values, ctx)?; + let partial = aggregate_values(aggregate_fn, values, logical_dtype, ctx)?; + + Ok(Some(partial)) + } +} + +/// Selects decimal aggregates whose child values produce the canonical partial. +fn aggregate_decimal( + aggregate_fn: &AggregateFnRef, + values: &ArrayRef, + logical_dtype: &DType, + ctx: &mut ExecutionCtx, +) -> VortexResult> { + if let Some(options) = aggregate_fn.as_opt::() { + // This kernel returns a batch total. Only the Stat::Sum path sums each batch from zero. + // Other variants add to the running sum and can overflow at different points. + if Stat::from_aggregate_fn(aggregate_fn) != Some(Stat::Sum) { + return Ok(None); + } + + let partial = decimal_sum_partial(values, options, logical_dtype, ctx)?; + + return Ok(Some(partial)); + } + + if !(aggregate_fn.is::() + || aggregate_fn.is::() + || aggregate_fn.is::() + || aggregate_fn.is::() + || aggregate_fn.is::()) + { + return Ok(None); + } + + let partial = aggregate_values(aggregate_fn, values, logical_dtype, ctx)?; - let partial_dtype = aggregate_fn.state_dtype(batch.dtype()).vortex_expect( - "an aggregate that accepts the values accepts the logical dtype, which is a wider \ - integer of the same signedness", - ); + Ok(Some(partial)) +} - Ok(Some(accumulator.flush()?.cast(&partial_dtype)?)) +/// Computes a child partial and casts it to the logical aggregate state dtype. +fn aggregate_values( + aggregate_fn: &AggregateFnRef, + values: &ArrayRef, + logical_dtype: &DType, + ctx: &mut ExecutionCtx, +) -> VortexResult { + let partial_dtype = aggregate_fn.state_dtype(logical_dtype).vortex_expect( + "the served aggregate accepts the logical dtype when it accepts the narrower values", + ); + + if let Some(options) = aggregate_fn.as_opt::() { + return min_max_partial(values, *options, logical_dtype, partial_dtype, ctx); } + + let mut accumulator = aggregate_fn.accumulator(values.dtype())?; + accumulator.accumulate(values, ctx)?; + + accumulator.flush()?.cast(&partial_dtype) +} + +/// Computes a decimal [`Sum`] batch partial without materializing the widened array. +/// +/// Checked arithmetic overflow or a batch total outside the result precision produces null. +/// Null rows are skipped, and empty or all-null batches produce zero. +fn decimal_sum_partial( + values: &ArrayRef, + options: &NumericalAggregateOpts, + logical_dtype: &DType, + ctx: &mut ExecutionCtx, +) -> VortexResult { + // Sum converts each stored value to the accumulator width selected by these logical dtypes + // before adding it. The child width never limits the sum. Widen preserves values, scale, and + // validity, so the additions, precision checks, and output dtype match the widened array. + let dtypes = AggregateDTypes::try_new(&Sum, options, logical_dtype.clone())?; + let args = dtypes.args(options); + let values = values.clone().execute::(ctx)?; + + let mut partial = Sum.empty_partial(args)?; + Sum.accumulate(args, &mut partial, &values, ctx)?; + + Sum.to_scalar(args, &partial) +} + +/// Builds a logical [`MinMax`] partial from the child's cached or computed extrema. +/// +/// [`min_max`] uses exact cached extrema before scanning the child. Extrema have the non-nullable +/// logical dtype, even for nullable inputs. A batch with no valid values has a null partial. +fn min_max_partial( + values: &ArrayRef, + options: NumericalAggregateOpts, + logical_dtype: &DType, + partial_dtype: DType, + ctx: &mut ExecutionCtx, +) -> VortexResult { + let Some(MinMaxResult { min, max }) = min_max(values, ctx, options)? else { + return Ok(Scalar::null(partial_dtype)); + }; + + let element_dtype = logical_dtype.as_nonnullable(); + let extrema = vec![min.cast(&element_dtype)?, max.cast(&element_dtype)?]; + + Ok(Scalar::struct_(partial_dtype, extrema)) } /// Returns whether the [`Sum`] of the values of `batch` equals the sum of the canonical batch. diff --git a/vortex-array/src/arrays/widen/tests/aggregates.rs b/vortex-array/src/arrays/widen/tests/aggregates.rs index 453c26e22bd..3ff601c4434 100644 --- a/vortex-array/src/arrays/widen/tests/aggregates.rs +++ b/vortex-array/src/arrays/widen/tests/aggregates.rs @@ -1,9 +1,15 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright the Vortex contributors +//! Tests for aggregate results and child-statistics reuse. +//! +//! Shared batch assertions compare integer and decimal [`Widen`] arrays with fresh canonical inputs. + +use std::fmt::Debug; use std::num::NonZeroUsize; use rstest::rstest; +use vortex_error::VortexExpect; use vortex_error::VortexResult; use super::SESSION; @@ -32,21 +38,27 @@ use crate::aggregate_fn::fns::last::Last; use crate::aggregate_fn::fns::max::Max; use crate::aggregate_fn::fns::min::Min; use crate::aggregate_fn::fns::min_max::MinMax; +use crate::aggregate_fn::fns::min_max::make_minmax_dtype; use crate::aggregate_fn::fns::null_count::NullCount; use crate::aggregate_fn::fns::sum::Sum; use crate::aggregate_fn::fns::sum_v2::SumV2; use crate::aggregate_fn::fns::uncompressed_size_in_bytes::UncompressedSizeInBytes; use crate::arrays::ChunkedArray; use crate::arrays::ConstantArray; +use crate::arrays::DecimalArray; use crate::arrays::PrimitiveArray; use crate::arrays::Widen; use crate::arrays::WidenArray; use crate::builtins::ArrayBuiltins; use crate::dtype::DType; +use crate::dtype::DecimalDType; use crate::dtype::Nullability; use crate::dtype::Nullability::NonNullable; use crate::dtype::Nullability::Nullable; use crate::dtype::PType; +use crate::expr::stats::Precision; +use crate::expr::stats::Stat; +use crate::scalar::DecimalValue; use crate::scalar::Scalar; /// Every aggregate that applies to integers, and whether the `Widen` aggregate kernel serves it. @@ -89,9 +101,8 @@ fn canonical_batch( .execute::(ctx) } -/// Runs the kernel that the session registers for `aggregate` on `Widen` over `array`, and returns -/// its partial, or `None` when no kernel is registered or the kernel declines. -fn kernel_partial( +/// Returns the registered [`Widen`] kernel's partial, or `None` if it is absent or declines. +pub(super) fn kernel_partial( aggregate: &AggregateFnRef, array: &ArrayRef, ctx: &mut ExecutionCtx, @@ -105,26 +116,26 @@ fn kernel_partial( } } -/// Accumulates `batches` of `dtype` with `aggregate` once as [`WidenArray::encode`] encodes them -/// and once as canonical arrays, and asserts that the running partials and the results are equal. +/// Checks encoded aggregates against canonical inputs, including kernel selection and partials. /// -/// On each batch that encodes to a `Widen`, the kernel must return a partial exactly when -/// `served`, and the partial must equal the partial of a fresh accumulator on the canonical batch. -/// Encoding and each accumulator get their own canonical batch, because both cache statistics on -/// the array they read, and an accumulator reads a cached statistic instead of the data. +/// `served` says whether the [`Widen`] kernel must return a partial. `build_batch` **must** return a +/// fresh array with its own statistics on every call. Reusing arrays could let aggregation read a +/// cached answer instead of testing the kernel. #[track_caller] -fn assert_aggregate_batches( +pub(super) fn assert_aggregate_batches( aggregate: &AggregateFnRef, served: bool, - batches: &[Vec>], + batches: &[T], dtype: &DType, + mut build_batch: impl FnMut(&T, &DType, &mut ExecutionCtx) -> VortexResult, + mut encode: impl FnMut(C, &mut ExecutionCtx) -> VortexResult, ctx: &mut ExecutionCtx, ) -> VortexResult<()> { let mut actual = aggregate.accumulator(dtype)?; let mut expected = aggregate.accumulator(dtype)?; for values in batches { - let encoded = WidenArray::encode(canonical_batch(values, dtype, ctx)?, ctx)?; + let encoded = encode(build_batch(values, dtype, ctx)?, ctx)?; if encoded.is::() { let partial = kernel_partial(aggregate, &encoded, ctx)?; @@ -132,13 +143,13 @@ fn assert_aggregate_batches( if let Some(partial) = partial { let mut reference = aggregate.accumulator(dtype)?; - reference.accumulate(&canonical_batch(values, dtype, ctx)?.into_array(), ctx)?; + reference.accumulate(&build_batch(values, dtype, ctx)?.into_array(), ctx)?; assert_eq!(partial, reference.flush()?, "{aggregate} on {values:?}"); } } actual.accumulate(&encoded, ctx)?; - expected.accumulate(&canonical_batch(values, dtype, ctx)?.into_array(), ctx)?; + expected.accumulate(&build_batch(values, dtype, ctx)?.into_array(), ctx)?; assert_eq!( actual.partial_scalar()?, expected.partial_scalar()?, @@ -230,7 +241,15 @@ fn test_aggregate_partials( let dtype = DType::Primitive(logical, nullability); for (aggregate, served) in aggregates() { - assert_aggregate_batches(&aggregate, served, &batches, &dtype, &mut ctx)?; + assert_aggregate_batches( + &aggregate, + served, + &batches, + &dtype, + |values, dtype, ctx| canonical_batch(values, dtype, ctx), + WidenArray::encode, + &mut ctx, + )?; } Ok(()) @@ -304,3 +323,46 @@ fn test_chunked_matches_canonical() -> VortexResult<()> { Ok(()) } + +/// Cached extrema deliberately differ from the data, so the result proves the cache was read. +#[rstest] +#[case::integer( + PrimitiveArray::from_iter([1i8, 2, 3]).into_array(), + DType::Primitive(PType::I64, NonNullable), + Scalar::primitive(-100i8, NonNullable), + Scalar::primitive(100i8, NonNullable), +)] +#[case::decimal( + DecimalArray::from_option_iter([Some(1i16), None], DecimalDType::new(4, 2)).into_array(), + DType::Decimal(DecimalDType::new(18, 2), Nullable), + Scalar::decimal(DecimalValue::I16(-9_999), DecimalDType::new(4, 2), NonNullable), + Scalar::decimal(DecimalValue::I16(9_999), DecimalDType::new(4, 2), NonNullable), +)] +fn test_min_max_reads_values_extrema( + #[case] values: ArrayRef, + #[case] logical_dtype: DType, + #[case] min: Scalar, + #[case] max: Scalar, +) -> VortexResult<()> { + let mut ctx = SESSION.create_execution_ctx(); + + for (stat, extremum) in [(Stat::Min, &min), (Stat::Max, &max)] { + let value = extremum.value().cloned(); + let value = value.vortex_expect("the planted extremum is not null"); + values.statistics().set(stat, Precision::Exact(value)); + } + + let array = WidenArray::try_new(values, logical_dtype.clone())?.into_array(); + let aggregate = MinMax.bind(NumericalAggregateOpts::default()); + + let partial = kernel_partial(&aggregate, &array, &mut ctx)?; + + let element_dtype = logical_dtype.as_nonnullable(); + let expected = Scalar::struct_( + make_minmax_dtype(&logical_dtype), + vec![min.cast(&element_dtype)?, max.cast(&element_dtype)?], + ); + assert_eq!(partial, Some(expected)); + + Ok(()) +} diff --git a/vortex-array/src/arrays/widen/tests/decimal_aggregates.rs b/vortex-array/src/arrays/widen/tests/decimal_aggregates.rs new file mode 100644 index 00000000000..ad2e3bb764b --- /dev/null +++ b/vortex-array/src/arrays/widen/tests/decimal_aggregates.rs @@ -0,0 +1,286 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Decimal aggregate results and overflow behavior. +//! +//! Compares decimal [`Widen`] arrays with canonical inputs and checks sum precision and statistics. + +use rstest::rstest; +use vortex_error::VortexExpect; +use vortex_error::VortexResult; + +use super::SESSION; +use super::aggregates::assert_aggregate_batches; +use super::aggregates::kernel_partial; +use crate::ArrayRef; +use crate::ExecutionCtx; +use crate::IntoArray; +use crate::VortexSessionExecute; +use crate::aggregate_fn::AggregateFnRef; +use crate::aggregate_fn::AggregateFnVTableExt; +use crate::aggregate_fn::EmptyOptions; +use crate::aggregate_fn::NumericalAggregateOpts; +use crate::aggregate_fn::fns::first::First; +use crate::aggregate_fn::fns::is_constant::IsConstant; +use crate::aggregate_fn::fns::is_sorted::IsSorted; +use crate::aggregate_fn::fns::is_sorted::IsSortedOptions; +use crate::aggregate_fn::fns::last::Last; +use crate::aggregate_fn::fns::max::Max; +use crate::aggregate_fn::fns::min::Min; +use crate::aggregate_fn::fns::min_max::MinMax; +use crate::aggregate_fn::fns::sum::Sum; +use crate::aggregate_fn::fns::sum_v2::SumV2; +use crate::arrays::ConstantArray; +use crate::arrays::DecimalArray; +use crate::arrays::Widen; +use crate::arrays::WidenArray; +use crate::arrays::widen::WidenArraySlotsExt; +use crate::dtype::BigCast; +use crate::dtype::DType; +use crate::dtype::DecimalDType; +use crate::dtype::Nullability; +use crate::dtype::Nullability::NonNullable; +use crate::dtype::Nullability::Nullable; +use crate::expr::stats::Precision; +use crate::expr::stats::Stat; +use crate::expr::stats::StatsProvider; +use crate::match_each_decimal_value_type; +use crate::scalar::DecimalValue; +use crate::scalar::Scalar; + +/// Decimal aggregates paired with whether the [`Widen`] kernel must return a partial. +fn aggregates() -> Vec<(AggregateFnRef, bool)> { + let options = NumericalAggregateOpts::default(); + + vec![ + (MinMax.bind(options), true), + (Min.bind(options), true), + (Max.bind(options), true), + (IsConstant.bind(EmptyOptions), true), + (IsSorted.bind(IsSortedOptions { strict: false }), true), + (IsSorted.bind(IsSortedOptions { strict: true }), true), + (Sum.bind(options), true), + (Sum.bind(NumericalAggregateOpts { skip_nans: false }), false), + (SumV2.bind(options), false), + (First.bind(EmptyOptions), false), + (Last.bind(EmptyOptions), false), + ] +} + +fn canonical_batch(values: &[Option], dtype: &DType) -> DecimalArray { + let decimal = *dtype + .as_decimal_opt() + .vortex_expect("decimal fixture has a decimal dtype"); + + match_each_decimal_value_type!(decimal.canonical_values_type(), |T| { + let values = values.iter().map(|value| { + value.map(|value| { + ::from(value) + .vortex_expect("fixture values fit the parent canonical width") + }) + }); + match dtype.nullability() { + Nullable => DecimalArray::from_option_iter(values, decimal), + NonNullable => DecimalArray::from_iter( + values.map(|value| { + value.vortex_expect("non-nullable fixture contains only valid values") + }), + decimal, + ), + } + }) +} + +/// Encodes a batch and asserts that it exercises the [`Widen`] kernel. +#[track_caller] +fn encode_batch(array: DecimalArray, ctx: &mut ExecutionCtx) -> VortexResult { + let encoded = WidenArray::encode_decimal(array, ctx)?; + assert!(encoded.is::()); + + Ok(encoded) +} + +/// Builds matching encoded and canonical sum inputs with a non-nullable `Decimal(4, 0)` dtype. +fn sum_batch(values: &[i8]) -> VortexResult<(ArrayRef, ArrayRef)> { + let child = DecimalArray::from_iter(values.iter().copied(), DecimalDType::new(2, 0)); + let widen = WidenArray::try_new( + child.into_array(), + DType::Decimal(DecimalDType::new(4, 0), NonNullable), + )?; + let canonical = DecimalArray::from_iter( + values.iter().map(|&value| -> i16 { value.into() }), + DecimalDType::new(4, 0), + ); + + Ok((widen.into_array(), canonical.into_array())) +} + +#[rstest] +#[case::extrema( + vec![ + vec![Some(-99), None, Some(99)], // i8 values with a null. + vec![Some(-9_999), Some(9_999)], // i16 extrema. + vec![Some(-999_999_999), Some(999_999_999)], // i32 extrema. + vec![Some(-999_999_999_999_999_999), Some(999_999_999_999_999_999)], // i64 extrema. + vec![Some(-(10i128.pow(38) - 1)), Some(10i128.pow(38) - 1)], // i128 extrema. + ], + Nullable, + -3, +)] +#[case::ascending( + vec![vec![Some(-99), Some(-1)], vec![Some(0), Some(99)]], + NonNullable, + 2, +)] +#[case::descending_boundary( + vec![vec![Some(0), Some(99)], vec![Some(-99), Some(-1)]], + Nullable, + 0, +)] +#[case::unsorted_batch(vec![vec![Some(99), Some(-99)]], NonNullable, 0)] +#[case::sum_exceeds_child_width( + vec![ + vec![Some(99), Some(98)], // The batch sum exceeds i8::MAX. + vec![Some(-99), Some(-98)], // The batch sum is below i8::MIN. + ], + NonNullable, + 0, +)] +#[case::equal_boundary(vec![vec![Some(7)], vec![Some(7)]], Nullable, 0)] +#[case::constant(vec![vec![Some(7), Some(7)], vec![Some(7)]], NonNullable, 0)] +#[case::changed_constant(vec![vec![Some(7), Some(7)], vec![Some(8)]], Nullable, 0)] +#[case::constant_with_null(vec![vec![Some(7), None, Some(7)]], Nullable, 0)] +#[case::empty(vec![vec![], vec![]], Nullable, 0)] +#[case::all_null(vec![vec![None], vec![None, None]], Nullable, 0)] +#[case::single_null(vec![vec![None]], Nullable, 0)] +#[case::single_null_then_sorted(vec![vec![None], vec![Some(1), Some(2)]], Nullable, 0)] +#[case::empty_then_constant(vec![vec![], vec![Some(7), Some(7)]], Nullable, 0)] +#[case::leading_and_trailing_nulls( + vec![vec![None, Some(-7)], vec![Some(7), None]], + Nullable, + 0, +)] +fn test_decimal_aggregate_partials( + #[case] batches: Vec>>, + #[case] nullability: Nullability, + #[case] scale: i8, +) -> VortexResult<()> { + let mut ctx = SESSION.create_execution_ctx(); + let dtype = DType::Decimal(DecimalDType::new(76, scale), nullability); + + for (aggregate, served) in aggregates() { + assert_aggregate_batches( + &aggregate, + served, + &batches, + &dtype, + |values, dtype, _ctx| Ok(canonical_batch(values, dtype)), + encode_batch, + &mut ctx, + )?; + } + + Ok(()) +} + +/// Overflow is checked after combining each batch with the running sum, for both signs. +#[rstest] +fn test_decimal_sum_running_overflow(#[values(1, -1)] sign: i8) -> VortexResult<()> { + let mut ctx = SESSION.create_execution_ctx(); + let aggregate = Sum.bind(NumericalAggregateOpts { skip_nans: true }); + let dtype = DType::Decimal(DecimalDType::new(4, 0), NonNullable); + let sum_dtype = DecimalDType::new(14, 0); + let unit: i64 = sign.into(); + let seed = unit * (10i64.pow(14) - 100); + let seed = Scalar::decimal(DecimalValue::I64(seed), sum_dtype, Nullable); + + let mut actual = aggregate.accumulator(&dtype)?; + let mut expected = aggregate.accumulator(&dtype)?; + actual.combine_partials(&seed)?; + expected.combine_partials(&seed)?; + + let fits = unit * (10i64.pow(14) - 1); + let batches = [ + ( + sign * 99, + Scalar::decimal(DecimalValue::I64(fits), sum_dtype, Nullable), + ), + (sign, Scalar::null(DType::Decimal(sum_dtype, Nullable))), + ]; + for (value, result) in batches { + let (widen, canonical) = sum_batch(&[value])?; + assert!(kernel_partial(&aggregate, &widen, &mut ctx)?.is_some()); + actual.accumulate(&widen, &mut ctx)?; + expected.accumulate(&canonical, &mut ctx)?; + + assert_eq!(actual.partial_scalar()?, expected.partial_scalar()?); + assert_eq!(actual.partial_scalar()?, result); + } + + Ok(()) +} + +/// A constant sum fits at the precision limit and becomes null after one more row. +/// The fitting sum exceeds the child's sum precision, so it also requires the logical sum dtype. +#[rstest] +#[case::fits(0)] +#[case::overflows(1)] +fn test_decimal_sum_at_precision_bound( + #[values(9, -9)] value: i8, + #[case] extra_rows: usize, +) -> VortexResult<()> { + let mut ctx = SESSION.create_execution_ctx(); + let aggregate = Sum.bind(NumericalAggregateOpts { skip_nans: true }); + let child_decimal = DecimalDType::new(1, 0); + let parent_decimal = DecimalDType::new(3, 0); + let dtype = DType::Decimal(parent_decimal, NonNullable); + let len = 1_111_111_111_111 + extra_rows; + let child_scalar = Scalar::decimal(DecimalValue::I8(value), child_decimal, NonNullable); + let child = ConstantArray::new(child_scalar, len).into_array(); + let widen = WidenArray::try_new(child, dtype.clone())?.into_array(); + + // NB: A kernel decline would materialize the rows of the `Widen`, so assert the kernel first. + assert!(kernel_partial(&aggregate, &widen, &mut ctx)?.is_some()); + + let mut actual = aggregate.accumulator(&dtype)?; + actual.accumulate(&widen, &mut ctx)?; + + let sum_dtype = DecimalDType::new(13, 0); + let unit: i64 = value.signum().into(); + let bound = unit * (10i64.pow(13) - 1); + let result = match extra_rows { + 0 => Scalar::decimal(DecimalValue::I64(bound), sum_dtype, Nullable), + _ => Scalar::null(DType::Decimal(sum_dtype, Nullable)), + }; + assert_eq!(actual.finish()?, result); + + Ok(()) +} + +/// A logical sum must ignore child sum statistics and leave them unchanged. +#[rstest] +#[case::cached(true)] +#[case::absent(false)] +fn test_decimal_sum_ignores_values_statistics(#[case] cached: bool) -> VortexResult<()> { + let mut ctx = SESSION.create_execution_ctx(); + let aggregate = Sum.bind(NumericalAggregateOpts { skip_nans: true }); + let planted = Scalar::decimal(DecimalValue::I64(-1), DecimalDType::new(12, 0), Nullable) + .into_value() + .vortex_expect("the planted sum is not null"); + let (widen, _) = sum_batch(&[40, 2])?; + let child = widen.as_::().values().clone(); + if cached { + child.statistics().set(Stat::Sum, Precision::Exact(planted)); + } + let child_sum = child.statistics().get(Stat::Sum); + + assert!(kernel_partial(&aggregate, &widen, &mut ctx)?.is_some()); + let mut accumulator = aggregate.accumulator(widen.dtype())?; + accumulator.accumulate(&widen, &mut ctx)?; + + let expected = Scalar::decimal(DecimalValue::I64(42), DecimalDType::new(14, 0), Nullable); + assert_eq!(accumulator.finish()?, expected); + assert_eq!(child.statistics().get(Stat::Sum), child_sum); + + Ok(()) +} diff --git a/vortex-array/src/arrays/widen/tests/mod.rs b/vortex-array/src/arrays/widen/tests/mod.rs index df67b13f0fa..337c8bc58eb 100644 --- a/vortex-array/src/arrays/widen/tests/mod.rs +++ b/vortex-array/src/arrays/widen/tests/mod.rs @@ -10,6 +10,7 @@ mod aggregates; mod compute; mod decimal; +mod decimal_aggregates; mod decimal_compare; mod decimal_fallback; mod encoding;