[parthchandra]

Which issue does this PR close?

Closes #7024.

Rationale for this change

A single min/max over a struct input sends the whole aggregate operator back to Spark. A common pattern is max(struct(order_col, payload_1, payload_2, ...)), used to pick the payload of the row with the largest ordering value. DataFusion already has a struct min/max accumulator. Only the Scala-side gate blocked it.

What changes are included in this PR?

• AggSerde.minMaxSupportLevel now accepts StructType:
  • Compatible when every field, at any depth, is a type native min/max supports, no field is float or double, and no field is nullable.
  • Incompatible (opt-in with spark.comet.expression.Min/Max.allowIncompatible) when some field is nullable. DataFusion skips a null field when comparing two structs, while Spark orders null as the smallest value.
  • Unsupported for float/double fields, non-default collations, and other unsupported leaf types such as arrays and timestamp_ntz.
• The native planner casts a struct input to Spark's declared type with Comet's Cast instead of DataFusion's CastExpr. Native expressions such as coalesce can report nullable fields where Spark declares non-nullable ones, and CastExpr rejects that narrowing. The aggregate keeps Spark's declared type, so its state matches what shuffle and scans hand to the final stage.
• Window min/max over structs still falls back.

How are these changes tested?

New SQL file tests that compare against Spark:

• min_max_struct.sql:
  • grouped, ungrouped and empty inputs
  • nested structs and whole-struct nulls
  • boolean, integral, decimal(38), date, timestamp, string (CJK) and binary fields
  • ties decided by later fields
  • PartialMerge via count(DISTINCT)
  • Spark partial to native final
  • fallback reasons for float, array and timestamp_ntz fields
  • Spark's analysis error for map fields
• min_max_struct_nullable.sql: the opt-in path. The known null-field difference is recorded as an ignored query.
• min_max_struct_timezone.sql: four session time zones, DST transitions, and values near the ends of the range.
• min_max_struct_no_comet_shuffle.sql: Comet shuffle disabled.

Plus a Rust unit test that the struct cast keeps values when it relabels field nullability.

[sunchao]

Reviewed 0047452b375a8b0336d26427fdaf9bd5113979f3 with five independent scopes covering correctness, resources, design/reuse, tests, and release/documentation compatibility. Requesting changes for the newly exposed accumulator accounting failure and the mixed-stage test gap described inline. The recursive type guards and explicit nullable-field opt-in otherwise look consistent with the documented semantics.

Please also update the window compatibility paragraph: it still says struct min/max falls back in both batch and window contexts. Supported structs now run as native batch aggregates, while window fallback remains unchanged.

Validation: all 61 current-head checks are successful, including the Comet Spark-profile matrix, Rust tests, and Spark SQL 3.4/4.1 suites; 18 other checks are skipped. Suite registration checks passed locally. The accumulator reproduction compiles the pinned DataFusion 55.2 source and uses cached DataFusion 55.1-fork / Arrow 59.2 support libraries. Its accumulator source is byte-identical, and I checked the relevant ordered-stream accounting in 55.2 separately. The otherwise identical one-line control completes where the baseline fails. This is a standalone DataFusion reproduction, not a full Spark/Comet run or an end-to-end performance benchmark.

[sunchao]

Could we fix or pick up DataFusion's struct accumulator accounting before enabling this path by default, or keep the affected path on Spark until then? In the pinned 55.2 MinMaxStructState::emit_to(EmitTo::First(n)), emitted state is subtracted using s.len() (one row per struct), although set_value charged get_array_memory_size() bytes. Ordered aggregation therefore continues reserving memory for groups it has already emitted. This is reachable through the normal native SortAggregate path.

With 64 groups carrying 64 KiB payloads, a component probe reports 4,222,912 bytes after emitting every group, then 8,445,824 after the next cohort, despite retaining no groups. A real ordered AggregateExec probe with 4,096 groups, 64-row batches, 4 KiB payloads and a 2 MiB pool fails with ResourcesExhausted; changing only the subtraction to the emitted arrays' memory size makes all 4,096 rows complete. The defect is upstream, but this gate newly exposes it to queries that previously stayed on Spark. Please include a bounded-memory, multi-batch sorted struct min/max regression with the fix. The review summary records the standalone harness's version boundary.

[parthchandra]

Thanks, good catch. Rather than wait for the upstream fix, struct min/max no longer uses DataFusion's struct accumulator. There is now a Comet one, SparkStructMinMax, in native/spark-expr/src/agg_funcs/min_max_struct.rs.

• Each batch is encoded once with Arrow's row format, ascending with nulls first (line 117). Each group keeps the bytes of its current winner, and memory is counted from those buffers (update, line 151).
• When groups are emitted, it subtracts exactly the bytes those groups held (evaluate, lines 281-284), so sorted aggregation stops holding memory for groups it already emitted.
• min_max_udaf sends struct types to it in native/core/src/execution/planner/aggregate.rs, line 927.

For the regression you asked for, sorted_aggregation_fits_a_small_pool (min_max_struct.rs, line 495) follows your setup:

• A Partial stage feeds a Final stage over sorted input: 4,096 groups of 4 KiB values, 64-row batches, a 2 MiB pool, with spilling to disk turned off.
• It asserts both stages run in DataFusion's sorted mode.
• With DataFusion's max_udaf swapped in, it fails with ResourcesExhausted in OrderedFinalAggregateStream. With the new accumulator it completes.

emitting_groups_releases_their_bytes (line 406) checks the accounting directly. It emits groups in two halves and checks the tracked bytes go back to zero each round.

A side effect: the row format orders struct fields exactly the way Spark does, with a null field below every value. So structs with nullable fields now match Spark, and they no longer need the opt-in. minMaxStructSupportLevel (aggregates.scala, line 1653) now only rejects float/double fields, collated strings and unsupported leaf types. The null-ordering cases are covered in min_max_struct_nullable.sql and in orders_null_fields_first_like_spark (min_max_struct.rs, line 370). Float and double fields still fall back, because the row format orders -0.0 below 0.0.

[sunchao]

Could we assert the partial/final engine placement for at least one struct case in each direction? Every query in this fixture uses spark_answer_only, so all four config combinations still pass if a fallback regression sends both stages to Spark. That leaves the new struct-state crossing contract untested even though result parity is checked. The existing CometAggregateSuite test avg empty partial buffers cross engines in both directions already shows the small native/Spark mode assertions needed here. A focused struct case using that pattern can establish the boundary while retaining these SQL fixtures for the broader parity matrix.

[parthchandra]

Agreed, spark_answer_only would hide a full fallback. I added struct min/max buffers cross engines in both directions to CometAggregateSuite. It follows the pattern of the avg test you pointed to.

• It turns off native final and asserts a native Partial with a Spark Final. Then it turns off native partial and asserts a Spark Partial with a native Final.
• It uses JVM columnar shuffle, so Spark's partial rows can reach the native final.
• The data has a null in the deciding field, so the null-first ordering is checked across the engine boundary too.

I kept the SQL file with its partial/final matrix (lines 23-24) for broader answer checks across the four combinations.

[sunchao]

Re-reviewed 0047452b375a8b0336d26427fdaf9bd5113979f3 across five independent scopes. This head is unchanged. Both existing requests remain:

• Fix the newly exposed struct accumulator accounting: prefix emission still subtracts row count from a byte counter, retaining phantom reservations in ordered aggregation.
• Assert the mixed partial/final engine placement: every mixed-stage query still uses spark_answer_only, so both stages falling back to Spark would pass.

The compatibility paragraph also still says struct min/max falls back in both batch and window contexts. No additional finding from this pass. Current-head checks remain green (61 successes, 18 skips). I rechecked the relevant source, fixtures and prior reproduction logs; the standalone harness's version boundary remains as documented in the previous review, and I did not rerun the full Spark/Comet suites. Keeping changes requested.

[andygrove]

I don't have additional comments beyond the existing review. I'll check back once those points are addressed.

[parthchandra]

Please also update the window compatibility paragraph: it still says struct min/max falls back in both batch and window contexts. Supported structs now run as native batch aggregates, while window fallback remains unchanged.

Updated.

[comphead]

Encoding each batch once with RowConverter and keeping the winning row bytes per group gives Spark's nulls-first struct order without the null-field opt-in, and it is the same approach as DuckDB's sort-key min/max. The description still describes the first revision: it says DataFusion's accumulator is reused, lists the allowIncompatible opt-in for nullable fields and a min_max_struct_no_comet_shuffle.sql that is not in the PR, and does not mention SparkStructMinMax, the new fixtures or the StructArray::try_new change in cast_struct_to_struct. Could it be refreshed before merge? Three comments.

[comphead]

Now that nullable fields run without the opt-in, this sentence no longer holds, since max(struct(ts, s, n)) at line 56 tests nullable fields here. min_max_struct_nullable.sql has no -- Config: header left, and two of its cases overlap this file: that query is a subset of its first query, and the struct-column table at the end repeats its last block with fewer nulls. Could the two files merge, dropping those duplicates?

[comphead]

Could you file the MinMaxStructState::emit_to accounting bug in DataFusion and link it here? set_value adds get_array_memory_size() bytes while EmitTo::First subtracts s.len() rows, and I could not find an upstream issue for it. The link would also tell the next reader why Comet keeps its own accumulator beyond null ordering.

[comphead]

update cannot return a negative delta, since clear() followed by extend_from_slice never shrinks the Vec's capacity. Returning usize would let both callers do self.value_bytes += delta instead of a signed accumulator and wrapping_add_signed, which also makes it clear that the counter only grows until evaluate subtracts what it emits.

[sunchao]

Re-reviewed ea79a1f364f0254a292c661506b86a5ed8d62b9d with five independent reviewers covering correctness/Spark semantics, resources/performance, design/reuse, test adequacy/maintainability, and release/documentation compatibility. The earlier phantom-reservation and mixed-stage assertion findings are addressed. The new accumulator releases emitted buffer capacities, and the Scala test now checks both native/Spark stage directions.

I am requesting changes for the measured wide-payload regression described inline. With the same Comet scan/sort/shuffle settings, enabling native Max made the local 4 KiB-payload workload about 1.31x slower ungrouped and 1.96x slower grouped. Results matched. This is a local pipeline comparison, not an isolated accumulator profile or a production performance claim.

I also read the newly arrived comments. The request to refresh the description and consolidate overlapping main/nullable fixtures is valid, so I have not duplicated that feedback.

Validation: all 61 current-head checks passed, with 18 skipped. I inspected the logs confirming all seven new accumulator tests, the cast tests, the mixed-stage Scala test, and the struct SQL fixtures ran. Suite registration passed locally. The extra local probes used JVM sources compiled from this head, its CI-built native library, Spark 4.1.3 and JDK 17. Full Rust/Spark suites were verified through CI rather than rerun locally.

[sunchao]

[P2] Avoid the wide-payload regression before enabling struct min/max by default

Could we address the cost of encoding every complete struct before choosing the winner, and add a benchmark for wide payloads? For the motivating max(struct(order_key, payload)) shape, this serializes the payload even when the first key already decides the comparison. A matched current-head control shows the newly enabled path is slower than keeping the aggregate on Spark.

I wrote these 200,000 rows to one Parquet file and read the same file for both variants:

SELECT id % 1000 AS g, 200000 - id AS k,
       CASE WHEN id % 10 = 0 THEN NULL
            ELSE repeat(sha2(CAST(id AS STRING), 256), 64) END AS payload
FROM range(200000)

Then I collected SELECT max(struct(k, payload)) FROM t and SELECT g, max(struct(k, payload)) FROM t GROUP BY g. With Spark 4.1.3/JDK 17, local[1], AQE off, one shuffle partition, Comet enabled, on-heap execution and JVM columnar shuffle, I changed only spark.comet.expression.Max.enabled. Plans confirmed the same Comet scan and, for grouped input, the same native sorts and columnar shuffle. The control uses Spark partial/final aggregates and the enabled variant uses Comet partial/final aggregates.

After two warmup rounds, five alternating measured rounds gave median collection times of 607 ms fallback vs 798 ms native ungrouped, and 2,075 ms fallback vs 4,059 ms native grouped. Every native sample exceeded the slowest fallback sample for that query, and all answers matched. Planning and result comparison were outside the timed region. These are local pipeline measurements using this head's compiled JVM sources and CI native library; they establish the regression for this shape without attributing all elapsed time to this function. The 64-byte payload case was faster than plain Spark, so a benchmark should retain both narrow and wide cases.