matrixorigin / matrixorigin/matrixone
[Enhancement]: Parallelize grouped-aggregation spill partition finalization
- Dominant language
- Go
- Stars
- 1.9k
- Forks
- 311
- Avg merge
- 1d 3h
- Merged PRs (30d)
- 768
Description
## Motivation
Grouped aggregation spill currently restores work through an operator-local bucket queue. One call selects one spill bucket, reads its records, reconstructs aggregate state, rebuilds a hash table, and may recursively repartition before another bucket can make progress.
This preserves correctness, but it leaves partition-level parallelism unused. A large or skewed owner can become the wall-clock bottleneck even when the query DOP is high and independent spill partitions are available.
This issue covers generic grouped aggregation. Exact DISTINCT state partitioning has additional ownership requirements tracked separately in #27698 and #27720.
## Current behavior
The current grouped-aggregation path:
1. builds a resident grouping hash table;
2. partitions resident groups into spill files after crossing the memory threshold;
3. queues non-empty spill buckets;
4. reloads one bucket in `loadSpilledData`;
5. deserializes group values and aggregate states;
6. rebuilds the grouping hash table and merges every record;
7. recursively re-spills a bucket when it still exceeds capacity;
8. only then advances to another queued bucket.
Independent buckets are therefore finalized serially inside one operator owner. Repeated serialization, hashing, reconstruction, and recursive reload amplify CPU and I/O cost.
## Required invariant
Once spill partitions have disjoint hash ownership, independent partitions may finalize concurrently, but:
- every logical group must be finalized exactly once;
- no partition may be visible to two finalization owners;
- aggregate merge order must preserve each aggregate function's declared semantics;
- total resident state, recovery reserve, open files, and spill bytes must remain bounded at query scope rather than multiplied by DOP;
- cancellation or failure must stop admission, unblock waiters, and release every partition, file, reservation, and worker;
- `Reset` and reuse must start a new generation with no state from the prior execution.
## Proposed execution model
1. Represent each sealed spill partition as an explicitly owned finalization task.
2. Schedule tasks through existing bounded pipeline/query scheduling primitives; do not create one goroutine per partition.
3. Derive finalization concurrency from:
- query DOP;
- available execution-memory reservation;
- estimated partition resident size;
- spill FD and disk reservations.
4. Finalize independent partitions concurrently.
5. Emit output through a bounded merge path with backpressure.
6. Release a partition's hash table, decoded state, read buffer, and file as soon as its output has been transferred.
7. Preserve partitioned blocks and precomputed hashes across spill/reload where the current wire and ownership contracts permit it, avoiding needless encode/decode/rehash cycles.
8. Repartition only the oversized partition; do not stop unrelated partitions from progressing.
A new scheduler or framework is not the default. Reuse existing query/pipeline task ownership unless it cannot express bounded independently owned partition work.
## State and ownership model
The design document must define:
- owner of queued, running, completed, failed, and canceled partitions;
- atomic state transitions and the single terminal cleanup owner;
- query-level memory and FD admission before a task starts;
- output backpressure and what happens when the consumer stops early;
- error propagation when one task fails while others are running or blocked;
- cancellation during read, decode, hash rebuild, merge, output, or recursive repartition;
- cleanup of partially created child partitions;
- `Reset`, `Free`, and repeated operator reuse generations;
- observability without per-key or unbounded partition-label metrics.
## Performance requirements
- Parallel execution must reduce wall time when at least two independent partitions fit the shared memory budget.
- Concurrency must automatically fall to one when only one partition can fit.
- Skewed partitions must not prevent smaller independent partitions from completing.
- No-spill and small-spill controls must not pay material scheduler, allocation, or synchronization overhead.
- Parallel reload must not increase total spill bytes merely to expose concurrency.
- CPU, read/write bytes, hash rebuilds, allocations, peak accounted memory, active tasks, and wall time must be measured.
## Validation matrix
| Dimension | Required cases |
|---|---|
| Partitions | empty, one, many, one oversized |
| Distribution | uniform and one hot partition |
| Aggregate | COUNT, SUM, variable-state aggregate, multiple aggregates |
| Keys | fixed width, VARCHAR, composite, NULL-containing |
| Memory | enough for 1, 2, and many concurrent finalizers |
| Repartition | none, one recursive level, bounded no-progress error |
| I/O | short read, corrupt/truncated record, injected read/write/close failure |
| Lifecycle | success, consumer early stop, error, cancellation at every phase, Reset/Free, reuse |
| Execution | DOP=1 and multi-DOP; one CN and multi-CN |
Unit tests must use small deterministic partitions and injected thresholds/failures. Do not add sleeps, large-data unit tests, or timing-only assertions. Scaling measurements belong in a benchmark/integration job.
## Acceptance criteria
1. At least two independent spill partitions finalize concurrently when admitted by the shared resource budget.
2. Results match an independent no-spill oracle across the validation matrix.
3. Peak accounted memory and recovery reserve remain within the documented query bound at every concurrency level.
4. Cancellation and all injected failures leave no files, goroutines, blocked channels, allocation debt, or reusable-state contamination.
5. DOP=1, no-spill, and low-cardinality controls do not regress materially.
6. A high-cardinality integration benchmark reports a meaningful wall-time reduction together with CPU, spill I/O, allocations, and peak memory.
7. EXPLAIN/observability exposes bounded counts for queued/running/finalized/repartitioned partitions and effective finalization concurrency.
## Related
- #27685
- #27728
- #27698
- #27720
- #20560
Contributor guide
Assessment
This issue has not been assessed yet.