mbutrovich commented on code in PR #4818: URL: https://github.com/apache/datafusion-comet/pull/4818#discussion_r3659324426
########## native/spark-expr/src/agg_funcs/kurtosis.rs: ########## @@ -0,0 +1,363 @@ +// 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; + +#[derive(Debug, PartialEq, Eq)] +pub struct Kurtosis { + name: String, + signature: Signature, + null_on_divide_by_zero: 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); + } +} + +impl Kurtosis { + pub fn new(name: impl Into<String>, null_on_divide_by_zero: bool) -> Self { + Self { + name: name.into(), + signature: Signature::numeric(1, Immutable), + null_on_divide_by_zero, + } + } +} + +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, + ))) + } + + // 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, + )), + ]) + } + + fn default_value(&self, _data_type: &DataType) -> Result<ScalarValue> { Review Comment: `default_value` returns `ScalarValue::Float64(None)`, which is what the `AggregateUDFImpl` trait default already produces for a `Float64` return type via `ScalarValue::try_from(data_type)` (`datafusion/expr/src/udaf.rs:828-830`). This override can be deleted. This is the same redundancy already flagged on `mode.rs:116` in #4782. Worth fixing in both so the pattern does not propagate to the next aggregate. ########## native/spark-expr/src/agg_funcs/kurtosis.rs: ########## @@ -0,0 +1,363 @@ +// 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; + +#[derive(Debug, PartialEq, Eq)] +pub struct Kurtosis { + name: String, + signature: Signature, + null_on_divide_by_zero: 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); + } +} + +impl Kurtosis { + pub fn new(name: impl Into<String>, null_on_divide_by_zero: bool) -> Self { + Self { + name: name.into(), + signature: Signature::numeric(1, Immutable), + null_on_divide_by_zero, + } + } +} + +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, + ))) + } + + // 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, + )), + ]) + } + + fn default_value(&self, _data_type: &DataType) -> Result<ScalarValue> { + Ok(ScalarValue::Float64(None)) + } +} + +/// Online update for the first four central moments. Direct port of Spark's +/// `CentralMomentAgg.updateExpressionsDef` for `momentOrder = 4`. +#[inline] +fn kurtosis_update( Review Comment: There is no `GroupsAccumulator` here, so grouped `kurtosis` runs through DataFusion's generic `GroupsAccumulatorAdapter`, one boxed `Accumulator` and a `ScalarValue` round trip per group per batch. That is a defensible starting point, but it is worth stating explicitly why, because the neighbouring central-moment aggregates in this same crate do have one: `VarianceGroupsAccumulator` (`variance.rs:268`) keeps flat `Vec<f64>` state and `StddevGroupsAccumulator` (`stddev.rs:206`) reuses it. 4817 in this same batch also ships a vectorized grouped accumulator and benchmarks it. Please either add one or record in the PR description that grouped kurtosis is adapter-backed, so the gap is a decision rather than an oversight. ########## native/spark-expr/src/agg_funcs/kurtosis.rs: ########## @@ -0,0 +1,363 @@ +// 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; + +#[derive(Debug, PartialEq, Eq)] +pub struct Kurtosis { + name: String, + signature: Signature, + null_on_divide_by_zero: 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); + } +} + +impl Kurtosis { + pub fn new(name: impl Into<String>, null_on_divide_by_zero: bool) -> Self { + Self { + name: name.into(), + signature: Signature::numeric(1, Immutable), + null_on_divide_by_zero, + } + } +} + +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, + ))) + } + + // 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, + )), + ]) + } + + fn default_value(&self, _data_type: &DataType) -> Result<ScalarValue> { + Ok(ScalarValue::Float64(None)) + } +} + +/// Online update for the first four central moments. Direct port of Spark's Review Comment: `kurtosis_update` and `kurtosis_merge` are free functions in this file, but `welford.rs` is exactly the module this crate uses to share moment recurrences across aggregates: it already holds `variance_update`, `variance_merge`, `covariance_update`, `covariance_merge` and `finalize_moments` (`welford.rs:27-141`), consumed by both `variance.rs` and `covariance.rs`. The order-4 recurrence subsumes the order-2 one, and `skewness` is the obvious next sibling and needs `[n, avg, m2, m3]` from the same recurrence. Putting these two functions in `welford.rs` alongside the existing ones keeps all the moment math in one place and means skewness is a `momentOrder = 3` evaluate on top of shared state rather than a third copy of the same algebra. ########## spark/src/test/resources/sql-tests/expressions/aggregate/kurtosis.sql: ########## @@ -0,0 +1,273 @@ +-- 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 + +-- ============================================================ +-- Skewness is Spark's sibling in CentralMomentAgg; we don't +-- implement it here, so it should fall back. (This documents the +-- boundary; if we add skewness later, the expect_fallback +-- becomes a plain query.) +-- ============================================================ + +query expect_fallback(unsupported Spark aggregate function: skewness) Review Comment: `expect_fallback(unsupported Spark aggregate function: skewness)` pins Comet's current lack of skewness support as expected behavior in the kurtosis fixture. That is the wrong place for it. It couples an unrelated expression's status to this file, and when skewness lands, this assertion fails in a fixture whose name gives no hint why. The comment above at `:161` already anticipates this. Either drop the query or move it to a fixture named for skewness. -- 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]
