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;