[
https://issues.apache.org/jira/browse/HIVE-30223?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Denys Kuzmenko updated HIVE-30223:
----------------------------------
Description:
A reduce-side GROUP BY reads a sorted shuffle: the map side sorts its partial
rows by key, the reducer merges the sorted runs and the GROUP BY streams over
them. Merging SUM, COUNT, MIN and
MAX partials needs the rows of a partition, not their order. Trino merges
them in a hash table over a hash-partitioned exchange and spills runs in hash
order under memory pressure. Hive pays
for the sort and the merge on every such edge.
ReduceSinkDeDuplication also moves a second aggregation into the merge's
reducer as a COMPLETE GROUP BY, which keeps the sort. TPC-DS q75's UNION
DISTINCT sorts 1.92B rows on 6 keys, and the
sort and merge take 12% of its CPU. q38 and q87 do the same for INTERSECT.
*Fix* ({{hive.optimize.groupby.unsorted.shuffle}}, default false)
* A reduce-side GROUP BY qualifies when it merges SUM, COUNT, MIN and MAX
partials by key (no DISTINCT, grouping sets or non-column keys) and its reducer
does not rely on key order: up to
the next ReduceSinks it holds only select, filter, map join, top n key and
hash aggregation. Such a GROUP BY reads an unordered partitioned edge with no
sort and aggregates in a hash table.
The edge keeps auto-reduce.
* Vectorized reducers only: a reducer that does not vectorize gets its sorted
shuffle back.
* The vectorized reducer batches the rows of an unordered input across keys.
Without that, every key would be forwarded as a batch of one row.
* When the table outgrows the hash table memory
({{hive.map.aggr.hash.percentmemory}} of the task's) or the GC canary dies, it
is written to a local run under the task's local dirs, in key
hash order (as in Trino's SpillableHashAggregationBuilder). The run is
streamed, with no copy of the table. At close, the runs merge by hash, with
keys sorted within a hash, through the
streaming mode with the same merge aggregators.
* A COMPLETE GROUP BY on all of the merge's keys, over the merge's rows
(through selects and filters), gets one row per group and stays as it is (q38,
q87). Otherwise it splits into a hash
aggregation and a merge over the unsorted partial results, and the two share
its hash table memory (q75). A reducer that does not vectorize gets the
COMPLETE GROUP BY back whole.
* A union is order-insensitive, so q38's and q87's union between the branch
aggregation and the next hash aggregation does not disqualify the edge.
*Measured* (TPC-DS 10 TB Parquet, 24 x r6id, same session, only the flag
changes)
||Query||off (s)||on (s)||
|q65|148.9|70.3|
|q97|100.7|73.0|
|q78|259.3|232.4|
> Reduce-side GROUP BY over an unsorted shuffle
> ---------------------------------------------
>
> Key: HIVE-30223
> URL: https://issues.apache.org/jira/browse/HIVE-30223
> Project: Hive
> Issue Type: Improvement
> Reporter: Denys Kuzmenko
> Priority: Major
>
> A reduce-side GROUP BY reads a sorted shuffle: the map side sorts its partial
> rows by key, the reducer merges the sorted runs and the GROUP BY streams over
> them. Merging SUM, COUNT, MIN and
> MAX partials needs the rows of a partition, not their order. Trino merges
> them in a hash table over a hash-partitioned exchange and spills runs in hash
> order under memory pressure. Hive pays
> for the sort and the merge on every such edge.
>
>
>
>
>
> ReduceSinkDeDuplication also moves a second aggregation into the merge's
> reducer as a COMPLETE GROUP BY, which keeps the sort. TPC-DS q75's UNION
> DISTINCT sorts 1.92B rows on 6 keys, and the
> sort and merge take 12% of its CPU. q38 and q87 do the same for INTERSECT.
>
>
>
>
>
> *Fix* ({{hive.optimize.groupby.unsorted.shuffle}}, default false)
>
>
> * A reduce-side GROUP BY qualifies when it merges SUM, COUNT, MIN and MAX
> partials by key (no DISTINCT, grouping sets or non-column keys) and its
> reducer does not rely on key order: up to
> the next ReduceSinks it holds only select, filter, map join, top n key and
> hash aggregation. Such a GROUP BY reads an unordered partitioned edge with no
> sort and aggregates in a hash table.
> The edge keeps auto-reduce.
>
>
> * Vectorized reducers only: a reducer that does not vectorize gets its
> sorted shuffle back.
>
> * The vectorized reducer batches the rows of an unordered input across
> keys. Without that, every key would be forwarded as a batch of one row.
>
> * When the table outgrows the hash table memory
> ({{hive.map.aggr.hash.percentmemory}} of the task's) or the GC canary dies,
> it is written to a local run under the task's local dirs, in key
> hash order (as in Trino's SpillableHashAggregationBuilder). The run is
> streamed, with no copy of the table. At close, the runs merge by hash, with
> keys sorted within a hash, through the
> streaming mode with the same merge aggregators.
>
>
> * A COMPLETE GROUP BY on all of the merge's keys, over the merge's rows
> (through selects and filters), gets one row per group and stays as it is
> (q38, q87). Otherwise it splits into a hash
> aggregation and a merge over the unsorted partial results, and the two
> share its hash table memory (q75). A reducer that does not vectorize gets the
> COMPLETE GROUP BY back whole.
> * A union is order-insensitive, so q38's and q87's union between the branch
> aggregation and the next hash aggregation does not disqualify the edge.
>
>
>
>
> *Measured* (TPC-DS 10 TB Parquet, 24 x r6id, same session, only the flag
> changes)
>
> ||Query||off (s)||on (s)||
>
>
> |q65|148.9|70.3|
>
>
> |q97|100.7|73.0|
>
>
> |q78|259.3|232.4|
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
