[Story][FEA] Timeseries-oriented cuDF-Polars: `OrderScheme` metadata, strict sort boundaries, and sort-aware execution

[rjzamora]

Context

Streaming window and time-oriented operators are tracked in #22032 — Streaming window support in cuDF-Polars (rolling, group_by_dynamic, ordered over(), shift, etc.). 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 will typically need to destroy that ordering to perform common group_by and join operations. Even when we do apply global sort operations, we don't keep track of this behavior in the executor, and will sometimes re-sort or shuffle data unnecessarily. Making ordering a first-class ChannelMetadata property (similar in spirit to how hash partitioning metadata is already tracked and preserved) should make the workloads highlighted in #22032 materially cheaper (less shuffle, less memory, simpler plans).

Dependency / alignment: rapidsmpf#853 — Add OrderScheme partitioning metadata (ordering in ChannelMetadata / partitioning). [DONE]

Sequencing

• Metadata convention: Align cuDF-Polars with rapidsmpf’s OrderScheme: what it means per partition, across ranks, null ordering, multi-column keys, and when it must be cleared vs preserved (Adopt OrderScheme metadata in cudf-polars #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 (Use basic OrderScheme metadata in sort_actor #22477)

---

• Allow OrderScheme to track multiple "orderings" at the same time (Refactor OrderScheme to track multiple Ordering descriptions #22896)
• Use 22896 to propagate "extra" ordering metadata through monotonic/unary column derivations ([FEA] Propagate "extra" ordering metadata through monotonic/unary column derivations #22999)

---

• Add general utility to enforce a local and global partitioning specification: We introduce enforce_partitioning that will abstracts-away the process of aligning incoming data with a specific Partitioning specification (whether that specification comprises HashScheme, OrderScheme, or a combination of these. Prerequisites:
  • Add OrderScheme.get_boundaries in RapidsMPF (Add OrderScheme.get_boundaries API rapidsmpf#1039): We need to be able to extract the boundaries table from an OrederScheme object in Python to sort the data with these boundaries.
  • Add adjust_orderscheme utility (Add adjust_ordering utility #22628): We need to be able to adjust the boundaries and/or strictness of an existing OrderScheme.
  • Add LocalRepartitioner.repartition_by_orderscheme utility (Add LocalRepartitioner.repartition_by_orderscheme #22522): We may have stratified partitioning. This utility will make enforce_partitioning easier to implement in a clean/consistent way for various partitioning schemes.
• Use enforce_partitioning to avoid unnecessary shuffling:
  • Use enforce_partitioning in groupby_actor
  • Use enforce_partitioning in join_actor
  • Use enforce_partitioning in over_actor
• Implement MergeSorted: This feature will depend on enforce_partitioning to align the partitioning of the input tables.

---

• Implement MapFunction(“hint_sorted”) (i.e. set_sorted()): This API allows the user to inform Polars that a table is already sorted. This means the user may pre-shuffle their dataset(s) on specific keys, and use set_sorted to declare that we don't need to re-sort on these keys. Prerequisites:
  • Add extract_orderscheme_partitioning utility (Extract OrderScheme metadata from a sorted Channel #22526): This utility will extract the boundaries table from a channel that we know to be sorted.