sunchao commented on code in PR #7191:
URL: https://github.com/apache/datafusion-comet/pull/7191#discussion_r4403603079
##########
spark/src/main/scala/org/apache/comet/serde/aggregates.scala:
##########
@@ -1639,14 +1643,70 @@ object AggSerde {
Unsupported(Some(ComparisonUtils.nonDefaultCollationReason("min/max")))
// DataFusion's grouped hash aggregate under-reports the memory these
values hold (#6936).
case _: StringType | BinaryType => Compatible()
+ case st: StructType => minMaxStructSupportLevel(st)
Review Comment:
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.
##########
spark/src/test/resources/sql-tests/expressions/aggregate/min_max_struct_mixed_stages.sql:
##########
@@ -0,0 +1,55 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Struct min/max with its partial and final stages on different engines. The
test-only configs
+-- below keep one stage on Spark, so the struct buffer crosses between Spark
and Comet in both
+-- directions: native partial -> Spark final, and Spark partial -> native
final. Min and Max allow
+-- both. The plan is mixed by design, so only the answer is checked.
+
+-- ConfigMatrix: spark.comet.testing.aggregate.partialMode.enabled=true,false
+-- ConfigMatrix: spark.comet.testing.aggregate.finalMode.enabled=true,false
+
+statement
+CREATE TABLE test_mm_struct_mixed(g int, ts bigint, s string, x int) USING
parquet
+
+statement
+INSERT INTO test_mm_struct_mixed VALUES
+ (1, 5, 'b', 1), (1, 9, '日本語', 2), (1, 9, '中文', 3),
+ (2, 9223372036854775807, 'a', 4), (2, -9223372036854775808, 'c', 5),
+ (3, NULL, NULL, NULL)
+
+query spark_answer_only
Review Comment:
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.
--
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]
