[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...)), used to pick the payload of the row with the largest ordering value.

What changes are included in this PR?

• AggSerde.minMaxSupportLevel accepts StructType when every field, at any depth, is a type native min/max supports other than float and double. Nullable fields are supported by default; no opt-in is needed. Structs with float/double fields, non-default collations, arrays, timestamp_ntz or an empty struct fall back.
• New accumulator SparkStructMinMax (native/spark-expr/src/agg_funcs/min_max_struct.rs), used instead of DataFusion's struct min/max:
  • Within each batch, each group's best row is picked with Arrow's struct comparator, which stops at the first field that differs. Only those candidates are encoded with Arrow's row format (ascending, nulls first), and each group keeps its winner's bytes.
  • Both orders match Spark's struct order, a null field below every value included.
  • Memory is counted from the stored buffers, and emitted groups release exactly their bytes. DataFusion 55.2's struct accumulator subtracts row counts instead (MinMaxStructState::emit_to(EmitTo::First(n)) subtracts row counts from a byte counter datafusion#26891), which makes sorted aggregation run out of memory.
• The native planner checks that a struct input matches the declared type apart from field nullability, and relabels it with Comet's Cast. cast_struct_to_struct now uses StructArray::try_new, so a null under a non-nullable field is an error instead of a panic.
• Window min/max over structs still falls back. compatibility/operators.md is updated.

Grouped struct min/max still sorts its input, because Spark plans it as a sort aggregate. With very wide payloads that sort dominates. Dropping it under a Comet sort aggregate is tracked in #7323.

How are these changes tested?

• Rust unit tests for the accumulator:
  • Spark's null-first order, empty and filtered groups, merge, and the ungrouped accumulator.
  • Emitted groups releasing their bytes.
  • Agreement between the comparator and the row bytes, and a brute-force check across batches.
  • A sorted Partial-to-Final aggregation that fits a 2 MiB pool. DataFusion's struct max fails this one.
• A Rust test for the cast error path.
• SQL file tests that compare against Spark:
  • min_max_struct.sql: field types and boundary values, nested structs, null fields at every level, ties, FILTER, PartialMerge, empty input, and fallback and error cases.
  • min_max_struct_timezone.sql: five session time zones.
  • min_max_struct_many_batches.sql: 64-row batches over 200k rows.
  • min_max_struct_mixed_stages.sql: partial and final on different engines.
  • A collated struct field in min_max_collation.sql.
• CometAggregateSuite: struct min/max buffers cross engines in both directions asserts native/Spark stage placement both ways.
• CometAggregateExpressionBenchmark: struct min/max with 64-byte and 4 KiB payloads, grouped and global.

[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]

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?

[parthchandra]

Merged. min_max_struct_nullable.sql is gone and its cases are in min_max_struct.sql, starting at line 119. I dropped the max(struct(ts, s, n)) query and the smaller struct-column block, and rewrote the header

[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.

[parthchandra]

Filed apache/datafusion#26891

[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.

[parthchandra]

Agreed. update now returns usize and both callers just add it to value_bytes.

[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]

Re-reviewed ea79a1f364f0254a292c661506b86a5ed8d62b9d with five independent scopes. The earlier byte-accounting and mixed-stage assertion findings are addressed. The accumulator subtracts emitted buffer capacities, the new tests cover prefix release and a small memory pool, and the Scala test asserts both native/Spark stage directions. I am resolving those two earlier threads.

The current-head wide-payload performance finding remains unresolved. The implementation still encodes every complete struct before selecting the winner, including payloads whose leading key already loses. The published matched pipeline benchmark applies to this exact unchanged head. I rechecked the source and its measurement boundary without rerunning or presenting those timings as new measurements.

The existing requests to consolidate overlapping main/nullable fixtures and refresh the outdated PR description are also valid. I have not duplicated those comments. All 61 current-head checks remain successful, with 18 skipped. Keeping changes requested for the measured performance regression.

[parthchandra]

The current-head wide-payload performance finding remains unresolved. The implementation still encodes every complete struct before selecting the winner, including payloads whose leading key already loses. The published matched pipeline benchmark applies to this exact unchanged head. I rechecked the source and its measurement boundary without rerunning or presenting those timings as new measurements.

@sunchao I filed #7323 for this. Is that okay as a follow up or would you prefer it be addressed in this PR?

[sunchao]

The accumulator now compares fields before encoding a winner. I checked its nullable and nested ordering, grouped updates across batches, and the Spark/Comet partial-state exchange. The previous wide-payload slowdown did not reproduce with only native Max toggled, all results matched, and CI passed for this commit. I’m approving the implementation.

Please also refresh the PR description. It still says nullable struct fields require allowIncompatible and lists min_max_struct_nullable.sql as a separate fixture. The current implementation handles supported nullable fields by default, and those tests are now in min_max_struct.sql.

[parthchandra]

@sunchao thanks for re-running it. I've refreshed the description: nullable fields no longer need the opt-in, the null cases are in min_max_struct.sql, and it links apache/datafusion#26891 and #7323.

[parthchandra]

@comphead, @andygrove, any more comments?