matrixorigin / matrixorigin/matrixone

[Enhancement]: Exploit proven storage/index order for streaming grouped LIMIT

Open
#27,857 0 comments 0 reactions 1 assignee Claimed by @XuPeng-SH View on GitHub
area/optimizer area/storage kind/enhancement needs-triage
Dominant language
Go
Stars
1.9k
Forks
311
Avg merge
1d 3h
Merged PRs (30d)
768

Description

## Summary

Investigate a proof-gated streaming aggregate for queries whose grouping order is compatible with a `CLUSTER BY` storage key. This is a separate follow-up from primary-key elimination (#27850) and non-PK unique-key elimination (#27856).

`CLUSTER BY` is not a uniqueness constraint. It may reduce hash/state or scan work only when the optimizer and storage layer can prove the exact scan stream order needed to finalize groups.

## First-principles distinction

```text
PRIMARY/UNIQUE proof
=> at most one row per group
=> Aggregate can be replaced by row expressions

CLUSTER BY order proof
=> rows of a group may be contiguous in one ordered stream
=> Aggregate is still required, but it may stream/finalize one group at a time
```

Current cluster-key metadata alone is insufficient. MatrixOne uses primary/cluster keys as sort keys during object merge, but object-local ordering does not by itself prove globally ordered scan output across:

- newly written or not-yet-merged objects;
- multiple objects/blocks scheduled in parallel;
- partitions, ranges or multiple CN workers;
- snapshots containing different object generations;
- deletes/updates and background compaction;
- forward versus reverse scan direction.

The optimization must consume an explicit physical order property, not infer global order from `TableDef.ClusterBy`.

## Initial eligible shapes

A first version may consider only:

1. scan output has a proven global lexicographic order matching a leading prefix of the complete grouping key;
2. all rows for each group reach exactly one streaming finalization owner, or ordered local streams are merged before finalization;
3. WHERE preserves the required order;
4. no query operator between scan and aggregate destroys or duplicates order;
5. bounded demand is unordered, or ORDER BY exactly matches the proven group-key order and direction;
6. HAVING, if supported, is evaluated after each group is complete and LIMIT/OFFSET counts only passing groups;
7. every fallback retains the established hash/spill aggregate.

Potential benefits:

- O(one active group) state for a fully ordered single stream;
- bounded state per proven ordered partition/range;
- early stop after `K + OFFSET` completed/passing groups for unordered LIMIT or compatible group-key ORDER BY;
- reduced spill/hash work even when a full scan is still required.

## Must fail closed

- `ORDER BY` aggregate results such as `ORDER BY COUNT(*) DESC`; unseen groups can still win;
- grouping by a non-leading cluster-key component;
- grouping expressions whose collation/coercion/order differs from the physical key;
- a stream assembled from sorted objects without a proven global merge order;
- overlapping key ranges assigned to independent final owners without an exact merge/finalization contract;
- reverse order without a proven reverse scan;
- partition-local order presented as table-global order;
- nullable/NaN/collation semantics not identical between storage ordering and SQL grouping;
- estimated clustering quality, zone maps or NDV used as an exact order proof.

## Required architecture

1. Define a reusable physical order property containing relation/tag, ordered expressions, direction, NULL/collation semantics, scope (block/object/partition/global), and whether key ranges overlap.
2. Have storage/access-path planning advertise only guarantees it can uphold for the exact snapshot and scan schedule.
3. Propagate/invalidate the property through Project, Filter, Exchange, Sort, Join, Union, partition pruning and distributed planning.
4. Let grouped aggregation select streaming mode only from that property.
5. Define ownership for group boundaries spanning batches, blocks, objects, partitions and CN streams.
6. Keep hash/spill aggregation as deterministic fallback on any missing or inconsistent property.

Do not add a query-text, table-name or raw `ClusterBy != nil` special case.

## Correctness matrix

| Dimension | Required cases |
| --- | --- |
| Key shape | single/composite cluster key; full key; leading prefix; non-prefix control |
| Cardinality | empty, one group, repeated rows/group, low/high cardinality |
| Object state | one object; many overlapping/non-overlapping objects; unmerged inserts; after compaction |
| Mutation | inserts, deletes, updates, snapshot reads and restart |
| Execution | DOP=1/multi-DOP; one/multi-CN; exchange/range assignment |
| Query | unordered LIMIT; matching ASC/DESC group-key order; aggregate order control; OFFSET |
| Filters | WHERE; HAVING with passing/non-passing groups; selective zone-map pruning |
| Types | NULL, signed/unsigned, DECIMAL, VARCHAR collations, CHAR padding, binary, temporal, composite key |
| Boundaries | Project, Sort, Join, Union/shared CTE, grouping sets, partitioned tables |
| Failure | cancellation/error at group boundary, fallback, Reset/Free/reuse if runtime mode is added |

Result comparisons must use multisets for unordered queries and exact order for ordered queries. They must include metadata and error status, not only row counts or plan text.

## Performance validation

On one NVMe dataset shared by base and candidate binaries:

1. create a non-unique cluster-key table with many rows per group and multiple objects;
2. test small/large LIMIT, OFFSET, selective WHERE/HAVING and matching group-key ORDER BY;
3. separately test `ORDER BY aggregate` and non-prefix grouping controls;
4. measure scan rows/bytes, active group state, hash allocations, spill bytes, merge/exchange work, CPU, peak RSS and wall-time medians;
5. force unmerged and compacted layouts to prove the optimization does not depend on accidental object order;
6. require a large-benefit case and no material fallback regression.

## Acceptance criteria

1. Cluster-key metadata alone never activates streaming or early stop.
2. A versioned physical-order contract proves exact scope and survives distributed plan construction.
3. Group boundaries crossing every batch/block/object/worker boundary are finalized exactly once.
4. LIMIT/OFFSET/HAVING consume only complete qualifying groups; aggregate-result ordering never stops early without an independent bound proof.
5. Unsupported or uncertain cases retain the current hash/spill plan.
6. Deterministic UT, storage/order-property tests, distributed BVT, same-data A/B, full service build, SCA and required CI pass.
7. The PR body provides a QA-executable setup for unmerged, compacted, single-CN and multi-CN layouts.

## Related

- Primary-key single-row elimination: #27850
- Non-PK unique-key follow-up: #27856
- Tracking/design decision: #27730
- ClickBench Q35 remains separate: #27685

## Ordered index sources

The same physical-order contract should cover more than `CLUSTER BY`:

- ordered primary/secondary index scans whose leading key prefix matches the grouping order;
- suffix ordering when every preceding index component is fixed by exact equality predicates;
- covering index scans, which can additionally reduce base-table reads but still require the order proof;
- forward and reverse scans only when direction and NULL/collation semantics match SQL exactly;
- a future loose/skip index scan that seeks group boundaries for eligible MIN/MAX or group-prefix queries.

Index metadata alone is not enough: the selected access path and distributed scan schedule must preserve one exact global stream or provide a correct ordered merge before group finalization. Bitmap, inverted, vector and full-text candidate indexes do not imply this order.

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.