NVIDIA / NVIDIA/cudf

[Story][FEA] Order-aware streaming execution in cuDF-Polars

Open
#22,128 1 comment 0 reactions 0 assignees View on GitHub
cudf-polars feature request
Dominant language
C++
Stars
9.8k
Forks
1.1k
Avg merge
3d 6m
Merged PRs (30d)
278

Description

### Context

Streaming window and time-oriented operators are tracked in https://github.com/rapidsai/cudf/issues/18633 and https://github.com/rapidsai/cudf/issues/22032. In practice these features almost always sit in **sort-based workflows** by time or logical sequence.

Today, even when data is already sorted on disk, the executor may destroy that ordering to perform common `group_by` and `join` operations. Even when we do apply global sort operations, we do not always preserve enough ordering metadata for downstream operators to avoid redundant sorts, shuffles, or repartitioning.

Making ordering a first-class `ChannelMetadata` property, similar in spirit to hash partitioning metadata, should make ordered/window and time-series workloads materially cheaper: less shuffle, less memory pressure, and simpler plans.

**Dependency / alignment:** https://github.com/rapidsai/rapidsmpf/pull/853 added `OrderScheme` partitioning metadata in rapidsmpf. [DONE]

### Sequencing

- [x] **Metadata convention**: align cuDF-Polars with rapidsmpf’s `OrderScheme`: per-partition meaning, across-rank meaning, null ordering, multi-column keys, and when ordering metadata must be cleared vs preserved.
- https://github.com/rapidsai/cudf/pull/22291

- [x] **`sort_actor` emits `OrderScheme`**: after sort, set metadata downstream operators can trust. The `sort_actor` should also avoid sorting data that is already sorted.
- https://github.com/rapidsai/cudf/pull/22477

- [x] **Track multiple orderings at once**: allow `OrderScheme` metadata to record more than one useful ordering.
- https://github.com/rapidsai/cudf/pull/22896

- [x] **Propagate extra ordering metadata** through monotonic/unary column derivations.
- https://github.com/rapidsai/cudf/issues/22999

- [x] **Add `OrderScheme.get_boundaries` in rapidsmpf**: expose the boundaries table from an `OrderScheme` object in Python.
- https://github.com/rapidsai/rapidsmpf/pull/1039

- [x] **Add `adjust_ordering` utility**: adjust the boundaries and/or strictness of an existing ordering.
- https://github.com/rapidsai/cudf/pull/22628

- [x] **Collect downstream partitioning hints during planning**: allow actor-graph construction to propagate useful downstream ordering/strict-partitioning requirements upstream. This lets scan/`hint_sorted`, sort, groupby, and join actors make better local decisions about preserving or materializing ordering.
- https://github.com/NVIDIA/cudf/issues/23705
- https://github.com/NVIDIA/cudf/pull/23729

- [ ] **Implement `MapFunction("hint_sorted")` / `set_sorted()` support**: allow users to declare that an input table is already sorted so the executor can attach and trust ordering metadata.
- In-memory support: https://github.com/NVIDIA/cudf/pull/23663 [DONE]
- Streaming metadata support: https://github.com/NVIDIA/cudf/pull/23748
- Tracking issue: https://github.com/NVIDIA/cudf/issues/21039
- **TODO**: Use `extract_orderscheme_partitioning` after https://github.com/NVIDIA/cudf/pull/22526 is merged.

- [x] **Expose Parquet column-chunk min/max statistics**: provide the metadata needed for Parquet scans to construct ordering boundaries without buffering the full input stream.
- https://github.com/NVIDIA/cudf/issues/23661
- https://github.com/NVIDIA/cudf/pull/23666

- [x] **Extract ordering boundaries from sorted streaming channels**: provide a fallback path for constructing `OrderScheme` metadata when scan-level metadata is unavailable.
- https://github.com/NVIDIA/cudf/pull/22526

- [ ] **Extract ordering boundaries from parquet metadata**: Use ordering requests to trigger parquet-metadata evaluation in the scan actor. This should be the "cheapest" way to attach `Ordering` metadata.
- **PR**: https://github.com/NVIDIA/cudf/pull/24199

- [x] **Use ordering metadata in `groupby_actor` planning**: allow `groupby_actor` to choose cheaper order-aware execution when the input ordering is compatible.
- Basic support: https://github.com/NVIDIA/cudf/pull/23306
- **TODO/Perf BUG**: The adjust-ordering branch is no longer taken in most cases (Fix: https://github.com/NVIDIA/cudf/pull/23927).

- [x] **(BLOCKER) Track/use local row-ordering metadata**: In order to propagate `Ordering` metadata correctly and effectively, we must keep track of both order-based "partitioning" **and** local ordering within the partition. Otherwise, we must either evict the `Ordering` metadata aggressively, or add work (that may be unnecessary) to ensure chunk-local groupby and join operations always preserve the ordering.
- PR: https://github.com/NVIDIA/cudf/pull/23844

- [x] **Use ordering metadata in `join_actor` planning**: allow `join_actor` to choose cheaper order-aware execution when one or both inputs have compatible ordering.
- https://github.com/NVIDIA/cudf/pull/23371
- **TODO/Perf BUG**: We are calling `adjust_ordering` unnecessarily in some cases (Fix: https://github.com/NVIDIA/cudf/pull/23927).

- [ ] **Evaluate ordering metadata in `over_actor` planning**: decide where grouped/window `over(...)` execution can avoid unnecessary shuffles or preserve useful ordering.

- [ ] **Evaluate local repartitioning by `OrderScheme`**: this may be useful for stratified or tiered partitioning, but is not the immediate planning abstraction.
- https://github.com/NVIDIA/cudf/pull/22522

- [ ] **Evaluate `MergeSorted` support**: decide whether merge-sorted execution needs a dedicated planning path and which ordering/repartitioning primitives it should use.

- [ ] **Evaluate order-aware concat/Union execution**: when concat inputs carry compatible `OrderScheme` metadata, use the known ordering/boundaries to avoid blind append/shuffle behavior and preserve or reconstruct ordered output metadata where possible.
- Related: https://github.com/NVIDIA/cudf/issues/22376

### Current Design Direction

Current PRs favor actor-specific planning guided by metadata and lightweight partitioning hints. The planning pass can tell upstream actors which ordering or strict partitioning would be useful downstream, but each actor still decides locally whether to no-op, preserve metadata, extract boundaries, call `adjust_ordering`, use tree/local execution, hash shuffle, or use a future order-aware path.

For now, prefer:
- keeping repartitioning decisions in actors
- factoring out shared helpers for common metadata checks and transformations
- avoiding a broad `enforce_partitioning` API until tiered partitioning semantics are clearer

### Expected Impact

This work is expected to reduce unnecessary sorting, shuffling, and memory pressure in:

- streaming `LazyFrame.rolling(...)`
- grouped range windows
- grouped `over(...)` execution
- order-aware grouped reductions
- order-aware joins
- future `join_asof` #24110
- user-declared sorted input via `set_sorted()`

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.