feat: run min and max over struct inputs natively

docs/source/user-guide/latest/compatibility/operators.md

@@ -167,11 +167,14 @@ incorrect result. When any single window expression in a `WindowExec` falls back
   (`stddev`, `stddev_pop`, `stddev_samp`, `var_pop`, `var_samp`, `corr`, `covar_pop`, `covar_samp`). These run
   natively as plain aggregations but not as window functions
   ([#4766](https://github.com/apache/datafusion-comet/issues/4766)).
-- `min` / `max` on timestamp-without-time-zone, interval, or nested (array / struct) input types, and `sum` /
-  `avg` on year-month or day-time interval input types. Windowed aggregates inherit the same input-type support as
-  the batch aggregates, so these fall back in both contexts.
+- `min` / `max` on timestamp-without-time-zone, interval, or array input types, and `sum` / `avg` on year-month
+  or day-time interval input types. Windowed aggregates inherit the same input-type support as the batch
+  aggregates, so these fall back in both contexts.
 - `min` / `max` on string or binary input types. These run natively as plain aggregations but not as window
   functions.
+- `min` / `max` on struct input types. A struct whose fields are all supported `min` / `max` types other than
+  `float` and `double` runs natively as a plain aggregation, but not as a window function. Other structs fall
+  back in both contexts.
 - `sum` or `avg` on `DECIMAL` with a sliding (non ever-expanding) frame, because the sliding path would wrap on
   overflow instead of returning Spark's `NULL`.
 - `avg` or `try_avg` on `DECIMAL(p, s)` with an expanding frame when `p >= 28`, because Spark can preserve

native/core/src/execution/planner/aggregate.rs

@@ -51,10 +51,11 @@ use datafusion_comet_proto::{
     spark_operator::{AggregateMode as ProtoAggregateMode, Operator},
 };
 use datafusion_comet_spark_expr::{
-    ApproxPercentile, Avg, AvgDecimal, BloomFilterAgg, CometCollectList, CometCollectSet,
-    Correlation, Covariance, HllPlusPlus, HllSketchAgg, HllUnionAgg, MaxMinBy, Mode, Regr,
-    RegrType, SparkBloomFilterVersion, SparkListAgg, SparkMinMax, SparkPercentile, SparkPivotFirst,
-    Stddev, SumDecimal, SumInteger, Variance, ZeroOrder,
+    ApproxPercentile, Avg, AvgDecimal, BloomFilterAgg, Cast, CometCollectList, CometCollectSet,
+    Correlation, Covariance, EvalMode, HllPlusPlus, HllSketchAgg, HllUnionAgg, MaxMinBy, Mode,
+    Regr, RegrType, SparkBloomFilterVersion, SparkCastOptions, SparkListAgg, SparkMinMax,
+    SparkPercentile, SparkPivotFirst, SparkStructMinMax, Stddev, SumDecimal, SumInteger, Variance,
+    ZeroOrder,
 };
 use jni::objects::{Global, JObject};
 
@@ -339,7 +340,7 @@ impl PhysicalPlanner {
                 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 = min_max_udaf(&datatype, false);
-                let child = Arc::new(CastExpr::new(child, datatype.clone(), None));
+                let child = Self::cast_min_max_child(child, &datatype, &schema)?;
 
                 AggregateExprBuilder::new(func, vec![child])
                     .schema(schema)
@@ -353,7 +354,7 @@ impl PhysicalPlanner {
                 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 = min_max_udaf(&datatype, true);
-                let child = Arc::new(CastExpr::new(child, datatype.clone(), None));
+                let child = Self::cast_min_max_child(child, &datatype, &schema)?;
 
                 AggregateExprBuilder::new(func, vec![child])
                     .schema(schema)
@@ -807,6 +808,38 @@ impl PhysicalPlanner {
         Ok(Arc::new(CastExpr::new(child, nullable_type, None)))
     }
 
+    /// Cast a min/max input to the aggregate's declared type. Native expressions such as
+    /// `struct(...)` can report nullable fields where Spark declares them non-nullable, and
+    /// DataFusion's `CastExpr` rejects that narrowing. For a struct, only nullability may differ:
+    /// Comet's `Cast` then relabels the fields without touching the data (and errors if a field
+    /// declared non-nullable holds a null), so the input matches the state that shuffle and scans
+    /// stamp with Spark's type. Any other difference is a planning error.
+    fn cast_min_max_child(
+        child: Arc<dyn PhysicalExpr>,
+        datatype: &DataType,
+        schema: &SchemaRef,
+    ) -> Result<Arc<dyn PhysicalExpr>, ExecutionError> {
+        if !matches!(datatype, DataType::Struct(_)) {
+            return Ok(Arc::new(CastExpr::new(child, datatype.clone(), None)));
+        }
+        let child_type = child.data_type(schema.as_ref())?;
+        if &child_type == datatype {
+            return Ok(child);
+        }
+        if !equal_ignoring_nullability(&child_type, datatype) {
+            return Err(GeneralError(format!(
+                "min/max input type {child_type} does not match the declared type {datatype}"
+            )));
+        }
+        Ok(Arc::new(Cast::new(
+            child,
+            datatype.clone(),
+            SparkCastOptions::new_without_timezone(EvalMode::Legacy),
+            None,
+            None,
+        )))
+    }
+
     pub(super) fn create_aggr_func_expr(
         name: &str,
         schema: SchemaRef,
@@ -823,6 +856,19 @@ impl PhysicalPlanner {
     }
 }
 
+/// True if two types match apart from struct field nullability and metadata.
+fn equal_ignoring_nullability(a: &DataType, b: &DataType) -> bool {
+    match (a, b) {
+        (DataType::Struct(fa), DataType::Struct(fb)) => {
+            fa.len() == fb.len()
+                && fa.iter().zip(fb.iter()).all(|(x, y)| {
+                    x.name() == y.name() && equal_ignoring_nullability(x.data_type(), y.data_type())
+                })
+        }
+        _ => a == b,
+    }
+}
+
 /// Return a copy of `data_type` with every nested field marked nullable. Map key fields are left
 /// non-nullable to preserve Arrow's map invariant. Primitive types are returned unchanged.
 fn make_all_fields_nullable(data_type: &DataType) -> DataType {
@@ -870,13 +916,15 @@ fn make_all_fields_nullable(data_type: &DataType) -> DataType {
     }
 }
 
-/// `min` or `max` over `data_type`: Spark's version for floats, which DataFusion orders
-/// differently, and DataFusion's for every other type.
+/// `min` or `max` over `data_type`: Spark's version for floats and structs, which DataFusion
+/// orders differently, and DataFusion's for every other type.
 pub(super) fn min_max_udaf(data_type: &DataType, is_max: bool) -> Arc<AggregateUDF> {
     // Spark has only 32- and 64-bit floating-point types. Keep Float16 on DataFusion's path so
     // an internal plan using it does not reach SparkMinMax, whose accumulators reject that type.
     if matches!(data_type, DataType::Float32 | DataType::Float64) {
         Arc::new(AggregateUDF::new_from_impl(SparkMinMax::new(is_max)))
+    } else if matches!(data_type, DataType::Struct(_)) {
+        Arc::new(AggregateUDF::new_from_impl(SparkStructMinMax::new(is_max)))
     } else if is_max {
         max_udaf()
     } else {
@@ -898,3 +946,53 @@ fn percentile_value(expr: &spark_expression::Expr) -> Result<f64, ExecutionError
         )),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use arrow::datatypes::{Fields, TimeUnit};
+    use std::collections::HashMap;
+
+    #[test]
+    fn equal_ignoring_nullability_allows_only_nullability_and_metadata() {
+        let inner = |nullable| {
+            DataType::Struct(Fields::from(vec![Field::new(
+                "s",
+                DataType::Utf8,
+                nullable,
+            )]))
+        };
+        let outer = |nullable, ts_type: DataType| {
+            DataType::Struct(Fields::from(vec![
+                Field::new("ts", ts_type, nullable),
+                Field::new("inner", inner(nullable), nullable),
+            ]))
+        };
+        let ts = DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into()));
+
+        assert!(equal_ignoring_nullability(
+            &outer(true, ts.clone()),
+            &outer(false, ts.clone())
+        ));
+
+        let with_metadata =
+            DataType::Struct(Fields::from(vec![Field::new("s", DataType::Utf8, true)
+                .with_metadata(HashMap::from([("k".to_string(), "v".to_string())]))]));
+        assert!(equal_ignoring_nullability(&with_metadata, &inner(false)));
+
+        // A different leaf type, timezone label or field name is not a relabel.
+        assert!(!equal_ignoring_nullability(
+            &outer(false, ts.clone()),
+            &outer(false, DataType::Int64)
+        ));
+        assert!(!equal_ignoring_nullability(
+            &outer(false, ts),
+            &outer(
+                false,
+                DataType::Timestamp(TimeUnit::Microsecond, Some("+05:30".into()))
+            )
+        ));
+        let renamed = DataType::Struct(Fields::from(vec![Field::new("t", DataType::Utf8, false)]));
+        assert!(!equal_ignoring_nullability(&renamed, &inner(false)));
+    }
+}