[
https://issues.apache.org/jira/browse/SPARK-58927?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-58927:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Major
> Labels: pull-request-available
>
> *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. Observed ~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
> surface).
> 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]
