From fb2f52bcc276c34ca61bbaff8e913aa3c2d22b79 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 1 Jul 2026 09:43:30 -0600 Subject: [PATCH 1/2] feat: support native mode aggregate function Add native support for the Spark `mode` aggregate, the most frequent value within a group. Spark breaks ties on the default `mode(col)` form non-deterministically (the chosen value depends on JVM hash-map iteration order), so the function is registered as Incompatible and opt-in via allowIncompatible; Comet resolves ties deterministically by returning the smallest tied value. NULLs are ignored, empty input returns NULL, and float keys are normalized to match Spark. The deterministic-flag and WITHIN GROUP forms fall back to Spark. Closes #3970 --- docs/source/user-guide/latest/expressions.md | 2 +- native/core/src/execution/planner.rs | 10 +- native/proto/src/proto/expr.proto | 6 + native/spark-expr/src/agg_funcs/mod.rs | 2 + native/spark-expr/src/agg_funcs/mode.rs | 496 ++++++++++++++++++ .../apache/comet/serde/QueryPlanSerde.scala | 1 + .../org/apache/comet/serde/aggregates.scala | 71 ++- .../apache/spark/sql/comet/operators.scala | 17 +- .../apache/comet/shims/CometTypeShim.scala | 6 + .../apache/comet/shims/CometTypeShim.scala | 6 + .../sql-tests/expressions/aggregate/mode.sql | 186 +++++++ 11 files changed, 795 insertions(+), 8 deletions(-) create mode 100644 native/spark-expr/src/agg_funcs/mode.rs create mode 100644 spark/src/test/resources/sql-tests/expressions/aggregate/mode.sql diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index ae9530d8ec..46ea5314fb 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -100,7 +100,7 @@ The tables below list every Spark built-in expression with its current status. | `median` | ✅ | Rewrites to `percentile(col, 0.5)`; falls back by default, opt-in via allowIncompatible ([#4719](https://github.com/apache/datafusion-comet/issues/4719)) | | `min` | ✅ | | | `min_by` | 🔜 | [#3841](https://github.com/apache/datafusion-comet/issues/3841) | -| `mode` | 🔜 | [#3970](https://github.com/apache/datafusion-comet/issues/3970) | +| `mode` | ✅ | `mode(col)` only; Spark breaks ties non-deterministically, so Comet returns the smallest tied value and falls back by default, opt-in via allowIncompatible ([#3970](https://github.com/apache/datafusion-comet/issues/3970)) | | `percentile` | ✅ | Single literal percentage on numeric input; array of percentages and a frequency argument fall back to Spark. Falls back by default, opt-in via allowIncompatible ([#4719](https://github.com/apache/datafusion-comet/issues/4719)) | | `percentile_cont` | ✅ | Spark 4.0+ `WITHIN GROUP (ORDER BY ...)`; ascending only, `DESC` falls back to Spark. Falls back by default, opt-in via allowIncompatible ([#4719](https://github.com/apache/datafusion-comet/issues/4719)) | | `percentile_disc` | 🔜 | Percentile aggregate | diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 049c8c13e9..3c56088877 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -130,8 +130,8 @@ use datafusion_comet_proto::{ use datafusion_comet_spark_expr::{ jvm_udf::JvmScalarUdfExpr, ArrayInsert, Avg, AvgDecimal, Cast, CheckOverflow, Correlation, Covariance, CreateNamedStruct, DecimalRescaleCheckOverflow, GetArrayStructFields, - GetStructField, IfExpr, ListExtract, NormalizeNaNAndZero, SparkCastOptions, Stddev, SumDecimal, - ToJson, UnboundColumn, Variance, WideDecimalBinaryExpr, WideDecimalOp, + GetStructField, IfExpr, ListExtract, Mode, NormalizeNaNAndZero, SparkCastOptions, Stddev, + SumDecimal, ToJson, UnboundColumn, Variance, WideDecimalBinaryExpr, WideDecimalOp, }; use itertools::Itertools; use jni::objects::{Global, JObject}; @@ -2643,6 +2643,12 @@ impl PhysicalPlanner { let func = AggregateUDF::new_from_impl(SparkCollectSet::new()); Self::create_aggr_func_expr("collect_set", schema, vec![child], func) } + AggExprStruct::Mode(expr) => { + let child = self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&schema))?; + let datatype = to_arrow_datatype(expr.datatype.as_ref().unwrap()); + let func = AggregateUDF::new_from_impl(Mode::new(datatype)); + Self::create_aggr_func_expr("mode", schema, vec![child], func) + } } } diff --git a/native/proto/src/proto/expr.proto b/native/proto/src/proto/expr.proto index 103af9bf10..532a58e7d6 100644 --- a/native/proto/src/proto/expr.proto +++ b/native/proto/src/proto/expr.proto @@ -144,6 +144,7 @@ message AggExpr { BloomFilterAgg bloomFilterAgg = 16; CollectSet collectSet = 17; Percentile percentile = 18; + Mode mode = 19; } // Optional filter expression for SQL FILTER (WHERE ...) clause. @@ -275,6 +276,11 @@ message CollectSet { DataType datatype = 2; } +message Mode { + Expr child = 1; + DataType datatype = 2; +} + enum EvalMode { LEGACY = 0; TRY = 1; diff --git a/native/spark-expr/src/agg_funcs/mod.rs b/native/spark-expr/src/agg_funcs/mod.rs index 2a0322e46c..aaa80b2287 100644 --- a/native/spark-expr/src/agg_funcs/mod.rs +++ b/native/spark-expr/src/agg_funcs/mod.rs @@ -19,6 +19,7 @@ mod avg; mod avg_decimal; mod correlation; mod covariance; +mod mode; mod stddev; mod sum_decimal; mod sum_int; @@ -29,6 +30,7 @@ pub use avg::Avg; pub use avg_decimal::AvgDecimal; pub use correlation::Correlation; pub use covariance::Covariance; +pub use mode::Mode; pub use stddev::Stddev; pub use sum_decimal::SumDecimal; pub use sum_int::SumInteger; diff --git a/native/spark-expr/src/agg_funcs/mode.rs b/native/spark-expr/src/agg_funcs/mode.rs new file mode 100644 index 0000000000..4576fc3203 --- /dev/null +++ b/native/spark-expr/src/agg_funcs/mode.rs @@ -0,0 +1,496 @@ +/* + * 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. + */ + +use arrow::array::{Array, ArrayRef, AsArray, BooleanArray, StructArray}; +use arrow::datatypes::{DataType, Field, FieldRef, Fields, Int64Type}; +use datafusion::common::{internal_datafusion_err, Result, ScalarValue}; +use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; +use datafusion::logical_expr::{ + Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature, Volatility, +}; +use datafusion::physical_expr::expressions::format_state_name; +use std::cmp::Ordering; +use std::collections::HashMap; +use std::mem::size_of; +use std::sync::Arc; + +/// Spark's `mode` aggregate: returns the most frequent value within a group, ignoring NULLs. +/// +/// Spark breaks ties on the default `mode(col)` form non-deterministically (the value is chosen +/// by JVM `OpenHashMap` iteration order), which a native hash map cannot reproduce bit-for-bit. +/// Comet resolves ties deterministically by returning the smallest value, so this function is +/// registered as `Incompatible` on the Scala side and is opt-in via `allowIncompatible`. +/// +/// Float keys are normalized before counting (`-0.0` becomes `0.0` and every `NaN` becomes a +/// canonical `NaN`) to match Spark's `NormalizeFloatingNumbers` behaviour so that counts agree. +/// +/// Spark's `Mode` is a `TypedImperativeAggregate` with a single aggregation-buffer attribute, so +/// the intermediate state is a single struct field `{ values: list, counts: list }` (a +/// parallel-array encoding of the frequency map) to keep the partial/final buffer schemas aligned +/// with Spark. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct Mode { + name: String, + signature: Signature, + data_type: DataType, +} + +impl Mode { + pub fn new(data_type: DataType) -> Self { + Self { + name: "mode".to_string(), + signature: Signature::any(1, Volatility::Immutable), + data_type, + } + } +} + +/// Fields of the single struct state column `{values: list, counts: list}`. +fn state_struct_fields(data_type: &DataType) -> Fields { + let values_list = DataType::List(Arc::new(Field::new_list_field(data_type.clone(), true))); + let counts_list = DataType::List(Arc::new(Field::new_list_field(DataType::Int64, true))); + Fields::from(vec![ + Field::new("values", values_list, false), + Field::new("counts", counts_list, false), + ]) +} + +/// Build the single-column struct state array holding one `{values, counts}` row per map. +fn build_state(data_type: &DataType, maps: &[&HashMap]) -> Result { + let mut value_lists = Vec::with_capacity(maps.len()); + let mut count_lists = Vec::with_capacity(maps.len()); + for map in maps { + let mut values = Vec::with_capacity(map.len()); + let mut counts = Vec::with_capacity(map.len()); + for (value, &count) in map.iter() { + values.push(value.clone()); + counts.push(ScalarValue::Int64(Some(count))); + } + value_lists.push(ScalarValue::List(ScalarValue::new_list( + &values, data_type, true, + ))); + count_lists.push(ScalarValue::List(ScalarValue::new_list( + &counts, + &DataType::Int64, + true, + ))); + } + let values = ScalarValue::iter_to_array(value_lists)?; + let counts = ScalarValue::iter_to_array(count_lists)?; + Ok(StructArray::new( + state_struct_fields(data_type), + vec![values, counts], + None, + )) +} + +impl AggregateUDFImpl for Mode { + fn name(&self) -> &str { + &self.name + } + + fn signature(&self) -> &Signature { + &self.signature + } + + fn return_type(&self, _arg_types: &[DataType]) -> Result { + Ok(self.data_type.clone()) + } + + fn default_value(&self, _data_type: &DataType) -> Result { + ScalarValue::try_from(&self.data_type) + } + + fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result> { + Ok(Box::new(ModeAccumulator::new(self.data_type.clone()))) + } + + fn state_fields(&self, _args: StateFieldsArgs) -> Result> { + Ok(vec![Arc::new(Field::new( + format_state_name(&self.name, "freq"), + DataType::Struct(state_struct_fields(&self.data_type)), + false, + ))]) + } + + fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool { + true + } + + fn create_groups_accumulator( + &self, + _args: AccumulatorArgs, + ) -> Result> { + Ok(Box::new(ModeGroupsAccumulator::new(self.data_type.clone()))) + } +} + +/// Normalize a scalar key so that Spark's floating-point normalization is honoured: `-0.0` and +/// `0.0` collapse to the same key and all `NaN` bit patterns collapse to a canonical `NaN`. +fn normalize_key(value: ScalarValue) -> ScalarValue { + /// Collapse `-0.0`/`0.0` and every `NaN` to a canonical form for one float variant. + macro_rules! normalize_float { + ($variant:path, $f:expr, $nan:expr) => { + if $f == 0.0 { + $variant(Some(0.0)) + } else if $f.is_nan() { + $variant(Some($nan)) + } else { + $variant(Some($f)) + } + }; + } + match value { + ScalarValue::Float32(Some(f)) => normalize_float!(ScalarValue::Float32, f, f32::NAN), + ScalarValue::Float64(Some(f)) => normalize_float!(ScalarValue::Float64, f, f64::NAN), + other => other, + } +} + +/// Add each non-null value in `array` to `map`, normalizing float keys. +fn count_values(map: &mut HashMap, array: &ArrayRef, idx: usize) -> Result<()> { + if array.is_null(idx) { + return Ok(()); + } + let key = normalize_key(ScalarValue::try_from_array(array, idx)?); + *map.entry(key).or_insert(0) += 1; + Ok(()) +} + +/// Fold row `row` of the struct-state columns (`{values, counts}`) into `map`. +fn merge_state_row( + map: &mut HashMap, + values_list: &arrow::array::ListArray, + counts_list: &arrow::array::ListArray, + row: usize, +) -> Result<()> { + if values_list.is_null(row) { + return Ok(()); + } + let values = values_list.value(row); + let counts = counts_list.value(row); + let counts = counts + .as_primitive_opt::() + .ok_or_else(|| internal_datafusion_err!("mode state counts must be Int64"))?; + for i in 0..values.len() { + if values.is_null(i) { + continue; + } + let key = normalize_key(ScalarValue::try_from_array(&values, i)?); + *map.entry(key).or_insert(0) += counts.value(i); + } + Ok(()) +} + +/// Pick the mode from a frequency map: the value with the highest count, breaking ties by the +/// smallest value. Returns a null scalar of `data_type` when the map is empty. +fn eval_mode(counts: &HashMap, data_type: &DataType) -> Result { + let mut best: Option<(&ScalarValue, i64)> = None; + for (value, &count) in counts.iter() { + let wins = match best { + None => true, + Some((best_value, best_count)) => { + count > best_count + || (count == best_count + && value.partial_cmp(best_value) == Some(Ordering::Less)) + } + }; + if wins { + best = Some((value, count)); + } + } + match best { + Some((value, _)) => Ok(value.clone()), + None => ScalarValue::try_from(data_type), + } +} + +/// Non-grouped accumulator backing global `mode` aggregation. +#[derive(Debug)] +pub struct ModeAccumulator { + counts: HashMap, + data_type: DataType, +} + +impl ModeAccumulator { + fn new(data_type: DataType) -> Self { + Self { + counts: HashMap::new(), + data_type, + } + } +} + +impl Accumulator for ModeAccumulator { + fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> { + let array = &values[0]; + for i in 0..array.len() { + count_values(&mut self.counts, array, i)?; + } + Ok(()) + } + + fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> { + let structs = states[0].as_struct(); + let values_list = structs.column(0).as_list::(); + let counts_list = structs.column(1).as_list::(); + for row in 0..structs.len() { + merge_state_row(&mut self.counts, values_list, counts_list, row)?; + } + Ok(()) + } + + fn state(&mut self) -> Result> { + let array = build_state(&self.data_type, &[&self.counts])?; + Ok(vec![ScalarValue::Struct(Arc::new(array))]) + } + + fn evaluate(&mut self) -> Result { + eval_mode(&self.counts, &self.data_type) + } + + fn size(&self) -> usize { + size_of_val(self) + self.counts.capacity() * size_of::<(ScalarValue, i64)>() + } +} + +/// Vectorized grouped accumulator: one frequency map per group. +#[derive(Debug)] +pub struct ModeGroupsAccumulator { + groups: Vec>, + data_type: DataType, +} + +impl ModeGroupsAccumulator { + fn new(data_type: DataType) -> Self { + Self { + groups: Vec::new(), + data_type, + } + } + + fn resize(&mut self, total_num_groups: usize) { + if self.groups.len() < total_num_groups { + self.groups.resize_with(total_num_groups, HashMap::new); + } + } +} + +impl GroupsAccumulator for ModeGroupsAccumulator { + fn update_batch( + &mut self, + values: &[ArrayRef], + group_indices: &[usize], + opt_filter: Option<&BooleanArray>, + total_num_groups: usize, + ) -> Result<()> { + self.resize(total_num_groups); + let array = &values[0]; + for (idx, &group_index) in group_indices.iter().enumerate() { + if let Some(f) = opt_filter { + if !f.is_valid(idx) || !f.value(idx) { + continue; + } + } + count_values(&mut self.groups[group_index], array, idx)?; + } + Ok(()) + } + + fn merge_batch( + &mut self, + values: &[ArrayRef], + group_indices: &[usize], + _opt_filter: Option<&BooleanArray>, + total_num_groups: usize, + ) -> Result<()> { + self.resize(total_num_groups); + let structs = values[0].as_struct(); + let values_list = structs.column(0).as_list::(); + let counts_list = structs.column(1).as_list::(); + for (row, &group_index) in group_indices.iter().enumerate() { + merge_state_row(&mut self.groups[group_index], values_list, counts_list, row)?; + } + Ok(()) + } + + fn evaluate(&mut self, emit_to: EmitTo) -> Result { + let emitted = emit_to.take_needed(&mut self.groups); + let mut results = Vec::with_capacity(emitted.len()); + for map in &emitted { + results.push(eval_mode(map, &self.data_type)?); + } + ScalarValue::iter_to_array(results) + } + + fn state(&mut self, emit_to: EmitTo) -> Result> { + let emitted = emit_to.take_needed(&mut self.groups); + let refs: Vec<&HashMap> = emitted.iter().collect(); + Ok(vec![Arc::new(build_state(&self.data_type, &refs)?)]) + } + + fn size(&self) -> usize { + size_of_val(self) + + self + .groups + .iter() + .map(|m| m.capacity() * size_of::<(ScalarValue, i64)>()) + .sum::() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use arrow::array::{Float64Array, Int32Array}; + use arrow::datatypes::Int32Type; + + fn i32_array(values: Vec>) -> ArrayRef { + Arc::new(Int32Array::from(values)) + } + + fn eval_acc(acc: &mut ModeAccumulator) -> ScalarValue { + acc.evaluate().unwrap() + } + + #[test] + fn most_frequent_value() { + let mut acc = ModeAccumulator::new(DataType::Int32); + acc.update_batch(&[i32_array(vec![Some(0), Some(10), Some(10)])]) + .unwrap(); + assert_eq!(eval_acc(&mut acc), ScalarValue::Int32(Some(10))); + } + + #[test] + fn nulls_are_ignored() { + let mut acc = ModeAccumulator::new(DataType::Int32); + acc.update_batch(&[i32_array(vec![ + Some(10), + None, + None, + None, + Some(10), + Some(7), + ])]) + .unwrap(); + assert_eq!(eval_acc(&mut acc), ScalarValue::Int32(Some(10))); + } + + #[test] + fn empty_input_is_null() { + let mut acc = ModeAccumulator::new(DataType::Int32); + acc.update_batch(&[i32_array(vec![None, None])]).unwrap(); + assert_eq!(eval_acc(&mut acc), ScalarValue::Int32(None)); + } + + #[test] + fn ties_break_to_smallest() { + let mut acc = ModeAccumulator::new(DataType::Int32); + // 10 and 20 each appear twice; Comet returns the smallest tied value. + acc.update_batch(&[i32_array(vec![Some(20), Some(10), Some(10), Some(20)])]) + .unwrap(); + assert_eq!(eval_acc(&mut acc), ScalarValue::Int32(Some(10))); + } + + /// Turn an accumulator's `Vec` state into the state arrays `merge_batch` consumes. + fn state_arrays(acc: &mut ModeAccumulator) -> Vec { + acc.state() + .unwrap() + .into_iter() + .map(|s| ScalarValue::iter_to_array(vec![s]).unwrap()) + .collect() + } + + #[test] + fn merge_matches_single_shot() { + let single = { + let mut a = ModeAccumulator::new(DataType::Int32); + a.update_batch(&[i32_array(vec![ + Some(1), + Some(1), + Some(2), + Some(3), + Some(3), + Some(3), + ])]) + .unwrap(); + eval_acc(&mut a) + }; + + let mut left = ModeAccumulator::new(DataType::Int32); + left.update_batch(&[i32_array(vec![Some(1), Some(1), Some(3)])]) + .unwrap(); + let lstate = state_arrays(&mut left); + + let mut right = ModeAccumulator::new(DataType::Int32); + right + .update_batch(&[i32_array(vec![Some(2), Some(3), Some(3)])]) + .unwrap(); + let rstate = state_arrays(&mut right); + + let mut merged = ModeAccumulator::new(DataType::Int32); + merged.merge_batch(&lstate).unwrap(); + merged.merge_batch(&rstate).unwrap(); + assert_eq!(eval_acc(&mut merged), single); + } + + #[test] + fn float_zero_and_nan_normalized() { + let mut acc = ModeAccumulator::new(DataType::Float64); + // -0.0 and 0.0 must count as one key. + let arr: ArrayRef = Arc::new(Float64Array::from(vec![ + Some(-0.0), + Some(0.0), + Some(0.0), + Some(1.5), + ])); + acc.update_batch(&[arr]).unwrap(); + assert_eq!(eval_acc(&mut acc), ScalarValue::Float64(Some(0.0))); + } + + #[test] + fn groups_accumulator_per_group_mode() { + let mut acc = ModeGroupsAccumulator::new(DataType::Int32); + let values = i32_array(vec![Some(5), Some(5), Some(9), Some(9), Some(9)]); + acc.update_batch(&[values], &[0, 0, 1, 1, 1], None, 2) + .unwrap(); + let result = acc.evaluate(EmitTo::All).unwrap(); + let result = result.as_primitive::(); + assert_eq!(result.value(0), 5); + assert_eq!(result.value(1), 9); + } + + #[test] + fn groups_accumulator_merge_roundtrip() { + // Partial over two groups, then merge its state into a fresh accumulator. + let mut partial = ModeGroupsAccumulator::new(DataType::Int32); + let values = i32_array(vec![Some(5), Some(5), Some(7), Some(9), Some(9), Some(9)]); + partial + .update_batch(&[values], &[0, 0, 0, 1, 1, 1], None, 2) + .unwrap(); + let state = partial.state(EmitTo::All).unwrap(); + + let mut final_acc = ModeGroupsAccumulator::new(DataType::Int32); + final_acc.merge_batch(&state, &[0, 1], None, 2).unwrap(); + let result = final_acc.evaluate(EmitTo::All).unwrap(); + let result = result.as_primitive::(); + assert_eq!(result.value(0), 5); + assert_eq!(result.value(1), 9); + } +} 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 44142e75ed..06ed92eb14 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -389,6 +389,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { classOf[Last] -> CometLast, classOf[Max] -> CometMax, classOf[Min] -> CometMin, + classOf[Mode] -> CometMode, classOf[Percentile] -> CometPercentile, classOf[StddevPop] -> CometStddevPop, classOf[StddevSamp] -> CometStddevSamp, 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 5710232cb4..8d93d83287 100644 --- a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala +++ b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala @@ -22,14 +22,14 @@ package org.apache.comet.serde import scala.jdk.CollectionConverters._ import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Literal} -import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, Average, BitAndAgg, BitOrAgg, BitXorAgg, BloomFilterAggregate, CentralMomentAgg, CollectSet, Corr, Count, Covariance, CovPopulation, CovSample, First, Last, Max, Min, Percentile, StddevPop, StddevSamp, Sum, VariancePop, VarianceSamp} +import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, Average, BitAndAgg, BitOrAgg, BitXorAgg, BloomFilterAggregate, CentralMomentAgg, CollectSet, Corr, Count, Covariance, CovPopulation, CovSample, First, Last, Max, Min, Mode, Percentile, StddevPop, StddevSamp, Sum, VariancePop, VarianceSamp} import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.types.{ByteType, DecimalType, DoubleType, IntegerType, LongType, NumericType, ShortType, StringType} +import org.apache.spark.sql.types.{BooleanType, ByteType, DataType, DateType, DecimalType, DoubleType, FloatType, IntegerType, LongType, NumericType, ShortType, StringType, TimestampNTZType, TimestampType} import org.apache.comet.CometConf.COMET_EXEC_STRICT_FLOATING_POINT import org.apache.comet.CometSparkSessionExtensions.{isSpark41Plus, withFallbackReason} import org.apache.comet.serde.QueryPlanSerde.{evalModeToProto, exprToProto, serializeDataType} -import org.apache.comet.shims.CometEvalModeUtil +import org.apache.comet.shims.{CometEvalModeUtil, CometTypeShim} object CometMin extends CometAggregateExpressionSerde[Min] { @@ -823,6 +823,71 @@ object CometCollectSet extends CometAggregateExpressionSerde[CollectSet] { } } +object CometMode extends CometAggregateExpressionSerde[Mode] with CometTypeShim { + + private val tieBreakReason = + "mode breaks ties non-deterministically in Spark (the result depends on JVM hash-map" + + " iteration order); Comet returns the smallest of the tied values instead" + + " (https://github.com/apache/datafusion-comet/issues/3970)" + + override def getIncompatibleReasons(): Seq[String] = Seq(tieBreakReason) + + private def isSupportedType(dt: DataType): Boolean = dt match { + case BooleanType => true + case ByteType | ShortType | IntegerType | LongType => true + case FloatType | DoubleType => true + case _: DecimalType => true + case DateType | TimestampType | TimestampNTZType => true + case StringType => true + case _ => false + } + + override def getSupportLevel(expr: Mode): SupportLevel = { + if (modeHasUnsupportedOrdering(expr)) { + // `mode(col, deterministic)` and `mode() WITHIN GROUP (ORDER BY col)` carry deterministic + // ordered tie-breaking that Comet does not implement yet (Spark 4.0+ only). + Unsupported( + Some("mode with a deterministic flag or WITHIN GROUP ordering is not supported")) + } else if (hasNonDefaultStringCollation(expr.child.dataType)) { + // Native counting is not collation-aware, so non-UTF8_BINARY collations would group keys + // differently from Spark. + Unsupported( + Some( + "mode does not support non-UTF8_BINARY collations " + + "(https://github.com/apache/datafusion-comet/issues/2190)")) + } else if (!isSupportedType(expr.child.dataType)) { + Unsupported(Some(s"mode does not support input type ${expr.child.dataType}")) + } else { + Incompatible(Some(tieBreakReason)) + } + } + + override def convert( + aggExpr: AggregateExpression, + expr: Mode, + inputs: Seq[Attribute], + binding: Boolean, + conf: SQLConf): Option[ExprOuterClass.AggExpr] = { + val child = expr.child + val childExpr = exprToProto(child, inputs, binding) + val dataType = serializeDataType(child.dataType) + + if (childExpr.isDefined && dataType.isDefined) { + val builder = ExprOuterClass.Mode.newBuilder() + builder.setChild(childExpr.get) + builder.setDatatype(dataType.get) + Some( + ExprOuterClass.AggExpr + .newBuilder() + .setMode(builder) + .build()) + } else { + withFallbackReason(aggExpr, child) + None + } + } +} + object AggSerde { import org.apache.spark.sql.types._ diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala index e4d6b53770..dcf30080c5 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala @@ -30,7 +30,7 @@ import org.apache.spark.broadcast.Broadcast import org.apache.spark.rdd.RDD import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.{Ascending, Attribute, AttributeSet, Expression, ExpressionSet, Generator, NamedExpression, SortOrder} -import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, AggregateMode, CollectSet, Final, First, Last, Partial, PartialMerge, Percentile} +import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, AggregateMode, CollectSet, Final, First, Last, Mode, Partial, PartialMerge, Percentile} import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide} import org.apache.spark.sql.catalyst.plans._ import org.apache.spark.sql.catalyst.plans.physical._ @@ -43,7 +43,7 @@ import org.apache.spark.sql.execution.aggregate.{BaseAggregateExec, HashAggregat import org.apache.spark.sql.execution.exchange.ReusedExchangeExec import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, BroadcastNestedLoopJoinExec, HashJoin, ShuffledHashJoinExec, SortMergeJoinExec} import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics} -import org.apache.spark.sql.types.{ArrayType, BooleanType, ByteType, DataType, DateType, DecimalType, DoubleType, FloatType, IntegerType, LongType, MapType, ShortType, StringType, TimestampNTZType, TimestampType} +import org.apache.spark.sql.types.{ArrayType, BooleanType, ByteType, DataType, DateType, DecimalType, DoubleType, FloatType, IntegerType, LongType, MapType, ShortType, StringType, StructField, StructType, TimestampNTZType, TimestampType} import org.apache.spark.sql.vectorized.ColumnarBatch import org.apache.spark.util.SerializableConfiguration import org.apache.spark.util.io.ChunkedByteBuffer @@ -1925,6 +1925,19 @@ object CometObjectHashAggregateExec // Comet casts the child to double, so the native state is ArrayType(DoubleType). val nativeStateType = ArrayType(DoubleType, containsNull = true) output(bufferIdx) = output(bufferIdx).withDataType(nativeStateType) + case m: Mode => + // Comet's native mode accumulator keeps a frequency map encoded as parallel arrays + // (see ModeAccumulator in native/spark-expr): a struct of the distinct values and their + // counts. + val elementType = m.child.dataType + val nativeStateType = StructType( + Seq( + StructField( + "values", + ArrayType(elementType, containsNull = true), + nullable = false), + StructField("counts", ArrayType(LongType, containsNull = true), nullable = false))) + output(bufferIdx) = output(bufferIdx).withDataType(nativeStateType) case _ => } bufferIdx += bufferAttrs.length diff --git a/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala b/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala index 97320be9e7..4d19bf35eb 100644 --- a/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala +++ b/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala @@ -21,12 +21,18 @@ package org.apache.comet.shims import scala.annotation.nowarn +import org.apache.spark.sql.catalyst.expressions.aggregate.Mode import org.apache.spark.sql.types.{DataType, StructType} trait CometTypeShim { @nowarn // Spark 4 feature; stubbed to false in Spark 3.x for compatibility. def isStringCollationType(dt: DataType): Boolean = false + // `mode() WITHIN GROUP (ORDER BY ...)` and the deterministic-flag form (which set `reverseOpt`) + // are Spark 4.0 features; Spark 3.x `Mode` is always the plain `mode(col)` form. + @nowarn + def modeHasUnsupportedOrdering(expr: Mode): Boolean = false + @nowarn // Spark 4 feature; stubbed to false in Spark 3.x for compatibility. def hasNonDefaultStringCollation(dt: DataType): Boolean = false diff --git a/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala b/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala index 1d4a9f601e..535f4012af 100644 --- a/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala +++ b/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala @@ -19,10 +19,16 @@ package org.apache.comet.shims +import org.apache.spark.sql.catalyst.expressions.aggregate.Mode import org.apache.spark.sql.execution.datasources.VariantMetadata import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StringType, StructType} trait CometTypeShim { + // `reverseOpt` is set for `mode() WITHIN GROUP (ORDER BY col [DESC])` and the + // `mode(col, deterministic)` form, both of which carry ordered tie-breaking that Comet does not + // implement yet. The plain `mode(col)` form leaves it as `None`. + def modeHasUnsupportedOrdering(expr: Mode): Boolean = expr.reverseOpt.isDefined + // A `StringType` carries collation metadata in Spark 4.0. Only non-default (non-UTF8_BINARY) // collations have semantics Comet's byte-level hashing/sorting/equality cannot honor. The // default `StringType` object is `StringType(UTF8_BINARY_COLLATION_ID)`, so comparing diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/mode.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/mode.sql new file mode 100644 index 0000000000..db7d1d2b90 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/aggregate/mode.sql @@ -0,0 +1,186 @@ +-- 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. + +-- Comet's `mode` is opt-in via allowIncompatible because Spark breaks ties non-deterministically. +-- Every compared query below has a single value with the strictly-highest frequency per group so +-- that Comet's smallest-value tie-break agrees with Spark's arbitrary choice. +-- Config: spark.comet.expression.Mode.allowIncompatible=true + +-- ============================================================ +-- Setup: tables +-- ============================================================ + +statement +CREATE TABLE mode_int(v int, grp string) USING parquet + +statement +INSERT INTO mode_int VALUES + (10, 'a'), (10, 'a'), (7, 'a'), (NULL, 'a'), + (5, 'b'), (5, 'b'), (5, 'b'), (9, 'b'), (NULL, 'b'), + (NULL, 'c'), (NULL, 'c') + +statement +CREATE TABLE mode_all_null(v int) USING parquet + +statement +INSERT INTO mode_all_null VALUES (NULL), (NULL) + +-- ============================================================ +-- Global aggregate (no GROUP BY): unique mode +-- ============================================================ + +query +SELECT mode(v) FROM mode_int + +-- ============================================================ +-- GROUP BY: unique mode per group; NULLs ignored +-- ============================================================ + +query +SELECT grp, mode(v) FROM mode_int GROUP BY grp ORDER BY grp + +-- ============================================================ +-- All-NULL input returns NULL +-- ============================================================ + +query +SELECT mode(v) FROM mode_all_null + +-- ============================================================ +-- Mixed with other aggregates +-- ============================================================ + +query +SELECT grp, mode(v), count(*), sum(v) FROM mode_int GROUP BY grp ORDER BY grp + +-- ============================================================ +-- HAVING clause +-- ============================================================ + +query +SELECT grp, mode(v) FROM mode_int GROUP BY grp HAVING count(v) > 3 ORDER BY grp + +-- ============================================================ +-- Boolean +-- ============================================================ + +statement +CREATE TABLE mode_bool(v boolean, grp string) USING parquet + +statement +INSERT INTO mode_bool VALUES + (true, 'a'), (true, 'a'), (false, 'a'), (NULL, 'a'), + (false, 'b'), (false, 'b'), (true, 'b') + +query +SELECT grp, mode(v) FROM mode_bool GROUP BY grp ORDER BY grp + +-- ============================================================ +-- Byte / Short / Long +-- ============================================================ + +statement +CREATE TABLE mode_nums(b tinyint, s smallint, l bigint, grp string) USING parquet + +statement +INSERT INTO mode_nums VALUES + (1, 100, 1000000000000, 'a'), (1, 100, 1000000000000, 'a'), (2, 200, 2000000000000, 'a'), + (3, 300, 3000000000000, 'b'), (3, 300, 3000000000000, 'b'), (4, 400, 4000000000000, 'b') + +query +SELECT grp, mode(b), mode(s), mode(l) FROM mode_nums GROUP BY grp ORDER BY grp + +-- ============================================================ +-- Float / Double: -0.0 and 0.0 normalize to one key +-- ============================================================ + +statement +CREATE TABLE mode_double(v double, grp string) USING parquet + +statement +INSERT INTO mode_double VALUES + (1.5, 'a'), (1.5, 'a'), (2.5, 'a'), (NULL, 'a'), + (CAST(0.0 AS DOUBLE), 'b'), (CAST(-0.0 AS DOUBLE), 'b'), (CAST(-0.0 AS DOUBLE), 'b'), (7.0, 'b') + +query +SELECT grp, mode(v) FROM mode_double GROUP BY grp ORDER BY grp + +-- ============================================================ +-- Decimal +-- ============================================================ + +statement +CREATE TABLE mode_decimal(v decimal(10,2), grp string) USING parquet + +statement +INSERT INTO mode_decimal VALUES + (1.50, 'a'), (1.50, 'a'), (2.50, 'a'), (NULL, 'a'), + (99999999.99, 'b'), (99999999.99, 'b'), (0.00, 'b') + +query +SELECT grp, mode(v) FROM mode_decimal GROUP BY grp ORDER BY grp + +-- ============================================================ +-- String +-- ============================================================ + +statement +CREATE TABLE mode_string(v string, grp string) USING parquet + +statement +INSERT INTO mode_string VALUES + ('hello', 'a'), ('hello', 'a'), ('world', 'a'), (NULL, 'a'), + ('', 'b'), ('', 'b'), ('x', 'b') + +query +SELECT grp, mode(v) FROM mode_string GROUP BY grp ORDER BY grp + +-- ============================================================ +-- Date / Timestamp +-- ============================================================ + +statement +CREATE TABLE mode_temporal(d date, t timestamp, grp string) USING parquet + +statement +INSERT INTO mode_temporal VALUES + (DATE '2024-01-01', TIMESTAMP '2024-01-01 00:00:00', 'a'), + (DATE '2024-01-01', TIMESTAMP '2024-01-01 00:00:00', 'a'), + (DATE '2024-06-15', TIMESTAMP '2024-06-15 12:30:00', 'a'), + (DATE '1970-01-01', TIMESTAMP '1970-01-01 00:00:00', 'b'), + (DATE '1970-01-01', TIMESTAMP '1970-01-01 00:00:00', 'b'), + (DATE '2000-12-31', TIMESTAMP '2000-12-31 23:59:59', 'b') + +query +SELECT grp, mode(d), mode(t) FROM mode_temporal GROUP BY grp ORDER BY grp + +-- ============================================================ +-- Unsupported input type falls back to Spark +-- +-- A single row keeps the result deterministic: Spark's mode on BinaryType compares Array[Byte] +-- keys by reference, so any binary multiset with repeats is a full tie and returns an arbitrary +-- value. One row avoids that while still exercising the unsupported-type fallback. +-- ============================================================ + +statement +CREATE TABLE mode_binary(v binary) USING parquet + +statement +INSERT INTO mode_binary VALUES (X'CAFE') + +query expect_fallback(does not support input type) +SELECT mode(v) FROM mode_binary From 47b0dfd6801fb09502970340c815954f58bfccf9 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 4 Aug 2026 08:44:11 -0600 Subject: [PATCH 2/2] fix: make mode's -0.0 key handling match the Spark version, plus review cleanups Addresses review feedback on #4782. The main item was the `-0.0` normalization in `normalize_key`. The review asked to stop collapsing `-0.0` into `0.0`, because Spark keys the frequency map on `java.lang.Double.equals` (via `OpenHashSet`), which distinguishes the two, and `NormalizeFloatingNumbers` never touches aggregate arguments. That is correct for Spark 3.4 through 4.1, but Spark 4.2.0 changed it: SPARK-57329 treats the split `-0.0`/`0.0` counts as a correctness bug and normalizes the key at update time. The fix landed in branch-4.2 after v4.2.0-rc1, so released 4.2.0 has it. Correct behaviour is therefore version-dependent, and neither always-collapsing nor never-collapsing is right across the profiles this repo builds. Added a `normalize_neg_zero` flag to the `Mode` proto message, set from `isSpark42Plus` in the serde (same pattern as `BloomFilterVersion` and `setIsSpark4Plus`), and gated the fold on it natively. `NaN` canonicalization stays unconditional, since `doubleToLongBits` collapses `NaN` on every supported version. `ScalarValue`'s `PartialEq`/`Hash` for floats are both bit-based, so once `NaN` is canonical the zeros stay distinct without further work. Test fixtures used `CAST(-0.0 AS DOUBLE)`, which does not produce a negative zero: an unsuffixed `-0.0` is a DecimalType literal and Decimal has no signed zero, so the column contained only `+0.0` and the existing signed-zero coverage was vacuous. Switched to `-0.0D`. Verified the new fixture is non-vacuous by forcing the flag to the wrong value and confirming it fails. Also in this commit: - correct the doc comment, which claimed the old behaviour matched `NormalizeFloatingNumbers`, and record which Spark comparison path governs `mode` versus `max_by`/`min_by` so the two are not "fixed" to match each other - drop the redundant `default_value` override - count key heap bytes in `size()` so string/binary/decimal modes do not under-report to the memory pool - assert the non-empty-groups invariant at both grouped emit sites - note that the `ScalarValue` frequency map is intentionally type-generic - add a `timestamp_ntz` compared query - add mode_within_group.sql pinning the Spark 4.x ordered forms as fallbacks, including that `ModeBuilder` rewrites `mode(col, false)` to the plain form so it still runs natively - TODO recording that the ASC WITHIN GROUP form could be Compatible --- native/core/src/execution/planner.rs | 3 +- native/proto/src/proto/expr.proto | 4 + native/spark-expr/src/agg_funcs/mode.rs | 247 ++++++++++++++---- .../org/apache/comet/serde/aggregates.scala | 9 +- .../sql-tests/expressions/aggregate/mode.sql | 71 ++++- .../aggregate/mode_within_group.sql | 70 +++++ 6 files changed, 348 insertions(+), 56 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/expressions/aggregate/mode_within_group.sql diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 3c56088877..becb6c35db 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -2646,7 +2646,8 @@ impl PhysicalPlanner { AggExprStruct::Mode(expr) => { let child = self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&schema))?; let datatype = to_arrow_datatype(expr.datatype.as_ref().unwrap()); - let func = AggregateUDF::new_from_impl(Mode::new(datatype)); + let func = + AggregateUDF::new_from_impl(Mode::new(datatype, expr.normalize_neg_zero)); Self::create_aggr_func_expr("mode", schema, vec![child], func) } } diff --git a/native/proto/src/proto/expr.proto b/native/proto/src/proto/expr.proto index 532a58e7d6..0b801f88a1 100644 --- a/native/proto/src/proto/expr.proto +++ b/native/proto/src/proto/expr.proto @@ -279,6 +279,10 @@ message CollectSet { message Mode { Expr child = 1; DataType datatype = 2; + // Whether `-0.0` should be folded into `0.0` before it is used as a frequency-map key. + // Spark only started doing this in 4.2.0 (SPARK-57329), so this tracks the Spark version + // Comet is running against. See `Mode` in the spark-expr crate for the full rationale. + bool normalize_neg_zero = 3; } enum EvalMode { diff --git a/native/spark-expr/src/agg_funcs/mode.rs b/native/spark-expr/src/agg_funcs/mode.rs index 4576fc3203..efa3667874 100644 --- a/native/spark-expr/src/agg_funcs/mode.rs +++ b/native/spark-expr/src/agg_funcs/mode.rs @@ -37,8 +37,25 @@ use std::sync::Arc; /// Comet resolves ties deterministically by returning the smallest value, so this function is /// registered as `Incompatible` on the Scala side and is opt-in via `allowIncompatible`. /// -/// Float keys are normalized before counting (`-0.0` becomes `0.0` and every `NaN` becomes a -/// canonical `NaN`) to match Spark's `NormalizeFloatingNumbers` behaviour so that counts agree. +/// # Float keys +/// +/// Spark keys the frequency map on the boxed input value and compares keys with +/// `OpenHashSet`'s `_data(pos) equals k` (`core/.../util/collection/OpenHashSet.scala:122`), i.e. +/// `java.lang.Double.equals`, which is defined via `doubleToLongBits`. That collapses every `NaN` +/// bit pattern to one key but keeps `-0.0` and `0.0` apart. Note that +/// `NormalizeFloatingNumbers` does *not* apply here: its `apply` only rewrites `WINDOW` and +/// `JOIN` patterns, so an aggregate's argument reaches `Mode` un-normalized. +/// +/// Spark 4.2.0 changed this. SPARK-57329 ("mode() returns incorrect result when input contains +/// both -0.0 and 0.0") treats the split `-0.0`/`0.0` counts as a bug and normalizes the key at +/// update time, so from 4.2.0 on the two fold into a single key. `normalize_neg_zero` therefore +/// tracks the Spark version Comet is running against: it is `false` for Spark 3.4 through 4.1 and +/// `true` for 4.2.0+. `NaN` canonicalization is unconditional because every supported version +/// collapses `NaN` via `doubleToLongBits`. +/// +/// Do not "simplify" this to always normalize: `max_by`/`min_by` need the opposite treatment, +/// because they compare the ordering column with `SQLOrderingUtil.compareDoubles`, which ties +/// `-0.0 == 0.0` on every version. /// /// Spark's `Mode` is a `TypedImperativeAggregate` with a single aggregation-buffer attribute, so /// the intermediate state is a single struct field `{ values: list, counts: list }` (a @@ -49,14 +66,17 @@ pub struct Mode { name: String, signature: Signature, data_type: DataType, + /// Whether `-0.0` folds into `0.0` before being used as a key (Spark 4.2.0+; SPARK-57329). + normalize_neg_zero: bool, } impl Mode { - pub fn new(data_type: DataType) -> Self { + pub fn new(data_type: DataType, normalize_neg_zero: bool) -> Self { Self { name: "mode".to_string(), signature: Signature::any(1, Volatility::Immutable), data_type, + normalize_neg_zero, } } } @@ -113,12 +133,11 @@ impl AggregateUDFImpl for Mode { Ok(self.data_type.clone()) } - fn default_value(&self, _data_type: &DataType) -> Result { - ScalarValue::try_from(&self.data_type) - } - fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result> { - Ok(Box::new(ModeAccumulator::new(self.data_type.clone()))) + Ok(Box::new(ModeAccumulator::new( + self.data_type.clone(), + self.normalize_neg_zero, + ))) } fn state_fields(&self, _args: StateFieldsArgs) -> Result> { @@ -137,20 +156,28 @@ impl AggregateUDFImpl for Mode { &self, _args: AccumulatorArgs, ) -> Result> { - Ok(Box::new(ModeGroupsAccumulator::new(self.data_type.clone()))) + Ok(Box::new(ModeGroupsAccumulator::new( + self.data_type.clone(), + self.normalize_neg_zero, + ))) } } -/// Normalize a scalar key so that Spark's floating-point normalization is honoured: `-0.0` and -/// `0.0` collapse to the same key and all `NaN` bit patterns collapse to a canonical `NaN`. -fn normalize_key(value: ScalarValue) -> ScalarValue { - /// Collapse `-0.0`/`0.0` and every `NaN` to a canonical form for one float variant. +/// Canonicalize a float key so that map lookups reproduce Spark's key equality. +/// +/// `ScalarValue`'s `PartialEq`/`Hash` for `Float32`/`Float64` are both defined on `to_bits()`, so +/// distinct `NaN` bit patterns would otherwise be distinct keys and `-0.0` is naturally kept apart +/// from `0.0`. Collapsing `NaN` to one canonical value therefore reproduces `doubleToLongBits` +/// equality, which is what Spark's `OpenHashSet` uses. `-0.0` is folded into `0.0` only when +/// `normalize_neg_zero` is set, i.e. only on Spark 4.2.0+ (SPARK-57329); see [`Mode`]. +fn normalize_key(value: ScalarValue, normalize_neg_zero: bool) -> ScalarValue { macro_rules! normalize_float { ($variant:path, $f:expr, $nan:expr) => { - if $f == 0.0 { - $variant(Some(0.0)) - } else if $f.is_nan() { + if $f.is_nan() { $variant(Some($nan)) + } else if normalize_neg_zero && $f == 0.0 { + // `-0.0 == 0.0` in IEEE 754, so this catches negative zero only. + $variant(Some(0.0)) } else { $variant(Some($f)) } @@ -163,12 +190,22 @@ fn normalize_key(value: ScalarValue) -> ScalarValue { } } -/// Add each non-null value in `array` to `map`, normalizing float keys. -fn count_values(map: &mut HashMap, array: &ArrayRef, idx: usize) -> Result<()> { +/// Add each non-null value in `array` to `map`, canonicalizing float keys. +/// +/// The map is intentionally keyed on the type-generic `ScalarValue` rather than a monomorphized +/// `HashMap, _>`: `mode` supports every primitive type plus decimal, string and +/// the temporal types, so one generic map is simpler than a kernel per type. Revisit if the hot +/// primitive paths ever show up in a profile. +fn count_values( + map: &mut HashMap, + array: &ArrayRef, + idx: usize, + normalize_neg_zero: bool, +) -> Result<()> { if array.is_null(idx) { return Ok(()); } - let key = normalize_key(ScalarValue::try_from_array(array, idx)?); + let key = normalize_key(ScalarValue::try_from_array(array, idx)?, normalize_neg_zero); *map.entry(key).or_insert(0) += 1; Ok(()) } @@ -179,6 +216,7 @@ fn merge_state_row( values_list: &arrow::array::ListArray, counts_list: &arrow::array::ListArray, row: usize, + normalize_neg_zero: bool, ) -> Result<()> { if values_list.is_null(row) { return Ok(()); @@ -192,7 +230,7 @@ fn merge_state_row( if values.is_null(i) { continue; } - let key = normalize_key(ScalarValue::try_from_array(&values, i)?); + let key = normalize_key(ScalarValue::try_from_array(&values, i)?, normalize_neg_zero); *map.entry(key).or_insert(0) += counts.value(i); } Ok(()) @@ -221,18 +259,33 @@ fn eval_mode(counts: &HashMap, data_type: &DataType) -> Result } } +/// Heap bytes held by the frequency map's keys, on top of the map's own slot allocation. +/// +/// `HashMap::capacity` only accounts for the inline `(ScalarValue, i64)` slots, which misses the +/// `String`/`Vec`/boxed-decimal payloads behind variable-length keys. Under-reporting those +/// would hide real memory from the pool that drives spill decisions. +fn map_size(map: &HashMap) -> usize { + map.capacity() * size_of::<(ScalarValue, i64)>() + + map + .keys() + .map(|k| k.size().saturating_sub(size_of::())) + .sum::() +} + /// Non-grouped accumulator backing global `mode` aggregation. #[derive(Debug)] pub struct ModeAccumulator { counts: HashMap, data_type: DataType, + normalize_neg_zero: bool, } impl ModeAccumulator { - fn new(data_type: DataType) -> Self { + fn new(data_type: DataType, normalize_neg_zero: bool) -> Self { Self { counts: HashMap::new(), data_type, + normalize_neg_zero, } } } @@ -241,7 +294,7 @@ impl Accumulator for ModeAccumulator { fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> { let array = &values[0]; for i in 0..array.len() { - count_values(&mut self.counts, array, i)?; + count_values(&mut self.counts, array, i, self.normalize_neg_zero)?; } Ok(()) } @@ -251,7 +304,13 @@ impl Accumulator for ModeAccumulator { let values_list = structs.column(0).as_list::(); let counts_list = structs.column(1).as_list::(); for row in 0..structs.len() { - merge_state_row(&mut self.counts, values_list, counts_list, row)?; + merge_state_row( + &mut self.counts, + values_list, + counts_list, + row, + self.normalize_neg_zero, + )?; } Ok(()) } @@ -266,7 +325,7 @@ impl Accumulator for ModeAccumulator { } fn size(&self) -> usize { - size_of_val(self) + self.counts.capacity() * size_of::<(ScalarValue, i64)>() + size_of_val(self) + map_size(&self.counts) } } @@ -275,13 +334,15 @@ impl Accumulator for ModeAccumulator { pub struct ModeGroupsAccumulator { groups: Vec>, data_type: DataType, + normalize_neg_zero: bool, } impl ModeGroupsAccumulator { - fn new(data_type: DataType) -> Self { + fn new(data_type: DataType, normalize_neg_zero: bool) -> Self { Self { groups: Vec::new(), data_type, + normalize_neg_zero, } } @@ -308,7 +369,12 @@ impl GroupsAccumulator for ModeGroupsAccumulator { continue; } } - count_values(&mut self.groups[group_index], array, idx)?; + count_values( + &mut self.groups[group_index], + array, + idx, + self.normalize_neg_zero, + )?; } Ok(()) } @@ -325,13 +391,23 @@ impl GroupsAccumulator for ModeGroupsAccumulator { let values_list = structs.column(0).as_list::(); let counts_list = structs.column(1).as_list::(); for (row, &group_index) in group_indices.iter().enumerate() { - merge_state_row(&mut self.groups[group_index], values_list, counts_list, row)?; + merge_state_row( + &mut self.groups[group_index], + values_list, + counts_list, + row, + self.normalize_neg_zero, + )?; } Ok(()) } fn evaluate(&mut self, emit_to: EmitTo) -> Result { let emitted = emit_to.take_needed(&mut self.groups); + // `ScalarValue::iter_to_array` errors on an empty iterator. The grouped-aggregate stream + // never emits zero groups, so this is unreachable; assert it rather than leaving the + // dependency implicit. + debug_assert!(!emitted.is_empty(), "mode: evaluate called with no groups"); let mut results = Vec::with_capacity(emitted.len()); for map in &emitted { results.push(eval_mode(map, &self.data_type)?); @@ -341,17 +417,15 @@ impl GroupsAccumulator for ModeGroupsAccumulator { fn state(&mut self, emit_to: EmitTo) -> Result> { let emitted = emit_to.take_needed(&mut self.groups); + // As in `evaluate`: `build_state` funnels into `ScalarValue::iter_to_array`, which needs a + // non-empty iterator. + debug_assert!(!emitted.is_empty(), "mode: state called with no groups"); let refs: Vec<&HashMap> = emitted.iter().collect(); Ok(vec![Arc::new(build_state(&self.data_type, &refs)?)]) } fn size(&self) -> usize { - size_of_val(self) - + self - .groups - .iter() - .map(|m| m.capacity() * size_of::<(ScalarValue, i64)>()) - .sum::() + size_of_val(self) + self.groups.iter().map(map_size).sum::() } } @@ -371,7 +445,7 @@ mod tests { #[test] fn most_frequent_value() { - let mut acc = ModeAccumulator::new(DataType::Int32); + let mut acc = ModeAccumulator::new(DataType::Int32, false); acc.update_batch(&[i32_array(vec![Some(0), Some(10), Some(10)])]) .unwrap(); assert_eq!(eval_acc(&mut acc), ScalarValue::Int32(Some(10))); @@ -379,7 +453,7 @@ mod tests { #[test] fn nulls_are_ignored() { - let mut acc = ModeAccumulator::new(DataType::Int32); + let mut acc = ModeAccumulator::new(DataType::Int32, false); acc.update_batch(&[i32_array(vec![ Some(10), None, @@ -394,14 +468,14 @@ mod tests { #[test] fn empty_input_is_null() { - let mut acc = ModeAccumulator::new(DataType::Int32); + let mut acc = ModeAccumulator::new(DataType::Int32, false); acc.update_batch(&[i32_array(vec![None, None])]).unwrap(); assert_eq!(eval_acc(&mut acc), ScalarValue::Int32(None)); } #[test] fn ties_break_to_smallest() { - let mut acc = ModeAccumulator::new(DataType::Int32); + let mut acc = ModeAccumulator::new(DataType::Int32, false); // 10 and 20 each appear twice; Comet returns the smallest tied value. acc.update_batch(&[i32_array(vec![Some(20), Some(10), Some(10), Some(20)])]) .unwrap(); @@ -420,7 +494,7 @@ mod tests { #[test] fn merge_matches_single_shot() { let single = { - let mut a = ModeAccumulator::new(DataType::Int32); + let mut a = ModeAccumulator::new(DataType::Int32, false); a.update_batch(&[i32_array(vec![ Some(1), Some(1), @@ -433,40 +507,109 @@ mod tests { eval_acc(&mut a) }; - let mut left = ModeAccumulator::new(DataType::Int32); + let mut left = ModeAccumulator::new(DataType::Int32, false); left.update_batch(&[i32_array(vec![Some(1), Some(1), Some(3)])]) .unwrap(); let lstate = state_arrays(&mut left); - let mut right = ModeAccumulator::new(DataType::Int32); + let mut right = ModeAccumulator::new(DataType::Int32, false); right .update_batch(&[i32_array(vec![Some(2), Some(3), Some(3)])]) .unwrap(); let rstate = state_arrays(&mut right); - let mut merged = ModeAccumulator::new(DataType::Int32); + let mut merged = ModeAccumulator::new(DataType::Int32, false); merged.merge_batch(&lstate).unwrap(); merged.merge_batch(&rstate).unwrap(); assert_eq!(eval_acc(&mut merged), single); } - #[test] - fn float_zero_and_nan_normalized() { - let mut acc = ModeAccumulator::new(DataType::Float64); - // -0.0 and 0.0 must count as one key. - let arr: ArrayRef = Arc::new(Float64Array::from(vec![ + /// Input from the SPARK-57329 report: `-0.0` x2, `0.0` x2, `5.0` x3. The winner differs + /// depending on whether the two zeros share a key, so it pins each version's behaviour + /// without depending on how `-0.0` and `0.0` compare. + fn signed_zero_input() -> ArrayRef { + Arc::new(Float64Array::from(vec![ + Some(-0.0), Some(-0.0), Some(0.0), Some(0.0), - Some(1.5), - ])); - acc.update_batch(&[arr]).unwrap(); + Some(5.0), + Some(5.0), + Some(5.0), + ])) + } + + #[test] + fn signed_zeros_are_distinct_keys_before_spark_42() { + // Spark 3.4-4.1 key on `java.lang.Double.equals`, so counts are -0.0:2, 0.0:2, 5.0:3 and + // 5.0 wins outright. + let mut acc = ModeAccumulator::new(DataType::Float64, false); + acc.update_batch(&[signed_zero_input()]).unwrap(); + assert_eq!(eval_acc(&mut acc), ScalarValue::Float64(Some(5.0))); + } + + #[test] + fn signed_zeros_share_a_key_from_spark_42() { + // Spark 4.2.0+ normalizes the key (SPARK-57329), so counts are 0.0:4, 5.0:3 and the + // zero wins. + let mut acc = ModeAccumulator::new(DataType::Float64, true); + acc.update_batch(&[signed_zero_input()]).unwrap(); assert_eq!(eval_acc(&mut acc), ScalarValue::Float64(Some(0.0))); } + #[test] + fn nan_collapses_on_every_version() { + // `doubleToLongBits` maps every NaN to one key on all supported versions, so the two NaNs + // outvote the single 1.0 regardless of the -0.0 setting. + for normalize_neg_zero in [false, true] { + let mut acc = ModeAccumulator::new(DataType::Float64, normalize_neg_zero); + let arr: ArrayRef = Arc::new(Float64Array::from(vec![ + Some(f64::NAN), + Some(-f64::NAN), + Some(1.0), + ])); + acc.update_batch(&[arr]).unwrap(); + match eval_acc(&mut acc) { + ScalarValue::Float64(Some(v)) => assert!( + v.is_nan(), + "expected NaN with normalize_neg_zero={normalize_neg_zero}, got {v}" + ), + other => panic!("expected Float64(NaN), got {other:?}"), + } + } + } + + #[test] + fn signed_zero_key_survives_merge() { + // The partial/final split must not lose the distinction: each side sees one -0.0 and one + // 0.0, and 5.0 only wins if they stay separate through the merge. + let mut left = ModeAccumulator::new(DataType::Float64, false); + left.update_batch(&[ + Arc::new(Float64Array::from(vec![Some(-0.0), Some(0.0), Some(5.0)])) as ArrayRef, + ]) + .unwrap(); + let lstate = state_arrays(&mut left); + + let mut right = ModeAccumulator::new(DataType::Float64, false); + right + .update_batch(&[Arc::new(Float64Array::from(vec![ + Some(-0.0), + Some(0.0), + Some(5.0), + Some(5.0), + ])) as ArrayRef]) + .unwrap(); + let rstate = state_arrays(&mut right); + + let mut merged = ModeAccumulator::new(DataType::Float64, false); + merged.merge_batch(&lstate).unwrap(); + merged.merge_batch(&rstate).unwrap(); + assert_eq!(eval_acc(&mut merged), ScalarValue::Float64(Some(5.0))); + } + #[test] fn groups_accumulator_per_group_mode() { - let mut acc = ModeGroupsAccumulator::new(DataType::Int32); + let mut acc = ModeGroupsAccumulator::new(DataType::Int32, false); let values = i32_array(vec![Some(5), Some(5), Some(9), Some(9), Some(9)]); acc.update_batch(&[values], &[0, 0, 1, 1, 1], None, 2) .unwrap(); @@ -479,14 +622,14 @@ mod tests { #[test] fn groups_accumulator_merge_roundtrip() { // Partial over two groups, then merge its state into a fresh accumulator. - let mut partial = ModeGroupsAccumulator::new(DataType::Int32); + let mut partial = ModeGroupsAccumulator::new(DataType::Int32, false); let values = i32_array(vec![Some(5), Some(5), Some(7), Some(9), Some(9), Some(9)]); partial .update_batch(&[values], &[0, 0, 0, 1, 1, 1], None, 2) .unwrap(); let state = partial.state(EmitTo::All).unwrap(); - let mut final_acc = ModeGroupsAccumulator::new(DataType::Int32); + let mut final_acc = ModeGroupsAccumulator::new(DataType::Int32, false); final_acc.merge_batch(&state, &[0, 1], None, 2).unwrap(); let result = final_acc.evaluate(EmitTo::All).unwrap(); let result = result.as_primitive::(); 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 8d93d83287..5ff7d1540d 100644 --- a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala +++ b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala @@ -27,7 +27,7 @@ import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.{BooleanType, ByteType, DataType, DateType, DecimalType, DoubleType, FloatType, IntegerType, LongType, NumericType, ShortType, StringType, TimestampNTZType, TimestampType} import org.apache.comet.CometConf.COMET_EXEC_STRICT_FLOATING_POINT -import org.apache.comet.CometSparkSessionExtensions.{isSpark41Plus, withFallbackReason} +import org.apache.comet.CometSparkSessionExtensions.{isSpark41Plus, isSpark42Plus, withFallbackReason} import org.apache.comet.serde.QueryPlanSerde.{evalModeToProto, exprToProto, serializeDataType} import org.apache.comet.shims.{CometEvalModeUtil, CometTypeShim} @@ -846,6 +846,9 @@ object CometMode extends CometAggregateExpressionSerde[Mode] with CometTypeShim if (modeHasUnsupportedOrdering(expr)) { // `mode(col, deterministic)` and `mode() WITHIN GROUP (ORDER BY col)` carry deterministic // ordered tie-breaking that Comet does not implement yet (Spark 4.0+ only). + // TODO the ASC form (`reverseOpt = Some(false)`) returns the smallest tied value, which is + // exactly Comet's tie-break, so it could be served natively as `Compatible`. + // https://github.com/apache/datafusion-comet/issues/3970 Unsupported( Some("mode with a deterministic flag or WITHIN GROUP ordering is not supported")) } else if (hasNonDefaultStringCollation(expr.child.dataType)) { @@ -876,6 +879,10 @@ object CometMode extends CometAggregateExpressionSerde[Mode] with CometTypeShim val builder = ExprOuterClass.Mode.newBuilder() builder.setChild(childExpr.get) builder.setDatatype(dataType.get) + // Spark 4.2.0 (SPARK-57329) normalizes `-0.0` to `0.0` before keying the frequency map; + // earlier versions key the raw boxed value, where `java.lang.Double.equals` keeps `-0.0` + // and `0.0` apart. The native side has to match whichever Spark we are running against. + builder.setNormalizeNegZero(isSpark42Plus) Some( ExprOuterClass.AggExpr .newBuilder() diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/mode.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/mode.sql index db7d1d2b90..f32e4905d0 100644 --- a/spark/src/test/resources/sql-tests/expressions/aggregate/mode.sql +++ b/spark/src/test/resources/sql-tests/expressions/aggregate/mode.sql @@ -105,7 +105,15 @@ query SELECT grp, mode(b), mode(s), mode(l) FROM mode_nums GROUP BY grp ORDER BY grp -- ============================================================ --- Float / Double: -0.0 and 0.0 normalize to one key +-- Float / Double +-- +-- Whether `-0.0` and `0.0` share a frequency-map key is version-dependent: Spark 3.4-4.1 key on +-- `java.lang.Double.equals` and keep them apart, while Spark 4.2.0+ folds `-0.0` into `0.0` +-- (SPARK-57329). Each group below still has a unique winner under both behaviours. +-- +-- Negative zero must be written as `-0.0D`, not `CAST(-0.0 AS DOUBLE)`: an unsuffixed `-0.0` is a +-- DecimalType literal, and Decimal has no signed zero, so the cast yields `+0.0` and the column +-- would silently contain no negative zeros at all. -- ============================================================ statement @@ -114,11 +122,50 @@ CREATE TABLE mode_double(v double, grp string) USING parquet statement INSERT INTO mode_double VALUES (1.5, 'a'), (1.5, 'a'), (2.5, 'a'), (NULL, 'a'), - (CAST(0.0 AS DOUBLE), 'b'), (CAST(-0.0 AS DOUBLE), 'b'), (CAST(-0.0 AS DOUBLE), 'b'), (7.0, 'b') + (0.0D, 'b'), (-0.0D, 'b'), (-0.0D, 'b'), (7.0, 'b') query SELECT grp, mode(v) FROM mode_double GROUP BY grp ORDER BY grp +-- ============================================================ +-- Signed zeros: which value wins depends on whether -0.0 and 0.0 share a key +-- +-- The counts are -0.0:2, 0.0:2, 5.0:3 (the SPARK-57329 reproducer). Keeping the zeros apart makes +-- 5.0 the unique winner; folding them together makes the zero win with 4. The two candidate +-- answers differ in magnitude, so this catches the divergence even where a `-0.0` vs `0.0` +-- difference would not be visible. Neither behaviour produces a tie, so there is no dependence on +-- Spark's non-deterministic tie-break. +-- ============================================================ + +statement +CREATE TABLE mode_signed_zero(v double, grp string) USING parquet + +statement +INSERT INTO mode_signed_zero VALUES + (-0.0D, 'a'), (-0.0D, 'a'), + (0.0D, 'a'), (0.0D, 'a'), + (5.0D, 'a'), (5.0D, 'a'), (5.0D, 'a') + +query +SELECT grp, mode(v) FROM mode_signed_zero GROUP BY grp ORDER BY grp + +-- ============================================================ +-- NaN collapses to a single key on every supported version +-- +-- `doubleToLongBits` maps every NaN bit pattern to one value, so the two NaNs outvote the single +-- 1.0 and the mode is NaN. +-- ============================================================ + +statement +CREATE TABLE mode_nan(v double, grp string) USING parquet + +statement +INSERT INTO mode_nan VALUES + (CAST('NaN' AS DOUBLE), 'a'), (CAST('NaN' AS DOUBLE), 'a'), (1.0D, 'a') + +query +SELECT grp, mode(v) FROM mode_nan GROUP BY grp ORDER BY grp + -- ============================================================ -- Decimal -- ============================================================ @@ -168,6 +215,26 @@ INSERT INTO mode_temporal VALUES query SELECT grp, mode(d), mode(t) FROM mode_temporal GROUP BY grp ORDER BY grp +-- ============================================================ +-- TimestampNTZ (declared supported by isSupportedType, so exercise it directly) +-- ============================================================ + +statement +CREATE TABLE mode_ntz(t timestamp_ntz, grp string) USING parquet + +statement +INSERT INTO mode_ntz VALUES + (TIMESTAMP_NTZ '2024-01-01 00:00:00', 'a'), + (TIMESTAMP_NTZ '2024-01-01 00:00:00', 'a'), + (TIMESTAMP_NTZ '2024-06-15 12:30:00', 'a'), + (NULL, 'a'), + (TIMESTAMP_NTZ '1970-01-01 00:00:00', 'b'), + (TIMESTAMP_NTZ '1970-01-01 00:00:00', 'b'), + (TIMESTAMP_NTZ '2000-12-31 23:59:59', 'b') + +query +SELECT grp, mode(t) FROM mode_ntz GROUP BY grp ORDER BY grp + -- ============================================================ -- Unsupported input type falls back to Spark -- diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/mode_within_group.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/mode_within_group.sql new file mode 100644 index 0000000000..1843139648 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/aggregate/mode_within_group.sql @@ -0,0 +1,70 @@ +-- 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. + +-- The ordered forms of `mode` set `reverseOpt`, which gives them deterministic tie-breaking that +-- Comet does not implement, so they must fall back to Spark. `allowIncompatible` is enabled so +-- that the ordering check is the only thing that can trigger a fallback; without it every query +-- here would fall back for the unrelated tie-break reason and the assertions would be vacuous. +-- +-- `mode(col, reverse)` and `mode() WITHIN GROUP (ORDER BY col)` are Spark 4.0 features; on Spark +-- 3.x `Mode` only has the plain `mode(col)` form, which mode.sql covers. +-- MinSparkVersion: 4.0 +-- Config: spark.comet.expression.Mode.allowIncompatible=true + +statement +CREATE TABLE mode_ordered(v int, grp string) USING parquet + +statement +INSERT INTO mode_ordered VALUES + (10, 'a'), (10, 'a'), (7, 'a'), (NULL, 'a'), + (5, 'b'), (5, 'b'), (5, 'b'), (9, 'b') + +-- ============================================================ +-- Sentinel: the plain form still runs natively under this config, so a silent whole-expression +-- regression cannot make the fallback assertions below pass vacuously. +-- ============================================================ + +query +SELECT grp, mode(v) FROM mode_ordered GROUP BY grp ORDER BY grp + +-- ============================================================ +-- Deterministic-flag form +-- +-- `ModeBuilder` only sets `reverseOpt` for `mode(col, true)`; it rewrites `mode(col, false)` to the +-- plain `Mode(child)` (Mode.scala, `ModeBuilder.build`), so the `false` form must still run +-- natively. That asymmetry is the reason `modeHasUnsupportedOrdering` keys off `reverseOpt` rather +-- than off the argument count. +-- ============================================================ + +query +SELECT mode(v, false) FROM mode_ordered + +query expect_fallback(mode with a deterministic flag or WITHIN GROUP ordering is not supported) +SELECT mode(v, true) FROM mode_ordered + +-- ============================================================ +-- WITHIN GROUP form, ascending and descending +-- ============================================================ + +query expect_fallback(mode with a deterministic flag or WITHIN GROUP ordering is not supported) +SELECT mode() WITHIN GROUP (ORDER BY v) FROM mode_ordered + +query expect_fallback(mode with a deterministic flag or WITHIN GROUP ordering is not supported) +SELECT mode() WITHIN GROUP (ORDER BY v DESC) FROM mode_ordered + +query expect_fallback(mode with a deterministic flag or WITHIN GROUP ordering is not supported) +SELECT grp, mode() WITHIN GROUP (ORDER BY v) FROM mode_ordered GROUP BY grp ORDER BY grp