From 4feb48363dda52a75f27b078a39bc594403b9d1d Mon Sep 17 00:00:00 2001 From: Yunsong Wang Date: Fri, 18 Sep 2026 14:31:48 -0700 Subject: [PATCH 1/5] Reduce HashCSR hash join peak memory Pack representative rows and hash fingerprints into 32-bit map entries, index CSR offsets by build row, and reuse offsets during scatter. Avoid per-row numeric fill scratch while caching representatives for keys containing lists or strings to preserve construction performance. Use exact capacity at default and lower load factors, release temporary buffers before allocating outputs, and size full-join flags by build rows. Add collision, row-boundary, nested-null, partitioned-join, and peak-memory regression coverage. --- cpp/src/join/hash_join/hash_csr.cuh | 74 +++-- cpp/src/join/hash_join/hash_csr_kernels.cuh | 165 +++++++---- cpp/src/join/hash_join/hash_join.cu | 93 ++++-- cpp/src/join/hash_join/hash_join_impl.cuh | 21 +- .../hash_join/partitioned_join_retrieve.cu | 123 ++++---- cpp/src/join/hash_join/retrieve_impl.cuh | 166 ++++++----- cpp/src/join/hash_join/size_impl.cuh | 6 +- cpp/tests/CMakeLists.txt | 1 + cpp/tests/join/hash_join_memory_tests.cpp | 264 ++++++++++++++++++ 9 files changed, 658 insertions(+), 255 deletions(-) create mode 100644 cpp/tests/join/hash_join_memory_tests.cpp diff --git a/cpp/src/join/hash_join/hash_csr.cuh b/cpp/src/join/hash_join/hash_csr.cuh index 055422cd3e28..952898b3a047 100644 --- a/cpp/src/join/hash_join/hash_csr.cuh +++ b/cpp/src/join/hash_join/hash_csr.cuh @@ -11,62 +11,80 @@ #include #include +#include #include namespace cudf::detail { -/// One open-addressed slot: the row hash and the index of the build row that claimed it. -using hash_table_entry_type = cuco::pair; +/// A row index and the high bits of its hash share one 32-bit table entry. +using hash_table_entry_type = cuda::std::uint32_t; -/// Where a build row landed: the slot it claimed and its rank among the rows sharing that slot. -/// Computing the rank during the build lets retrieval index straight into the CSR without a -/// second pass. -using build_position_type = cuco::pair; - -/// Device-side view of the open-addressed table, linearly probed with a power-of-two capacity. +/// Device-side view of the open-addressed table. The low bits hold the representative build +/// row, and the remaining bits hold a hash fingerprint. Fingerprint matches always undergo +/// a full row comparison; reducing the fingerprint width cannot introduce false matches. struct hash_table_ref { hash_table_entry_type* entries; - cuda::std::uint32_t capacity; ///< Power of two, so the probe index is a mask instead of a modulo + cuda::std::uint32_t capacity; + cuda::std::uint32_t row_mask; + cuda::fast_mod_div modulo; - __device__ cuda::std::uint32_t mask() const { return capacity - 1; } + template + __device__ bool equal(cuco::pair key, + hash_table_entry_type entry, + Equal check_row_equality) const + { + return ((key.first ^ entry) & ~row_mask) == 0 && + check_row_equality(key, {key.first, static_cast(entry & row_mask)}); + } template - __device__ cuda::std::uint32_t insert(hash_table_entry_type key, Equal equal) const + __device__ size_type insert(cuco::pair key, Equal equal_rows) const { + auto const desired = (key.first & ~row_mask) | static_cast(key.second); + auto slot = key.first % modulo; for (cuda::std::uint32_t step = 0; step < capacity; ++step) { - auto const slot = (static_cast(key.first) + step) & mask(); auto entry_ref = cuda::atomic_ref{entries[slot]}; - auto old = hash_table_entry_type{hash_value_type{-1}, size_type{CUDF_SIZE_TYPE_SENTINEL}}; - if (entry_ref.compare_exchange_strong(old, key, cuda::memory_order_relaxed)) { return slot; } - if (equal(key, old)) { return slot; } + auto old = hash_table_entry_type{-1}; + if (entry_ref.compare_exchange_strong(old, desired, cuda::memory_order_relaxed)) { + return key.second; + } + if (equal(key, old, equal_rows)) { return static_cast(old & row_mask); } + ++slot; + if (slot == capacity) { slot = 0; } } - return capacity; + return CUDF_SIZE_TYPE_SENTINEL; } - template - __device__ cuda::std::uint32_t find(hash_table_entry_type key, Equal equal) const + template + __device__ size_type find(cuco::pair key, Equal equal_rows) const { + auto slot = key.first % modulo; for (cuda::std::uint32_t step = 0; step < capacity; ++step) { - auto const slot = (static_cast(key.first) + step) & mask(); auto const current = entries[slot]; - if (current.second == CUDF_SIZE_TYPE_SENTINEL) { return capacity; } - if (equal(key, current)) { return slot; } + if (current == hash_table_entry_type{-1}) { return CUDF_SIZE_TYPE_SENTINEL; } + // Under null_equality::UNEQUAL a nested row containing nulls need not equal itself. + // The fill pass must still find the row that claimed this entry during construction. + if constexpr (IsBuild) { + if (static_cast(current & row_mask) == key.second) { return key.second; } + } + if (equal(key, current, equal_rows)) { return static_cast(current & row_mask); } + ++slot; + if (slot == capacity) { slot = 0; } } - return capacity; + return CUDF_SIZE_TYPE_SENTINEL; } }; +/// CSR segments are indexed by representative build row, including zero-length segments for +/// rows that did not claim a table entry. This avoids an offset for every empty hash slot. struct csr_ref { - size_type const* cumulative_ends; + size_type const* offsets; size_type const* values; - __device__ size_type begin(size_type slot) const - { - return slot == 0 ? size_type{0} : cumulative_ends[slot - 1]; - } + __device__ size_type begin(size_type row) const { return offsets[row]; } - __device__ size_type size(size_type slot) const { return cumulative_ends[slot] - begin(slot); } + __device__ size_type size(size_type row) const { return offsets[row + 1] - offsets[row]; } }; } // namespace cudf::detail diff --git a/cpp/src/join/hash_join/hash_csr_kernels.cuh b/cpp/src/join/hash_join/hash_csr_kernels.cuh index f18cf71156a3..4e9eb3b1d396 100644 --- a/cpp/src/join/hash_join/hash_csr_kernels.cuh +++ b/cpp/src/join/hash_join/hash_csr_kernels.cuh @@ -29,54 +29,88 @@ constexpr thread_index_type hash_csr_outputs_per_lane = 32; template CUDF_KERNEL void hash_csr_build_count_kernel(size_type num_rows, bitmask_type const* valid_rows, - build_position_type* build_positions, - size_type* slot_counts, + size_type* counts, + size_type* representatives, hash_table_ref map, Equal equal, Hasher hasher) { auto const stride = grid_1d::grid_stride(); for (auto row = grid_1d::global_thread_id(); row < num_rows; row += stride) { - auto const index = static_cast(row); - if (valid_rows != nullptr && !cudf::bit_is_set(valid_rows, index)) { - build_positions[index] = {cuda::std::uint32_t{-1}, size_type{CUDF_SIZE_TYPE_SENTINEL}}; - continue; + auto const index = static_cast(row); + auto representative = size_type{CUDF_SIZE_TYPE_SENTINEL}; + if (valid_rows == nullptr || cudf::bit_is_set(valid_rows, index)) { + representative = map.insert(cuco::pair{hasher(index), index}, equal); } - - auto const slot = map.insert(hash_table_entry_type{hasher(index), index}, equal); - if (slot == map.capacity) { - build_positions[index] = {cuda::std::uint32_t{-1}, size_type{CUDF_SIZE_TYPE_SENTINEL}}; - continue; + // Initialize excluded rows too: the cached fill pass only reads representatives. + if (representatives != nullptr) { representatives[index] = representative; } + if (representative == CUDF_SIZE_TYPE_SENTINEL) { continue; } + auto const peers = __match_any_sync(__activemask(), representative); + auto const lane = threadIdx.x % cudf::detail::warp_size; + if (lane == __ffs(peers) - 1) { + cuda::atomic_ref{counts[representative]}.fetch_add( + __popc(peers), cuda::memory_order_relaxed); } - auto slot_count_ref = cuda::atomic_ref{slot_counts[slot]}; - auto const rank = slot_count_ref.fetch_add(size_type{1}, cuda::memory_order_relaxed); - build_positions[index] = {slot, rank}; } } +__device__ inline void hash_csr_scatter_build_row(size_type index, + size_type representative, + size_type* offsets, + size_type* values) +{ + auto const peers = __match_any_sync(__activemask(), representative); + auto const lane = threadIdx.x % cudf::detail::warp_size; + auto const leader = __ffs(peers) - 1; + size_type end{}; + if (lane == leader) { + end = cuda::atomic_ref{offsets[representative]}.fetch_sub( + __popc(peers), cuda::memory_order_relaxed); + } + end = __shfl_sync(peers, end, leader); + auto const rank = __popc(peers & ((cuda::std::uint32_t{1} << lane) - 1)); + values[end - rank - 1] = index; +} + +CUDF_KERNEL void hash_csr_build_fill_cached_kernel(size_type num_rows, + size_type const* representatives, + size_type* offsets, + size_type* values) +{ + auto const stride = grid_1d::grid_stride(); + for (auto row = grid_1d::global_thread_id(); row < num_rows; row += stride) { + auto const index = static_cast(row); + auto const representative = representatives[index]; + if (representative == CUDF_SIZE_TYPE_SENTINEL) { continue; } + hash_csr_scatter_build_row(index, representative, offsets, values); + } +} + +template CUDF_KERNEL void hash_csr_build_fill_kernel(size_type num_rows, - build_position_type const* build_positions, - size_type const* cumulative_ends, - size_type* values) + bitmask_type const* valid_rows, + size_type* offsets, + size_type* values, + hash_table_ref map, + Equal equal, + Hasher hasher) { auto const stride = grid_1d::grid_stride(); for (auto row = grid_1d::global_thread_id(); row < num_rows; row += stride) { - auto const index = static_cast(row); - auto const position = build_positions[index]; - if (position.first == cuda::std::uint32_t{-1}) { continue; } - auto const slot = position.first; - auto const rank = position.second; - auto const begin = slot == 0 ? size_type{0} : cumulative_ends[slot - 1]; - values[begin + rank] = index; + auto const index = static_cast(row); + if (valid_rows != nullptr && !cudf::bit_is_set(valid_rows, index)) { continue; } + auto const representative = map.find(cuco::pair{hasher(index), index}, equal); + if (representative == CUDF_SIZE_TYPE_SENTINEL) { continue; } + hash_csr_scatter_build_row(index, representative, offsets, values); } } template CUDF_KERNEL void hash_csr_probe_count_kernel(size_type num_rows, bitmask_type const* valid_rows, - size_type* probe_slots, + size_type* probe_groups, size_type* match_counts, - cuda::std::uint32_t* matched_slots, + cuda::std::uint32_t* matched_groups, cuda::std::uint64_t* matched_build_rows, hash_table_ref map, csr_ref csr, @@ -86,29 +120,29 @@ CUDF_KERNEL void hash_csr_probe_count_kernel(size_type num_rows, auto const stride = grid_1d::grid_stride(); for (auto row = grid_1d::global_thread_id(); row < num_rows; row += stride) { auto const index = static_cast(row); - auto slot = map.capacity; + auto group = size_type{CUDF_SIZE_TYPE_SENTINEL}; if (valid_rows == nullptr || cudf::bit_is_set(valid_rows, index)) { - slot = map.find(hash_table_entry_type{hasher(index), index}, equal); + group = map.find(cuco::pair{hasher(index), index}, equal); } - auto const found = slot != map.capacity; - auto const count = found ? csr.size(static_cast(slot)) : size_type{0}; - if (probe_slots != nullptr) { - probe_slots[index] = found ? static_cast(slot) : CUDF_SIZE_TYPE_SENTINEL; + auto const found = group != CUDF_SIZE_TYPE_SENTINEL; + auto const count = found ? csr.size(static_cast(group)) : size_type{0}; + if (probe_groups != nullptr) { + probe_groups[index] = found ? static_cast(group) : CUDF_SIZE_TYPE_SENTINEL; } if (match_counts != nullptr) { match_counts[index] = IsOuter ? cuda::std::max(count, size_type{1}) : count; } - // Only right and full joins consume the matched-row tally, and `matched_slots` is null for + // Only right and full joins consume the matched-row tally, and `matched_groups` is null for // every other kind, so this whole block compiles away outside outer joins rather than costing // a branch per probe row. if constexpr (IsOuter) { - if (found && matched_slots != nullptr) { - auto matched_slot_ref = - cuda::atomic_ref{matched_slots[slot]}; + if (found && matched_groups != nullptr) { + auto matched_group_ref = + cuda::atomic_ref{matched_groups[group]}; auto expected = cuda::std::uint32_t{0}; - if (matched_slot_ref.compare_exchange_strong( + if (matched_group_ref.compare_exchange_strong( expected, cuda::std::uint32_t{1}, cuda::memory_order_relaxed)) { cuda::atomic_ref{*matched_build_rows} .fetch_add(static_cast(count), cuda::memory_order_relaxed); @@ -121,8 +155,8 @@ CUDF_KERNEL void hash_csr_probe_count_kernel(size_type num_rows, template void launch_hash_csr_build_count_kernel(size_type num_rows, bitmask_type const* valid_rows, - build_position_type* build_positions, - size_type* slot_counts, + size_type* counts, + size_type* representatives, hash_table_ref map, Equal equal, Hasher hasher, @@ -131,29 +165,48 @@ void launch_hash_csr_build_count_kernel(size_type num_rows, if (num_rows == 0) { return; } auto const config = grid_1d{num_rows, hash_csr_block_size}; hash_csr_build_count_kernel<<>>( - num_rows, valid_rows, build_positions, slot_counts, map, equal, hasher); + num_rows, valid_rows, counts, representatives, map, equal, hasher); + CUDF_CUDA_TRY(cudaGetLastError()); +} + +inline void launch_hash_csr_build_fill_cached_kernel(size_type num_rows, + size_type const* representatives, + size_type* offsets, + size_type* values, + cuda::stream_ref stream) +{ + if (num_rows == 0) { return; } + auto const config = grid_1d{num_rows, hash_csr_block_size}; + hash_csr_build_fill_cached_kernel<<>>(num_rows, representatives, offsets, values); CUDF_CUDA_TRY(cudaGetLastError()); } -inline void launch_hash_csr_build_fill_kernel(size_type num_rows, - build_position_type const* build_positions, - size_type const* cumulative_ends, - size_type* values, - cuda::stream_ref stream) +template +void launch_hash_csr_build_fill_kernel(size_type num_rows, + bitmask_type const* valid_rows, + size_type* offsets, + size_type* values, + hash_table_ref map, + Equal equal, + Hasher hasher, + cuda::stream_ref stream) { if (num_rows == 0) { return; } auto const config = grid_1d{num_rows, hash_csr_block_size}; hash_csr_build_fill_kernel<<>>( - num_rows, build_positions, cumulative_ends, values); + num_rows, valid_rows, offsets, values, map, equal, hasher); CUDF_CUDA_TRY(cudaGetLastError()); } template void launch_hash_csr_probe_count_kernel(size_type num_rows, bitmask_type const* valid_rows, - size_type* probe_slots, + size_type* probe_groups, size_type* match_counts, - cuda::std::uint32_t* matched_slots, + cuda::std::uint32_t* matched_groups, cuda::std::uint64_t* matched_build_rows, hash_table_ref map, csr_ref csr, @@ -166,9 +219,9 @@ void launch_hash_csr_probe_count_kernel(size_type num_rows, hash_csr_probe_count_kernel <<>>(num_rows, valid_rows, - probe_slots, + probe_groups, match_counts, - matched_slots, + matched_groups, matched_build_rows, map, csr, @@ -182,7 +235,7 @@ CUDF_KERNEL void hash_csr_retrieve_kernel(cuda::std::int64_t output_size, size_type num_probe_rows, cuda::std::int64_t outputs_per_warp, cuda::std::int64_t const* offsets, - size_type const* probe_slots, + size_type const* probe_groups, csr_ref csr, size_type left_index_offset, size_type* left_indices, @@ -218,16 +271,16 @@ CUDF_KERNEL void hash_csr_retrieve_kernel(cuda::std::int64_t output_size, offsets + last_probe + 2, output_index) - offsets - 1); - auto const slot = probe_slots[probe_row]; + auto const group = probe_groups[probe_row]; left_indices[output_index] = probe_row + left_index_offset; if constexpr (IsOuter) { - if (slot == CUDF_SIZE_TYPE_SENTINEL) { + if (group == CUDF_SIZE_TYPE_SENTINEL) { right_indices[output_index] = JoinNoMatch; continue; } } auto const local_match = static_cast(output_index - offsets[probe_row]); - right_indices[output_index] = csr.values[csr.begin(slot) + local_match]; + right_indices[output_index] = csr.values[csr.begin(group) + local_match]; } } } @@ -236,7 +289,7 @@ template void launch_hash_csr_retrieve_kernel(cuda::std::int64_t output_size, size_type num_probe_rows, cuda::std::int64_t const* offsets, - size_type const* probe_slots, + size_type const* probe_groups, csr_ref csr, size_type left_index_offset, size_type* left_indices, @@ -259,7 +312,7 @@ void launch_hash_csr_retrieve_kernel(cuda::std::int64_t output_size, num_probe_rows, outputs_per_warp, offsets, - probe_slots, + probe_groups, csr, left_index_offset, left_indices, diff --git a/cpp/src/join/hash_join/hash_join.cu b/cpp/src/join/hash_join/hash_join.cu index de27a204dfe5..ee2b6506824f 100644 --- a/cpp/src/join/hash_join/hash_join.cu +++ b/cpp/src/join/hash_join/hash_join.cu @@ -21,10 +21,12 @@ #include #include +#include #include #include +#include #include #include #include @@ -50,6 +52,12 @@ bool is_trivial_join(table_view const& left, table_view const& right, join_kind } namespace { +bool has_list_or_string(column_view const& column) +{ + return column.type().id() == type_id::LIST || column.type().id() == type_id::STRING || + std::any_of(column.child_begin(), column.child_end(), has_list_or_string); +} + cuda::std::uint32_t hash_csr_capacity(size_type rows, double load_factor) { auto const checked = checked_load_factor(load_factor); @@ -62,7 +70,11 @@ cuda::std::uint32_t hash_csr_capacity(size_type rows, double load_factor) CUDF_EXPECTS(capacity <= std::numeric_limits::max(), "HashCSR table capacity is not representable", std::overflow_error); - return static_cast(capacity); + // Avoid power-of-two rounding at the default and lower load factors. Retain the extra + // headroom of rounded capacities at higher load factors, where linear probing is sensitive + // to occupancy (in particular, load_factor == 1 must not produce an almost-full table). + return static_cast(checked <= CUCO_DESIRED_LOAD_FACTOR ? requested + : capacity); } } // namespace @@ -100,22 +112,26 @@ hash_join::hash_join(cudf::table_view const& right, 0xff, _impl->_entries.size() * sizeof(hash_table_entry_type), stream.get())); - CUDF_CUDA_TRY(cudaMemsetAsync(_impl->_cumulative_ends.data(), - 0, - _impl->_cumulative_ends.size() * sizeof(size_type), - stream.get())); + CUDF_CUDA_TRY(cudaMemsetAsync( + _impl->_offsets.data(), 0, _impl->_offsets.size() * sizeof(size_type), stream.get())); auto const temp_mr = cudf::get_current_device_resource_ref(); - auto const row_bitmask = cudf::detail::bitmask_and(right, stream, temp_mr).first; + auto const row_bitmask = _nulls_equal == null_equality::UNEQUAL + ? cudf::detail::bitmask_and(right, stream, temp_mr).first + : rmm::device_buffer{0, stream, temp_mr}; auto const valid_rows = _nulls_equal == null_equality::UNEQUAL ? static_cast(row_bitmask.data()) : nullptr; - rmm::device_uvector build_positions(right.num_rows(), stream, temp_mr); + // Hashing and comparing variable-width rows again can dominate construction. Cache one + // representative index per row for keys containing lists or strings, including within structs. + auto const cache_representatives = std::any_of(right.begin(), right.end(), has_list_or_string); + auto representatives = rmm::device_uvector{ + cache_representatives ? static_cast(right.num_rows()) : 0, stream, temp_mr}; auto build = [&](auto equality, auto hasher) { launch_hash_csr_build_count_kernel(right.num_rows(), valid_rows, - build_positions.data(), - _impl->_cumulative_ends.data(), + _impl->_offsets.data(), + representatives.data(), _impl->hash_table(), equality, hasher, @@ -123,25 +139,46 @@ hash_join::hash_join(cudf::table_view const& right, }; dispatch_join_comparator( right, right, _preprocessed_right, _preprocessed_right, _has_nulls, _nulls_equal, build); - std::size_t temp_storage_bytes{}; - CUDF_CUDA_TRY(cub::DeviceScan::InclusiveSum(nullptr, - temp_storage_bytes, - _impl->_cumulative_ends.data(), - _impl->_cumulative_ends.data(), - _impl->_capacity, - stream.get())); - rmm::device_buffer temp_storage(temp_storage_bytes, stream, temp_mr); - CUDF_CUDA_TRY(cub::DeviceScan::InclusiveSum(temp_storage.data(), - temp_storage_bytes, - _impl->_cumulative_ends.data(), - _impl->_cumulative_ends.data(), - _impl->_capacity, - stream.get())); - launch_hash_csr_build_fill_kernel(right.num_rows(), - build_positions.data(), - _impl->_cumulative_ends.data(), - _impl->_values.data(), - stream); + { + std::size_t temp_storage_bytes{}; + CUDF_CUDA_TRY(cub::DeviceScan::InclusiveSum(nullptr, + temp_storage_bytes, + _impl->_offsets.data(), + _impl->_offsets.data(), + _impl->_offsets.size(), + stream.get())); + rmm::device_buffer temp_storage(temp_storage_bytes, stream, temp_mr); + CUDF_CUDA_TRY(cub::DeviceScan::InclusiveSum(temp_storage.data(), + temp_storage_bytes, + _impl->_offsets.data(), + _impl->_offsets.data(), + _impl->_offsets.size(), + stream.get())); + } + // The output array is not needed until the scan workspace has been released. + _impl->_values.resize(right.num_rows(), stream); + // Reuse each cumulative end as a scatter cursor. Once all rows in a group have been + // scattered, its cursor is the group's exclusive begin. No per-row positions are retained. + if (cache_representatives) { + launch_hash_csr_build_fill_cached_kernel(right.num_rows(), + representatives.data(), + _impl->_offsets.data(), + _impl->_values.data(), + stream); + return; + } + auto fill = [&](auto equality, auto hasher) { + launch_hash_csr_build_fill_kernel(right.num_rows(), + valid_rows, + _impl->_offsets.data(), + _impl->_values.data(), + _impl->hash_table(), + equality, + hasher, + stream); + }; + dispatch_join_comparator( + right, right, _preprocessed_right, _preprocessed_right, _has_nulls, _nulls_equal, fill); } template hash_join::hash_join( diff --git a/cpp/src/join/hash_join/hash_join_impl.cuh b/cpp/src/join/hash_join/hash_join_impl.cuh index 06b2d74d8111..b67688372097 100644 --- a/cpp/src/join/hash_join/hash_join_impl.cuh +++ b/cpp/src/join/hash_join/hash_join_impl.cuh @@ -11,8 +11,11 @@ #include +#include +#include #include +#include #include #include @@ -26,24 +29,30 @@ struct hash_join::impl { cuda::mr::any_resource mr) : _mr(std::move(mr)), _entries(capacity, stream, _mr), - _cumulative_ends(capacity, stream, _mr), - _values(rows, stream, _mr), - _capacity(capacity) + _offsets(static_cast(rows) + 1, stream, _mr), + _values(0, stream, _mr), + _capacity(capacity), + _row_mask( + (cuda::std::uint32_t{1} << cuda::std::bit_width(static_cast(rows))) - + 1), + _modulo(capacity) { } hash_table_ref hash_table() const { - return {const_cast(_entries.data()), _capacity}; + return {const_cast(_entries.data()), _capacity, _row_mask, _modulo}; } - csr_ref csr() const { return {_cumulative_ends.data(), _values.data()}; } + csr_ref csr() const { return {_offsets.data(), _values.data()}; } cuda::mr::any_resource _mr; rmm::device_uvector _entries; - rmm::device_uvector _cumulative_ends; + rmm::device_uvector _offsets; rmm::device_uvector _values; cuda::std::uint32_t _capacity; + cuda::std::uint32_t _row_mask; + cuda::fast_mod_div _modulo; }; } // namespace cudf::detail diff --git a/cpp/src/join/hash_join/partitioned_join_retrieve.cu b/cpp/src/join/hash_join/partitioned_join_retrieve.cu index f214eb5d6215..a1d359409402 100644 --- a/cpp/src/join/hash_join/partitioned_join_retrieve.cu +++ b/cpp/src/join/hash_join/partitioned_join_retrieve.cu @@ -24,6 +24,8 @@ #include +#include + namespace cudf::detail { template std::pair>, @@ -83,61 +85,68 @@ hash_join::partitioned_join_retrieve(join_kind join, validate_hash_join_probe(_right, left_partition_view, _has_nulls); auto const temp_mr = cudf::get_current_device_resource_ref(); - auto const preprocessed_left = - cudf::detail::row::equality::preprocessed_table::create(left_partition_view, stream, temp_mr); - - auto counts = cudf::detail::make_zeroed_device_uvector_async( - static_cast(partition_size) + 1, stream, temp_mr); - CUDF_CUDA_TRY( - cudf::detail::memcpy_async(counts.data(), - match_ctx._match_counts->data() + left_start_idx, - static_cast(partition_size) * sizeof(size_type), - stream)); - auto offsets = cudf::detail::make_zeroed_device_uvector_async( - static_cast(partition_size) + 1, stream, temp_mr); - auto const output_size = cudf::detail::sizes_to_offsets( - counts.begin(), counts.end(), offsets.begin(), 0, stream, temp_mr); - CUDF_EXPECTS(output_size >= 0, "Join output size overflowed", std::overflow_error); - - rmm::device_uvector probe_slots(partition_size, stream, temp_mr); - auto const row_bitmask = cudf::detail::bitmask_and(left_partition_view, stream, temp_mr).first; - auto const valid_rows = _nulls_equal == null_equality::UNEQUAL - ? static_cast(row_bitmask.data()) - : nullptr; - auto save_slots = [&](auto equality, auto hasher) { - if (join == join_kind::INNER_JOIN) { - launch_hash_csr_probe_count_kernel(partition_size, - valid_rows, - probe_slots.data(), - nullptr, - nullptr, - nullptr, - _impl->hash_table(), - _impl->csr(), - equality, - hasher, - stream); - } else { - launch_hash_csr_probe_count_kernel(partition_size, - valid_rows, - probe_slots.data(), - nullptr, - nullptr, - nullptr, - _impl->hash_table(), - _impl->csr(), - equality, - hasher, - stream); - } - }; - dispatch_join_comparator(_right, - left_partition_view, - _preprocessed_right, - preprocessed_left, - _has_nulls, - _nulls_equal, - save_slots); + // Release copied counts and probe preprocessing before allocating the output. + auto [offsets, probe_groups, output_size] = [&] { + auto const preprocessed_left = + cudf::detail::row::equality::preprocessed_table::create(left_partition_view, stream, temp_mr); + + auto [offsets, output_size] = [&] { + auto counts = cudf::detail::make_zeroed_device_uvector_async( + static_cast(partition_size) + 1, stream, temp_mr); + CUDF_CUDA_TRY( + cudf::detail::memcpy_async(counts.data(), + match_ctx._match_counts->data() + left_start_idx, + static_cast(partition_size) * sizeof(size_type), + stream)); + auto offsets = cudf::detail::make_zeroed_device_uvector_async( + static_cast(partition_size) + 1, stream, temp_mr); + auto const output_size = cudf::detail::sizes_to_offsets( + counts.begin(), counts.end(), offsets.begin(), 0, stream, temp_mr); + CUDF_EXPECTS(output_size >= 0, "Join output size overflowed", std::overflow_error); + return std::pair(std::move(offsets), output_size); + }(); + + rmm::device_uvector probe_groups(partition_size, stream, temp_mr); + auto const row_bitmask = cudf::detail::bitmask_and(left_partition_view, stream, temp_mr).first; + auto const valid_rows = _nulls_equal == null_equality::UNEQUAL + ? static_cast(row_bitmask.data()) + : nullptr; + auto save_groups = [&](auto equality, auto hasher) { + if (join == join_kind::INNER_JOIN) { + launch_hash_csr_probe_count_kernel(partition_size, + valid_rows, + probe_groups.data(), + nullptr, + nullptr, + nullptr, + _impl->hash_table(), + _impl->csr(), + equality, + hasher, + stream); + } else { + launch_hash_csr_probe_count_kernel(partition_size, + valid_rows, + probe_groups.data(), + nullptr, + nullptr, + nullptr, + _impl->hash_table(), + _impl->csr(), + equality, + hasher, + stream); + } + }; + dispatch_join_comparator(_right, + left_partition_view, + _preprocessed_right, + preprocessed_left, + _has_nulls, + _nulls_equal, + save_groups); + return std::tuple(std::move(offsets), std::move(probe_groups), output_size); + }(); auto left_indices = std::make_unique>( static_cast(output_size), stream, mr); @@ -150,7 +159,7 @@ hash_join::partitioned_join_retrieve(join_kind join, launch_hash_csr_retrieve_kernel(output_size, partition_size, offsets.data(), - probe_slots.data(), + probe_groups.data(), _impl->csr(), left_start_idx, left_indices->data(), @@ -160,7 +169,7 @@ hash_join::partitioned_join_retrieve(join_kind join, launch_hash_csr_retrieve_kernel(output_size, partition_size, offsets.data(), - probe_slots.data(), + probe_groups.data(), _impl->csr(), left_start_idx, left_indices->data(), diff --git a/cpp/src/join/hash_join/retrieve_impl.cuh b/cpp/src/join/hash_join/retrieve_impl.cuh index c03bc4d67c28..4c9155e96ec6 100644 --- a/cpp/src/join/hash_join/retrieve_impl.cuh +++ b/cpp/src/join/hash_join/retrieve_impl.cuh @@ -23,6 +23,8 @@ #include +#include + namespace cudf::detail { template @@ -64,88 +66,98 @@ hash_join::join_retrieve(cudf::table_view const& left, } } - auto const preprocessed_left = cudf::detail::row::equality::preprocessed_table::create( - left, stream, cudf::get_current_device_resource_ref()); - auto const temp_mr = cudf::get_current_device_resource_ref(); - auto match_counts = cudf::detail::make_zeroed_device_uvector_async( - static_cast(left.num_rows()) + 1, stream, temp_mr); - rmm::device_uvector probe_slots(left.num_rows(), stream, temp_mr); - // A full join appends the unmatched right rows, so track which build rows the probe matched to - // size the output exactly. The other join kinds do not need it and skip the extra atomics. - auto matched_slots = Join == join_kind::FULL_JOIN + std::optional unmatched_right_rows; + // Release retrieval scratch before full-join finalization allocates its match flags. + auto join_indices = [&] { + // Only offsets and group IDs are needed to emit the output. Keep counting scratch out + // of the output allocation's lifetime. + auto [offsets, probe_groups, actual_size] = [&] { + auto const preprocessed_left = cudf::detail::row::equality::preprocessed_table::create( + left, stream, cudf::get_current_device_resource_ref()); + auto match_counts = cudf::detail::make_zeroed_device_uvector_async( + static_cast(left.num_rows()) + 1, stream, temp_mr); + rmm::device_uvector probe_groups(left.num_rows(), stream, temp_mr); + // A full join appends the unmatched right rows, so track which build rows the probe matched + // to size the output exactly. Other join kinds skip these flags and their atomics. + auto matched_groups = Join == join_kind::FULL_JOIN ? cudf::detail::make_zeroed_device_uvector_async( - _impl->_capacity, stream, temp_mr) + _right.num_rows(), stream, temp_mr) : rmm::device_uvector{0, stream, temp_mr}; - auto matched_build_rows = cudf::detail::device_scalar(0, stream, temp_mr); - auto const row_bitmask = cudf::detail::bitmask_and(left, stream, temp_mr).first; - auto const valid_rows = _nulls_equal == null_equality::UNEQUAL - ? static_cast(row_bitmask.data()) - : nullptr; - - auto count_matches = [&](auto equality, auto hasher) { - launch_hash_csr_probe_count_kernel( - left.num_rows(), - valid_rows, - probe_slots.data(), - match_counts.data(), - Join == join_kind::FULL_JOIN ? matched_slots.data() : nullptr, - matched_build_rows.data(), - _impl->hash_table(), - _impl->csr(), - equality, - hasher, - stream); - }; - dispatch_join_comparator( - _right, left, _preprocessed_right, preprocessed_left, _has_nulls, _nulls_equal, count_matches); - - auto offsets = cudf::detail::make_zeroed_device_uvector_async( - static_cast(left.num_rows()) + 1, stream, temp_mr); - auto const actual_size = cudf::detail::sizes_to_offsets( - match_counts.begin(), match_counts.end(), offsets.begin(), 0, stream, temp_mr); - CUDF_EXPECTS(actual_size >= 0, "Join output size overflowed", std::overflow_error); - auto const join_size = static_cast(actual_size); - - // A full join appends one entry per unmatched right row. The count pass already tallied the - // matched build rows, so the exact output size is known here and both the allocation below and - // `finalize_full_join` can use it: no worst-case reservation and no grow-then-shrink. - auto const unmatched_right_rows = [&]() -> std::optional { - if constexpr (Join == join_kind::FULL_JOIN) { - // Every build row is tallied at most once, so the count never exceeds the row count and - // narrowing to `size_type` here is safe. - auto const matched = matched_build_rows.value(stream); - return static_cast(static_cast(_right.num_rows()) - matched); - } else { - return std::nullopt; - } + auto matched_build_rows = + cudf::detail::device_scalar(0, stream, temp_mr); + auto const row_bitmask = cudf::detail::bitmask_and(left, stream, temp_mr).first; + auto const valid_rows = _nulls_equal == null_equality::UNEQUAL + ? static_cast(row_bitmask.data()) + : nullptr; + + auto count_matches = [&](auto equality, auto hasher) { + launch_hash_csr_probe_count_kernel( + left.num_rows(), + valid_rows, + probe_groups.data(), + match_counts.data(), + Join == join_kind::FULL_JOIN ? matched_groups.data() : nullptr, + matched_build_rows.data(), + _impl->hash_table(), + _impl->csr(), + equality, + hasher, + stream); + }; + dispatch_join_comparator(_right, + left, + _preprocessed_right, + preprocessed_left, + _has_nulls, + _nulls_equal, + count_matches); + + auto offsets = cudf::detail::make_zeroed_device_uvector_async( + static_cast(left.num_rows()) + 1, stream, temp_mr); + auto const actual_size = cudf::detail::sizes_to_offsets( + match_counts.begin(), match_counts.end(), offsets.begin(), 0, stream, temp_mr); + CUDF_EXPECTS(actual_size >= 0, "Join output size overflowed", std::overflow_error); + + // The count pass already tallied matched build rows. Preserve the exact complement size + // for output allocation and full-join finalization before releasing its device scalar. + if constexpr (Join == join_kind::FULL_JOIN) { + // Each build row is tallied at most once, so narrowing to size_type is safe. + auto const matched = matched_build_rows.value(stream); + unmatched_right_rows = + static_cast(static_cast(_right.num_rows()) - matched); + } + return std::tuple(std::move(offsets), std::move(probe_groups), actual_size); + }(); + + auto const join_size = static_cast(actual_size); + auto const allocation_size = + join_size + static_cast(unmatched_right_rows.value_or(size_type{0})); + // For a full join the final size includes the unmatched right rows, so validate only now. + validate_output_size(allocation_size); + + auto left_indices = + std::make_unique>(allocation_size, stream, mr); + auto right_indices = + std::make_unique>(allocation_size, stream, mr); + left_indices->resize(join_size, stream); + right_indices->resize(join_size, stream); + cudf::prefetch::detail::prefetch(*left_indices, stream); + cudf::prefetch::detail::prefetch(*right_indices, stream); + + launch_hash_csr_retrieve_kernel(actual_size, + left.num_rows(), + offsets.data(), + probe_groups.data(), + _impl->csr(), + 0, + left_indices->data(), + right_indices->data(), + stream); + + return std::pair(std::move(left_indices), std::move(right_indices)); }(); - auto const allocation_size = - join_size + static_cast(unmatched_right_rows.value_or(size_type{0})); - // For a full join the final size includes the unmatched right rows, so validate only now. - validate_output_size(allocation_size); - - auto left_indices = std::make_unique>(allocation_size, stream, mr); - auto right_indices = - std::make_unique>(allocation_size, stream, mr); - left_indices->resize(join_size, stream); - right_indices->resize(join_size, stream); - cudf::prefetch::detail::prefetch(*left_indices, stream); - cudf::prefetch::detail::prefetch(*right_indices, stream); - - launch_hash_csr_retrieve_kernel(actual_size, - left.num_rows(), - offsets.data(), - probe_slots.data(), - _impl->csr(), - 0, - left_indices->data(), - right_indices->data(), - stream); - - auto join_indices = std::pair(std::move(left_indices), std::move(right_indices)); - if constexpr (Join == join_kind::FULL_JOIN) { // The HashCSR retrieve kernels do not mark matched right rows, so let `finalize_full_join` // derive the match flags from the emitted right indices. diff --git a/cpp/src/join/hash_join/size_impl.cuh b/cpp/src/join/hash_join/size_impl.cuh index 5b900bcf2cb5..83f1e6afb185 100644 --- a/cpp/src/join/hash_join/size_impl.cuh +++ b/cpp/src/join/hash_join/size_impl.cuh @@ -89,8 +89,8 @@ std::size_t hash_join::join_size(cudf::table_view const& left, auto const temp_mr = cudf::get_current_device_resource_ref(); auto match_counts = cudf::detail::make_zeroed_device_uvector_async(left.num_rows(), stream, temp_mr); - auto matched_slots = cudf::detail::make_zeroed_device_uvector_async( - _impl->_capacity, stream, temp_mr); + auto matched_groups = cudf::detail::make_zeroed_device_uvector_async( + _right.num_rows(), stream, temp_mr); auto matched_build_rows = cudf::detail::device_scalar(0, stream, temp_mr); auto const row_bitmask = cudf::detail::bitmask_and(left, stream, temp_mr).first; auto const valid_rows = _nulls_equal == null_equality::UNEQUAL @@ -102,7 +102,7 @@ std::size_t hash_join::join_size(cudf::table_view const& left, valid_rows, nullptr, match_counts.data(), - matched_slots.data(), + matched_groups.data(), matched_build_rows.data(), _impl->hash_table(), _impl->csr(), diff --git a/cpp/tests/CMakeLists.txt b/cpp/tests/CMakeLists.txt index 5df5fff2ae48..729f0e041588 100644 --- a/cpp/tests/CMakeLists.txt +++ b/cpp/tests/CMakeLists.txt @@ -176,6 +176,7 @@ ConfigureTest( join/mixed_join_tests.cu join/direct_join_tests.cpp join/distinct_join_tests.cpp + join/hash_join_memory_tests.cpp join/key_remapping_tests.cpp join/streaming_hash_join_tests.cpp GPUS 1 diff --git a/cpp/tests/join/hash_join_memory_tests.cpp b/cpp/tests/join/hash_join_memory_tests.cpp new file mode 100644 index 000000000000..8afd28a943e2 --- /dev/null +++ b/cpp/tests/join/hash_join_memory_tests.cpp @@ -0,0 +1,264 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +namespace { + +using cudf::size_type; +using join_pair = std::pair; +using join_result = std::pair>, + std::unique_ptr>>; +template +using column_wrapper = cudf::test::fixed_width_column_wrapper; + +std::vector sorted_host_pairs(join_result const& result) +{ + auto const stream = cudf::test::get_default_stream(); + auto const left = + cudf::detail::make_host_vector(cudf::device_span{*result.first}, stream); + auto const right = + cudf::detail::make_host_vector(cudf::device_span{*result.second}, stream); + std::vector pairs; + pairs.reserve(left.size()); + for (std::size_t i = 0; i < left.size(); ++i) { + pairs.emplace_back(left[i], right[i]); + } + std::sort(pairs.begin(), pairs.end()); + return pairs; +} + +struct HashJoinMemoryTest : public cudf::test::BaseFixture {}; + +TEST_F(HashJoinMemoryTest, EqualFingerprintsStillCompareKeys) +{ + // These different int64 values have the same full Murmur3 hash, and therefore any shorter + // fingerprint also collides. Repeated rows must remain in separate equality groups. + constexpr int64_t first = 2708834937922957298; + constexpr int64_t second = 9018382015938582086; + column_wrapper collision_keys{{first, second}}; + auto const hashes = cudf::hashing::murmurhash3_x86_32(cudf::table_view{{collision_keys}}); + auto const host_hashes = cudf::detail::make_host_vector( + cudf::device_span{hashes->view().data(), 2}, + cudf::test::get_default_stream()); + ASSERT_EQ(host_hashes[0], host_hashes[1]); + // Seven build rows use 14 slots at the default load factor. Both keys start at the final + // slot, so insertion and lookup of one distinct key must wrap across slot zero. + ASSERT_EQ(host_hashes[0] % 14, 13); + + column_wrapper right{{first, second, first, second, second, first, second}}; + column_wrapper left{{second, first, int64_t{0}}}; + cudf::hash_join joiner{cudf::table_view{{right}}, cudf::null_equality::EQUAL}; + auto const result = joiner.left_join(cudf::table_view{{left}}); + std::vector const expected{ + {0, 1}, {0, 3}, {0, 4}, {0, 6}, {1, 0}, {1, 2}, {1, 5}, {2, cudf::JoinNoMatch}}; + EXPECT_EQ(sorted_host_pairs(result), expected); + + column_wrapper only_first{{first, first}}; + column_wrapper only_second{{second}}; + cudf::hash_join missing_joiner{cudf::table_view{{only_first}}, cudf::null_equality::EQUAL}; + EXPECT_EQ(missing_joiner.inner_join_size(cudf::table_view{{only_second}}), 0); +} + +TEST_F(HashJoinMemoryTest, RowIndexBitBoundaries) +{ + // Include the largest row ID on either side of changes in the number of row-index bits. + for (size_type const num_rows : {1, 2, 3, 255, 256, 257, 65535, 65536, 65537}) { + SCOPED_TRACE(num_rows); + std::vector keys(num_rows); + std::iota(keys.begin(), keys.end(), 0); + column_wrapper right(keys.begin(), keys.end()); + auto const table = cudf::table_view{{right}}; + cudf::hash_join joiner{table, cudf::null_equality::EQUAL}; + auto const result = joiner.inner_join(table); + auto const pairs = sorted_host_pairs(result); + ASSERT_EQ(pairs.size(), static_cast(num_rows)); + for (size_type i = 0; i < num_rows; ++i) { + ASSERT_EQ(pairs[i], std::make_pair(i, i)); + } + } +} + +TEST_F(HashJoinMemoryTest, UnevenGroupsNullsAndPartitions) +{ + // Groups cross warp and block boundaries, with gaps between their representative row IDs. + std::vector build_keys; + for (int32_t key = 0; key < 10; ++key) { + build_keys.insert(build_keys.end(), key * key + 1, key); + } + std::rotate(build_keys.begin(), build_keys.begin() + 37, build_keys.end()); + std::vector build_valid(build_keys.size(), true); + for (std::size_t i = 0; i < build_valid.size(); i += 7) { + build_valid[i] = false; + } + std::vector const probe_keys{8, 3, 0, 10, 8, 1, 5}; + std::vector const probe_valid{true, false, true, true, true, true, true}; + column_wrapper right(build_keys.begin(), build_keys.end(), build_valid.begin()); + column_wrapper left(probe_keys.begin(), probe_keys.end(), probe_valid.begin()); + auto const right_table = cudf::table_view{{right}}; + auto const left_table = cudf::table_view{{left}}; + + for (auto const nulls : {cudf::null_equality::EQUAL, cudf::null_equality::UNEQUAL}) { + SCOPED_TRACE(nulls == cudf::null_equality::EQUAL ? "nulls equal" : "nulls unequal"); + std::vector expected_inner; + std::vector expected_left; + std::vector matched_build(build_keys.size(), false); + for (size_type probe = 0; probe < static_cast(probe_keys.size()); ++probe) { + bool matched = false; + for (size_type build = 0; build < static_cast(build_keys.size()); ++build) { + auto const equal = + probe_valid[probe] && build_valid[build] + ? probe_keys[probe] == build_keys[build] + : !probe_valid[probe] && !build_valid[build] && nulls == cudf::null_equality::EQUAL; + if (equal) { + expected_inner.emplace_back(probe, build); + expected_left.emplace_back(probe, build); + matched_build[build] = true; + matched = true; + } + } + if (!matched) { expected_left.emplace_back(probe, cudf::JoinNoMatch); } + } + auto expected_full = expected_left; + for (size_type build = 0; build < static_cast(build_keys.size()); ++build) { + if (!matched_build[build]) { expected_full.emplace_back(cudf::JoinNoMatch, build); } + } + std::sort(expected_full.begin(), expected_full.end()); + + cudf::hash_join joiner{right_table, nulls}; + EXPECT_EQ(joiner.inner_join_size(left_table), expected_inner.size()); + EXPECT_EQ(joiner.left_join_size(left_table), expected_left.size()); + EXPECT_EQ(joiner.full_join_size(left_table), expected_full.size()); + EXPECT_EQ(sorted_host_pairs(joiner.inner_join(left_table)), expected_inner); + EXPECT_EQ(sorted_host_pairs(joiner.left_join(left_table)), expected_left); + EXPECT_EQ(sorted_host_pairs(joiner.full_join(left_table)), expected_full); + + auto matches = joiner.full_join_match_context(left_table); + auto partition = cudf::join_partition_context{ + std::make_unique(std::move(matches)), 0, 0}; + std::vector outputs; + for (size_type i = 0; i < left_table.num_rows(); ++i) { + partition.left_start_idx = i; + partition.left_end_idx = i + 1; + outputs.push_back(joiner.partitioned_full_join(partition)); + } + std::vector> left_parts; + std::vector> right_parts; + for (auto const& output : outputs) { + left_parts.emplace_back(*output.first); + right_parts.emplace_back(*output.second); + } + auto const finalized = cudf::hash_join::finalize_partitioned_full_join( + left_parts, right_parts, left_table.num_rows(), right_table.num_rows()); + EXPECT_EQ(sorted_host_pairs(finalized), expected_full); + } +} + +TEST_F(HashJoinMemoryTest, UnequalNestedNullsKeepAdjacentGroupsIntact) +{ + using lists = cudf::test::lists_column_wrapper; + using cudf::test::iterators::null_at; + // Rows containing a null child remain valid top-level rows but do not compare equal to + // themselves. Their CSR segments must still be filled without disturbing neighboring groups. + // The final top-level null row is excluded and must leave a sentinel in the build-row cache. + lists right{ + {{{2, 0}, null_at(1)}, {1}, {{2, 0}, null_at(1)}, {1}, {}, {{3, 0}, null_at(1)}, {}, {2}, {}}, + null_at(8)}; + lists left{{{1}, {{2, 0}, null_at(1)}, {}, {2}, {4}, {}}, null_at(5)}; + auto const left_table = cudf::table_view{{left}}; + cudf::hash_join joiner{cudf::table_view{{right}}, cudf::null_equality::UNEQUAL}; + std::vector const expected_inner{{0, 1}, {0, 3}, {2, 4}, {2, 6}, {3, 7}}; + EXPECT_EQ(joiner.inner_join_size(left_table), expected_inner.size()); + EXPECT_EQ(sorted_host_pairs(joiner.inner_join(left_table)), expected_inner); + + auto expected_full = expected_inner; + expected_full.insert(expected_full.end(), + {{1, cudf::JoinNoMatch}, + {4, cudf::JoinNoMatch}, + {5, cudf::JoinNoMatch}, + {cudf::JoinNoMatch, 0}, + {cudf::JoinNoMatch, 2}, + {cudf::JoinNoMatch, 5}, + {cudf::JoinNoMatch, 8}}); + std::sort(expected_full.begin(), expected_full.end()); + EXPECT_EQ(joiner.full_join_size(left_table), expected_full.size()); + EXPECT_EQ(sorted_host_pairs(joiner.full_join(left_table)), expected_full); +} + +TEST_F(HashJoinMemoryTest, ConstructorPeakIncludesTemporaryAllocations) +{ + constexpr size_type num_rows = 1 << 20; + auto const stream = cudf::test::get_default_stream(); + for (size_type const cardinality : {size_type{1}, num_rows}) { + SCOPED_TRACE(cardinality); + std::vector keys(num_rows); + for (size_type i = 0; i < num_rows; ++i) { + keys[i] = i % cardinality; + } + column_wrapper right(keys.begin(), keys.end()); + auto const measure_peak = [&](cudf::column_view input, + std::string const& key_type, + std::size_t bytes_per_row) { + // Input storage is outside the tracked resource. Track the persistent and current resources + // together so moving construction scratch between them cannot hide a memory regression. + auto tracked = rmm::mr::statistics_resource_adaptor{cudf::get_current_device_resource_ref()}; + cudf::test::scoped_current_device_resource current{tracked}; + { + cudf::hash_join joiner{cudf::table_view{{input}}, + cudf::nullable_join::NO, + cudf::null_equality::EQUAL, + 0.5, + stream, + tracked}; + stream.sync(); + auto const peak = tracked.get_bytes_counter().peak; + RecordProperty( + key_type + (cardinality == 1 ? "_all_same_peak_bytes" : "_unique_peak_bytes"), + std::to_string(peak)); + EXPECT_LE(peak, bytes_per_row * num_rows); + } + stream.sync(); + EXPECT_EQ(tracked.get_bytes_counter().value, 0); + }; + // Leave room for preprocessing and allocator/CUB variation while rejecting the previous + // 36-byte-per-row constructor peak. LIST and STRING keys retain a temporary representative. + measure_peak(right, "int32", 24); + auto const string_right = cudf::strings::from_integers(right, stream); + measure_peak(string_right->view(), "string", 32); + std::vector offsets(num_rows + 1); + std::iota(offsets.begin(), offsets.end(), 0); + column_wrapper list_offsets(offsets.begin(), offsets.end()); + auto const list_right = + cudf::make_lists_column(num_rows, list_offsets.release(), right.release(), 0, {}); + measure_peak(list_right->view(), "list_int32", 32); + } +} + +} // namespace From 815c3ca523e4b545f2a38ff1c00627f331912922 Mon Sep 17 00:00:00 2001 From: Yunsong Wang Date: Fri, 18 Sep 2026 15:03:01 -0700 Subject: [PATCH 2/5] Use cuda::std::pair for HashCSR query arguments Replace temporary cuco pairs in HashCSR lookup and equality with cuda::std::pair. The stored entries and atomic operations remain uint32_t, so the stronger pair alignment is unnecessary. Remove an unused join helper include and update the comparator comments. --- cpp/src/join/hash_join/dispatch.cuh | 14 +++++++------- cpp/src/join/hash_join/hash_csr.cuh | 9 +++++---- cpp/src/join/hash_join/hash_csr_kernels.cuh | 7 ++++--- cpp/src/join/hash_join/hash_join.cu | 1 - 4 files changed, 16 insertions(+), 15 deletions(-) diff --git a/cpp/src/join/hash_join/dispatch.cuh b/cpp/src/join/hash_join/dispatch.cuh index f68893c45cf2..f43622c42280 100644 --- a/cpp/src/join/hash_join/dispatch.cuh +++ b/cpp/src/join/hash_join/dispatch.cuh @@ -10,7 +10,7 @@ #include #include -#include +#include #include #include @@ -18,7 +18,7 @@ namespace cudf::detail { /** - * @brief Equality comparator for cuco hash table probing with row-level equality. + * @brief Equality comparator for hash table probing with row-level equality. */ template class pair_equal { @@ -26,8 +26,8 @@ class pair_equal { pair_equal(Equal check_row_equality) : _check_row_equality{std::move(check_row_equality)} {} __device__ __forceinline__ bool operator()( - cuco::pair const& lhs, - cuco::pair const& rhs) const noexcept + cuda::std::pair const& lhs, + cuda::std::pair const& rhs) const noexcept { using detail::row::lhs_index_type; using detail::row::rhs_index_type; @@ -41,7 +41,7 @@ class pair_equal { }; /** - * @brief Equality comparator for cuco hash table probing with primitive row equality. + * @brief Equality comparator for hash table probing with primitive row equality. */ class primitive_pair_equal { public: @@ -51,8 +51,8 @@ class primitive_pair_equal { } __device__ __forceinline__ bool operator()( - cuco::pair const& lhs, - cuco::pair const& rhs) const noexcept + cuda::std::pair const& lhs, + cuda::std::pair const& rhs) const noexcept { return lhs.first == rhs.first and _check_row_equality(lhs.second, rhs.second); } diff --git a/cpp/src/join/hash_join/hash_csr.cuh b/cpp/src/join/hash_join/hash_csr.cuh index 952898b3a047..f984856fc978 100644 --- a/cpp/src/join/hash_join/hash_csr.cuh +++ b/cpp/src/join/hash_join/hash_csr.cuh @@ -9,10 +9,10 @@ #include #include -#include #include #include #include +#include namespace cudf::detail { @@ -29,7 +29,7 @@ struct hash_table_ref { cuda::fast_mod_div modulo; template - __device__ bool equal(cuco::pair key, + __device__ bool equal(cuda::std::pair key, hash_table_entry_type entry, Equal check_row_equality) const { @@ -38,7 +38,8 @@ struct hash_table_ref { } template - __device__ size_type insert(cuco::pair key, Equal equal_rows) const + __device__ size_type insert(cuda::std::pair key, + Equal equal_rows) const { auto const desired = (key.first & ~row_mask) | static_cast(key.second); auto slot = key.first % modulo; @@ -57,7 +58,7 @@ struct hash_table_ref { } template - __device__ size_type find(cuco::pair key, Equal equal_rows) const + __device__ size_type find(cuda::std::pair key, Equal equal_rows) const { auto slot = key.first % modulo; for (cuda::std::uint32_t step = 0; step < capacity; ++step) { diff --git a/cpp/src/join/hash_join/hash_csr_kernels.cuh b/cpp/src/join/hash_join/hash_csr_kernels.cuh index 4e9eb3b1d396..9d0eaa53e985 100644 --- a/cpp/src/join/hash_join/hash_csr_kernels.cuh +++ b/cpp/src/join/hash_join/hash_csr_kernels.cuh @@ -17,6 +17,7 @@ #include #include #include +#include #include namespace cudf::detail { @@ -40,7 +41,7 @@ CUDF_KERNEL void hash_csr_build_count_kernel(size_type num_rows, auto const index = static_cast(row); auto representative = size_type{CUDF_SIZE_TYPE_SENTINEL}; if (valid_rows == nullptr || cudf::bit_is_set(valid_rows, index)) { - representative = map.insert(cuco::pair{hasher(index), index}, equal); + representative = map.insert(cuda::std::pair{hasher(index), index}, equal); } // Initialize excluded rows too: the cached fill pass only reads representatives. if (representatives != nullptr) { representatives[index] = representative; } @@ -99,7 +100,7 @@ CUDF_KERNEL void hash_csr_build_fill_kernel(size_type num_rows, for (auto row = grid_1d::global_thread_id(); row < num_rows; row += stride) { auto const index = static_cast(row); if (valid_rows != nullptr && !cudf::bit_is_set(valid_rows, index)) { continue; } - auto const representative = map.find(cuco::pair{hasher(index), index}, equal); + auto const representative = map.find(cuda::std::pair{hasher(index), index}, equal); if (representative == CUDF_SIZE_TYPE_SENTINEL) { continue; } hash_csr_scatter_build_row(index, representative, offsets, values); } @@ -122,7 +123,7 @@ CUDF_KERNEL void hash_csr_probe_count_kernel(size_type num_rows, auto const index = static_cast(row); auto group = size_type{CUDF_SIZE_TYPE_SENTINEL}; if (valid_rows == nullptr || cudf::bit_is_set(valid_rows, index)) { - group = map.find(cuco::pair{hasher(index), index}, equal); + group = map.find(cuda::std::pair{hasher(index), index}, equal); } auto const found = group != CUDF_SIZE_TYPE_SENTINEL; diff --git a/cpp/src/join/hash_join/hash_join.cu b/cpp/src/join/hash_join/hash_join.cu index ee2b6506824f..e131718a8c44 100644 --- a/cpp/src/join/hash_join/hash_join.cu +++ b/cpp/src/join/hash_join/hash_join.cu @@ -6,7 +6,6 @@ #include "common.cuh" #include "dispatch.cuh" #include "hash_csr_kernels.cuh" -#include "join/join_common_utils.cuh" #include #include From 838f3848da0d23691095bd33311c34e345c2c6ae Mon Sep 17 00:00:00 2001 From: Yunsong Wang Date: Fri, 18 Sep 2026 16:03:50 -0700 Subject: [PATCH 3/5] Remove added HashCSR join unit tests Remove the dedicated HashCSR memory regression test file and its JOIN_TEST registration. Retain the existing join suites and the implementation. --- cpp/tests/CMakeLists.txt | 1 - cpp/tests/join/hash_join_memory_tests.cpp | 264 ---------------------- 2 files changed, 265 deletions(-) delete mode 100644 cpp/tests/join/hash_join_memory_tests.cpp diff --git a/cpp/tests/CMakeLists.txt b/cpp/tests/CMakeLists.txt index f739e07f1aeb..d9b0c0d403d8 100644 --- a/cpp/tests/CMakeLists.txt +++ b/cpp/tests/CMakeLists.txt @@ -176,7 +176,6 @@ ConfigureTest( join/mixed_join_tests.cu join/direct_join_tests.cpp join/distinct_join_tests.cpp - join/hash_join_memory_tests.cpp join/key_remapping_tests.cpp join/streaming_hash_join_tests.cpp GPUS 1 diff --git a/cpp/tests/join/hash_join_memory_tests.cpp b/cpp/tests/join/hash_join_memory_tests.cpp deleted file mode 100644 index 8afd28a943e2..000000000000 --- a/cpp/tests/join/hash_join_memory_tests.cpp +++ /dev/null @@ -1,264 +0,0 @@ -/* - * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. - * SPDX-License-Identifier: Apache-2.0 - */ - -#include -#include -#include -#include -#include - -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include -#include - -#include -#include -#include -#include -#include -#include -#include - -namespace { - -using cudf::size_type; -using join_pair = std::pair; -using join_result = std::pair>, - std::unique_ptr>>; -template -using column_wrapper = cudf::test::fixed_width_column_wrapper; - -std::vector sorted_host_pairs(join_result const& result) -{ - auto const stream = cudf::test::get_default_stream(); - auto const left = - cudf::detail::make_host_vector(cudf::device_span{*result.first}, stream); - auto const right = - cudf::detail::make_host_vector(cudf::device_span{*result.second}, stream); - std::vector pairs; - pairs.reserve(left.size()); - for (std::size_t i = 0; i < left.size(); ++i) { - pairs.emplace_back(left[i], right[i]); - } - std::sort(pairs.begin(), pairs.end()); - return pairs; -} - -struct HashJoinMemoryTest : public cudf::test::BaseFixture {}; - -TEST_F(HashJoinMemoryTest, EqualFingerprintsStillCompareKeys) -{ - // These different int64 values have the same full Murmur3 hash, and therefore any shorter - // fingerprint also collides. Repeated rows must remain in separate equality groups. - constexpr int64_t first = 2708834937922957298; - constexpr int64_t second = 9018382015938582086; - column_wrapper collision_keys{{first, second}}; - auto const hashes = cudf::hashing::murmurhash3_x86_32(cudf::table_view{{collision_keys}}); - auto const host_hashes = cudf::detail::make_host_vector( - cudf::device_span{hashes->view().data(), 2}, - cudf::test::get_default_stream()); - ASSERT_EQ(host_hashes[0], host_hashes[1]); - // Seven build rows use 14 slots at the default load factor. Both keys start at the final - // slot, so insertion and lookup of one distinct key must wrap across slot zero. - ASSERT_EQ(host_hashes[0] % 14, 13); - - column_wrapper right{{first, second, first, second, second, first, second}}; - column_wrapper left{{second, first, int64_t{0}}}; - cudf::hash_join joiner{cudf::table_view{{right}}, cudf::null_equality::EQUAL}; - auto const result = joiner.left_join(cudf::table_view{{left}}); - std::vector const expected{ - {0, 1}, {0, 3}, {0, 4}, {0, 6}, {1, 0}, {1, 2}, {1, 5}, {2, cudf::JoinNoMatch}}; - EXPECT_EQ(sorted_host_pairs(result), expected); - - column_wrapper only_first{{first, first}}; - column_wrapper only_second{{second}}; - cudf::hash_join missing_joiner{cudf::table_view{{only_first}}, cudf::null_equality::EQUAL}; - EXPECT_EQ(missing_joiner.inner_join_size(cudf::table_view{{only_second}}), 0); -} - -TEST_F(HashJoinMemoryTest, RowIndexBitBoundaries) -{ - // Include the largest row ID on either side of changes in the number of row-index bits. - for (size_type const num_rows : {1, 2, 3, 255, 256, 257, 65535, 65536, 65537}) { - SCOPED_TRACE(num_rows); - std::vector keys(num_rows); - std::iota(keys.begin(), keys.end(), 0); - column_wrapper right(keys.begin(), keys.end()); - auto const table = cudf::table_view{{right}}; - cudf::hash_join joiner{table, cudf::null_equality::EQUAL}; - auto const result = joiner.inner_join(table); - auto const pairs = sorted_host_pairs(result); - ASSERT_EQ(pairs.size(), static_cast(num_rows)); - for (size_type i = 0; i < num_rows; ++i) { - ASSERT_EQ(pairs[i], std::make_pair(i, i)); - } - } -} - -TEST_F(HashJoinMemoryTest, UnevenGroupsNullsAndPartitions) -{ - // Groups cross warp and block boundaries, with gaps between their representative row IDs. - std::vector build_keys; - for (int32_t key = 0; key < 10; ++key) { - build_keys.insert(build_keys.end(), key * key + 1, key); - } - std::rotate(build_keys.begin(), build_keys.begin() + 37, build_keys.end()); - std::vector build_valid(build_keys.size(), true); - for (std::size_t i = 0; i < build_valid.size(); i += 7) { - build_valid[i] = false; - } - std::vector const probe_keys{8, 3, 0, 10, 8, 1, 5}; - std::vector const probe_valid{true, false, true, true, true, true, true}; - column_wrapper right(build_keys.begin(), build_keys.end(), build_valid.begin()); - column_wrapper left(probe_keys.begin(), probe_keys.end(), probe_valid.begin()); - auto const right_table = cudf::table_view{{right}}; - auto const left_table = cudf::table_view{{left}}; - - for (auto const nulls : {cudf::null_equality::EQUAL, cudf::null_equality::UNEQUAL}) { - SCOPED_TRACE(nulls == cudf::null_equality::EQUAL ? "nulls equal" : "nulls unequal"); - std::vector expected_inner; - std::vector expected_left; - std::vector matched_build(build_keys.size(), false); - for (size_type probe = 0; probe < static_cast(probe_keys.size()); ++probe) { - bool matched = false; - for (size_type build = 0; build < static_cast(build_keys.size()); ++build) { - auto const equal = - probe_valid[probe] && build_valid[build] - ? probe_keys[probe] == build_keys[build] - : !probe_valid[probe] && !build_valid[build] && nulls == cudf::null_equality::EQUAL; - if (equal) { - expected_inner.emplace_back(probe, build); - expected_left.emplace_back(probe, build); - matched_build[build] = true; - matched = true; - } - } - if (!matched) { expected_left.emplace_back(probe, cudf::JoinNoMatch); } - } - auto expected_full = expected_left; - for (size_type build = 0; build < static_cast(build_keys.size()); ++build) { - if (!matched_build[build]) { expected_full.emplace_back(cudf::JoinNoMatch, build); } - } - std::sort(expected_full.begin(), expected_full.end()); - - cudf::hash_join joiner{right_table, nulls}; - EXPECT_EQ(joiner.inner_join_size(left_table), expected_inner.size()); - EXPECT_EQ(joiner.left_join_size(left_table), expected_left.size()); - EXPECT_EQ(joiner.full_join_size(left_table), expected_full.size()); - EXPECT_EQ(sorted_host_pairs(joiner.inner_join(left_table)), expected_inner); - EXPECT_EQ(sorted_host_pairs(joiner.left_join(left_table)), expected_left); - EXPECT_EQ(sorted_host_pairs(joiner.full_join(left_table)), expected_full); - - auto matches = joiner.full_join_match_context(left_table); - auto partition = cudf::join_partition_context{ - std::make_unique(std::move(matches)), 0, 0}; - std::vector outputs; - for (size_type i = 0; i < left_table.num_rows(); ++i) { - partition.left_start_idx = i; - partition.left_end_idx = i + 1; - outputs.push_back(joiner.partitioned_full_join(partition)); - } - std::vector> left_parts; - std::vector> right_parts; - for (auto const& output : outputs) { - left_parts.emplace_back(*output.first); - right_parts.emplace_back(*output.second); - } - auto const finalized = cudf::hash_join::finalize_partitioned_full_join( - left_parts, right_parts, left_table.num_rows(), right_table.num_rows()); - EXPECT_EQ(sorted_host_pairs(finalized), expected_full); - } -} - -TEST_F(HashJoinMemoryTest, UnequalNestedNullsKeepAdjacentGroupsIntact) -{ - using lists = cudf::test::lists_column_wrapper; - using cudf::test::iterators::null_at; - // Rows containing a null child remain valid top-level rows but do not compare equal to - // themselves. Their CSR segments must still be filled without disturbing neighboring groups. - // The final top-level null row is excluded and must leave a sentinel in the build-row cache. - lists right{ - {{{2, 0}, null_at(1)}, {1}, {{2, 0}, null_at(1)}, {1}, {}, {{3, 0}, null_at(1)}, {}, {2}, {}}, - null_at(8)}; - lists left{{{1}, {{2, 0}, null_at(1)}, {}, {2}, {4}, {}}, null_at(5)}; - auto const left_table = cudf::table_view{{left}}; - cudf::hash_join joiner{cudf::table_view{{right}}, cudf::null_equality::UNEQUAL}; - std::vector const expected_inner{{0, 1}, {0, 3}, {2, 4}, {2, 6}, {3, 7}}; - EXPECT_EQ(joiner.inner_join_size(left_table), expected_inner.size()); - EXPECT_EQ(sorted_host_pairs(joiner.inner_join(left_table)), expected_inner); - - auto expected_full = expected_inner; - expected_full.insert(expected_full.end(), - {{1, cudf::JoinNoMatch}, - {4, cudf::JoinNoMatch}, - {5, cudf::JoinNoMatch}, - {cudf::JoinNoMatch, 0}, - {cudf::JoinNoMatch, 2}, - {cudf::JoinNoMatch, 5}, - {cudf::JoinNoMatch, 8}}); - std::sort(expected_full.begin(), expected_full.end()); - EXPECT_EQ(joiner.full_join_size(left_table), expected_full.size()); - EXPECT_EQ(sorted_host_pairs(joiner.full_join(left_table)), expected_full); -} - -TEST_F(HashJoinMemoryTest, ConstructorPeakIncludesTemporaryAllocations) -{ - constexpr size_type num_rows = 1 << 20; - auto const stream = cudf::test::get_default_stream(); - for (size_type const cardinality : {size_type{1}, num_rows}) { - SCOPED_TRACE(cardinality); - std::vector keys(num_rows); - for (size_type i = 0; i < num_rows; ++i) { - keys[i] = i % cardinality; - } - column_wrapper right(keys.begin(), keys.end()); - auto const measure_peak = [&](cudf::column_view input, - std::string const& key_type, - std::size_t bytes_per_row) { - // Input storage is outside the tracked resource. Track the persistent and current resources - // together so moving construction scratch between them cannot hide a memory regression. - auto tracked = rmm::mr::statistics_resource_adaptor{cudf::get_current_device_resource_ref()}; - cudf::test::scoped_current_device_resource current{tracked}; - { - cudf::hash_join joiner{cudf::table_view{{input}}, - cudf::nullable_join::NO, - cudf::null_equality::EQUAL, - 0.5, - stream, - tracked}; - stream.sync(); - auto const peak = tracked.get_bytes_counter().peak; - RecordProperty( - key_type + (cardinality == 1 ? "_all_same_peak_bytes" : "_unique_peak_bytes"), - std::to_string(peak)); - EXPECT_LE(peak, bytes_per_row * num_rows); - } - stream.sync(); - EXPECT_EQ(tracked.get_bytes_counter().value, 0); - }; - // Leave room for preprocessing and allocator/CUB variation while rejecting the previous - // 36-byte-per-row constructor peak. LIST and STRING keys retain a temporary representative. - measure_peak(right, "int32", 24); - auto const string_right = cudf::strings::from_integers(right, stream); - measure_peak(string_right->view(), "string", 32); - std::vector offsets(num_rows + 1); - std::iota(offsets.begin(), offsets.end(), 0); - column_wrapper list_offsets(offsets.begin(), offsets.end()); - auto const list_right = - cudf::make_lists_column(num_rows, list_offsets.release(), right.release(), 0, {}); - measure_peak(list_right->view(), "list_int32", 32); - } -} - -} // namespace From c2a6819565cd167d27217ab507e8849745546dc9 Mon Sep 17 00:00:00 2001 From: Yunsong Wang Date: Fri, 18 Sep 2026 16:22:47 -0700 Subject: [PATCH 4/5] Use slot terminology for HashCSR hash tables --- cpp/src/join/hash_join/hash_csr.cuh | 30 ++++++++++++----------- cpp/src/join/hash_join/hash_join.cu | 6 ++--- cpp/src/join/hash_join/hash_join_impl.cuh | 6 ++--- 3 files changed, 21 insertions(+), 21 deletions(-) diff --git a/cpp/src/join/hash_join/hash_csr.cuh b/cpp/src/join/hash_join/hash_csr.cuh index f984856fc978..06b4b6342839 100644 --- a/cpp/src/join/hash_join/hash_csr.cuh +++ b/cpp/src/join/hash_join/hash_csr.cuh @@ -16,25 +16,27 @@ namespace cudf::detail { -/// A row index and the high bits of its hash share one 32-bit table entry. -using hash_table_entry_type = cuda::std::uint32_t; +/// A row index and the high bits of its hash share one 32-bit hash table slot. +using hash_table_slot_type = cuda::std::uint32_t; /// Device-side view of the open-addressed table. The low bits hold the representative build /// row, and the remaining bits hold a hash fingerprint. Fingerprint matches always undergo /// a full row comparison; reducing the fingerprint width cannot introduce false matches. +/// For N build rows, bit_width(N) bits encode the row index. The all-ones row index is +/// outside [0, N), leaving the all-ones slot available as the empty sentinel. struct hash_table_ref { - hash_table_entry_type* entries; + hash_table_slot_type* slots; cuda::std::uint32_t capacity; cuda::std::uint32_t row_mask; cuda::fast_mod_div modulo; template __device__ bool equal(cuda::std::pair key, - hash_table_entry_type entry, + hash_table_slot_type slot, Equal check_row_equality) const { - return ((key.first ^ entry) & ~row_mask) == 0 && - check_row_equality(key, {key.first, static_cast(entry & row_mask)}); + return ((key.first ^ slot) & ~row_mask) == 0 && + check_row_equality(key, {key.first, static_cast(slot & row_mask)}); } template @@ -44,10 +46,10 @@ struct hash_table_ref { auto const desired = (key.first & ~row_mask) | static_cast(key.second); auto slot = key.first % modulo; for (cuda::std::uint32_t step = 0; step < capacity; ++step) { - auto entry_ref = - cuda::atomic_ref{entries[slot]}; - auto old = hash_table_entry_type{-1}; - if (entry_ref.compare_exchange_strong(old, desired, cuda::memory_order_relaxed)) { + auto slot_ref = + cuda::atomic_ref{slots[slot]}; + auto old = hash_table_slot_type{-1}; + if (slot_ref.compare_exchange_strong(old, desired, cuda::memory_order_relaxed)) { return key.second; } if (equal(key, old, equal_rows)) { return static_cast(old & row_mask); } @@ -62,10 +64,10 @@ struct hash_table_ref { { auto slot = key.first % modulo; for (cuda::std::uint32_t step = 0; step < capacity; ++step) { - auto const current = entries[slot]; - if (current == hash_table_entry_type{-1}) { return CUDF_SIZE_TYPE_SENTINEL; } + auto const current = slots[slot]; + if (current == hash_table_slot_type{-1}) { return CUDF_SIZE_TYPE_SENTINEL; } // Under null_equality::UNEQUAL a nested row containing nulls need not equal itself. - // The fill pass must still find the row that claimed this entry during construction. + // The fill pass must still find the row that claimed this slot during construction. if constexpr (IsBuild) { if (static_cast(current & row_mask) == key.second) { return key.second; } } @@ -78,7 +80,7 @@ struct hash_table_ref { }; /// CSR segments are indexed by representative build row, including zero-length segments for -/// rows that did not claim a table entry. This avoids an offset for every empty hash slot. +/// rows that did not claim a hash table slot. This avoids an offset for every empty hash slot. struct csr_ref { size_type const* offsets; size_type const* values; diff --git a/cpp/src/join/hash_join/hash_join.cu b/cpp/src/join/hash_join/hash_join.cu index e131718a8c44..6df2ed69d9bc 100644 --- a/cpp/src/join/hash_join/hash_join.cu +++ b/cpp/src/join/hash_join/hash_join.cu @@ -107,10 +107,8 @@ hash_join::hash_join(cudf::table_view const& right, CUDF_EXPECTS(0 != right.num_columns(), "Hash join right table is empty", std::invalid_argument); if (_is_empty) { return; } - CUDF_CUDA_TRY(cudaMemsetAsync(_impl->_entries.data(), - 0xff, - _impl->_entries.size() * sizeof(hash_table_entry_type), - stream.get())); + CUDF_CUDA_TRY(cudaMemsetAsync( + _impl->_slots.data(), 0xff, _impl->_slots.size() * sizeof(hash_table_slot_type), stream.get())); CUDF_CUDA_TRY(cudaMemsetAsync( _impl->_offsets.data(), 0, _impl->_offsets.size() * sizeof(size_type), stream.get())); diff --git a/cpp/src/join/hash_join/hash_join_impl.cuh b/cpp/src/join/hash_join/hash_join_impl.cuh index b67688372097..264ff3712347 100644 --- a/cpp/src/join/hash_join/hash_join_impl.cuh +++ b/cpp/src/join/hash_join/hash_join_impl.cuh @@ -28,7 +28,7 @@ struct hash_join::impl { cuda::stream_ref stream, cuda::mr::any_resource mr) : _mr(std::move(mr)), - _entries(capacity, stream, _mr), + _slots(capacity, stream, _mr), _offsets(static_cast(rows) + 1, stream, _mr), _values(0, stream, _mr), _capacity(capacity), @@ -41,13 +41,13 @@ struct hash_join::impl { hash_table_ref hash_table() const { - return {const_cast(_entries.data()), _capacity, _row_mask, _modulo}; + return {const_cast(_slots.data()), _capacity, _row_mask, _modulo}; } csr_ref csr() const { return {_offsets.data(), _values.data()}; } cuda::mr::any_resource _mr; - rmm::device_uvector _entries; + rmm::device_uvector _slots; rmm::device_uvector _offsets; rmm::device_uvector _values; cuda::std::uint32_t _capacity; From cd8454a9c9521f6f83ed31fa937af80e7c9e2ce4 Mon Sep 17 00:00:00 2001 From: Yunsong Wang Date: Mon, 21 Sep 2026 10:35:31 -0700 Subject: [PATCH 5/5] Use explicit maximum value for HashCSR empty slots --- cpp/src/join/hash_join/hash_csr.cuh | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/cpp/src/join/hash_join/hash_csr.cuh b/cpp/src/join/hash_join/hash_csr.cuh index 06b4b6342839..9d8def356a6a 100644 --- a/cpp/src/join/hash_join/hash_csr.cuh +++ b/cpp/src/join/hash_join/hash_csr.cuh @@ -12,6 +12,7 @@ #include #include #include +#include #include namespace cudf::detail { @@ -48,7 +49,7 @@ struct hash_table_ref { for (cuda::std::uint32_t step = 0; step < capacity; ++step) { auto slot_ref = cuda::atomic_ref{slots[slot]}; - auto old = hash_table_slot_type{-1}; + auto old = cuda::std::numeric_limits::max(); if (slot_ref.compare_exchange_strong(old, desired, cuda::memory_order_relaxed)) { return key.second; } @@ -65,7 +66,9 @@ struct hash_table_ref { auto slot = key.first % modulo; for (cuda::std::uint32_t step = 0; step < capacity; ++step) { auto const current = slots[slot]; - if (current == hash_table_slot_type{-1}) { return CUDF_SIZE_TYPE_SENTINEL; } + if (current == cuda::std::numeric_limits::max()) { + return CUDF_SIZE_TYPE_SENTINEL; + } // Under null_equality::UNEQUAL a nested row containing nulls need not equal itself. // The fill pass must still find the row that claimed this slot during construction. if constexpr (IsBuild) {