matrixorigin / matrixorigin/matrixone
[Enhancement]: Reuse canonical grouping hashes across shuffle and aggregation
- Dominant language
- Go
- Stars
- 1.9k
- Forks
- 311
- Avg merge
- 1d 3h
- Merged PRs (30d)
- 768
Description
## Motivation
Grouped aggregation can currently hash the same variable-width grouping bytes at more than one execution boundary:
1. `Shuffle` hashes a selected grouping key to choose an owner;
2. the receiving `Group` operator hashes the grouping identity again for hash-table lookup;
3. spill/repartition paths may hash reconstructed keys again.
#27744 replaces sampled VARCHAR ownership with a deterministic complete-key hash and fixes pathological owner skew. Its controlled 10M-row validation also exposes the remaining cost boundary: for an already healthy, non-spilling 32-byte VARCHAR shuffle, complete-key ownership changed the median from 0.549s to 0.589s (+7.3%), even though it reduced owner skew from 2.785x to 1.0024x. The hash microbenchmark itself is fast; repeated key-byte reads across operator boundaries are now the more general optimization opportunity.
This issue tracks a cost-aware contract for computing a canonical grouping hash once where profitable and reusing it across ownership, exchange, grouped hash-table admission, and compatible repartition/spill stages. It must not assume that carrying an extra hash is always cheaper than recomputation.
## Required invariant
For every eligible row, the carried hash must be derived from exactly the equality identity used by the consuming grouped hash table. It is only a lookup accelerator: hash collisions must still be resolved by canonical key equality.
Reuse must preserve:
- equal grouping identities reaching the same owner during one execution;
- SQL equality, NULL, collation, floating-point, decimal, constant, and grouping-set semantics;
- exact results under deliberate hash collisions;
- mixed-version execution and rollback;
- bounded batch/message memory and normal backpressure;
- clean ownership across success, error, cancellation, `Reset`, `Free`, and operator reuse.
A shuffle hash derived from only a subset of grouping columns cannot silently be reused as the full grouped-hash-table hash. The planner/executor must either compute the canonical full active grouping identity, prove a compatible composition, or retain the existing recomputation path.
## Design questions that must be resolved first
The design document must compare, rather than assume, at least these choices:
1. **Receiver recomputation:** current baseline; no additional exchange bytes.
2. **Batch sidecar hash vector:** carry a non-SQL-visible fixed-width hash with explicit batch ownership and serialization rules.
3. **Ordinary hidden vector:** reuse existing vector transport at the cost of schema/payload width and copy behavior.
4. **Local-only reuse:** reuse inside one process/pipeline while remote receivers recompute.
5. **Encoded/dictionary reuse:** cache hashes only where repeated encoded values already have a stable dictionary lifetime.
The selected contract must define:
- the canonical hash identity for single, composite, constant, nullable, and grouping-set keys;
- whether owner selection and grouped hash-table lookup use the same hash bits or compatible derivations;
- producer, batch/message, exchange, receiver, hash-table, and spill ownership;
- cloning, slicing, shrinking, unioning, marshaling, remote transport, and release behavior;
- protocol/capability negotiation for rolling upgrade and rollback;
- behavior when any intervening operator changes grouping expressions or row order;
- whether spill files persist the hash, and how stale/incompatible persisted hashes are rejected;
- bounded observability without per-key labels or hot-path logging.
Do not introduce a general batch-metadata framework unless multiple independent consumers establish a stable shared contract and the framework reduces total complexity.
## Cost model and adaptive boundary
Carrying a 64-bit hash adds up to 8 bytes per row before encoding and may increase allocation, copy, cache, network, and spill traffic. Recomputing instead costs another pass over the key bytes. Therefore the implementation must select reuse only inside a measured benefit envelope.
The cost model must consider:
- total canonical key bytes and variable-width dereferences;
- input rows, duplicate ratio, and expected surviving local groups;
- local versus remote exchange and number of receivers;
- row width and the proportional 8-byte payload increase;
- transport compression/encoding and memory-copy behavior;
- hardware/software hash cost;
- whether a compatible hash already exists for another required operation;
- spill probability and whether the hash can avoid reload/repartition work.
Short/fixed-width keys and narrow remote rows are mandatory negative controls. Falling back to receiver recomputation is expected whenever carrying the hash is not cheaper.
## Scope
In scope:
- generic grouped aggregation where canonical hash compatibility can be proven;
- local and remote shuffle/exchange boundaries;
- reuse by grouped hash-table admission;
- compatible reuse during generic grouped spill/repartition;
- planner/runtime selection between carry and recompute.
Out of scope:
- changing SQL semantics or accepting approximate hashes as equality;
- query-, schema-, benchmark-, data-shape-, or issue-specific branches;
- DISTINCT-key partitioning itself (#27720 and #27698);
- spill-finalization parallel scheduling (#27729);
- Aggregate-to-TopK fusion (#27730).
Those features may consume the resulting contract only where their canonical identity and ownership rules match.
## Validation matrix
| Dimension | Required cases |
|---|---|
| Key width | empty, 1-8B, 16B, 32B, 128B, 4KiB |
| Key type | fixed width, VARCHAR, binary, UTF-8, composite, constant |
| Semantics | duplicate keys, NULL, collation controls, signed zero/NaN as supported, deliberate collisions |
| Group shape | shuffle key equals full group identity; shuffle key is a strict subset; grouping sets/inactive keys |
| Cardinality | low NDV, medium NDV, nearly unique |
| Topology | DOP=1/multi-DOP; local exchange; one/multi-CN remote exchange |
| Compatibility | old producer/new consumer, new producer/old consumer, upgrade and rollback gates |
| Memory | no spill, injected spill, repartition/reload |
| Lifecycle | success, error, cancellation, early stop, `Reset`/`Free`, repeated reuse |
Correctness tests must compare results with forced recomputation and use an independent equality oracle. Unit tests must use small deterministic inputs and injected boundaries; do not add sleeps, large-data unit tests, or timing-only assertions.
## Performance requirements
- No per-row allocation, goroutine, lock, logging event, or metric label.
- The reuse path must remove the redundant key-byte hash pass at each claimed consumer; profiles/counters must prove this rather than infer it from wall time.
- Benchmarks must report CPU, allocations, copied and exchanged bytes, peak accounted memory, spill bytes, cache misses where available, and wall time.
- Results must compare forced recomputation, forced carry, and adaptive selection on the same binary and data.
- The existing complete-key distribution quality and mixed-version correctness from #27744 must not regress.
- Cases outside the measured benefit envelope must retain the cheaper existing path.
## Acceptance criteria
1. The design document defines canonical identity, ownership, lifecycle, compatibility, and the cost-based eligibility rule before production implementation.
2. Eligible grouped aggregation computes the canonical hash once and every claimed downstream stage consumes that value without rereading the full key solely to hash it.
3. Ineligible subset-key, semantic, protocol, or cost cases deterministically fall back to recomputation.
4. Results match forced recomputation across the validation matrix, including deliberate collisions and mixed versions.
5. The 10M-row VARCHAR control from #27728/#27685 demonstrates whether reuse removes the measured healthy-path overhead without increasing spill or owner skew; both favorable and unfavorable controls are reported.
6. Added hash payload, vector capacity, exchange bytes, and spill bytes are fully memory-accounted and released on every lifecycle path.
7. No short-key, fixed-width, low-NDV, narrow-row remote, or fallback control regresses materially under adaptive selection.
## Related
- #27728 and #27744 — complete-key ownership and the measured hash-cost boundary
- #27685 — high-cardinality grouped VARCHAR reproduction
- #27720 — DISTINCT canonical-key partitioning, with its own identity contract
- #27698 — bounded DISTINCT spill
- #27729 — grouped spill-partition finalization
- #27730 — Aggregate-to-TopK finalization fusion
- #20560 — broader shuffle execution-plan work
Contributor guide
Assessment
This issue has not been assessed yet.