PolarDB:Optimize aggregation and sorting

HashAgg

HashAgg uses a hash table to group and aggregate rows in a single pass:

1. Hash the grouping column value of each input row to locate the corresponding group in the hash table.

2. Apply the aggregate function to accumulate the row into that group.

3. After all input rows are processed, emit the aggregation results.

Because HashAgg holds all groups in memory simultaneously, it consumes more memory for high-cardinality GROUP BY columns.

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 ('?'))")

Two fields in the HashAgg output are worth noting:

• group — the GROUP BY columns. When two tables have a column with the same name, PolarDB-X 1.0 appends a numeric suffix to distinguish them (for example, name and name0 represent t1.name and t2.name).

• Aggregate function — written as output_column="FUNCTION()". In count(*)="COUNT()", count(*) is the output column name and COUNT() is the computation applied to each group.

To disable HashAgg for a specific query, add the SQL hint /*+TDDL:cmd_extra(ENABLE_HASH_AGG=false)*/.

SortAgg

SortAgg aggregates one group at a time by reading pre-sorted input sequentially:

1. Require sorted input — the plan includes a MergeSort or MemSort operator upstream.

2. Read rows one by one; accumulate into the current group while the grouping key matches.

3. When the grouping key changes, emit the result for the previous group and start a new one.

Because SortAgg processes only one group at a time, it uses less memory than HashAgg — making it preferable when memory is constrained or when input is already ordered.

The following EXPLAIN plan shows a query where the optimizer uses MemSort to satisfy SortAgg's ordering requirement:

> 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 for a specific query, add the SQL hint /*+TDDL:cmd_extra(ENABLE_SORT_AGG=false)*/.

Two-phase aggregation

Two-phase aggregation splits a single Agg operator into a PartialAgg phase (pushed down to each MySQL shard) and a Final Agg phase (runs at the coordinator). Each shard computes a partial result, and the coordinator merges those partial results into the final answer. This significantly reduces the volume of data transferred from shards to the coordinator.

The AVG function requires special handling: it splits into SUM and COUNT so each shard can compute additive partial values. The coordinator then sums the partial totals and divides to produce the final average.

The following EXPLAIN plan shows two-phase aggregation for avg(age) group by name across four shards:

> 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 (visible inside the LogicalView SQL). The coordinator-level HashAgg sums those partial totals, then the Project operator divides to produce the final average.

MemSort

MemSort runs a Quick Sort algorithm entirely in the coordinator's memory. It is used when sorting cannot be pushed down — typically when the ORDER BY columns span multiple tables joined at the coordinator level.

> explain select t1.name from t1 join t2 on t1.id = t2.id order by t1.name,t2.name;

Project(name="name")
  MemSort(sort="name ASC,name0 ASC")
    Project(name="name", name0="name0")
      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 ('?'))")

TopN

When a query has both ORDER BY and LIMIT, the optimizer combines the Sort and Limit operators into a single TopN operator. TopN maintains a fixed-size max-heap or min-heap keyed on the sort columns. As rows arrive, the heap retains only the top N rows, discarding the rest. When all input is consumed, the heap holds the final result — without sorting the entire input.

> explain select t1.name from t1 join t2 on t1.id = t2.id order by t1.name,t2.name limit 10;

Project(name="name")
  TopN(sort="name ASC,name0 ASC", offset=0, fetch=? 0)
    Project(name="name", name0="name0")
      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 ('?'))")

MergeSort

When the ORDER BY semantics allow it, PolarDB-X 1.0 pushes the sort down to each MySQL shard and performs only a final merge at the coordinator. This is MergeSort. MergeSort also acts as a data redistribution operator — similar to the Gather operator — collecting pre-sorted streams from shards and merging them into a globally sorted output.

Because MySQL does the sorting work on each shard in parallel, MergeSort reduces coordinator memory consumption and makes full use of shard-level compute.

> explain select name from t1 order by name;

MergeSort(sort="name ASC")
  LogicalView(tables="t1", shardCount=2, sql="SELECT `name` FROM `t1` AS `t1` ORDER BY `name`")

The ORDER BY name inside the LogicalView SQL confirms the sort is pushed to MySQL. The coordinator-level MergeSort operator performs only the final merge.