diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt index 77d06cf4d0f6..22ffeebbabfc 100644 --- a/cpp/CMakeLists.txt +++ b/cpp/CMakeLists.txt @@ -737,10 +737,11 @@ add_library( src/groupby/common/null_mask.cu src/groupby/common/utils.cpp src/groupby/groupby.cu - src/groupby/hash/compute_groupby.cu + src/groupby/hash/compute_groupby.cpp src/groupby/hash/compute_single_pass_aggs.cpp src/groupby/hash/compute_single_pass_aggs.cu src/groupby/hash/extract_single_pass_aggs.cpp + src/groupby/hash/group_keys.cu src/groupby/hash/groupby.cu src/groupby/hash/hash_compound_agg_finalizer.cu src/groupby/hash/output_utils.cu diff --git a/cpp/src/groupby/hash/compute_groupby.cpp b/cpp/src/groupby/hash/compute_groupby.cpp new file mode 100644 index 000000000000..5e9689e70eaf --- /dev/null +++ b/cpp/src/groupby/hash/compute_groupby.cpp @@ -0,0 +1,171 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include "compute_groupby.hpp" + +#include "compute_single_pass_aggs.hpp" +#include "extract_single_pass_aggs.hpp" +#include "group_keys.hpp" +#include "hash_compound_agg_finalizer.hpp" + +#include +#include +#include +#include + +#include +#include + +#include +#include +#include +#include +#include + +namespace cudf::groupby::detail::hash { + +namespace { + +// The result cache shares results across requests on the same values column. Normalize the +// extracted reductions with the same equality so an earlier compound request cannot choose the +// resource or nullability of a later requested result, or allocate a result that the cache drops. +auto extract_hash_groupby_aggs(std::span requests, + cuda::stream_ref stream) +{ + if (requests.size() <= 1) { return extract_single_pass_aggs(requests, stream); } + + using aggregation_set = + std::unordered_set>, + cudf::detail::pair_column_aggregation_hash, + cudf::detail::pair_column_aggregation_equal_to>; + aggregation_set requested; + for (auto const& request : requests) { + for (auto const& agg : request.aggregations) { + requested.emplace(request.values, *agg); + } + } + + auto [values, kinds, aggs, is_intermediate, has_compound] = + extract_single_pass_aggs(requests, stream); + aggregation_set extracted; + std::vector unique_values; + unique_values.reserve(aggs.size()); + for (std::size_t i = 0; i < aggs.size(); ++i) { + auto const key = + std::pair>{values.column(i), *aggs[i]}; + if (!extracted.insert(key).second) { continue; } + auto const output = unique_values.size(); + kinds[output] = kinds[i]; + is_intermediate[output] = !requested.contains(key); + if (output != i) { aggs[output] = std::move(aggs[i]); } + unique_values.push_back(values.column(i)); + } + kinds.resize(unique_values.size()); + aggs.resize(unique_values.size()); + is_intermediate.resize(unique_values.size()); + return std::tuple{table_view{unique_values}, + std::move(kinds), + std::move(aggs), + std::move(is_intermediate), + has_compound}; +} + +} // namespace + +std::unique_ptr compute_groupby(table_view const& keys, + std::span requests, + null_policy include_null_keys, + cudf::detail::result_cache* cache, + cuda::stream_ref stream, + cudf::memory_resources mr) +{ + auto const temp_mr = mr.get_temporary_mr(); + auto const temporary_resources = mr.as_temporary(); + + // Determine which grouping outputs the requested aggregations need. + auto const [values, agg_kinds, aggs, is_agg_intermediate, has_compound_aggs] = + extract_hash_groupby_aggs(requests, stream); + + // Counts without null filtering come directly from the group offsets. + auto const needs_reduction = [&] { + for (size_type i = 0; i < values.num_columns(); ++i) { + if (agg_kinds[i] != aggregation::COUNT_ALL && + (agg_kinds[i] != aggregation::COUNT_VALID || values.column(i).has_nulls())) { + return true; + } + } + return false; + }(); + auto const groups = group_keys( + keys, include_null_keys, !requests.empty(), needs_reduction, stream, temporary_resources); + + auto const gather_keys = [&] { + return cudf::detail::gather(keys, + groups.key_indices, + out_of_bounds_policy::DONT_CHECK, + cudf::negative_index_policy::NOT_ALLOWED, + stream, + mr); + }; + + // In case of no requests, we still need to generate a set of unique keys. + if (requests.empty()) { return gather_keys(); } + + auto const rows = groups.grouped_rows ? device_span{groups.grouped_rows->data(), + groups.grouped_rows->size()} + : device_span{}; + auto const offsets = + device_span{groups.group_offsets->data(), groups.group_offsets->size()}; + auto const grouped = + needs_reduction + ? make_grouped_rows(rows, offsets, stream, temporary_resources) + : grouped_rows{rows, + offsets, + cuda::device_buffer{stream, temp_mr}, + cuda::device_buffer{stream, temp_mr}, + cuda::device_buffer>{stream, temp_mr}, + cuda::device_buffer{stream, temp_mr}}; + auto results = + compute_single_pass_aggs(values, agg_kinds, is_agg_intermediate, grouped, stream, mr); + for (std::size_t i = 0; i < results.size(); ++i) { + cache->add_result(values.column(i), *aggs[i], std::move(results[i])); + } + + if (has_compound_aggs) { + // Requested M2 results must be cached on the output resource before VARIANCE or STD asks + // for an intermediate M2, regardless of the order of requests on a shared values column. + for (auto const& request : requests) { + auto const finalizer = hash_compound_agg_finalizer(request.values, cache, stream, mr); + for (auto const& agg : request.aggregations) { + if (agg->kind == aggregation::M2) { + cudf::detail::aggregation_dispatcher(agg->kind, finalizer, *agg); + } + } + } + for (auto const& request : requests) { + auto const& agg_v = request.aggregations; + auto const& col = request.values; + + // The finalizers only combine the single-pass results with linear transformations such as + // addition/multiplication (e.g. for variance/stddev); they do not aggregate further. + auto const finalizer = hash_compound_agg_finalizer(col, cache, stream, mr); + for (auto&& agg : agg_v) { + if (agg->kind == aggregation::VARIANCE || agg->kind == aggregation::STD) { + // Explicit M2 outputs were finalized above. Any missing M2 is only an intermediate + // for this ordinary groupby; the shared finalizer also serves streaming groupby. + auto const m2_agg = make_m2_aggregation(); + auto const m2_finalizer = + hash_compound_agg_finalizer(col, cache, stream, temporary_resources); + cudf::detail::aggregation_dispatcher(m2_agg->kind, m2_finalizer, *m2_agg); + } + cudf::detail::aggregation_dispatcher(agg->kind, finalizer, *agg); + } + } + } + + return gather_keys(); +} + +} // namespace cudf::groupby::detail::hash diff --git a/cpp/src/groupby/hash/compute_groupby.hpp b/cpp/src/groupby/hash/compute_groupby.hpp index 19114d0773a5..e4498f9db16c 100644 --- a/cpp/src/groupby/hash/compute_groupby.hpp +++ b/cpp/src/groupby/hash/compute_groupby.hpp @@ -27,26 +27,18 @@ namespace cudf::groupby::detail::hash { * of kernels. Then using these results, compound aggregations that require multiple passes will be * computed on top of them. All results are stored into the in/out parameter `cache`. * - * @tparam Equal Device row comparator type - * @tparam Hash Device row hasher type - * * @param keys Table whose rows act as the groupby keys * @param requests The set of columns to aggregate and the aggregations to perform - * @param skip_rows_with_nulls Flag indicating whether to ignore nulls or not - * @param d_row_equal Device row comparator - * @param d_row_hash Device row hasher + * @param include_null_keys Whether rows containing null keys participate in grouping * @param cache Dense aggregation results * @param stream CUDA stream used for device memory operations and kernel launches * @param mr Device memory resources used for the returned keys, aggregation results, and temporary * storage * @return Table of unique keys */ -template std::unique_ptr compute_groupby(table_view const& keys, std::span requests, - bool skip_rows_with_nulls, - Equal const& d_row_equal, - Hash const& d_row_hash, + null_policy include_null_keys, cudf::detail::result_cache* cache, cuda::stream_ref stream, cudf::memory_resources mr); diff --git a/cpp/src/groupby/hash/compute_groupby.cu b/cpp/src/groupby/hash/group_keys.cu similarity index 60% rename from cpp/src/groupby/hash/compute_groupby.cu rename to cpp/src/groupby/hash/group_keys.cu index 88629d46ec62..f0a7d0838ee2 100644 --- a/cpp/src/groupby/hash/compute_groupby.cu +++ b/cpp/src/groupby/hash/group_keys.cu @@ -3,24 +3,18 @@ * SPDX-License-Identifier: Apache-2.0 */ -#include "compute_groupby.hpp" -#include "compute_single_pass_aggs.hpp" -#include "extract_single_pass_aggs.hpp" +#include "group_keys.hpp" #include "groupby/common/utils.hpp" -#include "hash_compound_agg_finalizer.hpp" #include "hash_csr_kernels.cuh" #include "helpers.cuh" -#include #include #include -#include #include #include #include -#include +#include #include -#include #include #include @@ -30,9 +24,7 @@ #include #include #include -#include #include -#include #include #include #include @@ -42,70 +34,14 @@ #include #include #include -#include #include #include #include -#include #include -#include namespace cudf::groupby::detail::hash { - namespace { -// The result cache shares results across requests on the same values column. Normalize the -// extracted reductions with the same equality so an earlier compound request cannot choose the -// resource or nullability of a later requested result, or allocate a result that the cache drops. -auto extract_hash_groupby_aggs(std::span requests, - cuda::stream_ref stream) -{ - if (requests.size() <= 1) { return extract_single_pass_aggs(requests, stream); } - - using aggregation_set = - std::unordered_set>, - cudf::detail::pair_column_aggregation_hash, - cudf::detail::pair_column_aggregation_equal_to>; - aggregation_set requested; - for (auto const& request : requests) { - for (auto const& agg : request.aggregations) { - requested.emplace(request.values, *agg); - } - } - - auto [values, kinds, aggs, is_intermediate, has_compound] = - extract_single_pass_aggs(requests, stream); - aggregation_set extracted; - std::vector unique_values; - unique_values.reserve(aggs.size()); - for (std::size_t i = 0; i < aggs.size(); ++i) { - auto const key = - std::pair>{values.column(i), *aggs[i]}; - if (!extracted.insert(key).second) { continue; } - auto const output = unique_values.size(); - kinds[output] = kinds[i]; - is_intermediate[output] = !requested.contains(key); - if (output != i) { aggs[output] = std::move(aggs[i]); } - unique_values.push_back(values.column(i)); - } - kinds.resize(unique_values.size()); - aggs.resize(unique_values.size()); - is_intermediate.resize(unique_values.size()); - return std::tuple{table_view{unique_values}, - std::move(kinds), - std::move(aggs), - std::move(is_intermediate), - has_compound}; -} - -/// The keys grouped by the HashCSR build. -struct grouped_keys { - cuda::device_buffer key_rows; ///< First `num_groups` entries identify group keys - cuda::device_buffer group_offsets; ///< `num_groups + 1` offsets into `grouped_rows` - cuda::device_buffer grouped_rows; ///< Input rows reordered so groups are contiguous - size_type num_groups; ///< Number of representative rows in `key_rows` -}; - std::size_t hash_csr_capacity(size_type num_rows) { auto const requested = @@ -270,7 +206,7 @@ grouped_keys group_keys(size_type num_rows, // The occupied slots, in slot order, are the groups: without aggregations the slots hold the // one row wanted for each group, otherwise the slot indices lead to the counts and rows. - cuda::device_buffer key_rows(stream, mr.get_output_mr()); + cuda::device_buffer key_indices(stream, mr.get_output_mr()); cuda::device_buffer group_slots(stream, temp_mr); size_type num_groups{}; bool count_by_representative{}; @@ -313,11 +249,11 @@ grouped_keys group_keys(size_type num_rows, } if (!need_group_offsets) { - key_rows = cuda::device_buffer{ + key_indices = cuda::device_buffer{ stream, mr.get_output_mr(), std::min(num_rows, capacity), cuda::no_init}; - auto const key_rows_end = thrust::copy_if( - policy, slots.data(), slots.data() + slots.size(), key_rows.data(), is_occupied_fn{}); - num_groups = static_cast(cuda::std::distance(key_rows.data(), key_rows_end)); + auto const key_indices_end = thrust::copy_if( + policy, slots.data(), slots.data() + slots.size(), key_indices.data(), is_occupied_fn{}); + num_groups = static_cast(cuda::std::distance(key_indices.data(), key_indices_end)); } else { group_slots = cuda::device_buffer{ stream, temp_mr, std::min(num_rows, capacity), cuda::no_init}; @@ -338,17 +274,14 @@ grouped_keys group_keys(size_type num_rows, // The retry overwrites the table, so release it instead of copying it while growing. slots = cuda::device_buffer{stream, temp_mr}; slot_counts = cuda::device_buffer{stream, temp_mr}; - key_rows = cuda::device_buffer{stream, mr.get_output_mr()}; + key_indices = cuda::device_buffer{stream, mr.get_output_mr()}; group_slots = cuda::device_buffer{stream, temp_mr}; overflow.reset(); capacity = full_capacity; } if (!need_group_offsets) { - return {std::move(key_rows), - cuda::device_buffer{stream, mr.get_output_mr()}, - cuda::device_buffer{stream, mr.get_output_mr()}, - num_groups}; + return {std::move(key_indices), std::nullopt, std::nullopt, num_groups}; } // Every row is an included singleton group, so input order already forms a valid grouping. @@ -359,36 +292,38 @@ grouped_keys group_keys(size_type num_rows, group_slots = cuda::device_buffer{stream, temp_mr}; overflow.reset(); - key_rows = cuda::device_buffer{ + key_indices = cuda::device_buffer{ stream, mr.get_output_mr(), static_cast(num_rows), cuda::no_init}; cuda::device_buffer group_offsets( stream, mr.get_output_mr(), static_cast(num_rows) + 1, cuda::no_init); - cuda::device_buffer grouped_rows( - stream, mr.get_output_mr(), need_grouped_rows ? num_rows : 0, cuda::no_init); + std::optional> grouped_rows; + if (need_grouped_rows) { + grouped_rows.emplace(stream, mr.get_output_mr(), num_rows, cuda::no_init); + } auto const singleton_outputs = cuda::tabulate_output_iterator{ - [key_rows = key_rows.data(), + [key_indices = key_indices.data(), group_offsets = group_offsets.data(), - grouped_rows = grouped_rows.data(), + grouped_rows = grouped_rows ? grouped_rows->data() : nullptr, num_rows] __device__(cuda::std::ptrdiff_t index, size_type value) -> void { group_offsets[index] = value; if (index < num_rows) { - key_rows[index] = value; + key_indices[index] = value; if (grouped_rows != nullptr) { grouped_rows[index] = value; } } }}; thrust::sequence( policy, singleton_outputs, singleton_outputs + group_offsets.size(), size_type{0}); - return {std::move(key_rows), std::move(group_offsets), std::move(grouped_rows), num_groups}; + return {std::move(key_indices), std::move(group_offsets), std::move(grouped_rows), num_groups}; } auto const slot_rows = slots.data(); - key_rows = cuda::device_buffer{ + key_indices = cuda::device_buffer{ stream, mr.get_output_mr(), static_cast(num_groups), cuda::no_init}; if (count_by_representative) { - thrust::copy(policy, group_slots.data(), group_slots.data() + num_groups, key_rows.data()); + thrust::copy(policy, group_slots.data(), group_slots.data() + num_groups, key_indices.data()); } else { thrust::gather( - policy, group_slots.data(), group_slots.data() + num_groups, slot_rows, key_rows.data()); + policy, group_slots.data(), group_slots.data() + num_groups, slot_rows, key_indices.data()); } slots = cuda::device_buffer{stream, temp_mr}; @@ -398,10 +333,7 @@ grouped_keys group_keys(size_type num_rows, auto const group_counts = cuda::make_permutation_iterator(slot_counts.data(), group_slots.data()); thrust::inclusive_scan(policy, group_counts, group_counts + num_groups, group_offsets.data() + 1); if (!need_grouped_rows) { - return {std::move(key_rows), - std::move(group_offsets), - cuda::device_buffer{stream, mr.get_output_mr()}, - num_groups}; + return {std::move(key_indices), std::move(group_offsets), std::nullopt, num_groups}; } auto num_grouped_rows = num_rows; @@ -424,143 +356,60 @@ grouped_keys group_keys(size_type num_rows, launch_hash_csr_fill_kernel( num_rows, positions.data(), slot_counts.data(), grouped_rows.data(), stream); - return {std::move(key_rows), std::move(group_offsets), std::move(grouped_rows), num_groups}; + return {std::move(key_indices), std::move(group_offsets), std::move(grouped_rows), num_groups}; } } // namespace -template -std::unique_ptr
compute_groupby(table_view const& keys, - std::span requests, - bool skip_rows_with_nulls, - Equal const& d_row_equal, - Hash const& d_row_hash, - cudf::detail::result_cache* cache, - cuda::stream_ref stream, - cudf::memory_resources mr) +grouped_keys group_keys(table_view const& keys, + null_policy include_null_keys, + bool need_group_offsets, + bool need_grouped_rows, + cuda::stream_ref stream, + cudf::memory_resources mr) { - auto const num_rows = keys.num_rows(); - auto const temp_mr = mr.get_temporary_mr(); - auto const temporary_resources = mr.as_temporary(); - - [[maybe_unused]] auto [row_bitmask_data, row_bitmask] = + CUDF_EXPECTS(need_group_offsets || !need_grouped_rows, + "Grouped rows require group offsets", + std::invalid_argument); + auto const num_rows = keys.num_rows(); + auto const skip_rows_with_nulls = + cudf::has_nulls(keys) && include_null_keys == null_policy::EXCLUDE; + auto const has_null = nullate::DYNAMIC{cudf::has_nested_nulls(keys)}; + auto preprocessed_keys = + cudf::detail::row::hash::preprocessed_table::create(keys, stream, mr.get_temporary_mr()); + auto const comparator = cudf::detail::row::equality::self_comparator{preprocessed_keys}; + auto const row_hash = cudf::detail::row::hash::row_hasher{std::move(preprocessed_keys)}; + auto const d_row_hash = row_hash.device_hasher(has_null); + auto [row_bitmask_data, row_bitmask] = skip_rows_with_nulls ? cudf::groupby::detail::compute_row_bitmask(keys, stream) : std::pair, bitmask_type const*>{ - cudf::create_null_mask(0, mask_state::UNALLOCATED, stream, temp_mr), nullptr}; - - // Determine which grouping outputs the requested aggregations need. - auto const [values, agg_kinds, aggs, is_agg_intermediate, has_compound_aggs] = - extract_hash_groupby_aggs(requests, stream); - - // Counts without null filtering come directly from the group offsets. - auto const needs_reduction = [&] { - for (size_type i = 0; i < values.num_columns(); ++i) { - if (agg_kinds[i] != aggregation::COUNT_ALL && - (agg_kinds[i] != aggregation::COUNT_VALID || values.column(i).has_nulls())) { - return true; - } - } - return false; - }(); - auto const groups = group_keys(num_rows, - row_bitmask, - d_row_equal, - d_row_hash, - !requests.empty(), - needs_reduction, - num_rows < hash_csr_min_rows_to_estimate - ? std::nullopt - : small_key_domain_capacity(keys, skip_rows_with_nulls), - stream, - temporary_resources); - - auto const key_rows = device_span{groups.key_rows.data(), - static_cast(groups.num_groups)}; - auto const gather_keys = [&] { - return cudf::detail::gather(keys, - key_rows, - out_of_bounds_policy::DONT_CHECK, - cudf::negative_index_policy::NOT_ALLOWED, - stream, - mr); - }; - - // In case of no requests, we still need to generate a set of unique keys. - if (requests.empty()) { return gather_keys(); } - - auto const rows = - device_span{groups.grouped_rows.data(), groups.grouped_rows.size()}; - auto const offsets = - device_span{groups.group_offsets.data(), groups.group_offsets.size()}; - auto const grouped = - needs_reduction - ? make_grouped_rows(rows, offsets, stream, temporary_resources) - : grouped_rows{rows, - offsets, - cuda::device_buffer{stream, temp_mr}, - cuda::device_buffer{stream, temp_mr}, - cuda::device_buffer>{stream, temp_mr}, - cuda::device_buffer{stream, temp_mr}}; - auto results = - compute_single_pass_aggs(values, agg_kinds, is_agg_intermediate, grouped, stream, mr); - for (std::size_t i = 0; i < results.size(); ++i) { - cache->add_result(values.column(i), *aggs[i], std::move(results[i])); + cudf::create_null_mask(0, mask_state::UNALLOCATED, stream, mr.get_temporary_mr()), + nullptr}; + std::optional domain_capacity; + if (num_rows >= hash_csr_min_rows_to_estimate) { + domain_capacity = small_key_domain_capacity(keys, skip_rows_with_nulls); } - - if (has_compound_aggs) { - // Requested M2 results must be cached on the output resource before VARIANCE or STD asks - // for an intermediate M2, regardless of the order of requests on a shared values column. - for (auto const& request : requests) { - auto const finalizer = hash_compound_agg_finalizer(request.values, cache, stream, mr); - for (auto const& agg : request.aggregations) { - if (agg->kind == aggregation::M2) { - cudf::detail::aggregation_dispatcher(agg->kind, finalizer, *agg); - } - } - } - for (auto const& request : requests) { - auto const& agg_v = request.aggregations; - auto const& col = request.values; - - // The finalizers only combine the single-pass results with linear transformations such as - // addition/multiplication (e.g. for variance/stddev); they do not aggregate further. - auto const finalizer = hash_compound_agg_finalizer(col, cache, stream, mr); - for (auto&& agg : agg_v) { - if (agg->kind == aggregation::VARIANCE || agg->kind == aggregation::STD) { - // Explicit M2 outputs were finalized above. Any missing M2 is only an intermediate - // for this ordinary groupby; the shared finalizer also serves streaming groupby. - auto const m2_agg = make_m2_aggregation(); - auto const m2_finalizer = - hash_compound_agg_finalizer(col, cache, stream, temporary_resources); - cudf::detail::aggregation_dispatcher(m2_agg->kind, m2_finalizer, *m2_agg); - } - cudf::detail::aggregation_dispatcher(agg->kind, finalizer, *agg); - } - } + if (cudf::detail::has_nested_columns(keys)) { + return group_keys(num_rows, + row_bitmask, + comparator.equal_to(has_null, null_equality::EQUAL), + d_row_hash, + need_group_offsets, + need_grouped_rows, + domain_capacity, + stream, + mr); } - - return gather_keys(); + return group_keys(num_rows, + row_bitmask, + comparator.equal_to(has_null, null_equality::EQUAL), + d_row_hash, + need_group_offsets, + need_grouped_rows, + domain_capacity, + stream, + mr); } -template std::unique_ptr
compute_groupby( - table_view const& keys, - std::span requests, - bool skip_rows_with_nulls, - row_comparator_t const& d_row_equal, - row_hash_t const& d_row_hash, - cudf::detail::result_cache* cache, - cuda::stream_ref stream, - cudf::memory_resources mr); - -template std::unique_ptr
compute_groupby( - table_view const& keys, - std::span requests, - bool skip_rows_with_nulls, - nullable_row_comparator_t const& d_row_equal, - row_hash_t const& d_row_hash, - cudf::detail::result_cache* cache, - cuda::stream_ref stream, - cudf::memory_resources mr); - } // namespace cudf::groupby::detail::hash diff --git a/cpp/src/groupby/hash/group_keys.hpp b/cpp/src/groupby/hash/group_keys.hpp new file mode 100644 index 000000000000..798118d95a5a --- /dev/null +++ b/cpp/src/groupby/hash/group_keys.hpp @@ -0,0 +1,70 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#pragma once + +#include +#include +#include +#include + +#include +#include + +#include +#include +#include + +namespace cudf::groupby::detail::hash { + +/// The keys grouped by the HashCSR build. +class grouped_keys { + cuda::device_buffer _key_storage; + + public: + grouped_keys(cuda::device_buffer&& indices, + std::optional>&& offsets, + std::optional>&& rows, + size_type num_groups) + : _key_storage{std::move(indices)}, + key_indices{_key_storage.data(), static_cast(num_groups)}, + group_offsets{std::move(offsets)}, + grouped_rows{std::move(rows)} + { + } + + grouped_keys(grouped_keys const&) = delete; + grouped_keys& operator=(grouped_keys const&) = delete; + grouped_keys(grouped_keys&&) = default; + grouped_keys& operator=(grouped_keys&&) = default; + + /// Input row indices for unique keys; size equals the number of groups. + device_span key_indices; + /// `key_indices.size() + 1` group offsets, present when requested. + std::optional> group_offsets; + /// Input row indices in contiguous groups, present when requested. + std::optional> grouped_rows; +}; + +/** + * @brief Groups input keys with HashCSR, optionally materializing group offsets and grouped rows. + * + * @param keys Table whose rows define groups + * @param include_null_keys Whether rows containing null keys participate in grouping + * @param need_group_offsets Whether to return offsets for each group + * @param need_grouped_rows Whether to return grouped input row indices; requires group offsets + * @param stream CUDA stream used for device operations + * @param mr Memory resources for grouping outputs and temporary allocations + * @throws std::invalid_argument If grouped rows are requested without group offsets + * @return Representative rows, group offsets, and grouped row indices as requested + */ +grouped_keys group_keys(table_view const& keys, + null_policy include_null_keys, + bool need_group_offsets, + bool need_grouped_rows, + cuda::stream_ref stream, + cudf::memory_resources mr); + +} // namespace cudf::groupby::detail::hash diff --git a/cpp/src/groupby/hash/groupby.cu b/cpp/src/groupby/hash/groupby.cu index 5eb6316c93ac..411a41bec807 100644 --- a/cpp/src/groupby/hash/groupby.cu +++ b/cpp/src/groupby/hash/groupby.cu @@ -7,12 +7,10 @@ #include "compute_single_pass_aggs.hpp" #include "extract_single_pass_aggs.hpp" #include "groupby/common/utils.hpp" -#include "helpers.cuh" #include #include #include -#include #include #include #include @@ -22,44 +20,12 @@ #include +#include #include #include #include namespace cudf::groupby::detail::hash { -namespace { - -std::unique_ptr
dispatch_groupby(table_view const& keys, - std::span requests, - cudf::detail::result_cache* cache, - bool const keys_have_nulls, - null_policy const include_null_keys, - cuda::stream_ref stream, - cudf::memory_resources mr) -{ - auto const null_keys_are_equal = null_equality::EQUAL; - auto const has_null = nullate::DYNAMIC{cudf::has_nested_nulls(keys)}; - auto const skip_rows_with_nulls = keys_have_nulls and include_null_keys == null_policy::EXCLUDE; - - auto preprocessed_keys = - cudf::detail::row::hash::preprocessed_table::create(keys, stream, mr.get_temporary_mr()); - auto const comparator = cudf::detail::row::equality::self_comparator{preprocessed_keys}; - auto const row_hash = cudf::detail::row::hash::row_hasher{std::move(preprocessed_keys)}; - auto const d_row_hash = row_hash.device_hasher(has_null); - - if (cudf::detail::has_nested_columns(keys)) { - auto const d_row_equal = comparator.equal_to(has_null, null_keys_are_equal); - return compute_groupby( - keys, requests, skip_rows_with_nulls, d_row_equal, d_row_hash, cache, stream, mr); - } else { - auto const d_row_equal = comparator.equal_to(has_null, null_keys_are_equal); - return compute_groupby( - keys, requests, skip_rows_with_nulls, d_row_equal, d_row_hash, cache, stream, mr); - } -} - -} // namespace - /** * @brief Indicates if a set of aggregation requests can be satisfied with a * hash-based groupby implementation. @@ -98,7 +64,7 @@ std::pair, std::vector> groupby( cudf::detail::result_cache cache(requests.size()); std::unique_ptr
unique_keys = - dispatch_groupby(keys, requests, &cache, cudf::has_nulls(keys), include_null_keys, stream, mr); + compute_groupby(keys, requests, include_null_keys, &cache, stream, mr); return std::pair(std::move(unique_keys), extract_results(requests, cache, stream, mr.get_output_mr())); diff --git a/python/cudf/cudf/core/groupby/groupby.py b/python/cudf/cudf/core/groupby/groupby.py index 21e1be4fa442..970a70d176b5 100644 --- a/python/cudf/cudf/core/groupby/groupby.py +++ b/python/cudf/cudf/core/groupby/groupby.py @@ -19,7 +19,7 @@ import pylibcudf as plc from cudf.api.types import is_list_like, is_scalar -from cudf.core._internals import aggregation, sorting +from cudf.core._internals import aggregation, copying, sorting from cudf.core.abc import Serializable from cudf.core.column import access_columns from cudf.core.column.column import ( @@ -631,6 +631,7 @@ def __init__( self._dropna = dropna self._group_keys = group_keys self._selection = None + self._group_ordering: tuple[list[int], ColumnBase | None] | None = None if isinstance(self._by, _Grouping): self._by._obj = self.obj @@ -652,20 +653,18 @@ def __iter__(self): group_names, offsets, _, grouped_values = self._grouped() if isinstance(group_names, Index): group_names = group_names.to_pandas() - if self._sort or len(offsets) <= 2: + if ( + self._sort + or get_option("mode.pandas_compatible") + or len(offsets) <= 2 + ): order = range(len(offsets) - 1) else: - # libcudf returns groups sorted by key, but with ``sort=False`` - # pandas iterates groups in order of first appearance. Reorder by - # the earliest original row position in each group (group order - # matches between the two ``_groups`` calls since the grouping is - # identical). + # Preserve the iterator's first-appearance ordering used by + # ngroup and stack without ordering other native group results. pos_offsets, _, (positions,) = self._groups( [self._range_column_from_obj] ) - # Gather the earliest original row position of each group and sort - # the groups by it entirely on the device; only the small ``order`` - # array (one entry per group) is copied back to the host. first_pos = positions.take(as_column(pos_offsets[:-1])) order = first_pos.argsort().to_numpy() for i in order: @@ -955,10 +954,20 @@ def _groupby(self): return self._groupby_manager def _groups( - self, values: Iterable[ColumnBase] + self, values: Iterable[ColumnBase], *, order_groups: bool | None = None ) -> tuple[list[int], list[ColumnBase], list[ColumnBase]]: + if order_groups is None: + order_groups = self._sort or get_option("mode.pandas_compatible") # Materialize iterator to avoid consuming it during access context setup values_list = list(values) + # First-appearance ordering needs the input position of every grouped + # row. Gather a range column along with the values instead of grouping + # a second time. + need_positions = ( + order_groups and not self._sort and self._group_ordering is None + ) + if need_positions: + values_list.append(self._range_column_from_obj) key_dtypes = [col.dtype for col in self.grouping._key_columns] value_dtypes = [col.dtype for col in values_list] with access_columns(*values_list, mode="read", scope="internal"): @@ -973,25 +982,70 @@ def _groups( plc_table ) - return ( - offsets, + key_columns = [ + ColumnBase.create(col, dtype) + for col, dtype in zip( + grouped_keys.columns(), key_dtypes, strict=True + ) + ] + value_columns = ( [ ColumnBase.create(col, dtype) for col, dtype in zip( - grouped_keys.columns(), key_dtypes, strict=True + grouped_values.columns(), value_dtypes, strict=True ) - ], - ( - [ - ColumnBase.create(col, dtype) - for col, dtype in zip( - grouped_values.columns(), value_dtypes, strict=True - ) - ] - if grouped_values is not None - else [] - ), + ] + if grouped_values is not None + else [] ) + positions = value_columns.pop() if need_positions else None + # get_groups already returns groups in sorted key order. + if not order_groups or self._sort or len(offsets) <= 2: + return offsets, key_columns, value_columns + + # Cache the requested group order, preserving row order within each group. + if self._group_ordering is None: + group_starts = as_column(offsets[:-1]) + assert positions is not None + first_positions = positions.take(group_starts) + group_order = first_positions.argsort().to_numpy() + + if np.array_equal(group_order, np.arange(len(offsets) - 1)): + self._group_ordering = (offsets, None) + else: + old_offsets = np.asarray(offsets, dtype=SIZE_TYPE_DTYPE) + sizes = np.diff(old_offsets)[group_order] + new_offsets = np.zeros_like(old_offsets) + np.cumsum(sizes, out=new_offsets[1:]) + # Every row of a group moves by the same amount, so repeat the + # per-group shift over the group sizes and add it to the new + # positions. + shifts = Index( + old_offsets[:-1][group_order] - new_offsets[:-1] + ) + row_order = cp.arange(offsets[-1], dtype=SIZE_TYPE_DTYPE) + row_order += shifts.repeat(sizes).values + self._group_ordering = ( + new_offsets.tolist(), + as_column(row_order), + ) + + offsets, row_order = self._group_ordering + if row_order is not None: + columns = key_columns + value_columns + reordered = [ + ColumnBase.create(col, original.dtype) + for col, original in zip( + copying.gather(columns, row_order), # type: ignore[arg-type] + columns, + strict=True, + ) + ] + key_columns, value_columns = ( + reordered[: len(key_columns)], + reordered[len(key_columns) :], + ) + return offsets, key_columns, value_columns def _aggregate( self, values: tuple[ColumnBase, ...], aggregations @@ -1390,12 +1444,11 @@ def agg(self, func=None, *args, engine=None, engine_kwargs=None, **kwargs): result = DataFrame._from_data(data, index=result_index) - if self._sort: + is_scan = _is_all_scan_aggregate(normalized_aggs) + if self._sort and not is_scan: result = result.sort_index() else: - if get_option( - "mode.pandas_compatible" - ) and not _is_all_scan_aggregate(normalized_aggs): + if get_option("mode.pandas_compatible") and not is_scan: # Even with `sort=False`, pandas guarantees that # groupby preserves the order of rows within each group. left_cols = self.grouping.keys.drop_duplicates()._columns @@ -1464,7 +1517,6 @@ def agg(self, func=None, *args, engine=None, engine_kwargs=None, **kwargs): ) ) - is_scan = _is_all_scan_aggregate(normalized_aggs) if not self._as_index and not is_scan: result = result.reset_index() if is_scan: @@ -1696,7 +1748,9 @@ def _head_tail(self, n, *, take_head: bool, preserve_order: bool): # into the grouping, but that probably requires a new # aggregation scheme in libcudf. This is probably "fast # enough" for most reasonable input sizes. - _, offsets, _, group_values = self._grouped() + _, offsets, _, group_values = self._grouped( + order_groups=False if preserve_order else None + ) group_offsets = np.asarray(offsets, dtype=SIZE_TYPE_DTYPE) size_per_group = np.diff(group_offsets) # "Out of bounds" n for the group size either means no entries @@ -1724,7 +1778,9 @@ def _head_tail(self, n, *, take_head: bool, preserve_order: bool): # Can't use _mimic_pandas_order because we need to # subsample the gather map from the full input ordering, # rather than permuting the gather map of the output. - _, _, (ordering,) = self._groups([self._range_column_from_obj]) + _, _, (ordering,) = self._groups( + [self._range_column_from_obj], order_groups=False + ) # Invert permutation from original order to groups on the # subset of entries we want. gather_map = ordering.take(to_take).argsort() @@ -2256,11 +2312,14 @@ def deserialize(cls, header, frames): ) return cls(obj, grouping, **kwargs) - def _grouped(self, *, include_groups: bool = True): + def _grouped( + self, *, include_groups: bool = True, order_groups: bool | None = None + ): from cudf.core.dataframe import DataFrame offsets, grouped_key_cols, grouped_value_cols = self._groups( - itertools.chain(self.obj.index._columns, self.obj._columns) + itertools.chain(self.obj.index._columns, self.obj._columns), + order_groups=order_groups, ) grouped_keys = _index_from_data(dict(enumerate(grouped_key_cols))) if isinstance(self.grouping.keys, MultiIndex): @@ -2281,7 +2340,7 @@ def _grouped(self, *, include_groups: bool = True): selection is None or col_name not in selection ): del grouped_values[col_name] - group_names = grouped_keys.unique().sort_values() + group_names = grouped_keys.take(as_column(offsets[:-1])) return (group_names, offsets, grouped_keys, grouped_values) def _normalize_aggs( @@ -2561,11 +2620,11 @@ def _post_process_chunk_results( # Every chunk result is indexed like its input chunk, i.e. # the UDF acted as a transform. pandas restores the original # row order in this case (GroupBy._concat_objects) regardless - # of ``sort``. The concatenated chunks are in key-sorted + # of ``sort``. The concatenated chunks are in Python's # group order, so gather back through the inverse of the # grouping permutation. _, _, (positions,) = self._groups( - [self._range_column_from_obj] + [self._range_column_from_obj], order_groups=True ) result = result.take(positions.argsort().values) return result @@ -2710,37 +2769,9 @@ def mult(df): if not callable(func): raise TypeError(f"type {type(func)} is not callable") group_names, offsets, group_keys, grouped_values = self._grouped( - include_groups=include_groups + include_groups=include_groups, order_groups=True ) - if not self._sort and len(offsets) > 2: - # libcudf returns groups sorted by key, but with ``sort=False`` - # pandas processes groups in order of first appearance. Permute - # the grouped layout accordingly so both engines and the result - # assembly see pandas' iteration order. - pos_offsets, _, (positions,) = self._groups( - [self._range_column_from_obj] - ) - first_pos = positions.take(as_column(pos_offsets[:-1])) - group_order = first_pos.argsort().to_numpy() - sizes = np.diff(np.asarray(offsets, dtype=SIZE_TYPE_DTYPE)) - row_order = as_column( - np.concatenate( - [ - np.arange( - offsets[i], offsets[i + 1], dtype=SIZE_TYPE_DTYPE - ) - for i in group_order - ] - ) - ) - group_names = group_names.take(group_order) - group_keys = group_keys.take(row_order) - grouped_values = grouped_values.take(row_order) - new_offsets = np.zeros(len(sizes) + 1, dtype=SIZE_TYPE_DTYPE) - np.cumsum(sizes[group_order], out=new_offsets[1:]) - offsets = new_offsets.tolist() - if engine == "auto": if _can_be_jitted(grouped_values, func, args): engine = "jit" @@ -2767,10 +2798,9 @@ def mult(df): else: raise ValueError(f"Unsupported engine '{engine}'") - # No final sort: group-keyed results are already produced in - # sorted group-key order, and pandas preserves the UDF's - # within-group row order (and a transform's original row order) - # regardless of ``sort`` (pandas GH52444). + # Group-keyed results already use the requested group order. Preserve + # the UDF's within-group row order and a transform's original row + # order regardless of ``sort`` (pandas GH52444). if self._as_index is False: result = result.reset_index() return result @@ -2964,7 +2994,7 @@ def describe(self, percentiles=None, include=None, exclude=None): 1 370.0 30 2 24.0 90 3 26.0 80 - >>> gdf.groupby('Score').describe() + >>> gdf.groupby('Score', sort=True).describe() Speed count mean std min 25% 50% 75% max Score @@ -3101,7 +3131,7 @@ def cov(self, min_periods=0, ddof=1, numeric_only: bool = False): def _cov_or_corr(self, func, method_name): """ Internal function that is called by either corr() or cov() - for sort groupby correlation and covariance computations, + for groupby correlation and covariance computations, respectively. """ # create expanded dataframe consisting all combinations of the @@ -3685,7 +3715,9 @@ def _mimic_pandas_order( # result coming back from libcudf has null_count few rows than # the input, so we must produce an ordering from the full # input range. - _, _, (ordering,) = self._groups([self._range_column_from_obj]) + _, _, (ordering,) = self._groups( + [self._range_column_from_obj], order_groups=False + ) if self._dropna and any( c.has_nulls(include_nan=True) > 0 for c in self.grouping._key_columns diff --git a/python/cudf/cudf/pandas/scripts/pandas-testing-plugin.py b/python/cudf/cudf/pandas/scripts/pandas-testing-plugin.py index dc6a25b3174e..775f8db2b641 100644 --- a/python/cudf/cudf/pandas/scripts/pandas-testing-plugin.py +++ b/python/cudf/cudf/pandas/scripts/pandas-testing-plugin.py @@ -1853,8 +1853,6 @@ def pytest_unconfigure(config): "tests/groupby/test_groupby.py::test_groups_repr_truncates[1-{0: [0], ...}]": "TODO: Add a reason for failure", "tests/groupby/test_groupby.py::test_groups_repr_truncates[4-{0: [0], 1: [1], 2: [2], 3: [3], ...}]": "TODO: Add a reason for failure", "tests/groupby/test_groupby.py::test_groups_repr_truncates[5-{0: [0], 1: [1], 2: [2], 3: [3], 4: [4]}]": "TODO: Add a reason for failure", - "tests/groupby/test_groupby.py::test_groups_sort_dropna[False-False]": "AssertionError: numpy array are different", - "tests/groupby/test_groupby.py::test_groups_sort_dropna[False-True]": "AssertionError: numpy array are different", "tests/groupby/test_groupby.py::test_ops_not_as_index[idxmin]": "TODO: Add a reason for failure", "tests/groupby/test_groupby.py::test_ops_not_as_index[size]": "TODO: Add a reason for failure", "tests/groupby/test_groupby.py::test_single_element_listlike_level_grouping[level_arg0-False]": "AssertionError: assert ['x', 'y'] == [('x',), ('y',)]", diff --git a/python/cudf/cudf/tests/groupby/test_apply.py b/python/cudf/cudf/tests/groupby/test_apply.py index ff32a2499859..6d82f8f2f74a 100644 --- a/python/cudf/cudf/tests/groupby/test_apply.py +++ b/python/cudf/cudf/tests/groupby/test_apply.py @@ -979,15 +979,15 @@ def test_groupby_apply_series_preserves_multiindex_names(): def test_groupby_group_keys(group_keys, by): gdf = cudf.DataFrame( { - "A": "a a a a b b".split(), - "B": [1, 1, 2, 2, 3, 3], - "C": [4, 6, 5, 9, 8, 7], + "A": "b a b a a a".split(), + "B": [3, 1, 3, 1, 2, 2], + "C": [8, 4, 7, 6, 5, 9], } ) pdf = gdf.to_pandas() - g_group = gdf.groupby(by, group_keys=group_keys) - p_group = pdf.groupby(by, group_keys=group_keys) + g_group = gdf.groupby(by, group_keys=group_keys, sort=False) + p_group = pdf.groupby(by, group_keys=group_keys, sort=False) actual = g_group[["B", "C"]].apply(lambda x: x / x.sum()) expected = p_group[["B", "C"]].apply(lambda x: x / x.sum()) diff --git a/python/cudf/cudf/tests/groupby/test_head_tail.py b/python/cudf/cudf/tests/groupby/test_head_tail.py index 78e66079be76..28433be3e6bc 100644 --- a/python/cudf/cudf/tests/groupby/test_head_tail.py +++ b/python/cudf/cudf/tests/groupby/test_head_tail.py @@ -54,10 +54,9 @@ def expected(df, n, take_head, preserve_order): else: return g.tail(n=n) else: - # We groupby "a" which is the first column. This - # possibly relies on an implementation detail that for - # integer group keys, cudf produces groups in sorted - # (ascending) order. + # Build the reference in ascending key order. The comparison + # below normalizes group order while preserving row order + # within every group. keyfunc = operator.itemgetter(0) if take_head or n == 0: # Head does group[:n] as does tail for n == 0 @@ -88,6 +87,9 @@ def test_head_tail(df, n, take_head, expected, preserve_order): actual = df.groupby("a").head(n=n, preserve_order=preserve_order) else: actual = df.groupby("a").tail(n=n, preserve_order=preserve_order) + if not preserve_order: + # Native sort=False does not guarantee a particular group order. + actual = actual.to_pandas().sort_values("a", kind="stable") assert_eq(actual, expected) diff --git a/python/cudf/cudf/tests/groupby/test_size.py b/python/cudf/cudf/tests/groupby/test_size.py index 41b756a55e36..08276c61a6c7 100644 --- a/python/cudf/cudf/tests/groupby/test_size.py +++ b/python/cudf/cudf/tests/groupby/test_size.py @@ -1,5 +1,6 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import numpy as np import pandas as pd import pytest @@ -34,14 +35,18 @@ def test_size_series_masked_dtype(dtype): assert_groupby_results_equal(expect, got) -def test_apply_sort_false_first_appearance_order(): +@pytest.mark.parametrize( + "keys", + [[3, 1, 3, 2, 1], np.random.default_rng(2026).integers(0, 32, 10000)], +) +def test_apply_sort_false_first_appearance_order(keys): # groups are processed in order of first appearance with sort=False, # like pandas - pdf = pd.DataFrame({"k": [3, 1, 3, 2, 1], "v": [1, 2, 3, 4, 5]}) + pdf = pd.DataFrame({"k": keys, "v": range(len(keys))}) gdf = cudf.from_pandas(pdf) expect = pdf.groupby("k", sort=False)["v"].apply(lambda s: s.sum()) got = gdf.groupby("k", sort=False)["v"].apply(lambda s: s.sum()) - assert list(got.index.to_pandas()) == [3, 1, 2] + assert list(got.index.to_pandas()) == list(pdf["k"].drop_duplicates()) assert_groupby_results_equal(expect, got) diff --git a/python/cudf_polars/cudf_polars/dsl/expressions/rolling.py b/python/cudf_polars/cudf_polars/dsl/expressions/rolling.py index fe90941d61e7..719e5b473067 100644 --- a/python/cudf_polars/cudf_polars/dsl/expressions/rolling.py +++ b/python/cudf_polars/cudf_polars/dsl/expressions/rolling.py @@ -861,7 +861,7 @@ def _( def _reorder_to_input( self, row_id: plc.Column, - by_cols: list[Column], + grouper: plc.groupby.GroupBy, n_rows: int, rank_tables: list[plc.Table], rank_out_names: list[str], @@ -872,14 +872,13 @@ def _reorder_to_input( ) -> list[Column]: # Reorder scan results from grouped-order back to input row order if order_index is None: - key_orders = [k.order for k in by_cols] - key_nulls = [k.null_order for k in by_cols] - order_index = plc.sorting.stable_sorted_order( - plc.Table([*(c.obj for c in by_cols), row_id]), - [*key_orders, plc.types.Order.ASCENDING], - [*key_nulls, plc.types.NullOrder.AFTER], - stream=stream, + # The grouper keeps input order within each group but places the + # groups in an unspecified order, so ask it for the row ids in + # the order it produced the results rather than sorting the keys. + _, _, grouped_row_ids = grouper.get_groups( + plc.Table([row_id]), stream=stream ) + order_index = grouped_row_ids.columns()[0] return [ Column( @@ -1150,7 +1149,7 @@ def _apply_ordered_unary_op( ) return self._reorder_to_input( row_id, - by_cols, + grouper, df.num_rows, tables, names, @@ -1367,7 +1366,7 @@ def do_evaluate( # noqa: D102 broadcasted_cols.extend( self._reorder_to_input( row_id, - by_cols, + grouper, df.num_rows, tables, names, @@ -1387,7 +1386,7 @@ def do_evaluate( # noqa: D102 broadcasted_cols.extend( self._reorder_to_input( row_id, - by_cols, + grouper, df.num_rows, tables, names, @@ -1436,7 +1435,7 @@ def do_evaluate( # noqa: D102 broadcasted_cols.extend( self._reorder_to_input( row_id, - by_cols, + grouper, df.num_rows, tables, names,