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: [email protected]
For additional commands, e-mail: [email protected]
