Merge remote-tracking branch 'upstream/main' into cudf_polars/skill/expressions_impl

Author: mroeschke

cpp/libcudf_streaming/include/cudf_streaming/channel_metadata.hpp

@@ -1,6 +1,6 @@
 /**
  * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
- * reserved. SPDX-License-Identifier: Apache-2.0
+ * SPDX-License-Identifier: Apache-2.0
  */
 
 #pragma once
@@ -60,7 +60,7 @@ struct order_key {
 };
 
 /**
- * @brief Order-based partitioning scheme for sorted/range-partitioned data.
+ * @brief A valid ordering description for sorted/range-partitioned data.
  *
  * Data is partitioned by value ranges based on predetermined boundaries.
  * For N partitions, there are N-1 boundary rows:
@@ -78,17 +78,17 @@ struct order_key {
  * half-open key range (partition keys do not straddle chunk interiors). When false,
  * a chunk may contain keys spanning multiple partitions.
  */
-struct order_scheme {
+struct ordering {
   std::vector<order_key> keys;              ///< Sort keys (column, order, null_order per entry).
   std::shared_ptr<table_chunk> boundaries;  ///< N-1 boundary rows for N partitions.
   /// See struct-level note on `strict_boundaries` semantics.
   bool strict_boundaries{false};
 
-  /// @brief Default constructor. Produces an invalid (empty) scheme.
-  order_scheme() = default;
+  /// @brief Default constructor. Produces an invalid (empty) ordering.
+  ordering() = default;
 
   /**
-   * @brief Construct a validated order_scheme.
+   * @brief Construct a validated ordering.
    *
    * @param keys Non-empty sort keys; size must equal `boundaries->shape().second`.
    * @param boundaries Non-null, device-resident boundary table (N-1 rows for N
@@ -97,36 +97,67 @@ struct order_scheme {
    * @throws std::invalid_argument if `keys` is empty, `boundaries` is null or not
    * device-resident, or `keys.size() != boundaries->shape().second`.
    */
-  order_scheme(std::vector<order_key> keys,
-               std::shared_ptr<table_chunk> boundaries,
-               bool strict_boundaries = false);
+  ordering(std::vector<order_key> keys,
+           std::shared_ptr<table_chunk> boundaries,
+           bool strict_boundaries = false);
 
   /**
-   * @brief Return a new order_scheme with updated key column indices, sharing
-   * boundaries.
-   *
-   * The new key count must match the existing boundary column count.
+   * @brief Return a new ordering with updated key column indices, sharing
+   * boundary rows.
    *
    * @param new_keys Replacement sort keys; size must equal
    * `boundaries->shape().second`.
-   * @return A new order_scheme with `new_keys` and the same `boundaries` and
-   *         `strict_boundaries`.
-   * @throws std::invalid_argument if `new_keys` is empty or size mismatches boundaries.
+   * @return A new ordering with `new_keys` and the same boundaries and
+   * strictness.
+   * @throws std::invalid_argument if `new_keys` is empty or size mismatches
+   * boundaries.
    */
-  [[nodiscard]] order_scheme with_keys(std::vector<order_key> new_keys) const;
+  [[nodiscard]] ordering with_keys(std::vector<order_key> new_keys) const;
 
   /**
-   * @brief Check whether boundary values are aligned with another scheme.
+   * @brief Check whether boundary values are aligned with another ordering.
    *
-   * @param other The order_scheme to compare against.
+   * @param other The ordering to compare against.
    * @param br Buffer resource used for temporary allocations during comparison.
-   * @return True when both schemes have matching boundary values and strict_boundaries
-   * attributes, and the schemes are otherwise compatible (same order and null_order).
+   * @return True when both orderings have matching boundary values and
+   * strict_boundaries attributes, and are otherwise compatible (same order and
+   * null_order).
    */
-  [[nodiscard]] bool boundaries_aligned_with(order_scheme const& other,
+  [[nodiscard]] bool boundaries_aligned_with(ordering const& other,
                                              rapidsmpf::BufferResource& br) const;
 };
 
+/**
+ * @brief Order-based partitioning scheme for sorted/range-partitioned data.
+ *
+ * An order_scheme advertises that the same stream is sorted/range-partitioned
+ * with respect to any individual ordering it contains. Consumers are
+ * responsible for selecting the ordering that is relevant to a particular
+ * operation.
+ */
+struct order_scheme {
+  std::vector<ordering> orderings;  ///< Ordering descriptions valid for the stream.
+
+  /// @brief Default constructor. Produces an invalid (empty) scheme.
+  order_scheme() = default;
+
+  /**
+   * @brief Construct a validated single-ordering order_scheme.
+   *
+   * See `ordering` for parameter semantics.
+   */
+  order_scheme(std::vector<order_key> keys,
+               std::shared_ptr<table_chunk> boundaries,
+               bool strict_boundaries = false);
+
+  /**
+   * @brief Construct a validated multi-ordering order_scheme.
+   *
+   * @param orderings Non-empty sequence of orderings valid for the stream.
+   */
+  explicit order_scheme(std::vector<ordering> orderings);
+};
+
 /**
  * @brief Partitioning specification for a single hierarchical level.
  *
@@ -180,7 +211,7 @@ struct partitioning_spec {
 
   /**
    * @brief Create a spec for order/range partitioning.
-   * @param o The order scheme to use. `o.keys` must be non-empty; otherwise
+   * @param o The order scheme to use. `o.orderings` must be non-empty; otherwise
    * throws `std::invalid_argument`.
    * @return A partitioning_spec with type ORDER.
    */

cpp/libcudf_streaming/src/channel_metadata.cpp

@@ -1,6 +1,6 @@
 /**
  * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
- * reserved. SPDX-License-Identifier: Apache-2.0
+ * SPDX-License-Identifier: Apache-2.0
  */
 
 #include <cudf/aggregation.hpp>
@@ -21,36 +21,39 @@
 #include <utility>
 
 namespace cudf_streaming {
+namespace {
 
-order_scheme::order_scheme(std::vector<order_key> keys,
-                           std::shared_ptr<table_chunk> boundaries,
-                           bool strict_boundaries)
-  : keys{std::move(keys)}, boundaries{std::move(boundaries)}, strict_boundaries{strict_boundaries}
+void validate_ordering(ordering const& ordering)
 {
   RAPIDSMPF_EXPECTS(
-    !this->keys.empty(), "OrderScheme: keys must not be empty", std::invalid_argument);
+    !ordering.keys.empty(), "ordering: keys must not be empty", std::invalid_argument);
   RAPIDSMPF_EXPECTS(
-    this->boundaries != nullptr, "OrderScheme: boundaries must not be null", std::invalid_argument);
-  RAPIDSMPF_EXPECTS(this->boundaries->is_available(),
-                    "OrderScheme: boundaries must be device-resident",
-                    std::invalid_argument);
-  RAPIDSMPF_EXPECTS(this->keys.size() == static_cast<std::size_t>(this->boundaries->shape().second),
-                    "OrderScheme: number of keys must match number of boundary columns",
+    ordering.boundaries != nullptr, "ordering: boundaries must not be null", std::invalid_argument);
+  RAPIDSMPF_EXPECTS(ordering.boundaries->is_available(),
+                    "ordering: boundaries must be device-resident",
                     std::invalid_argument);
+  RAPIDSMPF_EXPECTS(
+    ordering.keys.size() == static_cast<std::size_t>(ordering.boundaries->shape().second),
+    "ordering: number of keys must match number of boundary columns",
+    std::invalid_argument);
 }
 
-partitioning_spec partitioning_spec::from_order(order_scheme o)
+}  // namespace
+
+ordering::ordering(std::vector<order_key> keys,
+                   std::shared_ptr<table_chunk> boundaries,
+                   bool strict_boundaries)
+  : keys{std::move(keys)}, boundaries{std::move(boundaries)}, strict_boundaries{strict_boundaries}
 {
-  return {.type = type::ORDER, .hash = std::nullopt, .order = std::move(o)};
+  validate_ordering(*this);
 }
 
-order_scheme order_scheme::with_keys(std::vector<order_key> new_keys) const
+ordering ordering::with_keys(std::vector<order_key> new_keys) const
 {
-  return order_scheme(std::move(new_keys), boundaries, strict_boundaries);
+  return ordering{std::move(new_keys), boundaries, strict_boundaries};
 }
 
-bool order_scheme::boundaries_aligned_with(order_scheme const& other,
-                                           rapidsmpf::BufferResource& br) const
+bool ordering::boundaries_aligned_with(ordering const& other, rapidsmpf::BufferResource& br) const
 {
   if (strict_boundaries != other.strict_boundaries ||
       boundaries->shape() != other.boundaries->shape()) {
@@ -85,6 +88,30 @@ bool order_scheme::boundaries_aligned_with(order_scheme const& other,
   return true;
 }
 
+order_scheme::order_scheme(std::vector<order_key> keys,
+                           std::shared_ptr<table_chunk> boundaries,
+                           bool strict_boundaries)
+  : order_scheme(
+      std::vector<ordering>{ordering{std::move(keys), std::move(boundaries), strict_boundaries}})
+{
+}
+
+order_scheme::order_scheme(std::vector<ordering> orderings) : orderings{std::move(orderings)}
+{
+  RAPIDSMPF_EXPECTS(
+    !this->orderings.empty(), "order_scheme: orderings must not be empty", std::invalid_argument);
+  for (auto const& ordering : this->orderings) {
+    RAPIDSMPF_EXPECTS(!ordering.keys.empty(),
+                      "order_scheme: ordering entries must not be empty",
+                      std::invalid_argument);
+  }
+}
+
+partitioning_spec partitioning_spec::from_order(order_scheme o)
+{
+  return {.type = type::ORDER, .hash = std::nullopt, .order = std::move(o)};
+}
+
 rapidsmpf::streaming::Message to_message(std::uint64_t sequence_number,
                                          std::unique_ptr<channel_metadata> m)
 {

cpp/libcudf_streaming/tests/streaming/test_channel_metadata.cpp

@@ -1,6 +1,6 @@
 /**
  * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
- * reserved. SPDX-License-Identifier: Apache-2.0
+ * SPDX-License-Identifier: Apache-2.0
  */
 
 #include <cudf_test/cudf_gtest.hpp>
@@ -50,6 +50,11 @@ TEST_F(StreamingChannelMetadata, OrderSchemeCtorRejectsNullBoundaries)
                std::invalid_argument);
 }
 
+TEST_F(StreamingChannelMetadata, OrderSchemeCtorRejectsEmptyOrderings)
+{
+  EXPECT_THROW(static_cast<void>(order_scheme(std::vector<ordering>{})), std::invalid_argument);
+}
+
 TEST_F(StreamingChannelMetadata, PartitioningSpec)
 {
   // None
@@ -173,14 +178,14 @@ class StreamingChannelMetadataGPU : public ::testing::Test {
   }
 };
 
-TEST_F(StreamingChannelMetadataGPU, OrderSchemeReplaceKeys)
+TEST_F(StreamingChannelMetadataGPU, OrderingReplaceKeys)
 {
   order_key k0{0, cudf::order::ASCENDING, cudf::null_order::BEFORE};
   order_key k5{5, cudf::order::DESCENDING, cudf::null_order::AFTER};
 
-  auto b = make_chunk({100, 200});
-  order_scheme o1({k0}, b);
-  auto o2 = o1.with_keys({k5});
+  auto b      = make_chunk({100, 200});
+  ordering o1 = ordering({k0}, b);
+  auto o2     = o1.with_keys({k5});
 
   EXPECT_EQ(o2.keys[0].column_index, 5);
   EXPECT_EQ(o2.keys[0].order, cudf::order::DESCENDING);
@@ -192,22 +197,40 @@ TEST_F(StreamingChannelMetadataGPU, OrderSchemeReplaceKeys)
   EXPECT_THROW(static_cast<void>(o1.with_keys({k0, k5})), std::invalid_argument);
 }
 
-TEST_F(StreamingChannelMetadataGPU, OrderSchemeBoundariesAlignedWith)
+TEST_F(StreamingChannelMetadataGPU, OrderSchemeMultipleOrderings)
+{
+  order_key k0{0, cudf::order::ASCENDING, cudf::null_order::BEFORE};
+  order_key k2{2, cudf::order::DESCENDING, cudf::null_order::AFTER};
+
+  auto b0 = make_chunk({100, 200});
+  auto b1 = make_chunk({300, 400});
+  order_scheme o({ordering{{k0}, b0, true}, ordering{{k2}, b1, false}});
+
+  ASSERT_EQ(o.orderings.size(), 2);
+  EXPECT_EQ(o.orderings[0].keys[0], k0);
+  EXPECT_EQ(o.orderings[0].boundaries.get(), b0.get());
+  EXPECT_TRUE(o.orderings[0].strict_boundaries);
+  EXPECT_EQ(o.orderings[1].keys[0], k2);
+  EXPECT_EQ(o.orderings[1].boundaries.get(), b1.get());
+  EXPECT_FALSE(o.orderings[1].strict_boundaries);
+}
+
+TEST_F(StreamingChannelMetadataGPU, OrderingBoundariesAlignedWith)
 {
   order_key k0{0, cudf::order::ASCENDING, cudf::null_order::BEFORE};
   order_key k3{3, cudf::order::ASCENDING, cudf::null_order::BEFORE};
 
-  order_scheme o1({k0}, make_chunk({100, 200}));
-  order_scheme o2({k0}, make_chunk({100, 200}));
+  ordering o1({k0}, make_chunk({100, 200}));
+  ordering o2({k0}, make_chunk({100, 200}));
   EXPECT_TRUE(o1.boundaries_aligned_with(o2, *br));
 
-  order_scheme o_shifted({k3}, make_chunk({100, 200}));
+  ordering o_shifted({k3}, make_chunk({100, 200}));
   EXPECT_TRUE(o1.boundaries_aligned_with(o_shifted, *br));
 
-  order_scheme o_strict({k0}, make_chunk({100, 200}), /*strict=*/true);
+  ordering o_strict({k0}, make_chunk({100, 200}), /*strict_boundaries=*/true);
   EXPECT_FALSE(o1.boundaries_aligned_with(o_strict, *br));
 
-  order_scheme o_diff({k0}, make_chunk({100, 300}));
+  ordering o_diff({k0}, make_chunk({100, 300}));
   EXPECT_FALSE(o1.boundaries_aligned_with(o_diff, *br));
 }
 
@@ -219,10 +242,10 @@ TEST_F(StreamingChannelMetadataGPU, PartitioningSpecOrder)
   auto spec = partitioning_spec::from_order(o);
   EXPECT_EQ(spec.type, partitioning_spec::type::ORDER);
   EXPECT_TRUE(spec.order.has_value());
-  EXPECT_EQ(spec.order->keys[0].column_index, 0);
+  EXPECT_EQ(spec.order->orderings[0].keys[0].column_index, 0);
 
   // Type checks only (partitioning_spec::operator== removed; ORDER value comparison
-  // requires boundaries_aligned_with on the order_scheme directly)
+  // requires boundaries_aligned_with on the ordering directly)
   EXPECT_EQ(spec.type, partitioning_spec::type::ORDER);
   EXPECT_NE(spec.type, partitioning_spec::from_hash(hash_scheme{{0}, 16}).type);
   EXPECT_NE(spec.type, partitioning_spec::none().type);