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

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

• 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

• 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

• Track multiple orderings at once: allow OrderScheme metadata to record more than one useful ordering.

  • https://github.com/rapidsai/cudf/pull/22896

• Propagate extra ordering metadata through monotonic/unary column derivations.

  • https://github.com/rapidsai/cudf/issues/22999

• Add OrderScheme.get_boundaries in rapidsmpf: expose the boundaries table from an OrderScheme object in Python.

  • https://github.com/rapidsai/rapidsmpf/pull/1039

• Add adjust_ordering utility: adjust the boundaries and/or strictness of an existing ordering.

  • https://github.com/rapidsai/cudf/pull/22628

• 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.

• 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

• 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

• 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).

• (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

• 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()