parthchandra commented on code in PR #7191: URL: https://github.com/apache/datafusion-comet/pull/7191#discussion_r4406780788
########## 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: 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. ########## 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: 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. -- 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]
