matrixorigin / matrixorigin/matrixone

[Performance]: Add cost-based hash partition execution for ordinary window functions

Open
#27,943 3 comments 0 reactions 1 assignee Assigned to @iamlinjunhong View on GitHub
ai-light kind/performance phase/testing
Dominant language
Go
Stars
1.9k
Forks
311
Avg merge
1d 3h
Merged PRs (30d)
768

Description

### Is there an existing issue for performance?

- [x] I have checked the existing issues.

### Environment

- Version or commit-id: 89740fae15dbba2cbe104348659694283cb160a6 (main, 2026-08-31)
- Scope: ordinary analytic window functions using OVER(...), not TIME WINDOW

### Details of Performance

Ordinary window execution is currently primarily sort-partition based.

For each window expression with PARTITION BY, the planner inserts a PARTITION node. The generic physical path sorts each input scope by the partition keys, merges the sorted streams on the coordinator to produce complete partitions, and then lets Window sort each partition by its window ORDER BY keys. Multiple window expressions can repeat this partitioning and sorting work.

Sort partitioning is desirable when the input already provides a compatible order or when predictable spill behavior is more important. It is not always the best choice when no useful input order exists, especially for large inputs, partition-only windows, or many small partitions.

Add a hash-partition physical path for ordinary windows while preserving window cardinality: every input row must remain in the output. This is not a rewrite to ordinary GROUP BY. The implementation needs a row-to-group mapping so that complete partitions can be evaluated and emitted without reducing N input rows to G groups.

Candidate physical plans:

1. SortPartition: local sort by partition keys, coordinator merge, Window evaluation.
2. HashPartition: gather input, hash partition on the coordinator, Window evaluation.
3. ShuffleHashPartition: hash shuffle by partition keys, then hash partition and Window evaluation on partition owners.

The optimizer should choose the algorithm and distribution topology after statistics are available. The decision should consider:

- input row count;
- composite partition-key NDV and expected/max partition size;
- partition-key width and row width;
- whether a compatible input ordering or distribution can be reused;
- presence and cost of per-partition window ORDER BY;
- DOP, CN count, skew, and available partition parallelism;
- estimated hash-table, row-id, buffered-row, and spill memory;
- whether compatible window specifications can share partitioning/order work.

Do not directly reuse the ordinary GROUP BY shuffle cost formula: Window preserves N output rows, so its reduction, network, buffering, and merge costs differ from aggregation.

The first implementation may use conservative hard guards:

- require a non-empty PARTITION BY;
- require hash-compatible key types and the same equality semantics as the existing comparator;
- prefer SortPartition for already ordered input, unsupported key semantics, unknown/unsafe memory estimates, or expected hash spill;
- expose the selected algorithm and estimates in EXPLAIN.

The execution layer should enforce the memory bound. Since ordinary Window is already blocking at the partition boundary, a local HashPartition implementation may fall back to sorting buffered input before publishing output if actual hash state exceeds its budget. The optimizer must still own distributed topology selection.

### Correctness and lifecycle requirements

- Preserve partition equality for NULL, CHAR padding, decimals, floating NaN, JSON, arrays, and composite keys. Initially exclude types whose hash equality differs from the existing partition comparator.
- Preserve all ordinary window families: aggregate, ranking, and value functions; ORDER BY and frame evaluation remain within each complete partition.
- Ensure every partition is owned by exactly one evaluator in distributed execution.
- Preserve cancellation, error, Reset/reuse, and Free cleanup paths.
- Do not rely on the incidental output order of a window query without an outer ORDER BY.
- Avoid OOM under skew or high NDV; account for or spill all retained state.

### Validation and acceptance criteria

- Planner tests cover semantic guards and forced Sort/Hash candidates.
- Operator tests compare HashPartition and SortPartition results across nullable and composite keys, ordered/unordered windows, skew, empty input, cancellation, and reuse.
- Distributed tests prove that one logical partition cannot be split across evaluators.
- EXPLAIN reports the selected partition algorithm and relevant estimates.
- Benchmarks cover uniform and skewed data with varying N, NDV, key width, row width, ORDER BY presence, and single-/multi-CN execution.
- Enable AUTO selection only after benchmarks establish a stable crossover and show no material regression for already ordered inputs or memory-constrained cases.

Related: #20655 describes the sort-based window path and the case where compatible clustering/order should be reused.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.