[jira] [Created] (SPARK-58927) Support hash-based aggregation for collated grouping keys

["Ganesha S (Jira)" <ji...@apache.org>]

Ganesha S created SPARK-58927:
---------------------------------

             Summary: Support hash-based aggregation for collated grouping keys
                 Key: SPARK-58927
                 URL: https://issues.apache.org/jira/browse/SPARK-58927
             Project: Spark
          Issue Type: Bug
          Components: SQL
    Affects Versions: 4.3.0
            Reporter: Ganesha S


*Problem*
A GROUP BY (or any aggregation) on a non-binary collated key, e.g., UTF8_LCASE,  is planned as SortAggregateExec, which sorts the entire input and can spill. Casting the same key to UTF8_BINARY uses HashAggregateExec and is dramatically faster. A user reported ~197s (with disk spill, a sort over 229M rows) for a UTF8_LCASE grouping key vs. ~39s (no spill) for the same query with the key cast back to UTF8_BINARY, on the same 200M-row input.

*Root cause*

Hash-based aggregation (HashAggregateExec / ObjectHashAggregateExec) keys its in-memory map on the binary representation of the grouping keys. This is only correct when every grouping key is binary-stable (UnsafeRowUtils.isBinaryStable). Non-binary collations are not binary-stable, under UTF8_LCASE, 'a ' and 'A' are equal but have different bytes, so Aggregate.supportsHashAggregate returns false, and the planner falls back to SortAggregateExec.

Notably, Spark already solved the equivalent problem for hash joins in SPARK-48000 by injecting CollationKey into the join keys (RewriteCollationJoin), but the aggregation path was never given the same treatment. So collated joins hash while collated aggregations sort.

 *Proposal*

 Mirror RewriteCollationJoin for aggregation. Add an optimizer rule RewriteCollationAggregate that:
- injects CollationKey into non-binary-stable grouping keys, so grouping happens on the collation-normalized (binary-stable) bytes; and
- preserves any original grouping value referenced in the output by wrapping it in First(...) — an arbitrary representative of each collation-equal group (semantically equivalent to the representative row of a sort-based aggregate surfaces).

 Because carrying the original string key via First yields a non-mutable aggregation buffer (which precludes HashAggregateExec), the planner is updated to route these rewritten aggregations to ObjectHashAggregateExec, which supports non-mutable buffers and falls back to sort-only under memory pressure, instead of SortAggregateExec. This avoids the full-input sort/spill.

 Guarded by a new config `spark.sql.collation.hashAggregation.enabled` (default true); setting it to false restores the previous sort-based behavior.

 Result: collated GROUP BY avoids the mandatory sort, matching the performance profile of the UTF8_BINARY cast while keeping collation-correct grouping and output.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org