Support MIN/MAX over StructType in native aggregates #7024

Summary

Comet does not convert min/max aggregates whose input is a StructType. The
whole aggregate operator (HashAggregate, ObjectHashAggregate, or SortAggregate)
falls back to Spark when any aggregate expression is unsupported, so a single
max(struct(...)) keeps the entire aggregation on the JVM.

The native engine already supports this: DataFusion 55.2.0 (the version Comet
pins) has MinMaxStructAccumulator (added upstream in apache/datafusion#15667),
including the grouped-accumulator path. The only thing blocking it is one gate on
the Scala side. This issue is about opening that gate safely, with the right
guards and parity tests.

Motivation

The common "max-by-with-payload" pattern produces struct-typed min/max. To keep
the row that has the largest ordering value and carry associated columns along,
users write:

df.groupBy("key").agg(
  max(struct(col("ts"), col("payloadA"), col("payloadB"))).alias("latest"))

Here max is over struct(bigint, string, string). Spark plans this as a
SortAggregate, and because the aggregate is struct-typed Comet declines it, so
the whole aggregation stays on the JVM. Supporting struct min/max lets this run
natively. This pattern is common wherever a job emulates max_by/min_by on
Spark versions (or buffer shapes) where the struct trick is the practical option.

Current behavior (where it is blocked)

AggSerde.minMaxSupportLevel in
spark/src/main/scala/org/apache/comet/serde/aggregates.scala:

def minMaxSupportLevel(dt: DataType): SupportLevel = dt match {
  case _: StringType if isStringCollationType(dt) =>
    Unsupported(Some(ComparisonUtils.nonDefaultCollationReason("min/max")))
  case _: StringType | BinaryType => Compatible()
  case _ if minMaxDataTypeSupported(dt) => Compatible()
  case _ => Unsupported(Some(s"Unsupported data type: $dt"))   // StructType lands here
}

minMaxUnsupportedReasons also lists "arrays and structs" as unsupported.

Paths audited (what is already covered)

I traced the full conversion path. Everything except the one Scala gate is in
place.

1. Shared aggregate gate. QueryPlanSerde.aggExprToProto (line ~805) is the
   single entry for HashAggregate, ObjectHashAggregate, and SortAggregate; all
   three call getSupportLevel -> minMaxSupportLevel. One change covers all
   three operators.

2. Scala serde. CometMin / CometMax convert() serialize the datatype with
   serializeDataType(expr.dataType); that already emits StructType (typeId 16 +
   StructInfo with field names, datatypes, nullability, field IDs). No change
   needed there.

3. Native planner (aggregate path). planner.rs AggExprStruct::Min/Max
   (~line 3264/3278) call min_max_udaf(datatype, is_max), which routes non-float
   types to DataFusion's min_udaf()/max_udaf(). The child is wrapped in a
   CastExpr to the (same) struct datatype; struct -> struct cast is supported
   (cast.rs:401, cast_struct_to_struct). So the identity cast is fine.

4. Native aggregate exec. Comet uses DataFusion's AggregateExec
   (planner.rs:1767/1920) with Partial/Final/Single modes. State for struct
   min/max is a single struct column produced/merged by
   MinMaxStructAccumulator.state() / merge_batch. The restrictive
   spark_aggregate_state.rs path is only for collect_list/collect_set, not
   min/max, so it does not apply.

5. DataFusion accumulator. MinMaxStructAccumulator implements both
   Accumulator and GroupsAccumulator, skips null inputs (matches Spark's
   null-skipping for min/max), and reports memory via size() (tracks
   total_data_bytes from get_array_memory_size()).

6. Window path is NOT affected and stays off. CometWindowExec gates min/max on
   minMaxDataTypeSupported (no struct), which we are not changing. Good -
   avoids sliding-window retract concerns for structs.

Proposed change

Add a StructType case to minMaxSupportLevel (plus update
minMaxUnsupportedReasons). Classify it carefully:

• Compatible only when the struct contains no FloatType/DoubleType anywhere
  (recursively) and all leaf fields are themselves min/max-supported with default
  collation strings.
• Otherwise keep Unsupported (do not silently opt in), or use Incompatible
  (opt-in via spark.comet.expression.<Name>.allowIncompatible) if we want power
  users to enable the known-divergent cases.

Recurse into nested structs; require every leaf to pass the same gate.

Parity risks and required guards

DataFusion compares structs with partial_cmp_struct (datafusion/common),
which flattens fields and compares in order with Arrow lt/eq kernels. Two real
divergences from Spark's InterpretedOrdering:

1. Float/double NaN. Arrow follows IEEE (NaN comparisons are false); Spark orders
   NaN as the largest value. A float/double field inside the struct can pick a
   different row than Spark. GUARD: reject (or mark Incompatible) any struct that
   contains a float/double field, at any nesting depth.

2. Null fields. partial_cmp_struct skips a field when either side is null (the
   lt/eq result is invalid), effectively treating null as "indifferent".
   Spark orders null as the smallest value. On ties where the deciding field is
   null, the two engines can pick different rows. When the ordering field is
   non-null (e.g. a positive timestamp) this does not bite, but a general feature
   must document it (and ideally test it).

Also carry over the existing non-default-collation string guard for string fields
inside the struct.

Memory / performance note

MinMaxStructAccumulator stores one StructArray per group. At high group
cardinality the per-group StructArray overhead is larger than a primitive
accumulator's. size() does account total_data_bytes, so the memory pool sees
it, but worth a benchmark on a high-cardinality struct min/max to check memory
pressure and spill behavior. Compare against the string min/max memory caveat
noted in #6936.

Test plan

Tests must compare Comet results against Spark and live in SQL test files where
possible.

• Struct min/max with grouping, struct of (bigint, string, string). Verify
  exact parity with Spark.
• Nullable vs non-nullable struct fields; whole-struct null (from a guarding
  when(...)) and field-level nulls.
• Nested structs (struct containing a struct), and struct containing array/map
  leaves if we choose to allow them.
• Boundary values on the ordering field: min/max bigint, decimals at max
  precision, dates and timestamps including timestamp with timezone and
  timestamp_ntz.
• Non-ASCII / CJK strings in string fields (raw-byte vs collation ordering).
• Float/double field present -> confirm it falls back (or is gated) and does not
  silently diverge on NaN.
• Ties on the ordering field with differing null fields -> document/assert the
  chosen behavior.
• Mixed partial/final: supportsSparkPartialToNativeFinal and
  supportsNativePartialToSparkFinal are both true for Min/Max, so cover
  Spark-partial -> native-final and native-partial -> Spark-final; confirm the
  struct state format round-trips through Comet shuffle.
• Failure-path / error-message checks where Spark raises structured errors.

Version context

Verified against datafusion-functions-aggregate 55.2.0 (Comet's pinned
version). min_max.rs dispatches Struct(_) to MinMaxStructAccumulator
(lines 336/630), flatten recurses nested structs, and partial_cmp_struct
(datafusion/common/src/scalar/mod.rs) skips null fields and lets float NaN follow
Arrow IEEE. The parity classification above (guard against float/double, document
null-field ties) stands for 55.2.0.

Out of scope

• Window-function min/max over structs (separate gate, keep off).
• Rewriting max(struct(order, payload)) into native max_by/min_by (Comet has
  CometMaxMinBy, but it requires fixed-length value and ordering, so it does not
  cover this; and it would need a Catalyst rewrite rule).