comphead opened a new issue, #6909: URL: https://github.com/apache/datafusion-comet/issues/6909
### What is the problem the feature request solves? `CometMin` and `CometMax` accept only boolean, integral, floating-point, decimal, `DATE` and `TIMESTAMP` input (`AggSerde.minMaxDataTypeSupported`). Over a `STRING` or `BINARY` column the aggregate is `Unsupported`, so the whole aggregate operator falls back to Spark, and no setting changes that. This rarely mattered before #4565. A `min`/`max` buffer has the input's type, and Spark plans `HashAggregateExec` only when every buffer field is fixed-width, so a string or binary `min`/`max` gets `SortAggregateExec`, which Comet did not support. Such an aggregate reached Comet only next to a `TypedImperativeAggregate` such as `collect_list`, which switches Spark to `ObjectHashAggregateExec`. Now that #4565 runs `SortAggregateExec` natively, string `min`/`max` is what keeps this shape on Spark. String `first`/`last` under `SortAggregateExec` already run natively (`first_last.sql`), while `min_max.sql` still expects the fallback: ``` -- spark/src/test/resources/sql-tests/expressions/aggregate/min_max.sql query expect_fallback(Unsupported data type: StringType) SELECT min(i), max(i), min(d), max(d), min(s), max(s) FROM test_min_max ``` A common case is a grouped aggregate that takes `min`/`max` of a date stored as a string. Both the partial and the final `SortAggregate` fall back, so the plan sorts the full input on every grouping key, converts it to rows for the partial aggregate, shuffles it through the JVM columnar shuffle, then sorts and converts it again for the final aggregate. With many grouping keys and few duplicate groups, those sorts dominate the query and spill heavily. A grouped repro: ```sql CREATE TABLE t (k INT, s STRING, b BINARY) USING parquet; INSERT INTO t VALUES (1, 'b', X'02'), (1, 'a', X'01'), (2, NULL, NULL), (2, 'c', X'80'); SELECT k, min(s), max(s), min(b), max(b) FROM t GROUP BY k; ``` ### Describe the potential solution 1. Accept `StringType` and `BinaryType` in `AggSerde.minMaxSupportLevel`, which only `CometMin` and `CometMax` use. Leave `minMaxDataTypeSupported` as it is: `max_by`/`min_by` and the window `min`/`max` check in `CometWindowExec` call it directly, and they can follow separately. 2. Keep collated strings on Spark. From Spark 4.0, `min`/`max` compare a string with `binaryCompare` only when its collation supports binary ordering, and with `semanticCompare` otherwise. `serializeDataType` maps every `StringType` to the same proto type, so the check has to be explicit, as `supportedScalarSortElementType` does with `isStringCollationType`. 3. I expect no native change. `min_max_udaf` sends non-floating-point types to DataFusion's `min_udaf`/`max_udaf`, which in DataFusion 55.2 have a groups accumulator for `Utf8`, `LargeUtf8`, `Utf8View`, `Binary`, `LargeBinary` and `BinaryView` (`MinMaxBytesAccumulator`). It compares raw bytes. That matches Spark 3.4 through 4.1, where a default-collation string and a `BinaryType` value both compare with `ByteArray.compareBinary`: unsigned, byte by byte, shorter prefix first. The planner already casts the child to the declared type, so dictionary-encoded input reaches the accumulator as `Utf8`. Tests, in `min_max.sql` (turning the `expect_fallback` above into `query`) or a new file: - grouped and global aggregates, plus one next to `collect_list` so the `ObjectHashAggregateExec` shape is covered too - NULLs inside a group, an all-NULL group, an empty input, and `''` next to NULL - prefixes (`'a'`, `'ab'`), case (`'A'`, `'a'`), and characters whose UTF-8 byte order differs from their UTF-16 order, such as `'😀'` (U+1F600) and `'ｱ'` (U+FF71) - binary values on both sides of `X'7F'`/`X'80'`, which a signed comparison would order the other way - a low-cardinality string column, which Parquet writes dictionary-encoded - on Spark 4.0 and later, `min(s COLLATE UTF8_LCASE)`, which must fall back `CometMin` and `CometMax` also declare `supportsSparkPartialToNativeFinal` and `supportsNativePartialToSparkFinal`. Both should be re-checked with a string buffer. ### Additional context Not part of this issue: even when native, this shape still sorts its input twice, because Spark adds the sorts that `SortAggregateExec` requires. DataFusion's hash aggregate can hold string state, so a follow-up could drop the sort below a `CometSortAggregateExec` and let the native planner sort the aggregate's output instead, which it already does when the input ordering comes from outside the native plan. -- 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]
