parthchandra opened a new issue, #7024:
URL: https://github.com/apache/datafusion-comet/issues/7024
### What is the problem the feature request solves?
## 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
Real workload: `IcebergTimeSeriesDiscoveryTransformJob` (applebot/caspian).
Its
aggregation uses the "correlated max-by" struct trick in several places,
e.g.:
```scala
min(when(col(s"${src}FirstSeenMillis") > 0,
struct(col(s"${src}FirstSeenMillis"),
col(s"${src}FirstSeenRequestSource"),
col(s"${src}FirstSeenSourceUrl")))).alias(s"_${src}First")
max(struct(col("lastCrawledTsMillis"), col("isRedirect"), col("finalUrl"),
col("redirectLastTsMillis"),
col("authorPublishedTsMillis"))).alias("_crawledMax")
```
These are `min`/`max` over `struct(long, string, string, ...)`. They force
Spark
to a `SortAggregate` and, more to the point, they are struct-typed min/max,
so
Comet declines them and the operator stays on Spark. Supporting struct
min/max
lets this aggregation run natively without touching the application.
## Current behavior (where it is blocked)
`AggSerde.minMaxSupportLevel` in
`spark/src/main/scala/org/apache/comet/serde/aggregates.scala`:
```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. For the caspian job the
ordering
field (`ts`) is always `> 0` and non-null, so this does not bite there,
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 (native review)
`MinMaxStructAccumulator` stores one `StructArray` per group. At high group
cardinality (the caspian job groups by `discoveredUrl` across 16000 shuffle
partitions) 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
off-heap
pressure and spill behavior. Compare against the string min/max memory caveat
noted in apache/datafusion-comet#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 `(long, string, string)` - the
caspian
shape. 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 long, decimals at max
precision,
dates and timestamps including timestamp with timezone and timestamp_ntz,
and
the [1980, now] range this job validates.
- 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 / build context
- `#4565` (SortAggregateExec support) and `#6927` (native string/binary
min/max)
are on `apple/main-apple` but not in any release tag yet. The image
`...comet-1.2.0.0-apple-20261005.164715-1` was built 2026-10-05, before
both.
- 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) is byte-identical to 53.1.0 - null
fields
are still skipped and float NaN still follows 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).
- Any change to the caspian application.
### Describe the potential solution
_No response_
### Additional context
_No response_
### Willingness to contribute
None
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
