Optimize and execute aggregation operations-PolarDB(PolarDB)-阿里云帮助中心

PolarDB-X processes GROUP BY queries using two aggregation operators: HashAgg and SortAgg. The query optimizer selects the operator based on available memory and input data characteristics — HashAgg in most cases, SortAgg when input is pre-sorted or memory is constrained. To reduce data transferred across shards, the optimizer also applies two-phase aggregation, pushing partial aggregation down to each MySQL storage shard before combining results at the compute layer.

Supported aggregate functions

PolarDB-X supports the following aggregate functions:

COUNT
SUM
AVG
MAX
MIN
BIT_OR
BIT_XOR
GROUP_CONCAT

Aggregation operators

The operators described in this section run at the PolarDB-X compute layer and are not pushed down to LogicalView. When an aggregation is pushed down to LogicalView, MySQL at the storage layer handles execution directly.

HashAgg

HashAgg is the default aggregation operator. Unlike SortAgg, it does not require sorted input, but it stores all groups in memory simultaneously. It uses a hash table to group input rows:

Assigns each input row to a group based on the hash value of its grouping column.
Merges the row's data into the group using the specified aggregate function.
Repeats until all rows are processed, then returns the aggregation result.

The following example shows HashAgg in an EXPLAIN plan:

> explain select count(*) from t1 join t2 on t1.id = t2.id group by t1.name,t2.name;

Project(count(*)="count(*)")
  HashAgg(group="name,name0", count(*)="COUNT()")
    BKAJoin(condition="id = id", type="inner")
      Gather(concurrent=true)
        LogicalView(tables="t1", shardCount=2, sql="SELECT `id`, `name` FROM `t1` AS `t1`")
      Gather(concurrent=true)
        LogicalView(tables="t2_[0-3]", shardCount=4, sql="SELECT `id`, `name` FROM `t2` AS `t2` WHERE (`id` IN ('?'))")

The HashAgg node contains two key fields:

group: the columns used to group data. Duplicate column names are disambiguated with a numeric suffix — in name,name0, name refers to t1.name and name0 refers to t2.name.
Aggregate function: the output column name appears before the =, and the calculation method appears after it. In count(*)="COUNT()", count(*) is the output column name and COUNT() counts the input rows.

To disable HashAgg, use the hint /*+TDDL:cmd_extra(ENABLE_HASH_AGG=false)*/.

SortAgg

SortAgg uses less memory than HashAgg because it processes one group at a time instead of holding all groups in memory. It requires that input rows are already sorted by the grouping column, which is typically provided by a MergeSort or MemSort operator upstream.

SortAgg processes data as follows:

Reads sorted input rows in sequence.
Accumulates rows that belong to the current group.
Outputs the group's aggregation result when a row from a different group is encountered, then starts a new group.

The following example shows the query plan when an ORDER BY clause is added, which triggers a MemSort after aggregation to sort the final output:

> explain select count(*) from t1 join t2 on t1.id = t2.id group by t1.name,t2.name order by t1.name, t2.name;

Project(count(*)="count(*)")
  MemSort(sort="name ASC,name0 ASC")
    HashAgg(group="name,name0", count(*)="COUNT()")
      BKAJoin(condition="id = id", type="inner")
        Gather(concurrent=true)
          LogicalView(tables="t1", shardCount=2, sql="SELECT `id`, `name` FROM `t1` AS `t1`")
        Gather(concurrent=true)
          LogicalView(tables="t2_[0-3]", shardCount=4, sql="SELECT `id`, `name` FROM `t2` AS `t2` WHERE (`id` IN ('?'))")

To disable SortAgg, use the hint /*+TDDL:cmd_extra(ENABLE_SORT_AGG=false)*/.

Two-phase aggregation

Two-phase aggregation reduces data transfer between storage shards and the compute layer. In the partial aggregation phase, each MySQL shard pre-aggregates its local data. In the final aggregation phase, PolarDB-X combines the partial results.

The optimizer also rewrites complex functions to make partial aggregation possible. For example, AVG is split into SUM and COUNT so each shard can compute its contribution independently.

The following example shows how the optimizer rewrites an AVG query:

> explain select avg(age) from t2 group by name

Project(avg(age)="sum_pushed_sum / sum_pushed_count")
  HashAgg(group="name", sum_pushed_sum="SUM(pushed_sum)", sum_pushed_count="SUM(pushed_count)")
    Gather(concurrent=true)
      LogicalView(tables="t2_[0-3]", shardCount=4, sql="SELECT `name`, SUM(`age`) AS `pushed_sum`, COUNT(`age`) AS `pushed_count` FROM `t2` AS `t2` GROUP BY `name`")

Each shard computes SUM(age) and COUNT(age) locally and returns the partial results. The compute layer then divides sum_pushed_sum by sum_pushed_count to produce the final average.

When to use SortAgg

HashAgg is the default and handles most queries well. Consider SortAgg in these scenarios:

Memory pressure: the dataset is large enough that HashAgg exhausts available memory.
Pre-sorted input: the grouping columns are already sorted by an upstream sort operator. SortAgg can stream through sorted input without re-sorting.
Severely skewed data: when a small number of groups contain most of the data, hash table overhead in HashAgg reduces efficiency.