This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 99237e16b2 feat: support native aggregate function `mode` (#4782)
99237e16b2 is described below
commit 99237e16b2de61183076045cb2ded1c11d20ec49
Author: Andy Grove <[email protected]>
AuthorDate: Thu Sep 10 13:33:13 2026 -0600
feat: support native aggregate function `mode` (#4782)
* 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
* 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
* fix: account for group slot storage in mode's grouped accumulator size()
---------
Co-authored-by: test <[email protected]>
---
docs/source/user-guide/latest/expressions.md | 2 +-
native/core/src/execution/planner.rs | 9 +-
native/proto/src/proto/expr.proto | 10 +
native/spark-expr/src/agg_funcs/mod.rs | 2 +
native/spark-expr/src/agg_funcs/mode.rs | 675 +++++++++++++++++++++
.../org/apache/comet/serde/QueryPlanSerde.scala | 1 +
.../scala/org/apache/comet/serde/aggregates.scala | 78 ++-
.../org/apache/spark/sql/comet/operators.scala | 17 +-
.../org/apache/comet/shims/CometTypeShim.scala | 6 +
.../org/apache/comet/shims/CometTypeShim.scala | 6 +
.../sql-tests/expressions/aggregate/mode.sql | 253 ++++++++
.../expressions/aggregate/mode_within_group.sql | 70 +++
12 files changed, 1122 insertions(+), 7 deletions(-)
diff --git a/docs/source/user-guide/latest/expressions.md
b/docs/source/user-guide/latest/expressions.md
index 231d80affb..c29b995fe1 100644
--- a/docs/source/user-guide/latest/expressions.md
+++ b/docs/source/user-guide/latest/expressions.md
@@ -114,7 +114,7 @@ The tables below list every Spark built-in expression with
its current status.
| `median` | ✅ | — | Rewrites to `percentile(col, 0.5)` and runs natively for
supported percentile inputs |
| `min` | ✅ | Native | |
| `min_by` | 🔜 | — |
[#3841](https://github.com/apache/datafusion-comet/issues/3841) |
-| `mode` | 🔜 | — |
[#3970](https://github.com/apache/datafusion-comet/issues/3970) |
+| `mode` | ✅ | Native | `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` | ✅ | Native | Single literal percentage on numeric input runs
natively; array of percentages and a frequency argument fall back to Spark |
| `percentile_cont` | ✅ | — | Spark 4.0+ `WITHIN GROUP (ORDER BY ...)`;
ascending only runs natively, `DESC` falls back to Spark |
| `percentile_disc` | 🔜 | — | Percentile aggregate |
diff --git a/native/core/src/execution/planner.rs
b/native/core/src/execution/planner.rs
index 3b2080db57..37d5e74441 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -149,7 +149,7 @@ use datafusion_comet_proto::{
use datafusion_comet_spark_expr::{
jvm_udf::JvmScalarUdfExpr, ApproxPercentile, ArrayInsert, Avg, AvgDecimal,
Cast, CheckOverflow,
Correlation, Covariance, CreateNamedStruct, DecimalRescaleCheckOverflow,
GetArrayStructFields,
- GetStructField, HllPlusPlus, IfExpr, ListExtract, NormalizeNaNAndZero,
Regr, RegrType,
+ GetStructField, HllPlusPlus, IfExpr, ListExtract, Mode,
NormalizeNaNAndZero, Regr, RegrType,
SparkCastOptions, Stddev, SumDecimal, ToJson, UnboundColumn, Variance,
WideDecimalBinaryExpr,
WideDecimalOp,
};
@@ -3186,6 +3186,13 @@ impl PhysicalPlanner {
let func =
AggregateUDF::new_from_impl(HllPlusPlus::new(expr.precision));
Self::create_aggr_func_expr("approx_count_distinct", schema,
vec![child], func)
}
+ AggExprStruct::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,
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 10df7706eb..06fdf19925 100644
--- a/native/proto/src/proto/expr.proto
+++ b/native/proto/src/proto/expr.proto
@@ -163,6 +163,7 @@ message AggExpr {
HllPlusPlus hllpp = 20;
CollectList collectList = 21;
Regr regr = 22;
+ Mode mode = 23;
}
// Optional filter expression for SQL FILTER (WHERE ...) clause.
@@ -352,6 +353,15 @@ message HllPlusPlus {
int32 precision = 2;
}
+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 {
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 828f3e989a..a63a8ddb0a 100644
--- a/native/spark-expr/src/agg_funcs/mod.rs
+++ b/native/spark-expr/src/agg_funcs/mod.rs
@@ -22,6 +22,7 @@ mod correlation;
mod covariance;
mod hll_plus_plus;
mod hll_plus_plus_const;
+mod mode;
mod percentile;
mod quantile_summaries;
mod regr;
@@ -37,6 +38,7 @@ pub use avg_decimal::AvgDecimal;
pub use correlation::Correlation;
pub use covariance::Covariance;
pub use hll_plus_plus::{hllpp_precision, HllPlusPlus};
+pub use mode::Mode;
pub use percentile::SparkPercentile;
pub use regr::{Regr, RegrType};
pub use stddev::Stddev;
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..7cba965329
--- /dev/null
+++ b/native/spark-expr/src/agg_funcs/mode.rs
@@ -0,0 +1,675 @@
+/*
+ * 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, not_impl_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
+///
+/// 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<T>,
counts: list<i64> }` (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,
+ /// 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, normalize_neg_zero: bool) -> Self {
+ Self {
+ name: "mode".to_string(),
+ signature: Signature::any(1, Volatility::Immutable),
+ data_type,
+ normalize_neg_zero,
+ }
+ }
+}
+
+/// Fields of the single struct state column `{values: list<T>, counts:
list<i64>}`.
+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<ScalarValue, i64>]) ->
Result<StructArray> {
+ 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<DataType> {
+ Ok(self.data_type.clone())
+ }
+
+ fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result<Box<dyn
Accumulator>> {
+ Ok(Box::new(ModeAccumulator::new(
+ self.data_type.clone(),
+ self.normalize_neg_zero,
+ )))
+ }
+
+ fn state_fields(&self, _args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
+ 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<Box<dyn GroupsAccumulator>> {
+ Ok(Box::new(ModeGroupsAccumulator::new(
+ self.data_type.clone(),
+ self.normalize_neg_zero,
+ )))
+ }
+}
+
+/// 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.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))
+ }
+ };
+ }
+ 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`, canonicalizing float keys.
+///
+/// The map is intentionally keyed on the type-generic `ScalarValue` rather
than a monomorphized
+/// `HashMap<Hashable<T::Native>, _>`: `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<ScalarValue, i64>,
+ 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)?,
normalize_neg_zero);
+ *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<ScalarValue, i64>,
+ 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(());
+ }
+ let values = values_list.value(row);
+ let counts = counts_list.value(row);
+ let counts = counts
+ .as_primitive_opt::<Int64Type>()
+ .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)?,
normalize_neg_zero);
+ *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<ScalarValue, i64>, data_type: &DataType) ->
Result<ScalarValue> {
+ 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),
+ }
+}
+
+/// 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<u8>`/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<ScalarValue, i64>) -> usize {
+ map.capacity() * size_of::<(ScalarValue, i64)>()
+ + map
+ .keys()
+ .map(|k| k.size().saturating_sub(size_of::<ScalarValue>()))
+ .sum::<usize>()
+}
+
+/// Non-grouped accumulator backing global `mode` aggregation.
+#[derive(Debug)]
+pub struct ModeAccumulator {
+ counts: HashMap<ScalarValue, i64>,
+ data_type: DataType,
+ normalize_neg_zero: bool,
+}
+
+impl ModeAccumulator {
+ fn new(data_type: DataType, normalize_neg_zero: bool) -> Self {
+ Self {
+ counts: HashMap::new(),
+ data_type,
+ normalize_neg_zero,
+ }
+ }
+}
+
+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, self.normalize_neg_zero)?;
+ }
+ Ok(())
+ }
+
+ fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
+ let structs = states[0].as_struct();
+ let values_list = structs.column(0).as_list::<i32>();
+ let counts_list = structs.column(1).as_list::<i32>();
+ for row in 0..structs.len() {
+ merge_state_row(
+ &mut self.counts,
+ values_list,
+ counts_list,
+ row,
+ self.normalize_neg_zero,
+ )?;
+ }
+ Ok(())
+ }
+
+ fn state(&mut self) -> Result<Vec<ScalarValue>> {
+ let array = build_state(&self.data_type, &[&self.counts])?;
+ Ok(vec![ScalarValue::Struct(Arc::new(array))])
+ }
+
+ fn evaluate(&mut self) -> Result<ScalarValue> {
+ eval_mode(&self.counts, &self.data_type)
+ }
+
+ fn size(&self) -> usize {
+ size_of_val(self) + map_size(&self.counts)
+ }
+}
+
+/// Vectorized grouped accumulator: one frequency map per group.
+#[derive(Debug)]
+pub struct ModeGroupsAccumulator {
+ groups: Vec<HashMap<ScalarValue, i64>>,
+ data_type: DataType,
+ normalize_neg_zero: bool,
+}
+
+impl ModeGroupsAccumulator {
+ fn new(data_type: DataType, normalize_neg_zero: bool) -> Self {
+ Self {
+ groups: Vec::new(),
+ data_type,
+ normalize_neg_zero,
+ }
+ }
+
+ 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,
+ self.normalize_neg_zero,
+ )?;
+ }
+ Ok(())
+ }
+
+ fn merge_batch(
+ &mut self,
+ values: &[ArrayRef],
+ group_indices: &[usize],
+ total_num_groups: usize,
+ ) -> Result<()> {
+ self.resize(total_num_groups);
+ let structs = values[0].as_struct();
+ let values_list = structs.column(0).as_list::<i32>();
+ let counts_list = structs.column(1).as_list::<i32>();
+ for (row, &group_index) in group_indices.iter().enumerate() {
+ 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<ArrayRef> {
+ 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)?);
+ }
+ ScalarValue::iter_to_array(results)
+ }
+
+ fn state(&mut self, emit_to: EmitTo) -> Result<Vec<ArrayRef>> {
+ 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<ScalarValue, i64>> = emitted.iter().collect();
+ Ok(vec![Arc::new(build_state(&self.data_type, &refs)?)])
+ }
+
+ fn convert_to_state(
+ &self,
+ _values: &[ArrayRef],
+ _opt_filter: Option<&BooleanArray>,
+ ) -> Result<Vec<ArrayRef>> {
+ not_impl_err!("Input batch conversion to state not implemented")
+ }
+
+ fn size(&self) -> usize {
+ // `size_of_val(self)` only covers the `Vec`'s inline
pointer/len/capacity, not the slot
+ // array it owns. One empty `HashMap` per group is 48 bytes of slot
that `map_size` reports
+ // as 0, so a run with many sparse groups would otherwise be invisible
to the memory pool
+ // that drives spill decisions.
+ size_of_val(self)
+ + self.groups.capacity() * size_of::<HashMap<ScalarValue, i64>>()
+ + self.groups.iter().map(map_size).sum::<usize>()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::array::{Float64Array, Int32Array};
+ use arrow::datatypes::Int32Type;
+
+ fn i32_array(values: Vec<Option<i32>>) -> 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, false);
+ 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, false);
+ 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, 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, 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();
+ assert_eq!(eval_acc(&mut acc), ScalarValue::Int32(Some(10)));
+ }
+
+ /// Turn an accumulator's `Vec<ScalarValue>` state into the state arrays
`merge_batch` consumes.
+ fn state_arrays(acc: &mut ModeAccumulator) -> Vec<ArrayRef> {
+ 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, false);
+ 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, 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, 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, false);
+ merged.merge_batch(&lstate).unwrap();
+ merged.merge_batch(&rstate).unwrap();
+ assert_eq!(eval_acc(&mut merged), single);
+ }
+
+ /// 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(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, 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();
+ let result = acc.evaluate(EmitTo::All).unwrap();
+ let result = result.as_primitive::<Int32Type>();
+ 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, 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, false);
+ final_acc.merge_batch(&state, &[0, 1], 2).unwrap();
+ let result = final_acc.evaluate(EmitTo::All).unwrap();
+ let result = result.as_primitive::<Int32Type>();
+ assert_eq!(result.value(0), 5);
+ assert_eq!(result.value(1), 9);
+ }
+
+ #[test]
+ fn groups_accumulator_size_accounts_for_group_slots() {
+ // All-NULL input leaves every per-group map empty, so `map_size`
reports 0 for each and the
+ // `Vec` slot array is the only allocation. Reporting only
`size_of_val(self)` there hides
+ // the whole thing from the memory pool that drives spill decisions.
+ const GROUPS: usize = 10_000;
+ let mut acc = ModeGroupsAccumulator::new(DataType::Int32, false);
+ acc.update_batch(
+ &[i32_array(vec![None; GROUPS])],
+ &(0..GROUPS).collect::<Vec<_>>(),
+ None,
+ GROUPS,
+ )
+ .unwrap();
+
+ let slot_bytes = GROUPS * size_of::<HashMap<ScalarValue, i64>>();
+ assert!(
+ acc.size() >= size_of_val(&acc) + slot_bytes,
+ "size() = {} does not cover {slot_bytes} bytes of empty group
slots",
+ acc.size()
+ );
+ }
+}
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 1fa43f07de..e748e82cd7 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -415,6 +415,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[RegrIntercept] -> CometRegrIntercept,
classOf[RegrR2] -> CometRegrR2,
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 c811c42559..4f4b0f5f41 100644
--- a/spark/src/main/scala/org/apache/comet/serde/aggregates.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/aggregates.scala
@@ -22,16 +22,16 @@ package org.apache.comet.serde
import scala.jdk.CollectionConverters._
import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression,
Literal}
-import
org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression,
ApproximatePercentile, Average, BitAndAgg, BitOrAgg, BitXorAgg,
BloomFilterAggregate, CentralMomentAgg, CollectList, CollectSet, Corr, Count,
Covariance, CovPopulation, CovSample, First, HyperLogLogPlusPlus, Last, Max,
Min, Percentile, RegrIntercept, RegrR2, RegrReplacement, RegrSlope, RegrSXY,
StddevPop, StddevSamp, Sum, VariancePop, VarianceSamp}
+import
org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression,
ApproximatePercentile, Average, BitAndAgg, BitOrAgg, BitXorAgg,
BloomFilterAggregate, CentralMomentAgg, CollectList, CollectSet, Corr, Count,
Covariance, CovPopulation, CovSample, First, HyperLogLogPlusPlus, Last, Max,
Min, Mode, Percentile, RegrIntercept, RegrR2, RegrReplacement, RegrSlope,
RegrSXY, StddevPop, StddevSamp, Sum, VariancePop, VarianceSamp}
import org.apache.spark.sql.catalyst.util.ArrayData
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType,
DataType, DateType, DecimalType, DoubleType, FloatType, IntegerType, LongType,
NumericType, ShortType, StringType, TimestampNTZType, TimestampType}
import org.apache.comet.CometConf.COMET_EXEC_STRICT_FLOATING_POINT
-import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus,
isSpark41Plus, withFallbackReason}
+import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus,
isSpark41Plus, isSpark42Plus, withFallbackReason}
import org.apache.comet.expressions.CometEvalMode
import org.apache.comet.serde.QueryPlanSerde.{evalModeToProto, exprToProto,
serializeDataType}
-import org.apache.comet.shims.{CometCollectShim, CometEvalModeUtil}
+import org.apache.comet.shims.{CometCollectShim, CometEvalModeUtil,
CometTypeShim}
object CometMin extends CometAggregateExpressionSerde[Min] {
@@ -1116,6 +1116,78 @@ object CometApproxCountDistinct extends
CometAggregateExpressionSerde[HyperLogLo
}
}
+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).
+ // 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)) {
+ // 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)
+ // 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()
+ .setMode(builder)
+ .build())
+ } else {
+ withFallbackReason(aggExpr, "Child expression or data type not
supported")
+ None
+ }
+ }
+}
+
object AggSerde {
import org.apache.spark.sql.types._
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
index a4a4c9d49e..3aa6b7220d 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
@@ -31,7 +31,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Ascending, Attribute,
AttributeSeq, AttributeSet, Expression, ExpressionSet, Generator,
NamedExpression, SortOrder}
-import
org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression,
AggregateMode, CollectList, CollectSet, Final, Partial, PartialMerge,
Percentile}
+import
org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression,
AggregateMode, CollectList, CollectSet, Final, 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._
@@ -44,7 +44,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
@@ -1906,6 +1906,19 @@ trait CometBaseAggregate {
// 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 7280fdc9dc..1e5d1686f9 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
@@ -24,6 +24,7 @@ import java.nio.charset.{CharacterCodingException,
CodingErrorAction, StandardCh
import scala.annotation.nowarn
+import org.apache.spark.sql.catalyst.expressions.aggregate.Mode
import org.apache.spark.sql.types.{DataType, StructType}
import org.apache.spark.unsafe.types.UTF8String
@@ -31,6 +32,11 @@ 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 1270eebfe0..8392aa76af 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,11 +19,17 @@
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, VariantType}
import org.apache.spark.unsafe.types.UTF8String
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..f32e4905d0
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/expressions/aggregate/mode.sql
@@ -0,0 +1,253 @@
+-- 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
+--
+-- 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
+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'),
+ (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
+-- ============================================================
+
+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
+
+-- ============================================================
+-- 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
+--
+-- 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
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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
