diff --git a/cpp/include/cudf_test/base_fixture.hpp b/cpp/include/cudf_test/base_fixture.hpp index 003955369001..498c1efba5fd 100644 --- a/cpp/include/cudf_test/base_fixture.hpp +++ b/cpp/include/cudf_test/base_fixture.hpp @@ -66,7 +66,11 @@ struct BaseFixtureWithHarness : public BaseFixture { */ cudf::memory_resources resources() { return _harness.resources(); } - protected: + /** + * @brief Return the memory-resource harness used by this fixture. + */ + [[nodiscard]] memory_resource_test_harness& harness() noexcept { return _harness; } + memory_resource_test_harness _harness{mr()}; }; diff --git a/cpp/include/cudf_test/memory_resource_utilities.hpp b/cpp/include/cudf_test/memory_resource_utilities.hpp index 411b6a5c598a..aee54b3c127c 100644 --- a/cpp/include/cudf_test/memory_resource_utilities.hpp +++ b/cpp/include/cudf_test/memory_resource_utilities.hpp @@ -118,16 +118,28 @@ class memory_resource_test_harness { /** @brief Synchronize `stream` before leaving a failing-resource API scope. */ void synchronize(cuda::stream_ref stream = cudf::test::get_default_stream()) const; - /** @brief Assert that output allocations are live after synchronizing `stream`. */ - void expect_output_allocations_live( + /** + * @brief Assert that output allocations are live after synchronizing `stream`. + * + * @return Output-resource byte counters after the assertion + */ + rmm::mr::statistics_resource_adaptor::counter expect_output_allocations_live( cuda::stream_ref stream = cudf::test::get_default_stream()) const; - /** @brief Assert that temporary allocations were made after synchronizing `stream`. */ - void expect_temporary_allocation_activity( + /** + * @brief Assert that temporary allocations were made after synchronizing `stream`. + * + * @return Temporary-resource byte counters after the assertion + */ + rmm::mr::statistics_resource_adaptor::counter expect_temporary_allocation_activity( cuda::stream_ref stream = cudf::test::get_default_stream()) const; - /** @brief Assert that no temporary allocations remain live after synchronizing `stream`. */ - void expect_temporary_allocations_released( + /** + * @brief Assert that no temporary allocations remain live after synchronizing `stream`. + * + * @return Temporary-resource byte counters after the assertion + */ + rmm::mr::statistics_resource_adaptor::counter expect_temporary_allocations_released( cuda::stream_ref stream = cudf::test::get_default_stream()) const; /** diff --git a/cpp/include/cudf_test/tdigest_utilities.hpp b/cpp/include/cudf_test/tdigest_utilities.hpp index e6013c9abf9c..0e20e8bcae29 100644 --- a/cpp/include/cudf_test/tdigest_utilities.hpp +++ b/cpp/include/cudf_test/tdigest_utilities.hpp @@ -12,7 +12,6 @@ #include #include #include -#include #include #include @@ -20,6 +19,7 @@ #include #include +#include // for use with groupby and reduction aggregation tests. @@ -33,20 +33,16 @@ using expected_value = cuda::std::tuple; */ struct tdigest_gen { // @cond - template < - typename T, - typename Func, - typename std::enable_if_t() || cudf::is_fixed_point()>* = nullptr> + template std::unique_ptr operator()(Func op, column_view const& values, int delta) + requires(is_numeric() || is_fixed_point()) { return op(values, delta); } - template < - typename T, - typename Func, - typename std::enable_if_t() && !cudf::is_fixed_point()>* = nullptr> + template std::unique_ptr operator()(Func op, column_view const& values, int delta) + requires(!is_numeric() && !is_fixed_point()) { CUDF_FAIL("Invalid tdigest test type"); } @@ -65,12 +61,27 @@ inline T rand_range(T min, T max) return min + static_cast(frand() * (max - min)); } +/** + * @brief Generate a typed column from a bucketed percentile distribution. + * + * @param buckets Upper bound for each generated bucket + * @param sizes Number of values generated for each bucket + * @param t Data type of the returned column + * @param sorted Whether to sort the generated values before conversion + * @param stream CUDA stream used for device memory operations and kernel launches + * @param mr Memory resources used for returned and temporary allocations + * @return Generated column + */ inline std::unique_ptr generate_typed_percentile_distribution( std::vector const& buckets, std::vector const& sizes, data_type t, - bool sorted = false) + bool sorted = false, + cuda::stream_ref stream = get_default_stream(), + memory_resources mr = get_current_device_resource_ref()) { + auto const temporary_mr = mr.get_temporary_mr(); + auto const temporary_resources = memory_resources{temporary_mr, temporary_mr}; srand(0); std::vector values; @@ -87,8 +98,8 @@ inline std::unique_ptr generate_typed_percentile_distribution( if (sorted) { std::sort(values.begin(), values.end()); } - cudf::test::fixed_width_column_wrapper src(values.begin(), values.end()); - return cudf::cast(src, t); + fixed_width_column_wrapper src(values.begin(), values.end(), stream, temporary_resources); + return cast(src, t, stream, mr.get_output_mr()); } // "standardized" means the parameters sent into generate_typed_percentile_distribution. the intent @@ -96,31 +107,59 @@ inline std::unique_ptr generate_typed_percentile_distribution( // percentile_approx tests. std::vector // buckets{10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0}; std::vector // sizes{50000, 50000, 50000, 50000, 50000, 100000, 100000, 100000, 100000, 100000}; +/** + * @brief Generate the standardized percentile distribution used by T-digest tests. + * + * @param t Data type of the returned column + * @param sorted Whether to sort the generated values before conversion + * @param stream CUDA stream used for device memory operations and kernel launches + * @param mr Memory resources used for returned and temporary allocations + * @return Generated column + */ inline std::unique_ptr generate_standardized_percentile_distribution( - data_type t = data_type{type_id::FLOAT64}, bool sorted = false) + data_type t = data_type{type_id::FLOAT64}, + bool sorted = false, + cuda::stream_ref stream = get_default_stream(), + memory_resources mr = get_current_device_resource_ref()) { std::vector buckets{10.0f, 20.0f, 30.0f, 40.0f, 50.0f, 60.0f, 70.0f, 80.0, 90.0f, 100.0f}; std::vector b_sizes{ 50000, 50000, 50000, 50000, 50000, 100000, 100000, 100000, 100000, 100000}; - return generate_typed_percentile_distribution(buckets, b_sizes, t, sorted); + return generate_typed_percentile_distribution(buckets, b_sizes, t, sorted, stream, mr); } /** * @brief Compare a tdigest column against a sampling of expected values. + * + * @param tdv T-digest column to validate + * @param h_expected Expected centroid index, mean, and weight tuples + * @param stream CUDA stream used for device memory operations and kernel launches + * @param mr Memory resources used for temporary device allocations */ -void tdigest_sample_compare(cudf::tdigest::tdigest_column_view const& tdv, - std::vector const& h_expected); +void tdigest_sample_compare(tdigest::tdigest_column_view const& tdv, + std::vector const& h_expected, + cuda::stream_ref stream = get_default_stream(), + memory_resources mr = get_current_device_resource_ref()); /** * @brief Compare the min/max values of a tdigest against inputs. + * + * @tparam T Input element type + * @param tdv T-digest column to validate + * @param input_values Values whose extrema are expected in the T-digest + * @param stream CUDA stream used for device memory operations and kernel launches + * @param mr Memory resources used for temporary device allocations */ template -void tdigest_minmax_compare(cudf::tdigest::tdigest_column_view const& tdv, - cudf::column_view const& input_values) +void tdigest_minmax_compare(tdigest::tdigest_column_view const& tdv, + column_view const& input_values, + cuda::stream_ref stream = get_default_stream(), + memory_resources mr = get_current_device_resource_ref()) { - using ScalarType = cudf::scalar_type_t; + auto const temporary_mr = mr.get_temporary_mr(); + using ScalarType = scalar_type_t; - auto [col_min, col_max] = cudf::minmax(input_values); + auto [col_min, col_max] = minmax(input_values, stream, temporary_mr); auto min_scalar = static_cast(col_min.get()); auto max_scalar = static_cast(col_max.get()); @@ -146,109 +185,143 @@ struct expected_tdigest { /** * @brief Create an expected tdigest column given component inputs. + * + * @param groups T-digest component values for each output row + * @param stream CUDA stream used for device memory operations and kernel launches + * @param mr Memory resources used for returned and temporary allocations + * @return Expected T-digest column */ -std::unique_ptr make_expected_tdigest_column(std::vector const& groups); +std::unique_ptr make_expected_tdigest_column( + std::vector const& groups, + cuda::stream_ref stream = get_default_stream(), + memory_resources mr = get_current_device_resource_ref()); // shared test for groupby/reduction. template -void tdigest_simple_aggregation(Func op) +void tdigest_simple_aggregation(Func op, + cuda::stream_ref stream = get_default_stream(), + memory_resources mr = get_current_device_resource_ref()) { + auto const temporary_mr = mr.get_temporary_mr(); + auto const temporary_resources = memory_resources{temporary_mr, temporary_mr}; bool is_cpu_cluster_computation_disabled[2] = {true, false}; for (int idx = 0; idx < 2; idx++) { - cudf::tdigest::detail::is_cpu_cluster_computation_disabled = - is_cpu_cluster_computation_disabled[idx]; + tdigest::detail::is_cpu_cluster_computation_disabled = is_cpu_cluster_computation_disabled[idx]; // create a tdigest that has far fewer values in it than the delta value. this should result // in every value remaining uncompressed - cudf::test::fixed_width_column_wrapper values{126, 15, 1, 99, 67}; + fixed_width_column_wrapper values( + {126, 15, 1, 99, 67}, rmm::cuda_stream_view{stream}, temporary_resources); int const delta = 1000; - auto result = cudf::type_dispatcher( - static_cast(values).type(), tdigest_gen{}, op, values, delta); - - cudf::test::fixed_width_column_wrapper raw_mean({1, 15, 67, 99, 126}); - cudf::test::fixed_width_column_wrapper weight{1, 1, 1, 1, 1}; - auto mean = cudf::cast(raw_mean, data_type{type_id::FLOAT64}); + auto result = + type_dispatcher(static_cast(values).type(), tdigest_gen{}, op, values, delta); + + fixed_width_column_wrapper raw_mean( + {1, 15, 67, 99, 126}, rmm::cuda_stream_view{stream}, temporary_resources); + fixed_width_column_wrapper weight( + {1, 1, 1, 1, 1}, rmm::cuda_stream_view{stream}, temporary_resources); + auto mean = cast(raw_mean, data_type{type_id::FLOAT64}, stream, temporary_mr); double const min = 1; double const max = 126; auto expected = make_expected_tdigest_column({{*mean, weight, static_cast(static_cast(min)), - static_cast(static_cast(max))}}); + static_cast(static_cast(max))}}, + stream, + temporary_resources); - CUDF_TEST_EXPECT_COLUMNS_EQUAL(*result, *expected); + CUDF_TEST_EXPECT_COLUMNS_EQUAL(*result, *expected, debug_output_level::FIRST_ERROR, stream, mr); } } // shared test for groupby/reduction. template -void tdigest_simple_with_nulls_aggregation(Func op) +void tdigest_simple_with_nulls_aggregation(Func op, + cuda::stream_ref stream = get_default_stream(), + memory_resources mr = get_current_device_resource_ref()) { + auto const temporary_mr = mr.get_temporary_mr(); + auto const temporary_resources = memory_resources{temporary_mr, temporary_mr}; bool is_cpu_cluster_computation_disabled[2] = {true, false}; for (int idx = 0; idx < 2; idx++) { - cudf::tdigest::detail::is_cpu_cluster_computation_disabled = - is_cpu_cluster_computation_disabled[idx]; + tdigest::detail::is_cpu_cluster_computation_disabled = is_cpu_cluster_computation_disabled[idx]; // create a tdigest that has far fewer values in it than the delta value. this should result // in every value remaining uncompressed - cudf::test::fixed_width_column_wrapper values{{122, 15, 1, 99, 67, 101, 100, 84, 44, 2}, - {1, 0, 1, 0, 1, 0, 1, 0, 1, 0}}; + fixed_width_column_wrapper values({122, 15, 1, 99, 67, 101, 100, 84, 44, 2}, + {1, 0, 1, 0, 1, 0, 1, 0, 1, 0}, + rmm::cuda_stream_view{stream}, + temporary_resources); int const delta = 1000; - auto result = cudf::type_dispatcher( - static_cast(values).type(), tdigest_gen{}, op, values, delta); - - cudf::test::fixed_width_column_wrapper raw_mean({1, 44, 67, 100, 122}); - cudf::test::fixed_width_column_wrapper weight{1, 1, 1, 1, 1}; - auto mean = cudf::cast(raw_mean, data_type{type_id::FLOAT64}); + auto result = + type_dispatcher(static_cast(values).type(), tdigest_gen{}, op, values, delta); + + fixed_width_column_wrapper raw_mean( + {1, 44, 67, 100, 122}, rmm::cuda_stream_view{stream}, temporary_resources); + fixed_width_column_wrapper weight( + {1, 1, 1, 1, 1}, rmm::cuda_stream_view{stream}, temporary_resources); + auto mean = cast(raw_mean, data_type{type_id::FLOAT64}, stream, temporary_mr); double const min = 1; double const max = 122; auto expected = make_expected_tdigest_column({{*mean, weight, static_cast(static_cast(min)), - static_cast(static_cast(max))}}); + static_cast(static_cast(max))}}, + stream, + temporary_resources); - CUDF_TEST_EXPECT_COLUMNS_EQUAL(*result, *expected); + CUDF_TEST_EXPECT_COLUMNS_EQUAL(*result, *expected, debug_output_level::FIRST_ERROR, stream, mr); } } // shared test for groupby/reduction. template -void tdigest_simple_all_nulls_aggregation(Func op) +void tdigest_simple_all_nulls_aggregation(Func op, + cuda::stream_ref stream = get_default_stream(), + memory_resources mr = get_current_device_resource_ref()) { + auto const temporary_mr = mr.get_temporary_mr(); + auto const temporary_resources = memory_resources{temporary_mr, temporary_mr}; bool is_cpu_cluster_computation_disabled[2] = {true, false}; for (int idx = 0; idx < 2; idx++) { - cudf::tdigest::detail::is_cpu_cluster_computation_disabled = - is_cpu_cluster_computation_disabled[idx]; + tdigest::detail::is_cpu_cluster_computation_disabled = is_cpu_cluster_computation_disabled[idx]; // create a tdigest that has far fewer values in it than the delta value. this should result // in every value remaining uncompressed - cudf::test::fixed_width_column_wrapper values{{122, 15, 1, 99, 67, 101, 100, 84, 44, 2}, - {0, 0, 0, 0, 0, 0, 0, 0, 0, 0}}; + fixed_width_column_wrapper values({122, 15, 1, 99, 67, 101, 100, 84, 44, 2}, + {0, 0, 0, 0, 0, 0, 0, 0, 0, 0}, + rmm::cuda_stream_view{stream}, + temporary_resources); int const delta = 1000; - auto result = cudf::type_dispatcher( - static_cast(values).type(), tdigest_gen{}, op, values, delta); + auto result = + type_dispatcher(static_cast(values).type(), tdigest_gen{}, op, values, delta); // NOTE: an empty tdigest column still has 1 row. - auto expected = cudf::tdigest::detail::make_empty_tdigests_column( - 1, cudf::get_default_stream(), cudf::get_current_device_resource_ref()); + auto expected = tdigest::detail::make_empty_tdigests_column(1, stream, temporary_mr); - CUDF_TEST_EXPECT_COLUMNS_EQUAL(*result, *expected); + CUDF_TEST_EXPECT_COLUMNS_EQUAL(*result, *expected, debug_output_level::FIRST_ERROR, stream, mr); } } // Note: there is no need to test different types here as the internals of a tdigest are always // the same regardless of input. template -void tdigest_merge_simple(Func op, MergeFunc merge_op) +void tdigest_merge_simple(Func op, + MergeFunc merge_op, + cuda::stream_ref stream = get_default_stream(), + memory_resources mr = get_current_device_resource_ref()) { + auto const temporary_mr = mr.get_temporary_mr(); + auto const temporary_resources = memory_resources{temporary_mr, temporary_mr}; bool is_cpu_cluster_computation_disabled[2] = {true, false}; for (int idx = 0; idx < 2; idx++) { - cudf::tdigest::detail::is_cpu_cluster_computation_disabled = - is_cpu_cluster_computation_disabled[idx]; + tdigest::detail::is_cpu_cluster_computation_disabled = is_cpu_cluster_computation_disabled[idx]; - auto values = generate_standardized_percentile_distribution(data_type{type_id::FLOAT64}); + auto values = generate_standardized_percentile_distribution( + data_type{type_id::FLOAT64}, false, stream, temporary_resources); CUDF_EXPECTS(values->size() == 750000, "Unexpected distribution size"); - auto split_values = cudf::split(*values, {250000, 500000}); + auto split_values = split(*values, {250000, 500000}, stream); int const delta = 1000; @@ -270,9 +343,9 @@ void tdigest_merge_simple(Func op, MergeFunc merge_op) int const merge_delta = 1000; // merge them - auto merge_input = cudf::concatenate(part_views); + auto merge_input = cudf::concatenate(part_views, stream, temporary_mr); auto result = merge_op(*merge_input, merge_delta); - cudf::tdigest::tdigest_column_view tdv(*result); + tdigest::tdigest_column_view tdv(*result); // verify centroids std::vector expected{{0, 0.00013945158577498588, 2}, @@ -288,43 +361,41 @@ void tdigest_merge_simple(Func op, MergeFunc merge_op) {625, 98.20470345147104751504, 405}, {700, 99.96818381983835877236, 56}, {711, 99.99970905482754801596, 1}}; - tdigest_sample_compare(tdv, expected); + tdigest_sample_compare(tdv, expected, stream, mr); // verify min/max - tdigest_minmax_compare(tdv, *values); + tdigest_minmax_compare(tdv, *values, stream, mr); } } } // shared test for groupby/reduction. template -void tdigest_merge_empty(MergeFunc merge_op) +void tdigest_merge_empty(MergeFunc merge_op, + cuda::stream_ref stream = get_default_stream(), + memory_resources mr = get_current_device_resource_ref()) { + auto const temporary_mr = mr.get_temporary_mr(); bool is_cpu_cluster_computation_disabled[2] = {true, false}; for (int idx = 0; idx < 2; idx++) { - cudf::tdigest::detail::is_cpu_cluster_computation_disabled = - is_cpu_cluster_computation_disabled[idx]; + tdigest::detail::is_cpu_cluster_computation_disabled = is_cpu_cluster_computation_disabled[idx]; // 3 empty tdigests all in the same group - auto a = cudf::tdigest::detail::make_empty_tdigests_column( - 1, cudf::get_default_stream(), cudf::get_current_device_resource_ref()); - auto b = cudf::tdigest::detail::make_empty_tdigests_column( - 1, cudf::get_default_stream(), cudf::get_current_device_resource_ref()); - auto c = cudf::tdigest::detail::make_empty_tdigests_column( - 1, cudf::get_default_stream(), cudf::get_current_device_resource_ref()); + auto a = tdigest::detail::make_empty_tdigests_column(1, stream, temporary_mr); + auto b = tdigest::detail::make_empty_tdigests_column(1, stream, temporary_mr); + auto c = tdigest::detail::make_empty_tdigests_column(1, stream, temporary_mr); std::vector cols; cols.push_back(*a); cols.push_back(*b); cols.push_back(*c); - auto values = cudf::concatenate(cols); + auto values = cudf::concatenate(cols, stream, temporary_mr); auto const delta = 1000; auto result = merge_op(*values, delta); - auto expected = cudf::tdigest::detail::make_empty_tdigests_column( - 1, cudf::get_default_stream(), cudf::get_current_device_resource_ref()); + auto expected = tdigest::detail::make_empty_tdigests_column(1, stream, temporary_mr); - CUDF_TEST_EXPECT_COLUMNS_EQUAL(*expected, *result); + CUDF_TEST_EXPECT_COLUMNS_EQUAL(*expected, *result, debug_output_level::FIRST_ERROR, stream, mr); } } diff --git a/cpp/tests/reductions/tdigest_tests.cpp b/cpp/tests/reductions/tdigest_tests.cpp index 9ed6fd4eb95b..384462c4cdb0 100644 --- a/cpp/tests/reductions/tdigest_tests.cpp +++ b/cpp/tests/reductions/tdigest_tests.cpp @@ -1,10 +1,11 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2022-2025, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ #include #include +#include #include #include @@ -70,84 +71,139 @@ TYPED_TEST(ReductionTDigestAllTypes, AllNull) cudf::test::tdigest_simple_all_nulls_aggregation(reduce_op{}); } -struct ReductionTDigestMerge : public cudf::test::BaseFixture {}; +struct ReductionTDigestMerge : public cudf::test::BaseFixtureWithHarness {}; TEST_F(ReductionTDigestMerge, Simple) { cudf::test::tdigest_merge_simple(reduce_op{}, reduce_merge_op{}); } +TEST_F(ReductionTDigestMerge, TestUtilityMemoryResourceControl) +{ + auto stream = this->stream(); + auto& harness = this->harness(); + + // generate_typed_percentile_distribution: output lives on output MR, temps are released. + // Note: do not install a failing current resource here; cast still routes Thrust scratch + // through get_current_device_resource_ref(). + { + auto distribution = cudf::test::generate_typed_percentile_distribution( + {10.0}, {4}, cudf::data_type{cudf::type_id::FLOAT64}, false, stream, harness.resources()); + harness.synchronize(stream); + harness.expect_output_allocations_live(stream); + harness.expect_temporary_allocation_activity(stream); + harness.expect_temporary_allocations_released(stream); + } + harness.expect_no_live_allocations(stream); + + // Inputs stay on setup_mr so they do not affect output/temporary live-byte checks + cudf::test::fixed_width_column_wrapper means({1.0, 2.0}, stream, harness.setup_mr()); + cudf::test::fixed_width_column_wrapper weights({1.0, 1.0}, stream, harness.setup_mr()); + + { + auto expected = cudf::test::make_expected_tdigest_column( + {{means, weights, 1.0, 2.0}}, stream, harness.resources()); + auto const output_bytes_before = harness.expect_output_allocations_live(stream); + harness.expect_temporary_allocation_activity(stream); + auto const temporary_bytes_before = harness.expect_temporary_allocations_released(stream); + + cudf::tdigest::tdigest_column_view tdv(*expected); + cudf::test::tdigest_sample_compare( + tdv, {{0, 1.0, 1.0}, {1, 2.0, 1.0}}, stream, harness.resources()); + cudf::test::tdigest_minmax_compare(tdv, means, stream, harness.resources()); + + // Compare helpers must not allocate output; only temporary traffic + harness.synchronize(stream); + EXPECT_EQ(harness.output_mr().get_bytes_counter().value, output_bytes_before.value); + EXPECT_EQ(harness.output_mr().get_bytes_counter().total, output_bytes_before.total); + EXPECT_GT(harness.temporary_mr().get_bytes_counter().total, temporary_bytes_before.total); + harness.expect_temporary_allocations_released(stream); + } + EXPECT_GT(harness.setup_mr().get_bytes_counter().value, 0); +} + // tests an issue with the cluster generating code with a small number of centroids that have large // weights TEST_F(ReductionTDigestMerge, FewHeavyCentroids) { + auto stream = this->stream(); + auto mr = this->resources(); + auto setup_mr = cudf::memory_resources{this->harness().setup_mr()}; + // digest 1 - cudf::test::fixed_width_column_wrapper c0c{1.0, 2.0}; - cudf::test::fixed_width_column_wrapper c0w{100.0, 50.0}; - cudf::test::structs_column_wrapper c0s({c0c, c0w}); - cudf::test::fixed_width_column_wrapper c0_offsets{0, 2}; + cudf::test::fixed_width_column_wrapper c0c({1.0, 2.0}, stream, setup_mr); + cudf::test::fixed_width_column_wrapper c0w({100.0, 50.0}, stream, setup_mr); + cudf::test::structs_column_wrapper c0s({c0c, c0w}, {}, stream, setup_mr); + cudf::test::fixed_width_column_wrapper c0_offsets({0, 2}, stream, setup_mr); auto c0l = cudf::make_lists_column(1, c0_offsets.release(), c0s.release(), 0, rmm::device_buffer{}); - cudf::test::fixed_width_column_wrapper c0min{1.0}; - cudf::test::fixed_width_column_wrapper c0max{2.0}; + cudf::test::fixed_width_column_wrapper c0min({1.0}, stream, setup_mr); + cudf::test::fixed_width_column_wrapper c0max({2.0}, stream, setup_mr); std::vector> c0_children; c0_children.push_back(std::move(c0l)); c0_children.push_back(c0min.release()); c0_children.push_back(c0max.release()); // tdigest struct - auto c0 = cudf::make_structs_column(1, std::move(c0_children), 0, {}); + auto c0 = + cudf::make_structs_column(1, std::move(c0_children), 0, {}, stream, setup_mr.get_output_mr()); cudf::tdigest::tdigest_column_view tdv0(*c0); // digest 2 - cudf::test::fixed_width_column_wrapper c1c{3.0, 4.0}; - cudf::test::fixed_width_column_wrapper c1w{200.0, 50.0}; - cudf::test::structs_column_wrapper c1s({c1c, c1w}); - cudf::test::fixed_width_column_wrapper c1_offsets{0, 2}; + cudf::test::fixed_width_column_wrapper c1c({3.0, 4.0}, stream, setup_mr); + cudf::test::fixed_width_column_wrapper c1w({200.0, 50.0}, stream, setup_mr); + cudf::test::structs_column_wrapper c1s({c1c, c1w}, {}, stream, setup_mr); + cudf::test::fixed_width_column_wrapper c1_offsets({0, 2}, stream, setup_mr); auto c1l = cudf::make_lists_column(1, c1_offsets.release(), c1s.release(), 0, rmm::device_buffer{}); - cudf::test::fixed_width_column_wrapper c1min{3.0}; - cudf::test::fixed_width_column_wrapper c1max{4.0}; + cudf::test::fixed_width_column_wrapper c1min({3.0}, stream, setup_mr); + cudf::test::fixed_width_column_wrapper c1max({4.0}, stream, setup_mr); std::vector> c1_children; c1_children.push_back(std::move(c1l)); c1_children.push_back(c1min.release()); c1_children.push_back(c1max.release()); // tdigest struct - auto c1 = cudf::make_structs_column(1, std::move(c1_children), 0, {}); + auto c1 = + cudf::make_structs_column(1, std::move(c1_children), 0, {}, stream, setup_mr.get_output_mr()); std::vector views; views.push_back(*c0); views.push_back(*c1); - auto values = cudf::concatenate(views); + auto values = cudf::concatenate(views, stream, setup_mr.get_output_mr()); // merge auto scalar_result = cudf::reduce(*values, *cudf::make_merge_tdigest_aggregation(1000), - cudf::data_type{cudf::type_id::STRUCT}); + cudf::data_type{cudf::type_id::STRUCT}, + stream, + mr.get_output_mr()); // convert to a table auto tbl = static_cast(scalar_result.get())->view(); std::vector> cols; std::transform( - tbl.begin(), tbl.end(), std::back_inserter(cols), [](cudf::column_view const& col) { - return std::make_unique(col); + tbl.begin(), tbl.end(), std::back_inserter(cols), [&](cudf::column_view const& col) { + return std::make_unique(col, stream, mr.get_output_mr()); }); - auto result = cudf::make_structs_column(tbl.num_rows(), std::move(cols), 0, rmm::device_buffer()); + auto result = cudf::make_structs_column( + tbl.num_rows(), std::move(cols), 0, rmm::device_buffer(), stream, mr.get_output_mr()); // we expect to see exactly 4 centroids (the same inputs) with properly computed min/max. - cudf::test::fixed_width_column_wrapper ec{1.0, 2.0, 3.0, 4.0}; - cudf::test::fixed_width_column_wrapper ew{100.0, 50.0, 200.0, 50.0}; - cudf::test::structs_column_wrapper es({ec, ew}); - cudf::test::fixed_width_column_wrapper e_offsets{0, 4}; + cudf::test::fixed_width_column_wrapper ec({1.0, 2.0, 3.0, 4.0}, stream, setup_mr); + cudf::test::fixed_width_column_wrapper ew({100.0, 50.0, 200.0, 50.0}, stream, setup_mr); + cudf::test::structs_column_wrapper es({ec, ew}, {}, stream, setup_mr); + cudf::test::fixed_width_column_wrapper e_offsets({0, 4}, stream, setup_mr); auto el = cudf::make_lists_column(1, e_offsets.release(), es.release(), 0, rmm::device_buffer{}); - cudf::test::fixed_width_column_wrapper emin{1.0}; - cudf::test::fixed_width_column_wrapper emax{4.0}; + cudf::test::fixed_width_column_wrapper emin({1.0}, stream, setup_mr); + cudf::test::fixed_width_column_wrapper emax({4.0}, stream, setup_mr); std::vector> e_children; e_children.push_back(std::move(el)); e_children.push_back(emin.release()); e_children.push_back(emax.release()); // tdigest struct - auto expected = cudf::make_structs_column(1, std::move(e_children), 0, {}); + auto expected = + cudf::make_structs_column(1, std::move(e_children), 0, {}, stream, setup_mr.get_output_mr()); - CUDF_TEST_EXPECT_COLUMNS_EQUAL(*result, *expected); + CUDF_TEST_EXPECT_COLUMNS_EQUAL( + *result, *expected, cudf::test::debug_output_level::FIRST_ERROR, stream, mr); } diff --git a/cpp/tests/utilities/memory_resource_utilities.cpp b/cpp/tests/utilities/memory_resource_utilities.cpp index 06e57bb027a4..aba54157e6c2 100644 --- a/cpp/tests/utilities/memory_resource_utilities.cpp +++ b/cpp/tests/utilities/memory_resource_utilities.cpp @@ -65,24 +65,31 @@ scoped_current_device_resource memory_resource_test_harness::fail_on_current_dev void memory_resource_test_harness::synchronize(cuda::stream_ref stream) const { stream.sync(); } -void memory_resource_test_harness::expect_output_allocations_live(cuda::stream_ref stream) const +rmm::mr::statistics_resource_adaptor::counter +memory_resource_test_harness::expect_output_allocations_live(cuda::stream_ref stream) const { synchronize(stream); - EXPECT_GT(_output_mr.get_bytes_counter().value, 0); + auto const counter = _output_mr.get_bytes_counter(); + EXPECT_GT(counter.value, 0); + return counter; } -void memory_resource_test_harness::expect_temporary_allocation_activity( - cuda::stream_ref stream) const +rmm::mr::statistics_resource_adaptor::counter +memory_resource_test_harness::expect_temporary_allocation_activity(cuda::stream_ref stream) const { synchronize(stream); - EXPECT_GT(_temporary_mr.get_bytes_counter().total, 0); + auto const counter = _temporary_mr.get_bytes_counter(); + EXPECT_GT(counter.total, 0); + return counter; } -void memory_resource_test_harness::expect_temporary_allocations_released( - cuda::stream_ref stream) const +rmm::mr::statistics_resource_adaptor::counter +memory_resource_test_harness::expect_temporary_allocations_released(cuda::stream_ref stream) const { synchronize(stream); - EXPECT_EQ(_temporary_mr.get_bytes_counter().value, 0); + auto const counter = _temporary_mr.get_bytes_counter(); + EXPECT_EQ(counter.value, 0); + return counter; } void memory_resource_test_harness::expect_resource_usage(std::size_t expected_output_bytes, diff --git a/cpp/tests/utilities/tdigest_utilities.cpp b/cpp/tests/utilities/tdigest_utilities.cpp index 0195edf91b72..cb40f52af5a3 100644 --- a/cpp/tests/utilities/tdigest_utilities.cpp +++ b/cpp/tests/utilities/tdigest_utilities.cpp @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ @@ -12,7 +12,6 @@ #include #include #include -#include #include #include @@ -24,9 +23,12 @@ namespace cudf { namespace test { -void tdigest_sample_compare(cudf::tdigest::tdigest_column_view const& tdv, - std::vector const& h_expected) +void tdigest_sample_compare(tdigest::tdigest_column_view const& tdv, + std::vector const& h_expected, + cuda::stream_ref stream, + memory_resources mr) { + auto const temporary_mr = mr.get_temporary_mr(); column_view result_mean = tdv.means(); column_view result_weight = tdv.weights(); @@ -44,49 +46,62 @@ void tdigest_sample_compare(cudf::tdigest::tdigest_column_view const& tdv, return cuda::std::get<2>(ex); }); - auto d_expected_src = cudf::detail::make_device_uvector_async( - h_expected_src, cudf::get_default_stream(), cudf::get_current_device_resource_ref()); - auto d_expected_mean = cudf::detail::make_device_uvector_async( - h_expected_mean, cudf::get_default_stream(), cudf::get_current_device_resource_ref()); - auto d_expected_weight = cudf::detail::make_device_uvector_async( - h_expected_weight, cudf::get_default_stream(), cudf::get_current_device_resource_ref()); - - auto map = cudf::device_span(d_expected_src); - auto sampled_result_mean = - std::move(cudf::gather(cudf::table_view({result_mean}), map)->release().front()); - auto sampled_result_weight = - std::move(cudf::gather(cudf::table_view({result_weight}), map)->release().front()); - - auto expected_mean = cudf::device_span(d_expected_mean); - auto expected_weight = cudf::device_span(d_expected_weight); - CUDF_TEST_EXPECT_COLUMNS_EQUIVALENT(expected_mean, *sampled_result_mean); - CUDF_TEST_EXPECT_COLUMNS_EQUAL(expected_weight, *sampled_result_weight); + auto d_expected_src = + cudf::detail::make_device_uvector_async(h_expected_src, stream, temporary_mr); + auto d_expected_mean = + cudf::detail::make_device_uvector_async(h_expected_mean, stream, temporary_mr); + auto d_expected_weight = + cudf::detail::make_device_uvector_async(h_expected_weight, stream, temporary_mr); + + auto map = device_span(d_expected_src); + auto sampled_result_mean = std::move( + gather(table_view({result_mean}), map, out_of_bounds_policy::DONT_CHECK, stream, temporary_mr) + ->release() + .front()); + auto sampled_result_weight = std::move( + gather(table_view({result_weight}), map, out_of_bounds_policy::DONT_CHECK, stream, temporary_mr) + ->release() + .front()); + + auto expected_mean = device_span(d_expected_mean); + auto expected_weight = device_span(d_expected_weight); + CUDF_TEST_EXPECT_COLUMNS_EQUIVALENT( + expected_mean, *sampled_result_mean, debug_output_level::FIRST_ERROR, default_ulp, stream, mr); + CUDF_TEST_EXPECT_COLUMNS_EQUAL( + expected_weight, *sampled_result_weight, debug_output_level::FIRST_ERROR, stream, mr); } -std::unique_ptr make_expected_tdigest_column(std::vector const& groups) +std::unique_ptr make_expected_tdigest_column(std::vector const& groups, + cuda::stream_ref stream, + memory_resources mr) { + auto const temporary_mr = mr.get_temporary_mr(); + auto const temporary_resources = memory_resources{temporary_mr, temporary_mr}; std::vector> tdigests; // make an individual digest auto make_digest = [&](expected_tdigest const& tdigest) { std::vector> inner_children; - inner_children.push_back(std::make_unique(tdigest.mean)); - inner_children.push_back(std::make_unique(tdigest.weight)); + inner_children.push_back(std::make_unique(tdigest.mean, stream, temporary_mr)); + inner_children.push_back(std::make_unique(tdigest.weight, stream, temporary_mr)); // tdigest struct - auto tdigests = - cudf::make_structs_column(tdigest.mean.size(), std::move(inner_children), 0, {}); + auto tdigests = make_structs_column( + tdigest.mean.size(), std::move(inner_children), 0, {}, stream, temporary_mr); - auto offsets = cudf::test::fixed_width_column_wrapper({0, tdigest.mean.size()}); - auto list = cudf::make_lists_column(1, offsets.release(), std::move(tdigests), 0, {}); + auto offsets = fixed_width_column_wrapper( + {0, tdigest.mean.size()}, rmm::cuda_stream_view{stream}, temporary_resources); + auto list = make_lists_column(1, offsets.release(), std::move(tdigests), 0, {}); - auto min_col = cudf::test::fixed_width_column_wrapper({tdigest.min}); - auto max_col = cudf::test::fixed_width_column_wrapper({tdigest.max}); + auto min_col = fixed_width_column_wrapper( + {tdigest.min}, rmm::cuda_stream_view{stream}, temporary_resources); + auto max_col = fixed_width_column_wrapper( + {tdigest.max}, rmm::cuda_stream_view{stream}, temporary_resources); std::vector> children; children.push_back(std::move(list)); children.push_back(min_col.release()); children.push_back(max_col.release()); - return make_structs_column(1, std::move(children), 0, {}); + return make_structs_column(1, std::move(children), 0, {}, stream, temporary_mr); }; // build the individual digests @@ -99,7 +114,7 @@ std::unique_ptr make_expected_tdigest_column(std::vector const& c) { return c->view(); }); - return cudf::concatenate(views); + return cudf::concatenate(views, stream, mr.get_output_mr()); } } // namespace test