rich7420 commented on code in PR #4818: URL: https://github.com/apache/datafusion-comet/pull/4818#discussion_r4554645493
########## docs/source/contributor-guide/expression-audits/agg_funcs.md: ########## @@ -72,6 +72,15 @@ - Performance (tuned 2026-09-09, PR [#5803](https://github.com/apache/datafusion-comet/pull/5803)): as for [collect_list](#collect-list), grouped `collect_set` no longer goes through `GroupsAccumulatorAdapter`. `CollectSetGroupsAccumulator` encodes each batch once for all of its groups and deduplicates against an open-addressed index over a shared arena, replacing one hash table plus one owned `Row` per distinct value per group. 60-93% faster ([#5797](https://github.com/apache/datafusion-comet/issues/5797)). Benchmark: `benches/collect.rs`. - `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. Review Comment: Looks like `supportsMixedPartialFinal` got split into `supportsSparkPartialToNativeFinal` and `supportsNativePartialToSparkFinal` in #5421. Could this line use the new names? ########## 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<H: std::hash::Hasher>(&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<String>, 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<DataType> { + Ok(DataType::Float64) + } + + fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> { + 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<f64>` 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<Vec<FieldRef>> { + 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<Vec<ScalarValue>> { + 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<ScalarValue> { + 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()) Review Comment: Maybe reuse `ZeroDivisor` here like `corr` does? Then the error picks up the `kurtosis(v)` query context like Spark's, and we can drop the `ansi_enabled` bool. I tried it locally with an `EvalMode` field in place of `ansi_enabled`. The context matches Spark's on 3.5 and 4.1. ########## 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<H: std::hash::Hasher>(&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<String>, 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<DataType> { + Ok(DataType::Float64) + } + + fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> { + 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<f64>` 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<Vec<FieldRef>> { + 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<Vec<ScalarValue>> { + 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 { Review Comment: Could we keep empty partials here, like `VarianceAccumulator::merge_batch` does after #6076? With values near `5e77` and an all-NULL file, I got `NaN` from Spark but `-1.22` from Comet. Repro, using the two-file setup from `statistical aggregates merge large nearby values across partitions`: ```scala (0 until 10).map(i => (0, Option(5e77 * (1 + i * 1e-9)))).toDF("g", "v") .coalesce(1).write.mode("append").parquet(path) Seq((0, Option.empty[Double])).toDF("g", "v") .coalesce(1).write.mode("append").parquet(path) // SELECT kurtosis(v) FROM t -> Spark NaN, Comet -1.2242423537682872 ``` Dropping the `n2 == 0.0` check fixes it for me, and the existing tests still pass. ########## spark/src/test/resources/sql-tests/expressions/aggregate/kurtosis.sql: ########## @@ -0,0 +1,295 @@ +-- 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 Review Comment: Could the four `spark_answer_only` queries here just use plain `query`, and line 294 use `expect_error_class`? Otherwise they'd still pass if `kurtosis` fell back to Spark. With `spark.comet.expression.Kurtosis.enabled=false`, the current versions still pass for me. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
