Chapter 18.3 · Query Planning and Execution · 31 min read
Sort and Aggregate Nodes in the Cloudberry Executor
Tuplesort and its spill path, hash and group aggregation, the two-stage distributed aggregate, and where work_mem and statement_mem bite.
By Tushar Pednekar · · Verified against the Apache Cloudberry source tree, read September 2026
Spill files are the number one cause of a slow MPP query. This chapter shows exactly which node writes them and why.
Every other node in this chapter can, in principle, work one tuple at a time. These cannot: a sort must see the last row before it may emit the first, and a hash aggregate must keep one live transition state for every distinct key it has met. That is why §18.3 owns memory — work_mem, Cloudberry’s statement_mem, spilling to workfiles, and the EXPLAIN ANALYZE footer that tells you how close you came to the edge. It is also where Cloudberry’s numbers stop looking like PostgreSQL’s: one Sort node on a three-segment cluster is three independent sorts, and the Memory: figure beside Sort Method: is their sum, not any one of them. Getting that wrong by a factor of three is the most common misreading of a Cloudberry plan, so we pin it down before doing any arithmetic. Demos below use a table m183_t of one million rows distributed by id — about 334 000 rows per segment.
18.3.1 Sort: a thin node over tuplesort
nodeSort.c is 647 lines and hardly any of them sort anything. ExecSort (nodeSort.c:59) does three things on its first call: build a Tuplesortstate, drain the child in a tight loop (nodeSort.c:177-185), and hand the whole batch to tuplesort_performsort (:192). Every later call is a one-line pull from the finished sort (:238). All the interesting behaviour lives in src/backend/utils/sort/tuplesort.c (3306 lines), with the per-datatype glue split out into tuplesortvariants.c (1606 lines) since PostgreSQL 15. Inside, tuplesort is a small state machine: tuples accumulate in an array of SortTuple structs while a budget counter, availMem, is decremented by the true allocated size of each tuple. When the budget goes negative — the LACKMEM macro at tuplesort.c:404 — the array is sorted and dumped as a run onto a logical tape. Which state the sort ends in is exactly what Sort Method: reports, and the mapping is made at tuplesort.c:2630-2645.
Sort Method: stringthree of the four sort methods, from one query shape
SET optimizer = off; -- PostgreSQL planner, so the plan shape is stable
-- (a) a LIMIT makes the sort bounded: keep a 10-element heap, discard the rest
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF)
SELECT id, amt FROM m183_t ORDER BY amt DESC, id LIMIT 10;
-- (b) 66 820 rows per segment, and they all fit
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF)
SELECT id FROM m183_t WHERE g < 200 ORDER BY txt;
-- (c) identical query, but the whole statement gets 2 MB
SET statement_mem = '2MB';
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF)
SELECT id FROM m183_t WHERE g < 200 ORDER BY txt;
-- (a)
Limit (actual rows=10 loops=1)
-> Gather Motion 3:1 (slice1; segments: 3) (actual rows=10 loops=1)
Merge Key: amt, id
-> Limit (actual rows=10 loops=1)
-> Sort (actual rows=10 loops=1)
Sort Key: amt DESC, id
Sort Method: top-N heapsort Memory: 78kB
-> Seq Scan on m183_t (actual rows=334042 loops=1)
-- (b)
Gather Motion 3:1 (slice1; segments: 3) (actual rows=200000 loops=1)
Merge Key: txt
-> Sort (actual rows=66820 loops=1)
Sort Key: txt
Sort Method: quicksort Memory: 18593kB
-> Seq Scan on m183_t (actual rows=66820 loops=1)
Filter: (g < 200)
Rows Removed by Filter: 266347
Execution Time: 368.707 ms
-- (c)
-> Sort (actual rows=66820 loops=1)
Sort Key: txt
Sort Method: external merge Disk: 4992kB
-> Seq Scan on m183_t (actual rows=66820 loops=1)
Memory used: 2048kB
Execution Time: 449.423 ms
The fourth string, external sort, is the
TSS_SORTEDONTAPEcase: the sort spilled, but by the timeperformsortfinished all runs had been merged onto a single tape, so reading back is a sequential scan of one tape rather than an N-way merge. It needs a random-access sort (a scrollable cursor, say) and we could not produce one on this cluster. Note also that (b) → (c) costs only 449/369 ≈ 1.2× in wall clock: spilling is not free, but on a warm cache it is nothing like the catastrophe folklore suggests. This is an—enable-cassertbuild, so treat every timing here as a ratio on one machine, never as a benchmark.
18.3.2 Whose kilobytes are those — and where do they go?
show_sort_info (explain.c:3715) does not print a per-segment number. It walks the CdbExplain_NodeSummary the QD collected from every QE and prints agg->vsum, the total across all segments that ran the node (explain.c:3758). The per-segment figures exist, but only under VERBOSE, on a curiously unindented continuation line.
Memory: and Disk: are cluster sums; VERBOSE breaks them out
SET optimizer = off;
EXPLAIN (ANALYZE, VERBOSE, TIMING OFF, COSTS OFF)
SELECT id FROM m183_t WHERE g < 200 ORDER BY txt;
SET statement_mem = '2MB'; -- and again, spilling
EXPLAIN (ANALYZE, VERBOSE, TIMING OFF, COSTS OFF)
SELECT id FROM m183_t WHERE g < 200 ORDER BY txt;
-> Sort (actual rows=66820 loops=1)
Sort Key: m183_t.txt
Sort Method: quicksort Memory: 18593kB
Max Memory: 6205kB Avg Memory: 6197kB (3 segments)
-> Seq Scan on public.m183_t (actual rows=66820 loops=1)
-- the disk figure aggregates the same way
Sort Method: external merge Disk: 5184kB
Max Memory: 1728kB Avg Memory: 1728kB (3 segments)
6197 × 3 = 18 591 ≈ 18 593; 1728 × 3 = 5184 exactly. The rule for sort instrumentation is therefore: Memory: and Disk: are summed over segments — divide by the segment count before comparing to any memory budget. This is not the convention actual rows uses; §18.1 settles that one, and the two are genuinely different aggregations computed side by side in explain_gp.c. With a per-segment number in hand we can audit it, because a sort’s footprint has exactly two parts. The first is the memtuples array: one 24-byte SortTuple (tuplesort.h:142) per row, holding a pointer plus datum1 — the first sort key pre-extracted, or, when the type offers one, an abbreviated key, a pass-by-value proxy that answers most comparisons without touching the tuple at all (tuplesort.h:129). The array grows by doubling from 1024 slots (grow_memtuples, tuplesort.c:1078), so its real size is the next power of two above the row count. The second part is the tuples themselves, copied as MinimalTuples into a dedicated child context and charged at their true allocated size via GetMemoryChunkSpace (tuplesort.c:1203).
Both halves are predictable to the byte. For the (id int, txt text) projection above, txt runs 5 to 10 characters, so each MinimalTuple is MAXALIGN(15) = 16 bytes of header plus a 4-byte id plus a 1+N short varlena — 26 to 31 bytes — which the tuple context rounds to a 32-byte chunk and prefixes with a 16-byte MemoryChunk header: 48 bytes per row, flat. Multiply out, add the doubled array, and compare with what the cluster actually reported. Note the last two rows: 11 % more rows buys 43 % more memory, because the array crossed 65 536 slots and doubled. Sort memory is a staircase, not a line, and the riser is 24 bytes times the old array length. The 1 kB overshoots are tuplesort_get_stats rounding up — spaceUsed = (maxSpace + 1023) / 1024 at tuplesort.c:2625.
| rows / segment | memtuples slots | array kB | tuples kB at 48 B each | predicted kB | measured Max Memory |
|---|---|---|---|---|---|
| 33 407 | 65 536 | 1536 | 1566 | 3102 | 3102kB |
| 50 069 | 65 536 | 1536 | 2347 | 3883 | 3884kB |
| 60 087 | 65 536 | 1536 | 2817 | 4353 | 4353kB |
| 66 820 | 131 072 | 3072 | 3132 | 6204 | 6205kB |
18.3.3 A claim worth testing: the PostgreSQL 16 tuplesort rewrite
PostgreSQL 15 and 16 reworked tuplesort’s memory management, and the change is usually quoted as a large wall-clock win; a previous investigation on this very cluster concluded the reproducible effect is instead roughly 2.3–2.5× less sort memory with wall clock largely unchanged on MPP. We are now equipped to check the mechanism rather than repeat either number. The mechanism is one if. Unbounded sorts allocate caller tuples from a generation context, which hands out chunks at MAXALIGN(size) with no size classes and never frees individually; bounded sorts keep the old aset.c context, because a top-N heap pfrees tuples in arbitrary order and needs a free list.
src/backend/utils/sort/tuplesort.c:783
if (state->base.sortopt & TUPLESORT_ALLOWBOUNDED)
state->base.tuplecontext = AllocSetContextCreate(state->base.sortcontext,
"Caller tuples",
ALLOCSET_DEFAULT_SIZES);
else
state->base.tuplecontext = GenerationContextCreate(state->base.sortcontext,
"Caller tuples",
ALLOCSET_DEFAULT_SIZES);
That if is also an experiment we can run on this build, because adding a LIMIT flips the allocator without changing the algorithm — as long as the bound is large enough that the sort never switches to a heap (ExecSetTupleBound, execProcnode.c:1443, only sets bounded; the switch happens later, at twice the bound). aset.c rounds a request up to a power-of-two size class; generation rounds it to MAXALIGN(size + 1). So the saving depends entirely on where the tuple width falls relative to a class boundary — which makes it predictable. A 33-byte MinimalTuple costs 16 + 64 = 80 bytes under aset.c but 16 + 40 = 56 under generation; a 32-byte one costs 16 + 32 = 48 under aset.c against 16 + 40 = 56 under generation, i.e. generation is worse. Here are both, one column wider than the other by a single character:
33 462 rows per segment, same quicksort, allocator flipped by the LIMIT
-- m183_s.txt is 11 chars (t_len 32); m183_s.txt12 is 12 chars (t_len 33)
SET optimizer = off;
EXPLAIN (ANALYZE, VERBOSE, TIMING OFF, COSTS OFF) SELECT id FROM m183_s ORDER BY txt;
EXPLAIN (ANALYZE, VERBOSE, TIMING OFF, COSTS OFF) SELECT id FROM m183_s ORDER BY txt LIMIT 30000;
EXPLAIN (ANALYZE, VERBOSE, TIMING OFF, COSTS OFF) SELECT id FROM m183_s ORDER BY txt12;
EXPLAIN (ANALYZE, VERBOSE, TIMING OFF, COSTS OFF) SELECT id FROM m183_s ORDER BY txt12 LIMIT 30000;
-- t_len 32, unbounded -> GenerationContext, MAXALIGN(33) = 40 => 56 B/row
Sort Method: quicksort Memory: 10078kB
Max Memory: 3366kB Avg Memory: 3359kB (3 segments)
-- t_len 32, bounded -> AllocSetContext, size class = 32 => 48 B/row
Sort Method: quicksort Memory: 9297kB
Max Memory: 3105kB Avg Memory: 3099kB (3 segments)
work_mem: 3125kB Segments: 3 Max: 1046kB (segment 0) Workfile: (0 spilling)
-- t_len 33, unbounded -> GenerationContext, MAXALIGN(34) = 40 => 56 B/row
Sort Method: quicksort Memory: 10078kB
Max Memory: 3366kB Avg Memory: 3359kB (3 segments)
-- t_len 33, bounded -> AllocSetContext, size class = 64 => 80 B/row
Sort Method: quicksort Memory: 12422kB
Max Memory: 4151kB Avg Memory: 4140kB (3 segments)
Predicted for t_len 33: 1536 kB of array plus 33 462 × 56 = 1830 kB, i.e. 3366 kB with the generation context, against 1536 + 33 462 × 80 = 4150 kB with aset.c. Measured: 3366 kB and 4151 kB. The tuple-storage component improves by exactly 80/56 = 1.43×, the whole footprint by 1.23×. One byte narrower, at t_len 32, the generation context is 8 bytes per row worse (3366 against 3105 kB); and for the m183_t shape of §18.3.2 both allocators land on the same 32-byte chunk and the change buys nothing at all. So the honest verdict: the mechanism is real and measurable to the byte, but “2.3–2.5×” is not a property of PostgreSQL 16 — it is a property of a tuple width. The win is bounded by the gap between a power-of-two size class and the next 8-byte boundary, capping near 1.6× on the tuple component for a single-chunk tuple and less on the total. As for wall clock, where we could compare like with like — the same query spilling and not spilling in §18.3.1 — the difference was 1.2×. We could not reproduce a large wall-clock effect and would not expect to: here a sort is one of three concurrent sorts behind a Gather Motion, and the interconnect is happy to hide it.
Two caveats, both from this build. First,
—enable-cassertimpliesMEMORY_CONTEXT_CHECKING(pg_config_manual.h:283), which growsMemoryChunkfrom 8 to 16 bytes and makesgeneration.c:349reserve a sentinel byte viaMAXALIGN(size + 1). On a release build the same 33-byte tuple costs 8 + 40 = 48 rather than 56, and the generation advantage widens — so the direction of our finding is conservative, not optimistic. Second, thework_mem:line appears only for the bounded sorts becausetuplesort_get_statsfillsstats->workmemusedonly whenTUPLESORT_ALLOWBOUNDEDis set (tuplesort.c:2626). That is an instrumentation gap, not a difference in behaviour.
18.3.4 Spilling: it is statement_mem, not work_mem
Notice what we changed to force a spill: statement_mem, not work_mem. Setting work_mem = '1MB' on this cluster does nothing to a Sort. Under resource queues (gp_resource_manager = queue, gp_resqueue_memory_policy = eager_free here) the dispatched plan arrives with a per-node budget already stamped on it — Plan.operatorMemKB, carved out of statement_mem by the quota policy at memquota.c:341 — and the executor reads that, not the GUC. nodeSort.c:130 and :140 pass it straight into tuplesort_begin_* with no reference to work_mem at all.
src/backend/executor/execUtils.c:2314
uint64 PlanStateOperatorMemKB(const PlanState *ps)
{
uint64 result = 0;
if (ps->plan->operatorMemKB == 0)
/* statements that do not go through the resource queue */
result = work_mem;
else
result = ps->plan->operatorMemKB;
return result;
}
sweeping statement_mem to find the spill threshold for 66 820 rows per segment
SET optimizer = off;
-- for each M in 2MB 4MB 6MB 8MB 10MB 16MB:
SET statement_mem = :M;
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF)
SELECT id FROM m183_t WHERE g < 200 ORDER BY txt;
statement_mem=2MB Sort Method: external merge Disk: 4992kB 449 ms
statement_mem=4MB Sort Method: external merge Disk: 4992kB 380 ms
statement_mem=6MB Sort Method: quicksort Memory: 15420kB 408 ms
statement_mem=8MB Sort Method: quicksort Memory: 17469kB 344 ms
statement_mem=10MB Sort Method: quicksort Memory: 18593kB 379 ms
statement_mem=16MB Sort Method: quicksort Memory: 18593kB 364 ms
The threshold sits between 4 and 6 MB, and the 6 MB row is the one to stare at: Memory: 15420kB is a sum, 5140 kB per segment, comfortably inside a 6 MB per-statement budget. Read it as 15 MB and you would conclude the sort had blown three times its quota. Note too that 6 MB reports less memory than 10 MB for the identical sort — at 6 MB the budget runs out partway through, so the array stops doubling early and tuplesort trades slack for extra comparisons rather than spilling. While a sort is spilling the workfiles are visible cluster-wide: the four gp_toolkit views read a shared workfile manager on each segment, with gp_workfile_entries one row per open file set, gp_workfile_usage_per_query rolled up by session, and gp_workfile_usage_per_segment by segment. They are live, so you need a second session while the query runs, and which nodes may appear at all is decided by nodeSupportWorkfileCaching (explain_gp.c:1539): Sort, HashJoin, Material, and Agg — the last only when its strategy is AGG_HASHED.
session B watching session A’s external merge sort, sampled three times
-- session A: SET statement_mem='1MB';
-- SELECT count(*) FROM (SELECT g,k,amt,txt FROM m183_t ORDER BY txt, amt) x;
-- session B, repeatedly:
SELECT segid, slice, optype, size, prefix FROM gp_toolkit.gp_workfile_entries ORDER BY segid;
SELECT sess_id, segid, size, numfiles, substr(query,1,44) AS q
FROM gp_toolkit.gp_workfile_usage_per_query ORDER BY segid;
segid | slice | optype | size | prefix
-------+-------+---------+---------+-----------
0 | 1 | Logtape | 2392064 | Logtape_1
1 | 1 | Logtape | 2719744 | Logtape_1
2 | 1 | Logtape | 2555904 | Logtape_4
sess_id | segid | size | numfiles | q
---------+-------+---------+----------+----------------------------------------------
25 | 0 | 4390912 | 1 | SELECT count(*) FROM (SELECT g, k, amt, txt
25 | 1 | 4390912 | 1 | SELECT count(*) FROM (SELECT g, k, amt, txt
25 | 2 | 4390912 | 1 | SELECT count(*) FROM (SELECT g, k, amt, txt
-- ~1 s later -- ~2 s later
0 | 1 | Logtape | 5996544 | Logtape_1 0 | 1 | Logtape | 7372800 | Logtape_1
1 | 1 | Logtape | 5996544 | Logtape_1 1 | 1 | Logtape | 7372800 | Logtape_1
2 | 1 | Logtape | 5767168 | Logtape_4 2 | 1 | Logtape | 7372800 | Logtape_4
optypeisLogtapefor both sorts and hash aggregates, because both spill throughlogtape.c. Theslicecolumn tells you which executor is spilling — invaluable on a plan with several memory-hungry slices, since the footer reports only one number per slice.gp_workfile_mgr_used_diskspacegives the same totals without the per-query join.
18.3.5 Incremental sort
nodeIncrementalSort.c (1258 lines) exploits input already sorted on a prefix of the required keys: instead of one sort of everything, it sorts each group of equal prefix values separately, so memory is bounded by the largest group and rows can flow after the first group closes. Its state machine keeps two tuplesorts alive and hands work between them (switchToPresortedPrefixMode, nodeIncrementalSort.c:287), and show_incremental_sort_info (explain.c:3955) rolls the per-group statistics up into the Full-sort Groups: and Pre-sorted Groups: lines. It is off by default on this cluster (enable_incremental_sort = off), which is worth knowing before you go hunting for it. Below, forty prefix groups sorted in 28–41 kB apiece replace one monolithic 3341 kB sort (1114 kB per segment) — peak memory bounded by the largest group instead of by the whole input, and here twice as fast as well. Note that Peak Memory on those lines is genuinely a peak over groups, not a sum: a third aggregation convention on the same plan, which is the kind of inconsistency this chapter exists to warn you about.
an index supplies g, so only amt has to be sorted — per group
SET optimizer = off;
SET enable_incremental_sort = on;
SET enable_seqscan = off; SET enable_bitmapscan = off; -- force the ordered index path
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF)
SELECT g, amt, id FROM m183_t WHERE g < 40 ORDER BY g, amt;
Gather Motion 3:1 (slice1; segments: 3) (actual rows=40000 loops=1)
Merge Key: g, amt
-> Incremental Sort (actual rows=13358 loops=1)
Sort Key: g, amt
Presorted Key: g
Full-sort Groups: 40 Sort Method: quicksort Average Memory: 28kB Peak Memory: 28kB
Pre-sorted Groups: 40 Sort Method: quicksort Average Memory: 41kB Peak Memory: 41kB
-> Index Scan using m183_t_g_idx on m183_t (actual rows=13358 loops=1)
Index Cond: (g < 40)
Execution Time: 29.908 ms
-- the same rows through a plain Sort, for contrast:
Sort Method: quicksort Memory: 3341kB Execution Time: 61.950 ms
18.3.6 Four aggregation strategies
nodeAgg.c is 5249 lines, the second-largest file in the executor, and it implements one node with four strategies. ExecAgg (nodeAgg.c:2354) is a four-way switch that hides all of it, dispatching on node->phase->aggstrategy. Whichever branch runs, the same two things happen per group: advance_aggregates (nodeAgg.c:836) calls each aggregate’s transition function once per input row, keeping the running transValue in curaggcontext — hashcontext for hashed groups, aggcontexts[setno] otherwise (nodeAgg.c:481) — and finalize_aggregates (nodeAgg.c:1343) turns those states into output datums when the group closes.
AGG_PLAIN — No GROUP BY: one AggStatePerGroup for the whole input, one output row. Shown as Aggregate.
AGG_SORTED — Input arrives sorted on the grouping keys; detect a key change, finalize, reset. Memory is O(1) in the group count — the cost was paid by the Sort beneath. Shown as GroupAggregate.
AGG_HASHED — One hash-table entry per distinct key, each carrying its transition states in hashcontext. Memory is O(groups), so this is the strategy that can spill. Shown as HashAggregate.
AGG_MIXED — Grouping sets where some sets are hashed and one is fed from sorted input in the same pass, so several hash tables live at once. Shown as MixedAggregate.
steered by enable_hashagg: the same query as HashAggregate and as GroupAggregate
SET optimizer = off;
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF) SELECT g, count(*), sum(amt) FROM m183_t GROUP BY g;
SET enable_hashagg = off;
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF) SELECT g, count(*), sum(amt) FROM m183_t GROUP BY g;
-- enable_hashagg = on
Gather Motion 3:1 (slice1; segments: 3) (actual rows=1000 loops=1)
-> Finalize HashAggregate (actual rows=340 loops=1)
Group Key: g
Extra Text: (seg0) hash table(s): 1; chain length 2.4 avg, 5 max; using 337 of 512 buckets; total 0 expansions.
-> Redistribute Motion 3:3 (slice2; segments: 3) (actual rows=1020 loops=1)
Hash Key: g
-> Streaming Partial HashAggregate (actual rows=1000 loops=1)
Group Key: g
Extra Text: (seg0) hash table(s): 1; chain length 2.5 avg, 6 max; using 1000 of 2048 buckets; total 0 expansions.
-> Seq Scan on m183_t (actual rows=334042 loops=1)
Execution Time: 161.249 ms
-- enable_hashagg = off
Finalize GroupAggregate (actual rows=1000 loops=1)
Group Key: g
-> Gather Motion 3:1 (slice1; segments: 3) (actual rows=3000 loops=1)
Merge Key: g
-> Partial GroupAggregate (actual rows=1000 loops=1)
Group Key: g
-> Sort (actual rows=334042 loops=1)
Sort Key: g
Sort Method: quicksort Memory: 83739kB
-> Seq Scan on m183_t (actual rows=334042 loops=1)
Execution Time: 397.948 ms
The trade is stark: the hashed plan holds 1000 group states in a 2048-bucket table and materialises nothing, while the sorted plan holds nothing per group but must first sort all 334 042 rows per segment, at 83 739 kB summed (27 913 kB each) and 2.5× the wall clock. nodeGroup.c (255 lines) implements a fifth, aggregate-free variant — the bare Group node — which this planner never chose in our tests; even SELECT DISTINCT k with enable_hashagg = off came out as a GroupAggregate. That Extra Text: line, meanwhile, is Cloudberry-only, assembled by agg_hash_explain_extra_message (nodeAgg.c:5181) from counters that hash_agg_update_metrics (nodeAgg.c:2018) refreshes as the table fills. The QD keeps the message from one representative segment and labels it, which is why it says (seg0) while actual rows on the same line is a different aggregation over all three. Field by field:
| field | source | how to read it |
|---|---|---|
hash table(s): 1 | aggstate->num_hashes | one table per hashed grouping set; MixedAggregate shows more than one |
chain length 2.4 avg, 5 max | tuplehash_coll_stat | collision-chain length in the simplehash table; above roughly 3 average the table is crowded |
using 337 of 512 buckets | hashtab->members / hashtab->size | 337 distinct groups landed on seg0 in a 512-bucket table. Accumulated over rounds and batches, so it can exceed the true group count |
total 0 expansions | hashtab->num_expansions | how many times simplehash had to grow and rehash. Non-zero means the planner’s numGroups estimate was low |
N groups total in B batches, P spill partitions | hash_batches_used | printed only if the table ever spilled — see §18.3.7 |
disk usage: NKB | aggstate->hash_disk_used | per-segment workfile bytes for this node |
18.3.7 Hash-agg spill and hash_mem_multiplier
Before PostgreSQL 13 a hash aggregate that mis-estimated its group count simply overran work_mem. Now it partitions the input and spills, and the budget it spills against comes from hash_agg_set_limits (nodeAgg.c:1861). Cloudberry rewires the first lines: the limit is the smaller of work_mem and the plan’s operatorMemKB from §18.3.4, and only then is hash_mem_multiplier applied.
src/backend/executor/nodeAgg.c:1867
uint64 strict_memlimit = work_mem;
if (aggstate)
{
uint64 operator_mem = PlanStateOperatorMemKB((PlanState *) aggstate);
if (operator_mem < strict_memlimit)
strict_memlimit = operator_mem;
}
strict_memlimit = strict_memlimit * hash_mem_multiplier;
hash_mem_multiplier against a pinned work_mem = '1MB', 334 042 groups per segment
SET optimizer = off;
SET work_mem = '1MB'; -- small enough to win the min() against operatorMemKB
-- for each M in 1, 2, 4:
SET hash_mem_multiplier = :M;
EXPLAIN (ANALYZE, VERBOSE, TIMING OFF, COSTS OFF)
SELECT count(*) FROM (SELECT id, count(*) c FROM m183_t GROUP BY id) x;
-- hash_mem_multiplier = 1
work_mem: 3217kB Segments: 3 Max: 1073kB (segment 0) Workfile: (3 spilling)
Extra Text: (seg1) hash table(s): 1; 332791 groups total in 73 batches, 2606120 spill
partitions; disk usage: 13504KB; chain length 2.2 avg, 7 max; ...
-- hash_mem_multiplier = 2
work_mem: 6193kB Segments: 3 Max: 2065kB (segment 0) Workfile: (3 spilling)
Extra Text: (seg0) hash table(s): 1; 333167 groups total in 17 batches, 5090768 spill
partitions; disk usage: 12000KB; chain length 2.7 avg, 13 max; ...
-- hash_mem_multiplier = 4
work_mem: 12337kB Segments: 3 Max: 4113kB (segment 0) Workfile: (3 spilling)
Extra Text: (seg1) hash table(s): 1; 332791 groups total in 9 batches, 2280488 spill
partitions; disk usage: 13408KB; chain length 2.6 avg, 14 max; ...
1073, 2065, 4113 kB against a 1024 kB work_mem: 1×, 2×, 4× plus a fixed ~40 kB of table metadata, exactly as the source says — and the default on this cluster is 2, so hash nodes really do get twice what a sort node would. Batches fall 73 → 17 → 9: each doubling roughly halves the passes over spilled data, which is the whole point of the knob. Unlike statement_mem, hash_mem_multiplier costs nothing when the aggregate does not spill. Here is the same query under a 1 MB statement budget, with the footer telling you what it wanted:
the memory footer when an aggregate wants more than it got
SET optimizer = off;
SET statement_mem = '1MB';
EXPLAIN (ANALYZE, VERBOSE, TIMING OFF, COSTS OFF)
SELECT count(*) FROM (SELECT id, count(*) c FROM m183_t GROUP BY id) x;
-> HashAggregate (actual rows=334042 loops=1)
Group Key: m183_t.id
work_mem: 5905kB Segments: 3 Max: 1969kB (segment 0) Workfile: (3 spilling)
Work_mem wanted: 18747K bytes avg, 19345K bytes max (seg1)
to lessen workfile I/O affecting 3 workers.
Extra Text: (seg1) hash table(s): 1; 332791 groups total in 29 batches,
5150156 spill partitions; disk usage: 12000KB; chain length 2.7 avg, 11 max;
using 332791 of 933888 buckets; total 1 expansions.
-> Seq Scan on public.m183_t (actual rows=334042 loops=1)
* (slice1) Executor memory: 1676K bytes avg x 3x(0) workers, 1690K bytes max (seg2).
Work_mem: 1969K bytes max, 19345K bytes wanted.
Memory used: 1024kB
Memory wanted: 19744kB
The leading * on (slice1) is the fast signal that a slice wanted more memory than it was given, and the per-node work_mem: / Work_mem wanted: lines come from explain_gp.c:1611 and :1627. Where does 19345K bytes wanted come from? Not a model — hash_agg_update_metrics accumulates it while the thing is happening: instrument->workmemwanted += meta_mem on the first pass plus hashkey_mem on every pass (nodeAgg.c:2084-2088), i.e. the measured size the table would have reached had it been allowed to hold every group at once. It is the number to put into statement_mem if you want this query to stop spilling, and Memory wanted: 19744kB on the last line is its maximum over all slices — the one figure in the footer you can act on directly.
18.3.8 Two-stage aggregation, executed — and what makes it streaming
§16.4 explained how the planner splits an aggregate in two and §17.3 how ORCA does the same; the hashed plan in §18.3.6 is that split running. Each segment holds all 1000 values of g (the table is distributed by id), so a partial aggregate collapses 334 042 rows to 1000 partial group states locally. Those cross a Redistribute Motion hashed on g, and a finalize aggregate combines them: on the reporting segment 1020 partial rows arrive — 340 groups × 3 senders — and collapse to 340 final groups, with Extra Text independently confirming 337 groups on seg0. A three-hundred-fold reduction happens before the interconnect sees a byte; that, not the aggregation itself, is what the split buys. (§18.1 establishes how to read those actual rows counters across segments; here we rely only on their internal consistency.)
The first stage was labelled Streaming Partial HashAggregate, and that word is pure Cloudberry: explain.c:1879 prepends it whenever Agg->streaming is set, which the planner does from gp_use_streaming_hashagg (cdbgroupingpaths.c:1286, GUC at guc_gp.c:1904). A streaming hash aggregate refuses to spill. When its table fills it stops admitting new groups — p_isnew is passed as NULL once table_filled is set (nodeAgg.c:2305) — drains what it has downstream, then resets the tables and restarts on the rest of the input instead of calling agg_refill_hash_table (nodeAgg.c:2977-3008). So it only partly deduplicates: the same group may be emitted once per round. Since a partial aggregate’s output is re-aggregated downstream anyway that is perfectly correct, and the trade is disk I/O here against extra rows on the interconnect there. On a high-cardinality key with a 1 MB budget it goes exactly as the comment at nodeAgg.c:2351 promises:
the same two-stage plan with Streaming on and off, statement_mem = '1MB'
SET optimizer = off;
SET statement_mem = '1MB';
SET gp_eager_two_phase_agg = on; -- force the split on a key with ~1M distinct values
-- for each of on / off:
SET gp_use_streaming_hashagg = :S;
EXPLAIN (ANALYZE, VERBOSE, TIMING OFF, COSTS OFF)
SELECT count(*) FROM (SELECT txt, count(*) c FROM m183_t GROUP BY txt) y;
-- gp_use_streaming_hashagg = on
-> Redistribute Motion 3:3 (slice2; segments: 3) (actual rows=333742 loops=1)
-> Streaming Partial HashAggregate (actual rows=334042 loops=1)
work_mem: 2161kB Segments: 3 Max: 721kB (segment 0) Workfile: (0 spilling)
Extra Text: (seg2) hash table(s): 1; chain length 2.9 avg, 15 max;
using 334042 of 507904 buckets; total 0 expansions.
Execution Time: 964.481 ms
-- gp_use_streaming_hashagg = off
-> Redistribute Motion 3:3 (slice2; segments: 3) (actual rows=333742 loops=1)
-> Partial HashAggregate (actual rows=334042 loops=1)
work_mem: 3409kB Segments: 3 Max: 1201kB (segment 0) Workfile: (3 spilling)
Extra Text: (seg0) hash table(s): 1; 333167 groups total in 73 batches,
2621824 spill partitions; disk usage: 13792KB; chain length 2.3 avg, 13 max;
using 333167 of 1114112 buckets; total 1 expansions.
Execution Time: 1201.320 ms
Zero workfiles and 721 kB, against three spilling segments with 73 batches and 13.8 MB of disk each — same answer, 20 % less elapsed time. The reverse case is just as real: on a low-cardinality key we measured the streaming stage emitting 278 382 rows where the non-streaming stage emitted 9973, tripling the Motion’s payload for nothing, because nothing needed to spill in the first place. gp_use_streaming_hashagg defaults to on and is a sensible default for the first stage of a split aggregate, but it is a bet on the group count, not a free win.
18.3.9 Window functions
nodeWindowAgg.c (4014 lines) is the largest single-node file after nodeAgg.c, and the reason is frames. A window function may look at any row of its partition, so WindowAgg buffers the whole current partition in a tuplestore and hands the functions read pointers into it (nodeWindowAgg.c:16-18); WinFuncExprState carries a markptr/readptr pair per function so each can wander independently, update_frameheadpos / update_frametailpos track the frame edges, and WinSetMarkPosition lets the tuplestore discard rows that have fallen off the head. What it does not do is sort: WindowAgg requires its input already ordered by PARTITION BY then ORDER BY, which is why every window function sits directly on a Sort — and in MPP, on a Redistribute Motion beneath that, because a partition must be whole on one segment before it can be framed.
a real frame clause; both optimizers build the same three-node stack
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF)
SELECT k, id,
sum(amt) OVER (PARTITION BY k ORDER BY id
ROWS BETWEEN 2 PRECEDING AND CURRENT ROW)
FROM m183_t WHERE g < 5;
Gather Motion 3:1 (slice1; segments: 3) (actual rows=5000 loops=1)
-> WindowAgg (actual rows=3000 loops=1)
Partition By: k
Order By: id
-> Sort (actual rows=3000 loops=1)
Sort Key: k, id
Sort Method: quicksort Memory: 404kB
-> Redistribute Motion 3:3 (slice2; segments: 3) (actual rows=3000 loops=1)
Hash Key: k
-> Index Scan using m183_t_g_idx on m183_t (actual rows=1739 loops=1)
Index Cond: (g < 5)
Optimizer: GPORCA Execution Time: 9.978 ms
(identical shape with Optimizer: Postgres query optimizer, 42.433 ms)
There is an ORCA knob,
optimizer_force_window_hash_agg, mapped to theEopttraceEnableWindowHashAggtrace flag atCConfigParamMapping.cpp:330. Do not use it on this build. Turning it on makes ORCA emit aT_WindowHashAggplan node for whichExecInitNodehas no case —grep T_WindowHashAgg src/backend/executor/returns nothing — so evenEXPLAINdies withERROR: unrecognized node type: 63 (execProcnode.c:533). It is a declared developer option carrying the comment// TODO: remove it before mergeatguc_gp.c:3315, which is a fair summary of its status.
18.3.10 TupleSplit: several DISTINCTs in one pass
count(DISTINCT a) is easy in MPP: redistribute on a and every segment counts locally. count(DISTINCT a), count(DISTINCT b) is hard, because the two aggregates want incompatible redistributions. nodeTupleSplit.c (281 lines, Cloudberry-only) resolves it by turning one input row into several: for each of the N distinct-qualified aggregates it emits a copy with every column irrelevant to that aggregate set to NULL, tagged by a synthetic AggExprId carrying the aggregate’s index. Group by (AggExprId, a, b) and a single hash pass deduplicates all N distinct sets at once, with AggExprId in the redistribution key keeping the copies from colliding.
src/backend/executor/nodeTupleSplit.c:228
for (AttrNumber attno = 1; attno <= node->outerslot->tts_nvalid; attno++)
{
/* If the column is relevant to the current dqa, keep it */
if (bms_is_member(attno, node->dqa_split_bms[node->currentExprId]))
continue;
/* otherwise, null this column out */
isnull[attno - 1] = true;
}
result = ExecProject(node->ss.ps.ps_ProjInfo);
/* the next DQA to process */
node->currentExprId = (node->currentExprId + 1) % node->numDisDQAs;
The outer slot is fetched only when currentExprId wraps back to 0 (nodeTupleSplit.c:183), so the child is scanned once and each tuple is replayed N times from the same slot — the multiplication costs no extra I/O, and a per-DQA FILTER clause pushed down here can skip a copy outright (nodeTupleSplit.c:201-214). The node comes only from the PostgreSQL-planner path; ORCA refuses the query unless optimizer_enable_multiple_distinct_aggs (guc_gp.c:2523) is on, raising its fallback at CTranslatorQueryToDXL.cpp:2271. The working recipe is the one in src/test/regress/sql/gp_dqa.sql:
two DISTINCTs: ORCA falls back, the planner builds Streaming HashAggregate over TupleSplit
SET optimizer = off;
SET enable_hashagg = on; SET enable_groupagg = off;
SET gp_motion_cost_per_row = 2; -- as in src/test/regress/sql/gp_dqa.sql
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF)
SELECT count(distinct g), count(distinct k) FROM m183_t;
RESET ALL; SET optimizer_trace_fallback = on; -- now let ORCA try
EXPLAIN (COSTS OFF) SELECT count(distinct g), count(distinct k) FROM m183_t;
Finalize Aggregate (actual rows=1 loops=1)
-> Gather Motion 3:1 (slice1; segments: 3) (actual rows=3 loops=1)
-> Partial Aggregate (actual rows=1 loops=1)
-> Redistribute Motion 3:3 (slice2; segments: 3) (actual rows=1083 loops=1)
Hash Key: g, k, (AggExprId)
-> Streaming HashAggregate (actual rows=1050 loops=1)
Group Key: AggExprId, g, k
Extra Text: (seg0) hash table(s): 1; chain length 2.4 avg, 6 max;
using 1050 of 2048 buckets; total 0 expansions.
-> TupleSplit (actual rows=668084 loops=1)
Split by Col: (g), (k)
-> Seq Scan on m183_t (actual rows=334042 loops=1)
Optimizer: Postgres query optimizer Execution Time: 247.061 ms
-- under ORCA:
INFO: GPORCA failed to produce a plan, falling back to Postgres-based planner
DETAIL: Falling back to Postgres-based planner because GPORCA does not support the
following feature: Multiple Distinct Qualified Aggregates are disabled in the optimizer
Split by Col: (g), (k) names the two distinct sets, and actual rows = 668084 is precisely 2 × 334 042 — the multiplication, visible. Downstream, 1050 groups per segment cover both sets together (1000 values of g plus 50 of k), so the interconnect carries about a thousand rows rather than a million, twice. That is the trick: pay double on a local scan to avoid a second redistribution of the whole table. It completes the memory-hungry half of the executor — §18.4 takes up the other family of hash tables, the ones inside nodeHash.c, with their own batching, their own Extra Text and, in Cloudberry, their own deadlock-avoidance problem; §18.5 goes a level down to how the transition functions and frame conditions in this section are actually evaluated. The Motion nodes that carried every partial aggregate here are Chapter 19, and the cluster-wide limits that decide what statement_mem you are allowed in the first place are Chapter 21.
More in Query Planning and Execution
- From SQL to Segments: Cloudberry Query Processing
- Statistics and Selectivity Estimation in Cloudberry
- The Postgres Planner in Cloudberry, Step by Step
- Distributed Planning: Locus, Motion and Direct Dispatch
- ORCA Optimizer: Entry Points and Fallback Rules
- Inside ORCA's Cascades Engine: Groups and Jobs