feat: support native aggregate function mode

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 |

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)
+            }
         }
     }
 

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;

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;