diff --git a/docs/source/contributor-guide/expression-audits/agg_funcs.md b/docs/source/contributor-guide/expression-audits/agg_funcs.md index 164ca2a72c5..089584f8d7f 100644 --- a/docs/source/contributor-guide/expression-audits/agg_funcs.md +++ b/docs/source/contributor-guide/expression-audits/agg_funcs.md @@ -70,6 +70,15 @@ - Comet implementation: the native side delegates to `datafusion_spark::function::aggregate::collect::SparkCollectSet`, which wraps `DistinctArrayAggAccumulator` with `ignore_nulls = true` in a `NullToEmptyListAccumulator` so a final NULL accumulator state becomes an empty array. The `containsNull` mismatch against Spark's declared output type, and its rationale, are identical to [collect_list](#collect-list). - `CometCollectSet` reports `Incompatible` for float and double input when `spark.comet.exec.strictFloatingPoint=true`, because the native distinct comparison treats `NaN == NaN` and collapses repeated `NaN`s into a single element while Spark keeps each one. The native path for floating-point input is then opt-in via `spark.comet.expression.CollectSet.allowIncompatible=true`. All other input types are `Compatible`. +## kurtosis + +- Spark 3.4.3 (audited 2026-07-03): `Kurtosis(child, nullOnDivideByZero)` extends `CentralMomentAgg` with `momentOrder = 4`. Excess kurtosis (Fisher), formula `n * m4 / (m2 * m2) - 3.0`; empty group → `NULL`; `m2 == 0` → `NULL` when `nullOnDivideByZero=true` (default when `spark.sql.legacy.statisticalAggregate=false`) else `NaN`. Any numeric input is cast to `Double` by `ImplicitCastInputTypes`. +- Spark 3.5.8 (audited 2026-07-03): identical to 3.4.3. +- Spark 4.0.1 (audited 2026-07-03): identical to 3.4.3. No collation involvement. +- Spark 4.1.1 (audited 2026-07-03): identical to 3.4.3. +- `CometKurtosis` maps the aggregate to a Comet-owned `Kurtosis` UDAF whose intermediate state (`[n, avg, m2, m3, m4]` Float64) mirrors Spark's `CentralMomentAgg` buffer for `momentOrder = 4`, so Partial output produced by either engine has the same wire format. The Rust update/merge kernels are a direct port of Spark's `updateExpressionsDef` and `mergeExpressions`. `supportsMixedPartialFinal` is left at the default `false`, matching the conservative policy the other `CentralMomentAgg` serdes (`Variance`, `Stddev`) already use in the same file. +- Window use (`kurtosis(x) OVER (...)`) currently falls back to Spark: the window path doesn't wire the Comet aggregate for kurtosis today. + ## median - Spark 3.4.3 (audited 2026-06-24): `Median(child)` is a `RuntimeReplaceableAggregate` with `replacement = Percentile(child, Literal(0.5))`. Catalyst rewrites `median(x)` to `percentile(x, 0.5)` before Comet sees the plan, so it is served by `CometPercentile`. diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 2cf0894ceb8..c5305f81896 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -104,7 +104,7 @@ The tables below list every Spark built-in expression with its current status. | `first_value` | ✅ | Native | | | `grouping` | ✅ | — | Grouping indicator for ROLLUP/CUBE/GROUPING SETS | | `grouping_id` | ✅ | — | Grouping indicator for ROLLUP/CUBE/GROUPING SETS | -| `kurtosis` | 🔜 | — | Not yet implemented natively | +| `kurtosis` | ✅ | Native | Excess kurtosis (Fisher definition). | | `last` | ✅ | Native | | | `last_value` | ✅ | Native | | | `listagg` | 🔜 | — | String aggregation | diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 5cfe74aa412..2fcf83125c9 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -148,9 +148,9 @@ use datafusion_comet_proto::{ use datafusion_comet_spark_expr::{ jvm_udf::JvmScalarUdfExpr, ApproxPercentile, ArrayInsert, Avg, AvgDecimal, Cast, CheckOverflow, Correlation, Covariance, CreateNamedStruct, DecimalRescaleCheckOverflow, GetArrayStructFields, - GetStructField, HllPlusPlus, IfExpr, ListExtract, NormalizeNaNAndZero, Regr, RegrType, - SparkCastOptions, Stddev, SumDecimal, ToJson, UnboundColumn, Variance, WideDecimalBinaryExpr, - WideDecimalOp, + GetStructField, HllPlusPlus, IfExpr, Kurtosis, ListExtract, NormalizeNaNAndZero, Regr, + RegrType, SparkCastOptions, Stddev, SumDecimal, ToJson, UnboundColumn, Variance, + WideDecimalBinaryExpr, WideDecimalOp, }; use itertools::Itertools; use jni::objects::{Global, JObject}; @@ -3127,6 +3127,15 @@ impl PhysicalPlanner { let func = AggregateUDF::new_from_impl(HllPlusPlus::new(expr.precision)); Self::create_aggr_func_expr("approx_count_distinct", schema, vec![child], func) } + AggExprStruct::Kurtosis(expr) => { + let child = self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&schema))?; + let func = AggregateUDF::new_from_impl(Kurtosis::new( + "kurtosis", + expr.null_on_divide_by_zero, + expr.ansi_enabled, + )); + Self::create_aggr_func_expr("kurtosis", schema, vec![child], func) + } } } diff --git a/native/proto/src/proto/expr.proto b/native/proto/src/proto/expr.proto index 10df7706ebd..c8fd671e936 100644 --- a/native/proto/src/proto/expr.proto +++ b/native/proto/src/proto/expr.proto @@ -163,6 +163,7 @@ message AggExpr { HllPlusPlus hllpp = 20; CollectList collectList = 21; Regr regr = 22; + Kurtosis kurtosis = 23; } // Optional filter expression for SQL FILTER (WHERE ...) clause. @@ -257,6 +258,20 @@ message Stddev { StatisticsType stats_type = 4; } +// Excess kurtosis (Fisher definition: normal distribution -> 0). Spark's +// intermediate buffer is `[n, avg, m2, m3, m4]` of Float64 to match the +// `CentralMomentAgg` DeclarativeAggregate wire format for mixed +// Partial/Final execution with Spark. +message Kurtosis { + Expr child = 1; + bool null_on_divide_by_zero = 2; + // Spark builds the final `n * m4 / (m2 * m2) - 3.0` from a `Divide` whose eval mode is + // captured from the session at plan time, so a divisor that underflows to zero raises + // DIVIDE_BY_ZERO under ANSI and yields NULL otherwise. That is a different switch from + // `null_on_divide_by_zero`, which comes from `legacyStatisticalAggregate`. + bool ansi_enabled = 3; +} + message Correlation { Expr child1 = 1; Expr child2 = 2; diff --git a/native/spark-expr/src/agg_funcs/kurtosis.rs b/native/spark-expr/src/agg_funcs/kurtosis.rs new file mode 100644 index 00000000000..8c0761807c2 --- /dev/null +++ b/native/spark-expr/src/agg_funcs/kurtosis.rs @@ -0,0 +1,365 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Spark-compatible excess-kurtosis aggregate. +//! +//! Spark's `Kurtosis` is a `CentralMomentAgg` (`DeclarativeAggregate`) whose +//! intermediate buffer is `[n, avg, m2, m3, m4]` of Float64. This accumulator +//! mirrors that buffer exactly, using the same higher-order online update / +//! merge recurrences (Meng 2015) that `CentralMomentAgg` compiles into +//! catalyst expressions. Matching the wire format lets Spark's Partial and +//! Comet's Final (or vice versa) share intermediate state without a cast. +//! +//! Result formula (excess kurtosis, Fisher definition): +//! +//! * `n == 0` -> NULL +//! * `m2 == 0` -> NULL when `null_on_divide_by_zero`, else NaN +//! * otherwise -> `n * m4 / (m2 * m2) - 3.0` + +use std::mem::size_of; +use std::sync::Arc; + +use arrow::array::{ArrayRef, Float64Array}; +use arrow::datatypes::{DataType, Field, FieldRef}; +use datafusion::common::{downcast_value, Result, ScalarValue}; +use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; +use datafusion::logical_expr::Volatility::Immutable; +use datafusion::logical_expr::{Accumulator, AggregateUDFImpl, Signature}; +use datafusion::physical_expr::expressions::format_state_name; + +use crate::agg_funcs::welford::{moments4_merge, moments4_update}; +use crate::divide_by_zero_error; + +#[derive(Debug, PartialEq, Eq)] +pub struct Kurtosis { + name: String, + signature: Signature, + null_on_divide_by_zero: bool, + ansi_enabled: bool, +} + +impl std::hash::Hash for Kurtosis { + fn hash(&self, state: &mut H) { + self.name.hash(state); + self.signature.hash(state); + self.null_on_divide_by_zero.hash(state); + self.ansi_enabled.hash(state); + } +} + +impl Kurtosis { + pub fn new(name: impl Into, null_on_divide_by_zero: bool, ansi_enabled: bool) -> Self { + Self { + name: name.into(), + signature: Signature::numeric(1, Immutable), + null_on_divide_by_zero, + ansi_enabled, + } + } +} + +impl AggregateUDFImpl for Kurtosis { + fn name(&self) -> &str { + &self.name + } + + fn signature(&self) -> &Signature { + &self.signature + } + + fn return_type(&self, _arg_types: &[DataType]) -> Result { + Ok(DataType::Float64) + } + + fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result> { + Ok(Box::new(KurtosisAccumulator::new( + self.null_on_divide_by_zero, + self.ansi_enabled, + ))) + } + + // No `GroupsAccumulator`: grouped `kurtosis` deliberately runs through DataFusion's generic + // `GroupsAccumulatorAdapter`, which costs one boxed `Accumulator` and a `ScalarValue` round + // trip per group per batch. This is a gap relative to the neighbouring central-moment + // aggregates - `VarianceGroupsAccumulator` keeps flat `Vec` state and + // `StddevGroupsAccumulator` reuses it - and it is recorded here as a decision rather than an + // oversight. The vectorized version wants to land with `skewness`, since both are an + // `evaluate` over the same `[n, avg, m2, m3, m4]` state that `moments4_update` already + // maintains, and one flat-state accumulator should then serve all three. + + // Fields ordered to match Spark's `[n, avg, m2, m3, m4]` buffer so that a + // Spark-produced Partial state can be merged into a Comet-produced Final + // (and vice versa) without a schema conversion. + fn state_fields(&self, _args: StateFieldsArgs) -> Result> { + Ok(vec![ + Arc::new(Field::new( + format_state_name(&self.name, "n"), + DataType::Float64, + true, + )), + Arc::new(Field::new( + format_state_name(&self.name, "avg"), + DataType::Float64, + true, + )), + Arc::new(Field::new( + format_state_name(&self.name, "m2"), + DataType::Float64, + true, + )), + Arc::new(Field::new( + format_state_name(&self.name, "m3"), + DataType::Float64, + true, + )), + Arc::new(Field::new( + format_state_name(&self.name, "m4"), + DataType::Float64, + true, + )), + ]) + } +} + +#[derive(Debug)] +pub struct KurtosisAccumulator { + n: f64, + avg: f64, + m2: f64, + m3: f64, + m4: f64, + null_on_divide_by_zero: bool, + ansi_enabled: bool, +} + +impl KurtosisAccumulator { + pub fn new(null_on_divide_by_zero: bool, ansi_enabled: bool) -> Self { + Self { + n: 0.0, + avg: 0.0, + m2: 0.0, + m3: 0.0, + m4: 0.0, + null_on_divide_by_zero, + ansi_enabled, + } + } +} + +impl Accumulator for KurtosisAccumulator { + fn state(&mut self) -> Result> { + Ok(vec![ + ScalarValue::from(self.n), + ScalarValue::from(self.avg), + ScalarValue::from(self.m2), + ScalarValue::from(self.m3), + ScalarValue::from(self.m4), + ]) + } + + fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> { + let arr = downcast_value!(&values[0], Float64Array).iter().flatten(); + for value in arr { + let (n, avg, m2, m3, m4) = + moments4_update(self.n, self.avg, self.m2, self.m3, self.m4, value); + self.n = n; + self.avg = avg; + self.m2 = m2; + self.m3 = m3; + self.m4 = m4; + } + Ok(()) + } + + fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> { + let ns = downcast_value!(states[0], Float64Array); + let avgs = downcast_value!(states[1], Float64Array); + let m2s = downcast_value!(states[2], Float64Array); + let m3s = downcast_value!(states[3], Float64Array); + let m4s = downcast_value!(states[4], Float64Array); + + for i in 0..ns.len() { + let n2 = ns.value(i); + if n2 == 0.0 { + // Empty partial state contributes nothing and would produce + // divide-by-zero garbage in `delta_n`; skip it. + continue; + } + let (n, avg, m2, m3, m4) = moments4_merge( + self.n, + self.avg, + self.m2, + self.m3, + self.m4, + n2, + avgs.value(i), + m2s.value(i), + m3s.value(i), + m4s.value(i), + ); + self.n = n; + self.avg = avg; + self.m2 = m2; + self.m3 = m3; + self.m4 = m4; + } + Ok(()) + } + + fn evaluate(&mut self) -> Result { + if self.n == 0.0 { + return Ok(ScalarValue::Float64(None)); + } + if self.m2 == 0.0 { + return Ok(ScalarValue::Float64(if self.null_on_divide_by_zero { + None + } else { + Some(f64::NAN) + })); + } + // Spark's guard above is on `m2`, but the division is by `m2 * m2`, and that product can + // underflow to zero while `m2` itself is finite and non-zero (`1e-100` and `2e-100` give + // an `m2` of 5e-201, whose square is 0). Spark's `Divide` then sees a zero divisor and + // applies its own rule, which is the session's ANSI setting rather than + // `null_on_divide_by_zero`. Plain IEEE division here would return NaN instead. + let divisor = self.m2 * self.m2; + if divisor == 0.0 { + return if self.ansi_enabled { + Err(divide_by_zero_error().into()) + } else { + Ok(ScalarValue::Float64(None)) + }; + } + Ok(ScalarValue::Float64(Some(self.n * self.m4 / divisor - 3.0))) + } + + fn size(&self) -> usize { + size_of::() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn eval(values: &[f64], null_on_divide_by_zero: bool) -> Option { + let mut acc = KurtosisAccumulator::new(null_on_divide_by_zero, false); + let arr: ArrayRef = Arc::new(Float64Array::from(values.to_vec())); + acc.update_batch(&[arr]).unwrap(); + match acc.evaluate().unwrap() { + ScalarValue::Float64(v) => v, + other => panic!("expected Float64, got {other:?}"), + } + } + + #[test] + fn empty_group_returns_null() { + assert_eq!(eval(&[], true), None); + } + + #[test] + fn single_value_returns_divide_by_zero_result() { + // m2 == 0 with a single value => NULL when null_on_divide_by_zero, else NaN. + assert_eq!(eval(&[42.0], true), None); + let nan = eval(&[42.0], false).unwrap(); + assert!(nan.is_nan(), "expected NaN, got {nan}"); + } + + #[test] + fn matches_spark_example() { + // Spark's own example from ExpressionDescription: + // SELECT kurtosis(col) FROM VALUES (-10), (-20), (100), (1000) AS tab(col); + // => -0.7014368047529627 + let got = eval(&[-10.0, -20.0, 100.0, 1000.0], true).unwrap(); + assert!((got - -0.7014368047529627_f64).abs() < 1e-12, "got {got}"); + } + + #[test] + fn matches_spark_second_example() { + // SELECT kurtosis(col) FROM VALUES (1), (10), (100), (10), (1) as tab(col); + // => 0.19432323191699075 + let got = eval(&[1.0, 10.0, 100.0, 10.0, 1.0], true).unwrap(); + assert!((got - 0.19432323191699075_f64).abs() < 1e-12, "got {got}"); + } + + #[test] + fn merge_produces_same_result_as_single_batch() { + // Merging two partitions must reproduce the single-batch result. + let values = [-10.0_f64, -20.0, 100.0, 1000.0]; + let full = eval(&values, true).unwrap(); + + let arr_a: ArrayRef = Arc::new(Float64Array::from(values[..2].to_vec())); + let arr_b: ArrayRef = Arc::new(Float64Array::from(values[2..].to_vec())); + + let mut a = KurtosisAccumulator::new(true, false); + a.update_batch(&[arr_a]).unwrap(); + let state_a = a.state().unwrap(); + + let mut b = KurtosisAccumulator::new(true, false); + b.update_batch(&[arr_b]).unwrap(); + + // Represent partition-A state as five single-row Float64 arrays and merge. + let state_arrays: Vec = state_a + .into_iter() + .map(|sv| match sv { + ScalarValue::Float64(v) => { + Arc::new(Float64Array::from(vec![v.unwrap()])) as ArrayRef + } + other => panic!("unexpected state scalar {other:?}"), + }) + .collect(); + b.merge_batch(&state_arrays).unwrap(); + + let merged = match b.evaluate().unwrap() { + ScalarValue::Float64(Some(v)) => v, + other => panic!("expected Float64(Some(_)), got {other:?}"), + }; + assert!((merged - full).abs() < 1e-9, "merged={merged}, full={full}"); + } + + /// `m2` is finite and non-zero here, so Spark's `m2 === 0` guard does not fire, but + /// `m2 * m2` underflows to zero and Spark's `Divide` takes over. That means the session's + /// ANSI setting decides, not `null_on_divide_by_zero`. Plain IEEE division returned NaN. + #[test] + fn divisor_underflow_follows_spark_division_semantics() { + let values = [1e-100, 2e-100]; + + // Confirm the premise rather than assuming it: m2 != 0 but m2 * m2 == 0. + let mut probe = KurtosisAccumulator::new(true, false); + probe + .update_batch(&[Arc::new(Float64Array::from(values.to_vec())) as ArrayRef]) + .unwrap(); + assert_ne!(probe.m2, 0.0, "m2 must be non-zero for this case to bite"); + assert_eq!(probe.m2 * probe.m2, 0.0, "m2 * m2 must underflow to zero"); + + // ANSI off: NULL, for either value of null_on_divide_by_zero, because this path is + // governed by the Divide and not by the m2 == 0 branch. + for null_on_divide_by_zero in [true, false] { + assert_eq!(eval(&values, null_on_divide_by_zero), None); + } + + // ANSI on: DIVIDE_BY_ZERO. + let mut ansi = KurtosisAccumulator::new(true, true); + ansi.update_batch(&[Arc::new(Float64Array::from(values.to_vec())) as ArrayRef]) + .unwrap(); + let err = ansi.evaluate().unwrap_err().to_string(); + assert!( + err.contains("DIVIDE_BY_ZERO"), + "expected a DIVIDE_BY_ZERO error, got {err}" + ); + } +} diff --git a/native/spark-expr/src/agg_funcs/mod.rs b/native/spark-expr/src/agg_funcs/mod.rs index 828f3e989aa..04882ceb8d4 100644 --- a/native/spark-expr/src/agg_funcs/mod.rs +++ b/native/spark-expr/src/agg_funcs/mod.rs @@ -22,6 +22,7 @@ mod correlation; mod covariance; mod hll_plus_plus; mod hll_plus_plus_const; +mod kurtosis; mod percentile; mod quantile_summaries; mod regr; @@ -37,6 +38,7 @@ pub use avg_decimal::AvgDecimal; pub use correlation::Correlation; pub use covariance::Covariance; pub use hll_plus_plus::{hllpp_precision, HllPlusPlus}; +pub use kurtosis::Kurtosis; pub use percentile::SparkPercentile; pub use regr::{Regr, RegrType}; pub use stddev::Stddev; diff --git a/native/spark-expr/src/agg_funcs/welford.rs b/native/spark-expr/src/agg_funcs/welford.rs index bcc44b29887..edb126b55d8 100644 --- a/native/spark-expr/src/agg_funcs/welford.rs +++ b/native/spark-expr/src/agg_funcs/welford.rs @@ -156,3 +156,63 @@ pub(crate) fn covariance_merge( let new_c = c_a + c_b + delta1 * delta2 * count_a * count_b / new_count; (new_count, new_mean1, new_mean2, new_c) } + +/// Online update for the first four central moments `[n, avg, m2, m3, m4]`. Direct port of +/// Spark's `CentralMomentAgg.updateExpressionsDef` for `momentOrder = 4`. +/// +/// The order-4 recurrence subsumes the order-2 one above, so `skewness` (`momentOrder = 3`) +/// is an `evaluate` on top of this same state rather than another copy of the algebra. +#[inline] +pub(crate) fn moments4_update( + n: f64, + avg: f64, + m2: f64, + m3: f64, + m4: f64, + value: f64, +) -> (f64, f64, f64, f64, f64) { + let new_n = n + 1.0; + let delta = value - avg; + let delta_n = delta / new_n; + let new_avg = avg + delta_n; + let new_m2 = m2 + delta * (delta - delta_n); + let delta2 = delta * delta; + let delta_n2 = delta_n * delta_n; + let new_m3 = m3 - 3.0 * delta_n * new_m2 + delta * (delta2 - delta_n2); + let new_m4 = m4 - 4.0 * delta_n * new_m3 - 6.0 * delta_n2 * new_m2 + + delta * (delta * delta2 - delta_n * delta_n2); + (new_n, new_avg, new_m2, new_m3, new_m4) +} + +/// Merge two partial `[n, avg, m2, m3, m4]` states. Direct port of Spark's +/// `CentralMomentAgg.mergeExpressions` for `momentOrder = 4`. +#[inline] +#[allow(clippy::too_many_arguments)] +pub(crate) fn moments4_merge( + n1: f64, + avg1: f64, + m2_1: f64, + m3_1: f64, + m4_1: f64, + n2: f64, + avg2: f64, + m2_2: f64, + m3_2: f64, + m4_2: f64, +) -> (f64, f64, f64, f64, f64) { + let new_n = n1 + n2; + let delta = avg2 - avg1; + let delta_n = if new_n == 0.0 { 0.0 } else { delta / new_n }; + let new_avg = avg1 + delta_n * n2; + let new_m2 = m2_1 + m2_2 + delta * delta_n * n1 * n2; + let new_m3 = m3_1 + + m3_2 + + delta_n * delta_n * delta * n1 * n2 * (n1 - n2) + + 3.0 * delta_n * (n1 * m2_2 - n2 * m2_1); + let new_m4 = m4_1 + + m4_2 + + delta_n * delta_n * delta_n * delta * n1 * n2 * (n1 * n1 - n1 * n2 + n2 * n2) + + 6.0 * delta_n * delta_n * (n1 * n1 * m2_2 + n2 * n2 * m2_1) + + 4.0 * delta_n * (n1 * m3_2 - n2 * m3_1); + (new_n, new_avg, new_m2, new_m3, new_m4) +} diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala index 70646ce2e4c..10f2b2b2dac 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -411,6 +411,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { classOf[CovPopulation] -> CometCovPopulation, classOf[CovSample] -> CometCovSample, classOf[First] -> CometFirst, + classOf[Kurtosis] -> CometKurtosis, classOf[Last] -> CometLast, classOf[Max] -> CometMax, classOf[Min] -> CometMin, diff --git a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala index c811c42559b..0e0d7fe6983 100644 --- a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala +++ b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala @@ -22,7 +22,7 @@ package org.apache.comet.serde import scala.jdk.CollectionConverters._ import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression, Literal} -import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, ApproximatePercentile, Average, BitAndAgg, BitOrAgg, BitXorAgg, BloomFilterAggregate, CentralMomentAgg, CollectList, CollectSet, Corr, Count, Covariance, CovPopulation, CovSample, First, HyperLogLogPlusPlus, Last, Max, Min, Percentile, RegrIntercept, RegrR2, RegrReplacement, RegrSlope, RegrSXY, StddevPop, StddevSamp, Sum, VariancePop, VarianceSamp} +import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, ApproximatePercentile, Average, BitAndAgg, BitOrAgg, BitXorAgg, BloomFilterAggregate, CentralMomentAgg, CollectList, CollectSet, Corr, Count, Covariance, CovPopulation, CovSample, First, HyperLogLogPlusPlus, Kurtosis, Last, Max, Min, Percentile, RegrIntercept, RegrR2, RegrReplacement, RegrSlope, RegrSXY, StddevPop, StddevSamp, Sum, VariancePop, VarianceSamp} import org.apache.spark.sql.catalyst.util.ArrayData import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType, DataType, DateType, DecimalType, DoubleType, FloatType, IntegerType, LongType, NumericType, ShortType, StringType, TimestampNTZType, TimestampType} @@ -1116,6 +1116,44 @@ object CometApproxCountDistinct extends CometAggregateExpressionSerde[HyperLogLo } } +object CometKurtosis extends CometAggregateExpressionSerde[Kurtosis] { + + // Not marked safe for mixed partial/final: follows the same policy as `Variance` / `Stddev`, + // whose complex `[n, avg, m2, ...]` buffer is not certified compatible across engines. The + // native accumulator does mirror Spark's `[n, avg, m2, m3, m4]` wire format, so lifting this + // to `true` should be considered together with the other `CentralMomentAgg` serdes. + + override def convert( + aggExpr: AggregateExpression, + kurtosis: Kurtosis, + inputs: Seq[Attribute], + binding: Boolean, + conf: SQLConf): Option[ExprOuterClass.AggExpr] = { + val child = kurtosis.child + val childExpr = exprToProto(child, inputs, binding) + + if (childExpr.isDefined) { + val builder = ExprOuterClass.Kurtosis.newBuilder() + builder.setChild(childExpr.get) + builder.setNullOnDivideByZero(kurtosis.nullOnDivideByZero) + // Spark's evaluate expression divides by `m2 * m2`, and that `Divide` picks up its eval + // mode from the session. `m2` can be non-zero while `m2 * m2` underflows to zero, which + // the `m2 === 0` guard above it does not catch, so the native side needs to know whether + // that divisor should raise or return null. + builder.setAnsiEnabled(conf.ansiEnabled) + + Some( + ExprOuterClass.AggExpr + .newBuilder() + .setKurtosis(builder) + .build()) + } else { + withFallbackReason(aggExpr, "Child expression or data type not supported") + None + } + } +} + object AggSerde { import org.apache.spark.sql.types._ diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/kurtosis.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/kurtosis.sql new file mode 100644 index 00000000000..dd5a4d3813d --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/aggregate/kurtosis.sql @@ -0,0 +1,263 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to you under the Apache License, Version 2.0 (the +-- "License"); you may not use this file except in compliance +-- with the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, +-- software distributed under the License is distributed on an +-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +-- KIND, either express or implied. See the License for the +-- specific language governing permissions and limitations +-- under the License. + +-- ConfigMatrix: parquet.enable.dictionary=false,true + +-- ============================================================ +-- Setup +-- ============================================================ + +statement +CREATE TABLE k_dbl(v double, grp string) USING parquet + +statement +INSERT INTO k_dbl VALUES + (-10.0, 'g1'), (-20.0, 'g1'), (100.0, 'g1'), (1000.0, 'g1'), + (1.0, 'g2'), (10.0, 'g2'), (100.0, 'g2'), (10.0, 'g2'), (1.0, 'g2'), + (42.0, 'g3'), + (NULL, 'g4'), (NULL, 'g4'), + (7.0, 'g5'), (7.0, 'g5'), (7.0, 'g5') + +statement +CREATE TABLE k_int(v int, grp string) USING parquet + +statement +INSERT INTO k_int VALUES + (1, 'g1'), (10, 'g1'), (100, 'g1'), (10, 'g1'), (1, 'g1'), + (NULL, 'g2'), (5, 'g2') + +statement +CREATE TABLE k_dec(v decimal(10,2), grp string) USING parquet + +statement +INSERT INTO k_dec VALUES + (1.50, 'g1'), (2.50, 'g1'), (3.50, 'g1'), (4.50, 'g1') + +statement +CREATE TABLE k_empty(v double) USING parquet + +statement +CREATE TABLE k_spark_ex1(v double) USING parquet + +statement +INSERT INTO k_spark_ex1 VALUES (-10.0), (-20.0), (100.0), (1000.0) + +statement +CREATE TABLE k_spark_ex2(v double) USING parquet + +statement +INSERT INTO k_spark_ex2 VALUES (1.0), (10.0), (100.0), (10.0), (1.0) + +statement +CREATE TABLE k_single(v double) USING parquet + +statement +INSERT INTO k_single VALUES (42.0) + +statement +CREATE TABLE k_const(v double) USING parquet + +statement +INSERT INTO k_const VALUES (7.0), (7.0), (7.0) + +statement +CREATE TABLE k_lit(x int) USING parquet + +statement +INSERT INTO k_lit VALUES (1) + +-- ============================================================ +-- Spark's own example: matches -0.7014368047529627. +-- ============================================================ + +query +SELECT kurtosis(v) FROM k_spark_ex1 + +-- Spark's second example: matches 0.19432323191699075. +query +SELECT kurtosis(v) FROM k_spark_ex2 + +-- ============================================================ +-- GROUP BY over doubles: covers a "normal" group (g1), a heavier +-- group (g2), a single-value group (g3, m2=0 => NULL by default), +-- an all-NULL group (g4 => NULL), and constants (g5, m2=0). +-- ============================================================ + +query +SELECT grp, kurtosis(v) FROM k_dbl GROUP BY grp ORDER BY grp + +-- ============================================================ +-- Global aggregate (no GROUP BY). +-- ============================================================ + +query +SELECT kurtosis(v) FROM k_dbl + +-- Empty table returns NULL. +query +SELECT kurtosis(v) FROM k_empty + +-- ============================================================ +-- Integer input: promoted to Double by Spark's ImplicitCastInputTypes. +-- ============================================================ + +query +SELECT grp, kurtosis(v) FROM k_int GROUP BY grp ORDER BY grp + +-- ============================================================ +-- Decimal input. +-- ============================================================ + +query +SELECT grp, kurtosis(v) FROM k_dec GROUP BY grp ORDER BY grp + +-- ============================================================ +-- Literal argument (constant folded; still exercises planning). +-- ============================================================ + +query +SELECT kurtosis(1.0) FROM k_lit + +query +SELECT kurtosis(NULL) FROM k_lit + +-- ============================================================ +-- Divide-by-zero cases under default (nullOnDivideByZero=true): +-- single-value and all-equal groups both yield NULL. See +-- kurtosis_legacy.sql for the `legacyStatisticalAggregate=true` +-- variant that returns NaN instead. +-- ============================================================ + +query +SELECT kurtosis(v) FROM k_single + +query +SELECT kurtosis(v) FROM k_const + +-- ============================================================ +-- FILTER (WHERE ...) — Partial only carries the filter. +-- ============================================================ + +query +SELECT grp, kurtosis(v) FILTER (WHERE v > 0) FROM k_dbl GROUP BY grp ORDER BY grp + +-- ============================================================ +-- Additional coverage requested by the audit. +-- ============================================================ + +statement +CREATE TABLE k_float(v float, grp string) USING parquet + +statement +INSERT INTO k_float VALUES + (CAST(1.0 AS FLOAT), 'g1'), (CAST(10.0 AS FLOAT), 'g1'), + (CAST(100.0 AS FLOAT), 'g1'), (CAST(10.0 AS FLOAT), 'g1'), + (CAST(1.0 AS FLOAT), 'g1') + +statement +CREATE TABLE k_long(v bigint, grp string) USING parquet + +statement +INSERT INTO k_long VALUES + (1, 'g1'), (10, 'g1'), (100, 'g1'), (10, 'g1'), (1, 'g1') + +statement +CREATE TABLE k_wnd(k int, part string, v double) USING parquet + +statement +INSERT INTO k_wnd VALUES + (1, 'p1', 1.0), (2, 'p1', 1.0), (3, 'p1', 2.0), (4, 'p1', 2.0), + (5, 'p1', 3.0), (6, 'p1', 3.0), (7, 'p1', 3.0), + (8, 'p2', 1.0), (9, 'p2', 2.0), (10, 'p2', 5.0) + +-- Float input: promoted to Double by Spark's ImplicitCastInputTypes. +query +SELECT grp, kurtosis(v) FROM k_float GROUP BY grp ORDER BY grp + +-- BigInt input. +query +SELECT grp, kurtosis(v) FROM k_long GROUP BY grp ORDER BY grp + +-- Mixed with other CentralMomentAgg siblings in one query. +query +SELECT kurtosis(v), avg(v), stddev(v), count(*) FROM k_dbl WHERE v IS NOT NULL + +-- Window use: matches Spark's `DataFrameWindowFunctionsSuite` +-- "skewness and kurtosis functions in window" test. Comet's window +-- path doesn't wire `kurtosis` as a window aggregate today, so this +-- falls back to Spark. +query expect_fallback(is not supported for window function) +SELECT k, + kurtosis(v) OVER (PARTITION BY part ORDER BY k + ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) +FROM k_wnd ORDER BY k + +-- ============================================================ +-- Numerical stress: NaN/Infinity/-Infinity in a group. Spark +-- propagates these through arithmetic, so the aggregate is +-- expected to produce NaN or NULL rather than a finite value. +-- Use spark_answer_only because our Welford recurrence produces +-- byte-identical NaN payloads with a different bit pattern from +-- Spark's DeclarativeAggregate compilation; result-value equality +-- via Spark's own comparison is what the harness needs to see. +-- ============================================================ + +statement +CREATE TABLE k_nan(v double) USING parquet + +statement +INSERT INTO k_nan VALUES + (1.0), (2.0), (CAST('NaN' AS DOUBLE)), (3.0) + +statement +CREATE TABLE k_inf(v double) USING parquet + +statement +INSERT INTO k_inf VALUES + (1.0), (2.0), (CAST('Infinity' AS DOUBLE)), (3.0) + +statement +CREATE TABLE k_neg_inf(v double) USING parquet + +statement +INSERT INTO k_neg_inf VALUES + (1.0), (2.0), (CAST('-Infinity' AS DOUBLE)), (3.0) + +query spark_answer_only +SELECT kurtosis(v) FROM k_nan + +query spark_answer_only +SELECT kurtosis(v) FROM k_inf + +query spark_answer_only +SELECT kurtosis(v) FROM k_neg_inf + +-- ============================================================ +-- Large-magnitude inputs: Welford is numerically stable but the +-- final `n * m4 / (m2 * m2) - 3.0` can still differ from Spark's +-- codegen at high magnitudes. Use spark_answer_only. +-- ============================================================ + +statement +CREATE TABLE k_big(v double) USING parquet + +statement +INSERT INTO k_big VALUES + (1.0e15), (2.0e15), (3.0e15), (4.0e15), (5.0e15) + +query spark_answer_only +SELECT kurtosis(v) FROM k_big diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/kurtosis_legacy.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/kurtosis_legacy.sql new file mode 100644 index 00000000000..28e3424ac49 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/aggregate/kurtosis_legacy.sql @@ -0,0 +1,41 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to you under the Apache License, Version 2.0 (the +-- "License"); you may not use this file except in compliance +-- with the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, +-- software distributed under the License is distributed on an +-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +-- KIND, either express or implied. See the License for the +-- specific language governing permissions and limitations +-- under the License. + +-- Legacy statistical-aggregate semantics: divide-by-zero returns NaN +-- instead of NULL. Spark passes `!legacyStatisticalAggregate` as +-- `nullOnDivideByZero` when constructing the aggregate. +-- Config: spark.sql.legacy.statisticalAggregate=true + +statement +CREATE TABLE k_legacy_single(v double) USING parquet + +statement +INSERT INTO k_legacy_single VALUES (42.0) + +statement +CREATE TABLE k_legacy_const(v double) USING parquet + +statement +INSERT INTO k_legacy_const VALUES (7.0), (7.0), (7.0) + +-- Single-value group: m2 == 0. Expect NaN (not NULL). +query +SELECT kurtosis(v) FROM k_legacy_single + +-- All-equal group: same divide-by-zero shape. +query +SELECT kurtosis(v) FROM k_legacy_const