Skip to content

Commit 8b8dd84

Browse files
committed
Merge remote-tracking branch 'upstream/main' into mi_factorize
# Conflicts: # python/cudf/cudf/pandas/scripts/pandas-testing-plugin.py
2 parents bf33cd4 + 175fb17 commit 8b8dd84

34 files changed

Lines changed: 2606 additions & 508 deletions

File tree

cpp/libcudf_streaming/include/cudf_streaming/channel_metadata.hpp

Lines changed: 54 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
/**
22
* SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
3-
* reserved. SPDX-License-Identifier: Apache-2.0
3+
* SPDX-License-Identifier: Apache-2.0
44
*/
55

66
#pragma once
@@ -60,7 +60,7 @@ struct order_key {
6060
};
6161

6262
/**
63-
* @brief Order-based partitioning scheme for sorted/range-partitioned data.
63+
* @brief A valid ordering description for sorted/range-partitioned data.
6464
*
6565
* Data is partitioned by value ranges based on predetermined boundaries.
6666
* For N partitions, there are N-1 boundary rows:
@@ -78,17 +78,17 @@ struct order_key {
7878
* half-open key range (partition keys do not straddle chunk interiors). When false,
7979
* a chunk may contain keys spanning multiple partitions.
8080
*/
81-
struct order_scheme {
81+
struct ordering {
8282
std::vector<order_key> keys; ///< Sort keys (column, order, null_order per entry).
8383
std::shared_ptr<table_chunk> boundaries; ///< N-1 boundary rows for N partitions.
8484
/// See struct-level note on `strict_boundaries` semantics.
8585
bool strict_boundaries{false};
8686

87-
/// @brief Default constructor. Produces an invalid (empty) scheme.
88-
order_scheme() = default;
87+
/// @brief Default constructor. Produces an invalid (empty) ordering.
88+
ordering() = default;
8989

9090
/**
91-
* @brief Construct a validated order_scheme.
91+
* @brief Construct a validated ordering.
9292
*
9393
* @param keys Non-empty sort keys; size must equal `boundaries->shape().second`.
9494
* @param boundaries Non-null, device-resident boundary table (N-1 rows for N
@@ -97,36 +97,67 @@ struct order_scheme {
9797
* @throws std::invalid_argument if `keys` is empty, `boundaries` is null or not
9898
* device-resident, or `keys.size() != boundaries->shape().second`.
9999
*/
100-
order_scheme(std::vector<order_key> keys,
101-
std::shared_ptr<table_chunk> boundaries,
102-
bool strict_boundaries = false);
100+
ordering(std::vector<order_key> keys,
101+
std::shared_ptr<table_chunk> boundaries,
102+
bool strict_boundaries = false);
103103

104104
/**
105-
* @brief Return a new order_scheme with updated key column indices, sharing
106-
* boundaries.
107-
*
108-
* The new key count must match the existing boundary column count.
105+
* @brief Return a new ordering with updated key column indices, sharing
106+
* boundary rows.
109107
*
110108
* @param new_keys Replacement sort keys; size must equal
111109
* `boundaries->shape().second`.
112-
* @return A new order_scheme with `new_keys` and the same `boundaries` and
113-
* `strict_boundaries`.
114-
* @throws std::invalid_argument if `new_keys` is empty or size mismatches boundaries.
110+
* @return A new ordering with `new_keys` and the same boundaries and
111+
* strictness.
112+
* @throws std::invalid_argument if `new_keys` is empty or size mismatches
113+
* boundaries.
115114
*/
116-
[[nodiscard]] order_scheme with_keys(std::vector<order_key> new_keys) const;
115+
[[nodiscard]] ordering with_keys(std::vector<order_key> new_keys) const;
117116

118117
/**
119-
* @brief Check whether boundary values are aligned with another scheme.
118+
* @brief Check whether boundary values are aligned with another ordering.
120119
*
121-
* @param other The order_scheme to compare against.
120+
* @param other The ordering to compare against.
122121
* @param br Buffer resource used for temporary allocations during comparison.
123-
* @return True when both schemes have matching boundary values and strict_boundaries
124-
* attributes, and the schemes are otherwise compatible (same order and null_order).
122+
* @return True when both orderings have matching boundary values and
123+
* strict_boundaries attributes, and are otherwise compatible (same order and
124+
* null_order).
125125
*/
126-
[[nodiscard]] bool boundaries_aligned_with(order_scheme const& other,
126+
[[nodiscard]] bool boundaries_aligned_with(ordering const& other,
127127
rapidsmpf::BufferResource& br) const;
128128
};
129129

130+
/**
131+
* @brief Order-based partitioning scheme for sorted/range-partitioned data.
132+
*
133+
* An order_scheme advertises that the same stream is sorted/range-partitioned
134+
* with respect to any individual ordering it contains. Consumers are
135+
* responsible for selecting the ordering that is relevant to a particular
136+
* operation.
137+
*/
138+
struct order_scheme {
139+
std::vector<ordering> orderings; ///< Ordering descriptions valid for the stream.
140+
141+
/// @brief Default constructor. Produces an invalid (empty) scheme.
142+
order_scheme() = default;
143+
144+
/**
145+
* @brief Construct a validated single-ordering order_scheme.
146+
*
147+
* See `ordering` for parameter semantics.
148+
*/
149+
order_scheme(std::vector<order_key> keys,
150+
std::shared_ptr<table_chunk> boundaries,
151+
bool strict_boundaries = false);
152+
153+
/**
154+
* @brief Construct a validated multi-ordering order_scheme.
155+
*
156+
* @param orderings Non-empty sequence of orderings valid for the stream.
157+
*/
158+
explicit order_scheme(std::vector<ordering> orderings);
159+
};
160+
130161
/**
131162
* @brief Partitioning specification for a single hierarchical level.
132163
*
@@ -180,7 +211,7 @@ struct partitioning_spec {
180211

181212
/**
182213
* @brief Create a spec for order/range partitioning.
183-
* @param o The order scheme to use. `o.keys` must be non-empty; otherwise
214+
* @param o The order scheme to use. `o.orderings` must be non-empty; otherwise
184215
* throws `std::invalid_argument`.
185216
* @return A partitioning_spec with type ORDER.
186217
*/

cpp/libcudf_streaming/src/channel_metadata.cpp

Lines changed: 45 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
/**
22
* SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
3-
* reserved. SPDX-License-Identifier: Apache-2.0
3+
* SPDX-License-Identifier: Apache-2.0
44
*/
55

66
#include <cudf/aggregation.hpp>
@@ -21,36 +21,39 @@
2121
#include <utility>
2222

2323
namespace cudf_streaming {
24+
namespace {
2425

25-
order_scheme::order_scheme(std::vector<order_key> keys,
26-
std::shared_ptr<table_chunk> boundaries,
27-
bool strict_boundaries)
28-
: keys{std::move(keys)}, boundaries{std::move(boundaries)}, strict_boundaries{strict_boundaries}
26+
void validate_ordering(ordering const& ordering)
2927
{
3028
RAPIDSMPF_EXPECTS(
31-
!this->keys.empty(), "OrderScheme: keys must not be empty", std::invalid_argument);
29+
!ordering.keys.empty(), "ordering: keys must not be empty", std::invalid_argument);
3230
RAPIDSMPF_EXPECTS(
33-
this->boundaries != nullptr, "OrderScheme: boundaries must not be null", std::invalid_argument);
34-
RAPIDSMPF_EXPECTS(this->boundaries->is_available(),
35-
"OrderScheme: boundaries must be device-resident",
36-
std::invalid_argument);
37-
RAPIDSMPF_EXPECTS(this->keys.size() == static_cast<std::size_t>(this->boundaries->shape().second),
38-
"OrderScheme: number of keys must match number of boundary columns",
31+
ordering.boundaries != nullptr, "ordering: boundaries must not be null", std::invalid_argument);
32+
RAPIDSMPF_EXPECTS(ordering.boundaries->is_available(),
33+
"ordering: boundaries must be device-resident",
3934
std::invalid_argument);
35+
RAPIDSMPF_EXPECTS(
36+
ordering.keys.size() == static_cast<std::size_t>(ordering.boundaries->shape().second),
37+
"ordering: number of keys must match number of boundary columns",
38+
std::invalid_argument);
4039
}
4140

42-
partitioning_spec partitioning_spec::from_order(order_scheme o)
41+
} // namespace
42+
43+
ordering::ordering(std::vector<order_key> keys,
44+
std::shared_ptr<table_chunk> boundaries,
45+
bool strict_boundaries)
46+
: keys{std::move(keys)}, boundaries{std::move(boundaries)}, strict_boundaries{strict_boundaries}
4347
{
44-
return {.type = type::ORDER, .hash = std::nullopt, .order = std::move(o)};
48+
validate_ordering(*this);
4549
}
4650

47-
order_scheme order_scheme::with_keys(std::vector<order_key> new_keys) const
51+
ordering ordering::with_keys(std::vector<order_key> new_keys) const
4852
{
49-
return order_scheme(std::move(new_keys), boundaries, strict_boundaries);
53+
return ordering{std::move(new_keys), boundaries, strict_boundaries};
5054
}
5155

52-
bool order_scheme::boundaries_aligned_with(order_scheme const& other,
53-
rapidsmpf::BufferResource& br) const
56+
bool ordering::boundaries_aligned_with(ordering const& other, rapidsmpf::BufferResource& br) const
5457
{
5558
if (strict_boundaries != other.strict_boundaries ||
5659
boundaries->shape() != other.boundaries->shape()) {
@@ -85,6 +88,30 @@ bool order_scheme::boundaries_aligned_with(order_scheme const& other,
8588
return true;
8689
}
8790

91+
order_scheme::order_scheme(std::vector<order_key> keys,
92+
std::shared_ptr<table_chunk> boundaries,
93+
bool strict_boundaries)
94+
: order_scheme(
95+
std::vector<ordering>{ordering{std::move(keys), std::move(boundaries), strict_boundaries}})
96+
{
97+
}
98+
99+
order_scheme::order_scheme(std::vector<ordering> orderings) : orderings{std::move(orderings)}
100+
{
101+
RAPIDSMPF_EXPECTS(
102+
!this->orderings.empty(), "order_scheme: orderings must not be empty", std::invalid_argument);
103+
for (auto const& ordering : this->orderings) {
104+
RAPIDSMPF_EXPECTS(!ordering.keys.empty(),
105+
"order_scheme: ordering entries must not be empty",
106+
std::invalid_argument);
107+
}
108+
}
109+
110+
partitioning_spec partitioning_spec::from_order(order_scheme o)
111+
{
112+
return {.type = type::ORDER, .hash = std::nullopt, .order = std::move(o)};
113+
}
114+
88115
rapidsmpf::streaming::Message to_message(std::uint64_t sequence_number,
89116
std::unique_ptr<channel_metadata> m)
90117
{

cpp/libcudf_streaming/tests/streaming/test_channel_metadata.cpp

Lines changed: 36 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
/**
22
* SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
3-
* reserved. SPDX-License-Identifier: Apache-2.0
3+
* SPDX-License-Identifier: Apache-2.0
44
*/
55

66
#include <cudf_test/cudf_gtest.hpp>
@@ -50,6 +50,11 @@ TEST_F(StreamingChannelMetadata, OrderSchemeCtorRejectsNullBoundaries)
5050
std::invalid_argument);
5151
}
5252

53+
TEST_F(StreamingChannelMetadata, OrderSchemeCtorRejectsEmptyOrderings)
54+
{
55+
EXPECT_THROW(static_cast<void>(order_scheme(std::vector<ordering>{})), std::invalid_argument);
56+
}
57+
5358
TEST_F(StreamingChannelMetadata, PartitioningSpec)
5459
{
5560
// None
@@ -173,14 +178,14 @@ class StreamingChannelMetadataGPU : public ::testing::Test {
173178
}
174179
};
175180

176-
TEST_F(StreamingChannelMetadataGPU, OrderSchemeReplaceKeys)
181+
TEST_F(StreamingChannelMetadataGPU, OrderingReplaceKeys)
177182
{
178183
order_key k0{0, cudf::order::ASCENDING, cudf::null_order::BEFORE};
179184
order_key k5{5, cudf::order::DESCENDING, cudf::null_order::AFTER};
180185

181-
auto b = make_chunk({100, 200});
182-
order_scheme o1({k0}, b);
183-
auto o2 = o1.with_keys({k5});
186+
auto b = make_chunk({100, 200});
187+
ordering o1 = ordering({k0}, b);
188+
auto o2 = o1.with_keys({k5});
184189

185190
EXPECT_EQ(o2.keys[0].column_index, 5);
186191
EXPECT_EQ(o2.keys[0].order, cudf::order::DESCENDING);
@@ -192,22 +197,40 @@ TEST_F(StreamingChannelMetadataGPU, OrderSchemeReplaceKeys)
192197
EXPECT_THROW(static_cast<void>(o1.with_keys({k0, k5})), std::invalid_argument);
193198
}
194199

195-
TEST_F(StreamingChannelMetadataGPU, OrderSchemeBoundariesAlignedWith)
200+
TEST_F(StreamingChannelMetadataGPU, OrderSchemeMultipleOrderings)
201+
{
202+
order_key k0{0, cudf::order::ASCENDING, cudf::null_order::BEFORE};
203+
order_key k2{2, cudf::order::DESCENDING, cudf::null_order::AFTER};
204+
205+
auto b0 = make_chunk({100, 200});
206+
auto b1 = make_chunk({300, 400});
207+
order_scheme o({ordering{{k0}, b0, true}, ordering{{k2}, b1, false}});
208+
209+
ASSERT_EQ(o.orderings.size(), 2);
210+
EXPECT_EQ(o.orderings[0].keys[0], k0);
211+
EXPECT_EQ(o.orderings[0].boundaries.get(), b0.get());
212+
EXPECT_TRUE(o.orderings[0].strict_boundaries);
213+
EXPECT_EQ(o.orderings[1].keys[0], k2);
214+
EXPECT_EQ(o.orderings[1].boundaries.get(), b1.get());
215+
EXPECT_FALSE(o.orderings[1].strict_boundaries);
216+
}
217+
218+
TEST_F(StreamingChannelMetadataGPU, OrderingBoundariesAlignedWith)
196219
{
197220
order_key k0{0, cudf::order::ASCENDING, cudf::null_order::BEFORE};
198221
order_key k3{3, cudf::order::ASCENDING, cudf::null_order::BEFORE};
199222

200-
order_scheme o1({k0}, make_chunk({100, 200}));
201-
order_scheme o2({k0}, make_chunk({100, 200}));
223+
ordering o1({k0}, make_chunk({100, 200}));
224+
ordering o2({k0}, make_chunk({100, 200}));
202225
EXPECT_TRUE(o1.boundaries_aligned_with(o2, *br));
203226

204-
order_scheme o_shifted({k3}, make_chunk({100, 200}));
227+
ordering o_shifted({k3}, make_chunk({100, 200}));
205228
EXPECT_TRUE(o1.boundaries_aligned_with(o_shifted, *br));
206229

207-
order_scheme o_strict({k0}, make_chunk({100, 200}), /*strict=*/true);
230+
ordering o_strict({k0}, make_chunk({100, 200}), /*strict_boundaries=*/true);
208231
EXPECT_FALSE(o1.boundaries_aligned_with(o_strict, *br));
209232

210-
order_scheme o_diff({k0}, make_chunk({100, 300}));
233+
ordering o_diff({k0}, make_chunk({100, 300}));
211234
EXPECT_FALSE(o1.boundaries_aligned_with(o_diff, *br));
212235
}
213236

@@ -219,10 +242,10 @@ TEST_F(StreamingChannelMetadataGPU, PartitioningSpecOrder)
219242
auto spec = partitioning_spec::from_order(o);
220243
EXPECT_EQ(spec.type, partitioning_spec::type::ORDER);
221244
EXPECT_TRUE(spec.order.has_value());
222-
EXPECT_EQ(spec.order->keys[0].column_index, 0);
245+
EXPECT_EQ(spec.order->orderings[0].keys[0].column_index, 0);
223246

224247
// Type checks only (partitioning_spec::operator== removed; ORDER value comparison
225-
// requires boundaries_aligned_with on the order_scheme directly)
248+
// requires boundaries_aligned_with on the ordering directly)
226249
EXPECT_EQ(spec.type, partitioning_spec::type::ORDER);
227250
EXPECT_NE(spec.type, partitioning_spec::from_hash(hash_scheme{{0}, 16}).type);
228251
EXPECT_NE(spec.type, partitioning_spec::none().type);

0 commit comments

Comments
 (0)