Table of contents
Open Table of contents
A group can outlive a batch
Consider a shop counting sales by region:
SELECT /*+ STREAM_AGG() */ region, SUM(amount)
FROM sales
GROUP BY region;Suppose its input arrives ordered by region. TiDB reads rows in chunks, not one giant array. The first chunk ends halfway through the east group:
child chunk 1: (east, 12) (east, 7)
child chunk 2: (east, 4) (west, 5) (west, 8)
output rows: (east, 23) (west, 13)If the executor emits east = 19 at the end of chunk 1, it gives the wrong answer. It must hold the current aggregate until it knows the next group has started, even when that boundary appears in a later chunk. That is the core job of TiDB’s stream aggregation executor.
This walkthrough follows TiDB source at commit ab30f36. It focuses on four pieces: PhysicalStreamAgg, executorBuilder, VecGroupChecker, and StreamAggExec. The aggregate functions they call are an interface, not another execution pipeline explored here.
Who hands the group to whom?
planner executor construction
+---------------------+ +-------------------------+
| PhysicalStreamAgg |---------->| executorBuilder |
| ordered input needed| | wires child + group keys|
+---------------------+ +------------+------------+
|
+----------v-----------+
| StreamAggExec |
| owns current results |
+---+---------------+---+
| |
asks for boundaries pulls batches
| |
+--------v--------+ +---v---------+
| VecGroupChecker | | child exec |
| compare keys | | sorted rows |
+-----------------+ +-------------+The physical plan considers stream aggregation when it can require appropriately ordered child rows. Its ordinary candidate path uses possible child orderings; STREAM_AGG() also makes it consider an enforced ordering. A hint influences planning; it does not let the executor correctly aggregate unsorted rows by magic. A sort may be needed, and the optimizer can still have other constraints.
The builder creates the child first, passes the physical plan’s GroupByItems to a new VecGroupChecker, builds each AggFunc, and sets an optional default row for an empty input. That is the bridge from an optimizer decision to runnable code.
Watch the east group cross the boundary
At OpenSelf, StreamAggExec creates a reusable child chunk and allocates one partial result per aggregate function. An aggregate’s partial result is its running state: for SUM(amount), think of a running total rather than a completed output row.
When the parent calls Next, the executor resets the output chunk and repeats consumeOneGroup until it is full or there are no more groups. VecGroupChecker.SplitIntoGroups finds ranges of equal group keys within the current input chunk. It also remembers the previous chunk’s last key and reports whether the next chunk’s first group matches it (checker implementation).
For the example, the sequence is:
parent StreamAggExec VecGroupChecker child
|--- Next ------>| | |
| |--- pull chunk 1 ------------------------>|
| |<-- east 12, east 7 ----------------------|
| |--- split ----------->| east [0, 2) |
| | update SUM: 19; do not emit yet |
| |--- pull chunk 2 ------------------------>|
| |<-- east 4, west 5, west 8 ---------------|
| |--- split ----------->| first group east |
| | update SUM: 23; finalize east |
| | update west; finalize at end of input |
|<-- output chunk| [(east, 23), (west, 13)] |In the code, consumeCurGroupRowsAndFetchChild updates the pending group’s partial results before fetching the next child chunk. If the checker’s first group matches the previous chunk’s last group, consumeOneGroup includes those new rows too. Only after a different key appears can appendResult2Chunk append the final result and reset the partial state for the following group. Reaching end of input also flushes the last group.
There is a useful empty-input distinction: GROUP BY region returns no rows for an empty table. A final aggregate without group keys, such as SELECT COUNT(*) FROM sales, still needs one row (0). The builder’s DefaultVal condition and the executor’s end-of-input handling implement the distinction for the appropriate final aggregation plan.
Settings worth knowing before changing a query
STREAM_AGG()is a documented optimizer hint. Ordered input makes incremental groups possible, but forcing a sort can offset the memory advantage. CompareEXPLAIN ANALYZEplans rather than assuming the hint is faster.tidb_enable_vectorized_expressionis passed intoVecGroupCheckerby the builder to control eligible vectorized group-key evaluation; it does not create sorted input (builder, system variable).tidb_track_aggregate_memory_usagegoverns whether the stream executor attaches its memory tracker to the statement tracker. It does not mean no memory is used when tracking is off (executor, system variable).tidb_max_chunk_sizebounds the number of rows a chunk can hold. An input chunk ending mid-group is normal, not an error (session setting).
The executor keeps state for the current group rather than a hash entry for every group. That often saves memory, but large aggregate states, chunk buffers, and any preceding sort still cost memory. Finally, a stream aggregation’s input order is an implementation requirement, not a SQL promise that the result is ordered: add ORDER BY region if callers need that guarantee. In EXPLAIN, distinguish a StreamAgg under cop[tikv] from one at root; the Go executor above describes the latter, not TiKV’s separate coprocessor implementation (aggregation explain guide).