diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt index 4ba016d0da6c..4b5887d4ec19 100644 --- a/cpp/CMakeLists.txt +++ b/cpp/CMakeLists.txt @@ -879,12 +879,8 @@ add_library( src/join/key_remapping.cu src/join/mark_join.cu src/join/mixed_join.cu - src/join/mixed_join_kernel.cu - src/join/mixed_join_kernel_nulls.cu src/join/mixed_join_kernels_semi.cu src/join/mixed_join_semi.cu - src/join/mixed_join_size_kernel.cu - src/join/mixed_join_size_kernel_nulls.cu src/join/sort_merge_join.cu src/json/json_path.cu src/lists/contains.cu diff --git a/cpp/include/cudf/detail/join/join.hpp b/cpp/include/cudf/detail/join/join.hpp index bf9fd5d42def..92d43eb055a7 100644 --- a/cpp/include/cudf/detail/join/join.hpp +++ b/cpp/include/cudf/detail/join/join.hpp @@ -1,15 +1,51 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ #pragma once +#include +#include +#include #include +#include + +#include +#include +#include + +#include +#include +#include +#include namespace cudf { namespace detail { constexpr int DEFAULT_JOIN_CG_SIZE = 2; +/** + * @brief Internal `filter_join_indices` accepting a precomputed output size. + * + * Same semantics as `cudf::filter_join_indices`. When `output_size` is provided it is used directly + * to size the output, skipping the internal size-counting pass. The value must equal the size that + * the function would otherwise compute (for example the result of `filter_join_indices_output_size` + * for the same inputs); behavior is undefined otherwise. + * + * @param output_size Optional precomputed number of output rows; computed internally if not + * provided + */ +std::pair>, + std::unique_ptr>> +filter_join_indices(table_view const& left, + table_view const& right, + device_span left_indices, + device_span right_indices, + ast::expression const& predicate, + join_kind join_kind, + std::optional output_size, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr); + } // namespace detail } // namespace cudf diff --git a/cpp/include/cudf/join/join.hpp b/cpp/include/cudf/join/join.hpp index a0171b1d4833..333bd69b0912 100644 --- a/cpp/include/cudf/join/join.hpp +++ b/cpp/include/cudf/join/join.hpp @@ -17,7 +17,11 @@ #include +#include #include +#include +#include +#include /** * @file @@ -340,6 +344,9 @@ std::unique_ptr cross_join( * @param right_indices Device span of row indices in the right table from hash join. * @param predicate An AST expression that returns a boolean for each pair of rows. * @param join_kind The type of join operation. Must be INNER_JOIN, LEFT_JOIN, or FULL_JOIN. + * @param output_size Optional precomputed number of output rows. When provided, skips the internal + * size-counting pass. Behavior is undefined if it differs from the size the function would + * otherwise produce for the same inputs. * @param stream CUDA stream used for kernel launches and memory operations. * @param mr Device memory resource used to allocate output indices. * @@ -354,7 +361,37 @@ filter_join_indices(cudf::table_view const& left, cudf::device_span right_indices, cudf::ast::expression const& predicate, cudf::join_kind join_kind, - rmm::cuda_stream_view stream = cudf::get_default_stream(), + std::optional output_size = std::nullopt, + rmm::cuda_stream_view stream = cudf::get_default_stream(), + rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref()); + +/** + * @brief Filters join result indices based on a conditional predicate and join type. + * + * @deprecated Use the overload that accepts an optional output size instead. + * + * @param left The left table for predicate evaluation (conditional columns only). + * @param right The right table for predicate evaluation (conditional columns only). + * @param left_indices Device span of row indices in the left table from hash join. + * @param right_indices Device span of row indices in the right table from hash join. + * @param predicate An AST expression that returns a boolean for each pair of rows. + * @param join_kind The type of join operation. Must be INNER_JOIN, LEFT_JOIN, or FULL_JOIN. + * @param stream CUDA stream used for kernel launches and memory operations. + * @param mr Device memory resource used to allocate output indices. + * + * @return A pair of device vectors [filtered_left_indices, filtered_right_indices] + * corresponding to rows that satisfy the join semantics and predicate. + */ +[[deprecated("Use the overload that takes an optional output_size parameter.")]] +std::pair>, + std::unique_ptr>> +filter_join_indices(cudf::table_view const& left, + cudf::table_view const& right, + cudf::device_span left_indices, + cudf::device_span right_indices, + cudf::ast::expression const& predicate, + cudf::join_kind join_kind, + rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref()); /** @@ -362,15 +399,19 @@ filter_join_indices(cudf::table_view const& left, * the filtered index vectors. * * Runs the same predicate evaluation as `filter_join_indices` but skips the index - * materialization step, returning only the total number of pairs that would be - * emitted. The semantics per `join_kind` match `filter_join_indices`: - * - INNER_JOIN: number of pairs where the predicate evaluates to true. - * - LEFT_JOIN: predicate-passing pairs plus one entry per left row with no passing match. - * - FULL_JOIN: input pairs plus one extra entry per pair whose predicate failed - * (because failed matches split into `(left, JoinNoMatch)` and `(JoinNoMatch, right)`). - * - * The returned size may be passed as a precomputed hint to APIs that compose - * `filter_join_indices` (for example, the mixed join APIs). + * materialization step, returning the total number of pairs that would be emitted along with the + * per-output contribution counts whose sum is that total. The counts are laid out per `join_kind` + * so that each entry records how many output rows the corresponding input contributes: + * - INNER_JOIN: indexed per input pair; entry `i` is `1` if the predicate passes and `0` otherwise. + * - FULL_JOIN: indexed per input pair; entry `i` is `1` for a preserved pair (predicate passes or + * the pair already contains a `JoinNoMatch`) and `2` for a failed valid pair (which splits into + * `(left, JoinNoMatch)` and `(JoinNoMatch, right)`). + * - LEFT_JOIN: indexed per left row; each entry holds the number of passing pairs for that left + * row, floored to `1` to account for the synthetic `(left, JoinNoMatch)` entry. + * + * The returned size and contribution counts may be passed as a precomputed hint to APIs that + * compose `filter_join_indices` (for example, the mixed join APIs). The layout above is an + * implementation detail that callers should treat as opaque rather than rely upon. * * @throw std::invalid_argument if `join_kind` is not INNER_JOIN, LEFT_JOIN, or FULL_JOIN. * @throw std::invalid_argument if `left_indices` and `right_indices` have different sizes. @@ -383,17 +424,21 @@ filter_join_indices(cudf::table_view const& left, * @param predicate An AST expression that returns a boolean for each pair of rows. * @param join_kind The type of join operation. Must be INNER_JOIN, LEFT_JOIN, or FULL_JOIN. * @param stream CUDA stream used for kernel launches and memory operations. + * @param mr Device memory resource used to allocate the returned contribution counts. * - * @return The exact number of pairs that `filter_join_indices` would produce. + * @return A pair containing the exact number of pairs that `filter_join_indices` would produce + * and the per-output contribution counts that sum to that number. */ -[[nodiscard]] std::size_t filter_join_indices_output_size( +[[nodiscard]] std::pair>> +filter_join_indices_output_size( cudf::table_view const& left, cudf::table_view const& right, cudf::device_span left_indices, cudf::device_span right_indices, cudf::ast::expression const& predicate, cudf::join_kind join_kind, - rmm::cuda_stream_view stream = cudf::get_default_stream()); + rmm::cuda_stream_view stream = cudf::get_default_stream(), + rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref()); /** * @brief JIT-based filtering of join result indices using string predicate. diff --git a/cpp/src/join/filter_join_indices/filter_join_indices.cu b/cpp/src/join/filter_join_indices/filter_join_indices.cu index 10cfbd1a6d8e..908a8176517e 100644 --- a/cpp/src/join/filter_join_indices/filter_join_indices.cu +++ b/cpp/src/join/filter_join_indices/filter_join_indices.cu @@ -12,7 +12,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -38,8 +40,11 @@ #include #include #include +#include +#include #include +#include #include namespace cudf { @@ -53,6 +58,7 @@ filter_join_indices(cudf::table_view const& left, cudf::device_span right_indices, ast::expression const& predicate, join_kind join_kind, + std::optional output_size, rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr) { @@ -168,10 +174,13 @@ filter_join_indices(cudf::table_view const& left, auto valid_predicate = [=] __device__(size_type i) -> bool { return predicate_results_ptr[i]; }; auto const num_valid = - cudf::detail::count_if(cuda::counting_iterator{0}, - cuda::counting_iterator{static_cast(left_indices.size())}, - valid_predicate, - stream); + output_size.has_value() + ? *output_size + : cudf::detail::count_if( + cuda::counting_iterator{0}, + cuda::counting_iterator{static_cast(left_indices.size())}, + valid_predicate, + stream); if (num_valid == 0) { return make_empty_result(); } @@ -224,12 +233,12 @@ filter_join_indices(cudf::table_view const& left, auto const num_invalid = left.num_rows() - num_filter_passing; - // Find the number of indices passing the filter i.e. rows that are valid according to the - // predicate CUB APIs are used instead of Thrust to enable 64-bit operations on index vectors of - // size greater than integer limits - cudf::detail::device_scalar d_num_valid(stream, - cudf::get_current_device_resource_ref()); - { + auto const num_valid = [&]() -> std::size_t { + if (output_size.has_value()) { return *output_size - num_invalid; } + // CUB APIs are used instead of Thrust to enable 64-bit operations on index vectors of size + // greater than integer limits + cudf::detail::device_scalar d_num_valid(stream, + cudf::get_current_device_resource_ref()); auto const predicate_it = cuda::transform_iterator{predicate_results_ptr, cuda::proclaim_return_type( @@ -248,12 +257,12 @@ filter_join_indices(cudf::table_view const& left, d_num_valid.data(), left_indices.size(), stream.value()); - } - auto const num_valid = d_num_valid.value(stream); - auto const output_size = num_valid + num_invalid; - if (output_size == 0) { return make_empty_result(); } + return d_num_valid.value(stream); + }(); + auto const result_size = num_valid + num_invalid; + if (result_size == 0) { return make_empty_result(); } - auto [filtered_left_indices, filtered_right_indices] = make_result_vectors(output_size); + auto [filtered_left_indices, filtered_right_indices] = make_result_vectors(result_size); if (num_valid > 0) { auto input_iter = thrust::make_zip_iterator(cuda::std::tuple{left_indices.begin(), right_indices.begin()}); @@ -307,15 +316,18 @@ filter_join_indices(cudf::table_view const& left, // Count failed matches for output sizing auto const failed_matched_count = - cudf::detail::count_if(cuda::counting_iterator{0}, - cuda::counting_iterator{static_cast(left_indices.size())}, - is_failed_matched_pair, - stream); - auto const output_size = left_indices.size() + failed_matched_count; + output_size.has_value() + ? *output_size - left_indices.size() + : cudf::detail::count_if( + cuda::counting_iterator{0}, + cuda::counting_iterator{static_cast(left_indices.size())}, + is_failed_matched_pair, + stream); + auto const result_size = left_indices.size() + failed_matched_count; - if (output_size == 0) { return make_empty_result(); } + if (result_size == 0) { return make_empty_result(); } - auto [filtered_left_indices, filtered_right_indices] = make_result_vectors(output_size); + auto [filtered_left_indices, filtered_right_indices] = make_result_vectors(result_size); // Use two-step approach with optimized memory management // Step 1: Handle primary pairs @@ -361,13 +373,15 @@ filter_join_indices(cudf::table_view const& left, } } -std::size_t filter_join_indices_output_size(cudf::table_view const& left, - cudf::table_view const& right, - cudf::device_span left_indices, - cudf::device_span right_indices, - ast::expression const& predicate, - join_kind join_kind, - rmm::cuda_stream_view stream) +std::pair>> +filter_join_indices_output_size(cudf::table_view const& left, + cudf::table_view const& right, + cudf::device_span left_indices, + cudf::device_span right_indices, + ast::expression const& predicate, + join_kind join_kind, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) { // Validate inputs (same constraints as filter_join_indices) CUDF_EXPECTS(left_indices.size() == right_indices.size(), @@ -379,8 +393,12 @@ std::size_t filter_join_indices_output_size(cudf::table_view const& left, "filter_join_indices_output_size only supports INNER_JOIN, LEFT_JOIN, and FULL_JOIN.", std::invalid_argument); - if (left_indices.empty()) { return 0; } - if (join_kind == join_kind::LEFT_JOIN && left.num_rows() == 0) { return 0; } + auto empty_counts = [&]() { + return std::make_unique>(0, stream, mr); + }; + + if (left_indices.empty()) { return {0, empty_counts()}; } + if (join_kind == join_kind::LEFT_JOIN && left.num_rows() == 0) { return {0, empty_counts()}; } auto const has_nulls = predicate.may_evaluate_null(left, right, stream); @@ -399,16 +417,13 @@ std::size_t filter_join_indices_output_size(cudf::table_view const& left, detail::grid_1d const config(left_indices.size(), DEFAULT_JOIN_BLOCK_SIZE); auto const shmem_per_block = parser.shmem_per_thread * DEFAULT_JOIN_BLOCK_SIZE; - // The count kernel uses a single atomic counter. Allocate device_scalar zero-initialized. - cudf::detail::device_scalar d_count( - std::size_t{0}, stream, cudf::get_current_device_resource_ref()); - - // For LEFT_JOIN, allocate a zeroed per-left-row mark buffer; for others, pass nullptr. - auto left_passing_marks = cudf::detail::make_zeroed_device_uvector_async( - join_kind == join_kind::LEFT_JOIN ? static_cast(left.num_rows()) : 0, - stream, - cudf::get_current_device_resource_ref()); - auto* const marks_ptr = join_kind == join_kind::LEFT_JOIN ? left_passing_marks.data() : nullptr; + auto const counts_size = join_kind == join_kind::LEFT_JOIN + ? static_cast(left.num_rows()) + : left_indices.size(); + auto output_counts = + join_kind == join_kind::LEFT_JOIN + ? cudf::detail::make_zeroed_device_uvector_async(counts_size, stream, mr) + : rmm::device_uvector(counts_size, stream, mr); cudf::detail::dispatch_bool(has_nulls, [&](auto has_nulls_c) { cudf::detail::dispatch_bool(has_complex_type, [&](auto has_complex_c) { @@ -422,25 +437,27 @@ std::size_t filter_join_indices_output_size(cudf::table_view const& left, config, shmem_per_block, join_kind, - d_count.data(), - marks_ptr, + output_counts.data(), stream); }); }); - auto const num_predicate_passing = d_count.value(stream); - - switch (join_kind) { - case join_kind::INNER_JOIN: return num_predicate_passing; - case join_kind::FULL_JOIN: return left_indices.size() + num_predicate_passing; - case join_kind::LEFT_JOIN: { - auto const num_filter_passing = cudf::detail::count_if( - left_passing_marks.begin(), left_passing_marks.end(), cuda::std::identity{}, stream); - auto const num_invalid = static_cast(left.num_rows()) - num_filter_passing; - return num_predicate_passing + num_invalid; - } - default: CUDF_FAIL("Unsupported join kind for filter_join_indices_output_size"); + if (join_kind == join_kind::LEFT_JOIN) { + thrust::transform(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()), + output_counts.begin(), + output_counts.end(), + output_counts.begin(), + cuda::proclaim_return_type( + [] __device__(size_type count) { return count > 0 ? count : 1; })); } + + std::size_t const total = + thrust::reduce(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()), + output_counts.begin(), + output_counts.end(), + std::size_t{0}); + + return {total, std::make_unique>(std::move(output_counts))}; } } // namespace detail @@ -454,25 +471,44 @@ filter_join_indices(cudf::table_view const& left, cudf::device_span right_indices, ast::expression const& predicate, cudf::join_kind join_kind, + std::optional output_size, rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr) { CUDF_FUNC_RANGE(); return detail::filter_join_indices( - left, right, left_indices, right_indices, predicate, join_kind, stream, mr); + left, right, left_indices, right_indices, predicate, join_kind, output_size, stream, mr); +} + +std::pair>, + std::unique_ptr>> +filter_join_indices(cudf::table_view const& left, + cudf::table_view const& right, + cudf::device_span left_indices, + cudf::device_span right_indices, + ast::expression const& predicate, + cudf::join_kind join_kind, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) +{ + CUDF_FUNC_RANGE(); + return detail::filter_join_indices( + left, right, left_indices, right_indices, predicate, join_kind, std::nullopt, stream, mr); } -std::size_t filter_join_indices_output_size(cudf::table_view const& left, - cudf::table_view const& right, - cudf::device_span left_indices, - cudf::device_span right_indices, - ast::expression const& predicate, - cudf::join_kind join_kind, - rmm::cuda_stream_view stream) +std::pair>> +filter_join_indices_output_size(cudf::table_view const& left, + cudf::table_view const& right, + cudf::device_span left_indices, + cudf::device_span right_indices, + ast::expression const& predicate, + cudf::join_kind join_kind, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) { CUDF_FUNC_RANGE(); return detail::filter_join_indices_output_size( - left, right, left_indices, right_indices, predicate, join_kind, stream); + left, right, left_indices, right_indices, predicate, join_kind, stream, mr); } } // namespace cudf diff --git a/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel.cuh b/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel.cuh index e158e3883662..f835add8b016 100644 --- a/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel.cuh +++ b/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel.cuh @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ #pragma once @@ -18,23 +18,25 @@ #include -#include #include -#include - -#include namespace cudf::detail { /** - * @brief Counts the per-join-kind output size of `filter_join_indices` without materializing - * a per-pair boolean buffer. + * @brief Fills the per-output contribution counts of `filter_join_indices` without materializing + * the filtered index vectors. * - * Each thread accumulates a private partial count, the block aggregates with CUB, and each - * block adds its block-sum to `*count_out` exactly once via `cuda::atomic_ref`. For LEFT_JOIN, - * `left_passing_marks[left_row_index]` is additionally set to `true` for every left row that - * contributes to the count, which lets the host derive the number of synthetic JoinNoMatch - * entries. + * The total output size is the sum of `output_counts`, which is laid out per join kind so that + * each entry records how many output rows the corresponding input contributes: + * - INNER_JOIN: `output_counts` is indexed per input pair; entry `i` is `1` if the predicate + * passes and `0` otherwise. + * - FULL_JOIN: `output_counts` is indexed per input pair; entry `i` is `1` for a preserved pair + * (predicate passes or the pair already contains a `JoinNoMatch`) and `2` for a failed valid + * pair (which splits into `(left, JoinNoMatch)` and `(JoinNoMatch, right)`). + * - LEFT_JOIN: `output_counts` is indexed per left row; the kernel atomically accumulates the + * number of passing pairs for each left row. Left rows with no passing pair are floored to `1` + * by the host afterwards to account for the synthetic `(left, JoinNoMatch)` entry. The buffer + * must be zero-initialized before the launch. */ template CUDF_KERNEL __launch_bounds__(DEFAULT_JOIN_BLOCK_SIZE) void filter_join_indices_output_size_kernel( @@ -44,8 +46,7 @@ CUDF_KERNEL __launch_bounds__(DEFAULT_JOIN_BLOCK_SIZE) void filter_join_indices_ cudf::device_span right_indices, cudf::ast::detail::expression_device_view device_expression_data, cudf::join_kind join_kind, - std::size_t* count_out, - bool* left_passing_marks) + cudf::size_type* output_counts) { extern __shared__ char raw_intermediate_storage[]; auto* intermediate_storage = @@ -53,17 +54,12 @@ CUDF_KERNEL __launch_bounds__(DEFAULT_JOIN_BLOCK_SIZE) void filter_join_indices_ auto thread_intermediate_storage = &intermediate_storage[threadIdx.x * device_expression_data.num_intermediates]; - using BlockReduce = cub::BlockReduce; - __shared__ typename BlockReduce::TempStorage temp_storage; - auto const tid = cudf::detail::grid_1d::global_thread_id(); auto const stride = cudf::detail::grid_1d::grid_stride(); auto evaluator = cudf::ast::detail::expression_evaluator{ left_table, right_table, device_expression_data}; - cuda::std::size_t thread_local_count = 0; - for (auto i = tid; i < static_cast(left_indices.size()); i += stride) { auto const left_row_index = left_indices[i]; auto const right_row_index = right_indices[i]; @@ -85,35 +81,20 @@ CUDF_KERNEL __launch_bounds__(DEFAULT_JOIN_BLOCK_SIZE) void filter_join_indices_ } switch (join_kind) { - case cudf::join_kind::INNER_JOIN: - if (predicate_pass) { ++thread_local_count; } + case cudf::join_kind::INNER_JOIN: output_counts[i] = predicate_pass ? 1 : 0; break; + case cudf::join_kind::FULL_JOIN: + output_counts[i] = (both_valid && !predicate_pass) ? 2 : 1; break; case cudf::join_kind::LEFT_JOIN: - if (predicate_pass) { - ++thread_local_count; - // Mark the left row as "passing" so the host can derive how many left rows need a - // synthetic JoinNoMatch entry. For matched-passing pairs and for pre-existing - // (left, JoinNoMatch) entries from upstream hash_join.left_join the left index is a - // valid row index in [0, left_table.num_rows()). - if (left_row_index >= 0 && left_row_index < left_table.num_rows()) { - left_passing_marks[left_row_index] = true; - } + if (predicate_pass && left_row_index >= 0 && left_row_index < left_table.num_rows()) { + cuda::atomic_ref count_ref{ + output_counts[left_row_index]}; + count_ref.fetch_add(1, cuda::memory_order_relaxed); } break; - case cudf::join_kind::FULL_JOIN: - // Count failed matches: predicate false AND both indices valid. - if (both_valid && !predicate_pass) { ++thread_local_count; } - break; default: break; } } - - cuda::std::size_t const block_sum = BlockReduce(temp_storage).Sum(thread_local_count); - - if (threadIdx.x == 0) { - cuda::atomic_ref count_ref{*count_out}; - count_ref.fetch_add(block_sum, cuda::memory_order_relaxed); - } } template @@ -126,8 +107,7 @@ void launch_filter_output_size_kernel( cudf::detail::grid_1d const& config, std::size_t shmem_per_block, cudf::join_kind join_kind, - std::size_t* count_out, - bool* left_passing_marks, + cudf::size_type* output_counts, rmm::cuda_stream_view stream) { filter_join_indices_output_size_kernel @@ -138,8 +118,7 @@ void launch_filter_output_size_kernel( right_indices, device_expression_data, join_kind, - count_out, - left_passing_marks); + output_counts); CUDF_CUDA_TRY(cudaGetLastError()); } diff --git a/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel.hpp b/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel.hpp index b5b25e5e7a0e..b343694e88af 100644 --- a/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel.hpp +++ b/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel.hpp @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ #pragma once @@ -20,19 +20,15 @@ namespace cudf::detail { /** - * @brief Launches a kernel that counts the per-join-kind output size for `filter_join_indices`. + * @brief Launches a kernel that fills the per-output contribution counts for `filter_join_indices`. * - * For INNER_JOIN this is the number of pairs whose predicate evaluates to true. - * For LEFT_JOIN this is the number of input pairs whose predicate evaluates to true - * (including pre-existing unmatched pairs that are preserved); additionally, - * `left_passing_marks[left_row_index]` is set to `true` for every left row that - * contributes to that count (used by the host code to derive the number of left - * rows that need a synthetic JoinNoMatch entry). - * For FULL_JOIN this is the number of failed matched pairs (predicate false and - * both indices valid), which is added on top of `left_indices.size()` host-side. - * - * The kernel avoids materializing a per-pair boolean buffer; it folds the count - * directly into `count_out` via atomic increments. + * The total output size is the sum of `output_counts`. Its layout depends on the join kind: + * - INNER_JOIN: per input pair, `1` if the predicate passes and `0` otherwise. + * - FULL_JOIN: per input pair, `1` for a preserved pair and `2` for a failed valid pair (which + * splits into `(left, JoinNoMatch)` and `(JoinNoMatch, right)`). + * - LEFT_JOIN: per left row, the number of passing pairs (accumulated atomically). The host floors + * empty rows to `1` afterwards to account for the synthetic `(left, JoinNoMatch)` entry, so the + * buffer must be zero-initialized before the launch. * * @tparam has_nulls Indicates whether the expression may evaluate to null * @tparam has_complex_type Indicates whether the expression may contain complex types @@ -45,10 +41,9 @@ namespace cudf::detail { * @param[in] config Grid configuration for kernel launch * @param[in] shmem_per_block Amount of shared memory to allocate per block * @param[in] join_kind The join kind. Must be INNER_JOIN, LEFT_JOIN, or FULL_JOIN. - * @param[out] count_out Atomic counter for the per-kind count described above - * @param[out] left_passing_marks Byte buffer of size `left_table.num_rows()` used by LEFT_JOIN - * to mark left rows whose entries contribute to `count_out`. Must be zero-initialized - * before the kernel launch and may be `nullptr` for INNER_JOIN and FULL_JOIN. + * @param[out] output_counts Per-output contribution counts described above. Sized to + * `left_indices.size()` for INNER_JOIN and FULL_JOIN, and to `left_table.num_rows()` + * (zero-initialized) for LEFT_JOIN. * @param[in] stream CUDA stream on which to launch the kernel */ template @@ -61,8 +56,7 @@ void launch_filter_output_size_kernel( cudf::detail::grid_1d const& config, std::size_t shmem_per_block, cudf::join_kind join_kind, - std::size_t* count_out, - bool* left_passing_marks, + cudf::size_type* output_counts, rmm::cuda_stream_view stream); } // namespace cudf::detail diff --git a/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_complex.cu b/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_complex.cu index 3bb93635d553..b55a7e33d152 100644 --- a/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_complex.cu +++ b/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_complex.cu @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ @@ -16,7 +16,6 @@ template void launch_filter_output_size_kernel( cudf::detail::grid_1d const& config, std::size_t shmem_per_block, cudf::join_kind join_kind, - std::size_t* count_out, - bool* left_passing_marks, + cudf::size_type* output_counts, rmm::cuda_stream_view stream); } // namespace cudf::detail diff --git a/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_null_complex.cu b/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_null_complex.cu index 195feaeeb65a..794b6ae55b48 100644 --- a/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_null_complex.cu +++ b/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_null_complex.cu @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ @@ -16,7 +16,6 @@ template void launch_filter_output_size_kernel( cudf::detail::grid_1d const& config, std::size_t shmem_per_block, cudf::join_kind join_kind, - std::size_t* count_out, - bool* left_passing_marks, + cudf::size_type* output_counts, rmm::cuda_stream_view stream); } // namespace cudf::detail diff --git a/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_null_primitive.cu b/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_null_primitive.cu index babeb76a3f82..84358e6fe9d7 100644 --- a/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_null_primitive.cu +++ b/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_null_primitive.cu @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ @@ -16,7 +16,6 @@ template void launch_filter_output_size_kernel( cudf::detail::grid_1d const& config, std::size_t shmem_per_block, cudf::join_kind join_kind, - std::size_t* count_out, - bool* left_passing_marks, + cudf::size_type* output_counts, rmm::cuda_stream_view stream); } // namespace cudf::detail diff --git a/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_primitive.cu b/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_primitive.cu index f695652818fb..f2de1322e301 100644 --- a/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_primitive.cu +++ b/cpp/src/join/filter_join_indices/filter_join_indices_output_size_kernel_primitive.cu @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ @@ -16,7 +16,6 @@ template void launch_filter_output_size_kernel( cudf::detail::grid_1d const& config, std::size_t shmem_per_block, cudf::join_kind join_kind, - std::size_t* count_out, - bool* left_passing_marks, + cudf::size_type* output_counts, rmm::cuda_stream_view stream); } // namespace cudf::detail diff --git a/cpp/src/join/mixed_filter_join_common_utils.cuh b/cpp/src/join/mixed_filter_join_common_utils.cuh index d80638c81be8..829b5584a943 100644 --- a/cpp/src/join/mixed_filter_join_common_utils.cuh +++ b/cpp/src/join/mixed_filter_join_common_utils.cuh @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ #pragma once @@ -7,9 +7,14 @@ #include "mixed_join_common_utils.cuh" #include +#include + +#include #include +#include + namespace cudf::detail { /** diff --git a/cpp/src/join/mixed_join.cu b/cpp/src/join/mixed_join.cu index 79d359c7b01a..baeb35670a36 100644 --- a/cpp/src/join/mixed_join.cu +++ b/cpp/src/join/mixed_join.cu @@ -1,532 +1,128 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ -#include "join_common_utils.cuh" #include "join_common_utils.hpp" -#include "mixed_join_common_utils.cuh" -#include "mixed_join_kernel.hpp" -#include "mixed_join_size_kernel.hpp" -#include #include -#include -#include +#include #include -#include +#include #include #include -#include #include #include #include +#include #include #include +#include #include -#include -#include -#include -#include -#include +#include #include #include -#include +#include namespace cudf { namespace detail { namespace { -/** - * @brief Builds the hash table based on the given `build_table`. - * - * @tparam HashTable The type of the hash table - * - * @param build Table of columns used to build join hash. - * @param preprocessed_build shared_ptr to cudf::detail::row::equality::preprocessed_table - * for build - * @param hash_table Build hash table. - * @param has_nested_nulls Flag to denote if build or probe tables have nested nulls - * @param nulls_equal Flag to denote nulls are equal or not. - * @param bitmask Bitmask to denote whether a row is valid. - * @param stream CUDA stream used for device memory operations and kernel launches. - */ -template -void build_join_hash_table( - cudf::table_view const& build, - std::shared_ptr const& preprocessed_build, - HashTable& hash_table, - bool has_nested_nulls, - null_equality nulls_equal, - [[maybe_unused]] bitmask_type const* bitmask, - rmm::cuda_stream_view stream) -{ - CUDF_EXPECTS(0 != build.num_columns(), "Selected build dataset is empty", std::invalid_argument); - CUDF_EXPECTS(0 != build.num_rows(), "Build side table has no rows", std::invalid_argument); - - auto insert_rows = [&](auto const& build, auto const& d_hasher) { - auto const iter = cudf::detail::make_counting_transform_iterator(0, pair_fn{d_hasher}); - - if (nulls_equal == cudf::null_equality::EQUAL or not nullable(build)) { - hash_table.insert_async(iter, iter + build.num_rows(), stream.value()); - } else { - auto const stencil = cuda::counting_iterator{0}; - auto const pred = row_is_valid{bitmask}; - - hash_table.insert_if_async(iter, iter + build.num_rows(), stencil, pred, stream.value()); - } - }; - - auto const nulls = nullate::DYNAMIC{has_nested_nulls}; - - auto const row_hash = detail::row::hash::row_hasher{preprocessed_build}; - auto const d_hasher = row_hash.device_hasher(nulls); - - insert_rows(build, d_hasher); -} - -/** - * @brief Precomputes double hashing indices and row hash values for mixed join operations. - * - * This function exists as a performance optimization to work around the register spilling issue - * reported in https://github.com/NVIDIA/cuCollections/issues/761. The new cuco hash table - * implementation suffers from register spilling due to longer register live ranges, which can - * cause up to 20x performance degradation. - * - * By precomputing the double hashing indices (initial slot and step size) and row hash values - * in a separate pass, we reduce register pressure in the subsequent count and retrieve kernels. - * This approach yields approximately 20% speedup compared to the legacy multimap-based - * implementation. - * - * The tradeoff is that we cannot use cuco's device APIs directly in mixed join operations. - * Instead, we must reimplement the entire double hashing probing logic in cudf without relying - * on cuco's device APIs. This should be revisited and potentially removed once issue #761 is - * fully resolved. - * - * @param hash_table The cuco multiset hash table - * @param hash_probe Hash function for computing row hashes - * @param probe_table_num_rows Number of rows in the probe table - * @param stream CUDA stream for operations - * @param mr Memory resource for allocations - * @return A pair of device vectors: (input_pairs, hash_indices) where input_pairs contains - * (row_hash, row_index) pairs and hash_indices contains (initial_slot, step_size) pairs - */ -template -std::pair>, - rmm::device_uvector>> -precompute_mixed_join_data(mixed_multiset_type const& hash_table, - HashProbe const& hash_probe, - size_type probe_table_num_rows, - rmm::cuda_stream_view stream, - rmm::device_async_resource_ref mr) -{ - auto input_pairs = - rmm::device_uvector>(probe_table_num_rows, stream, mr); - auto hash_indices = rmm::device_uvector>( - probe_table_num_rows, stream, mr); - - auto const capacity = hash_table.capacity(); - auto const probe_hash_fn = hash_table.hash_function(); - static constexpr std::size_t bucket_size = mixed_multiset_type::bucket_size; - - auto const num_buckets = capacity / bucket_size; - auto const num_buckets_minus_one = num_buckets - 1; - - // Functor to pre-compute both input pairs and initial slots and step sizes for double hashing. - auto precompute_fn = [=] __device__(size_type i) { - auto const probe_key = cuco::pair{hash_probe(i), i}; - - // Use the probing scheme's hash functions for proper double hashing - auto const hash1_val = cuda::std::get<0>(probe_hash_fn)(probe_key); - auto const hash2_val = cuda::std::get<1>(probe_hash_fn)(probe_key); - - auto const init_idx = static_cast( - (static_cast(hash1_val) % num_buckets) * bucket_size); - auto const step_val = static_cast( - ((static_cast(hash2_val) % num_buckets_minus_one) + 1) * bucket_size); - - return cuda::std::pair{probe_key, cuda::std::pair{init_idx, step_val}}; - }; - - // Single transform to fill both arrays using zip iterator - thrust::transform( - rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()), - cuda::counting_iterator{0}, - cuda::counting_iterator{probe_table_num_rows}, - thrust::make_zip_iterator(cuda::std::make_tuple(input_pairs.begin(), hash_indices.begin())), - precompute_fn); - - return std::make_pair(std::move(input_pairs), std::move(hash_indices)); -} - -struct mixed_join_setup_data { - bool swap_tables; - size_type outer_num_rows; - cudf::nullate::DYNAMIC has_nulls; - ast::detail::expression_parser parser; - mixed_multiset_type hash_table; - std::shared_ptr preprocessed_build; - std::shared_ptr preprocessed_probe; - std::unique_ptr> left_conditional_view; - std::unique_ptr> - right_conditional_view; - detail::grid_1d config; - thread_index_type shmem_size_per_block; - row_equality equality_probe; - cudf::device_span> hash_table_storage; - rmm::device_uvector> input_pairs; - rmm::device_uvector> hash_indices; -}; - -mixed_join_setup_data setup_mixed_join_common(table_view const& left_equality, - table_view const& right_equality, - table_view const& left_conditional, - table_view const& right_conditional, - ast::expression const& binary_predicate, - null_equality compare_nulls, - join_kind join_type, - rmm::cuda_stream_view stream, - rmm::device_async_resource_ref mr) -{ - CUDF_EXPECTS(left_conditional.num_rows() == left_equality.num_rows(), - "The left conditional and equality tables must have the same number of rows."); - CUDF_EXPECTS(right_conditional.num_rows() == right_equality.num_rows(), - "The right conditional and equality tables must have the same number of rows."); - - auto const right_num_rows = right_conditional.num_rows(); - auto const left_num_rows = left_conditional.num_rows(); - auto const swap_tables = (join_type == join_kind::INNER_JOIN) && (right_num_rows > left_num_rows); - auto const outer_num_rows = swap_tables ? right_num_rows : left_num_rows; - - // If evaluating the expression may produce null outputs we create a nullable - // output column and follow the null-supporting expression evaluation code path. - auto const has_nulls = cudf::nullate::DYNAMIC{ - cudf::has_nulls(left_equality) || cudf::has_nulls(right_equality) || - binary_predicate.may_evaluate_null(left_conditional, right_conditional, stream)}; - - auto parser = ast::detail::expression_parser{ - binary_predicate, left_conditional, right_conditional, has_nulls, stream, mr}; - CUDF_EXPECTS(parser.output_type().id() == type_id::BOOL8, - "The expression must produce a boolean output.", - cudf::data_type_error); - - // TODO: The non-conditional join impls start with a dictionary matching, - // figure out what that is and what it's needed for (and if conditional joins - // need to do the same). - auto& probe = swap_tables ? right_equality : left_equality; - auto& build = swap_tables ? left_equality : right_equality; - - // Create hash table with load factor following hash join pattern - mixed_multiset_type hash_table{ - cuco::extent{static_cast(build.num_rows())}, - cudf::detail::CUCO_DESIRED_LOAD_FACTOR, - cuco::empty_key{cuco::pair{std::numeric_limits::max(), cudf::JoinNoMatch}}, - {}, - {}, - {}, - {}, - rmm::mr::polymorphic_allocator{}, - stream.value()}; - - // TODO: To add support for nested columns we will need to flatten in many - // places. However, this probably isn't worth adding any time soon since we - // won't be able to support AST conditions for those types anyway. - auto const row_bitmask = - cudf::detail::bitmask_and(build, stream, cudf::get_current_device_resource_ref()).first; - auto preprocessed_build = detail::row::equality::preprocessed_table::create(build, stream); - build_join_hash_table(build, - preprocessed_build, - hash_table, - has_nulls, - compare_nulls, - static_cast(row_bitmask.data()), - stream); - - auto left_conditional_view = table_device_view::create(left_conditional, stream); - auto right_conditional_view = table_device_view::create(right_conditional, stream); - - // For inner joins we support optimizing the join by launching one thread for - // whichever table is larger rather than always using the left table. - detail::grid_1d const config(outer_num_rows, DEFAULT_JOIN_BLOCK_SIZE); - auto const shmem_size_per_block = parser.shmem_per_thread * config.num_threads_per_block; - - auto preprocessed_probe = detail::row::equality::preprocessed_table::create(probe, stream); - auto const row_hash = cudf::detail::row::hash::row_hasher{preprocessed_probe}; - auto const hash_probe = row_hash.device_hasher(has_nulls); - auto const row_comparator = - cudf::detail::row::equality::two_table_comparator{preprocessed_probe, preprocessed_build}; - auto const equality_probe = row_comparator.equal_to(has_nulls, compare_nulls); - - // Precompute hash table storage and input data - auto hash_table_storage = cudf::device_span>{ - hash_table.data(), hash_table.capacity()}; - CUDF_EXPECTS(reinterpret_cast(hash_table_storage.data()) % - (2 * sizeof(cuco::pair)) == - 0, - "Hash table storage must be aligned to 2-element boundary"); - auto [input_pairs, hash_indices] = - precompute_mixed_join_data(hash_table, hash_probe, outer_num_rows, stream, mr); - - return {swap_tables, - outer_num_rows, - has_nulls, - std::move(parser), - std::move(hash_table), - std::move(preprocessed_build), - std::move(preprocessed_probe), - std::move(left_conditional_view), - std::move(right_conditional_view), - config, - shmem_size_per_block, - equality_probe, - hash_table_storage, - std::move(input_pairs), - std::move(hash_indices)}; -} /** - * @brief Helper function to compute the output size for mixed joins by launching count kernels. + * @brief Probes the equality hash table for the given join kind. * - * This function encapsulates the common logic needed by both mixed_join and - * compute_mixed_join_output_size to count the number of matches per row. + * The hash table is built on the right equality table and probed with the left equality table, + * yielding the index pairs that the conditional predicate is subsequently applied to. */ -std::pair>> -compute_mixed_join_matches_per_row( - cudf::nullate::DYNAMIC has_nulls, - table_device_view const& left_conditional_view, - table_device_view const& right_conditional_view, - bool is_outer_join, - bool swap_tables, - row_equality const& equality_probe, - cudf::device_span> hash_table_storage, - cuco::pair const* input_pairs, - cuda::std::pair const* hash_indices, - cudf::ast::detail::expression_device_view device_expression_data, - size_type outer_num_rows, - detail::grid_1d config, - thread_index_type shmem_size_per_block, - rmm::cuda_stream_view stream, - rmm::device_async_resource_ref mr) +std::pair>, + std::unique_ptr>> +equality_join_indices(cudf::hash_join const& hash_joiner, + table_view const& left_equality, + join_kind join_type, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) { - auto matches_per_row = std::make_unique>( - static_cast(outer_num_rows), stream, mr); - auto matches_per_row_span = cudf::device_span{ - matches_per_row->begin(), static_cast(outer_num_rows)}; - - if (has_nulls) { - launch_mixed_join_count(left_conditional_view, - right_conditional_view, - is_outer_join, - swap_tables, - equality_probe, - hash_table_storage, - input_pairs, - hash_indices, - device_expression_data, - matches_per_row_span, - config, - shmem_size_per_block, - stream); - } else { - launch_mixed_join_count(left_conditional_view, - right_conditional_view, - is_outer_join, - swap_tables, - equality_probe, - hash_table_storage, - input_pairs, - hash_indices, - device_expression_data, - matches_per_row_span, - config, - shmem_size_per_block, - stream); + switch (join_type) { + case join_kind::INNER_JOIN: return hash_joiner.inner_join(left_equality, {}, stream, mr); + case join_kind::LEFT_JOIN: return hash_joiner.left_join(left_equality, {}, stream, mr); + case join_kind::FULL_JOIN: return hash_joiner.full_join(left_equality, {}, stream, mr); + default: CUDF_FAIL("Invalid join kind."); } - - std::size_t const size = - thrust::reduce(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()), - matches_per_row_span.begin(), - matches_per_row_span.end(), - std::size_t{0}); - - return {size, std::move(matches_per_row)}; } + } // anonymous namespace std::pair>, std::unique_ptr>> -mixed_join( - table_view const& left_equality, - table_view const& right_equality, - table_view const& left_conditional, - table_view const& right_conditional, - ast::expression const& binary_predicate, - null_equality compare_nulls, - join_kind join_type, - std::optional>> const& output_size_data, - rmm::cuda_stream_view stream, - rmm::device_async_resource_ref mr) +mixed_join(table_view const& left_equality, + table_view const& right_equality, + table_view const& left_conditional, + table_view const& right_conditional, + ast::expression const& binary_predicate, + null_equality compare_nulls, + join_kind join_type, + output_size_data_type const& output_size_data, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) { CUDF_EXPECTS((join_type != join_kind::LEFT_SEMI_JOIN) && (join_type != join_kind::LEFT_ANTI_JOIN), "Left semi and anti joins should use mixed_join_semi."); + CUDF_EXPECTS(left_conditional.num_rows() == left_equality.num_rows(), + "The left conditional and equality tables must have the same number of rows."); + CUDF_EXPECTS(right_conditional.num_rows() == right_equality.num_rows(), + "The right conditional and equality tables must have the same number of rows."); - auto const right_num_rows = right_conditional.num_rows(); - auto const left_num_rows = left_conditional.num_rows(); - - // We can immediately filter out cases where the right table is empty. In - // some cases, we return all the rows of the left table with a corresponding - // null index for the right table; in others, we return an empty output. - if (right_num_rows == 0) { + // hash_join requires a non-empty build (right) table. + if (right_conditional.num_rows() == 0) { switch (join_type) { - // Left and full joins all return all the row indices from - // left with a corresponding NULL from the right. case join_kind::LEFT_JOIN: case join_kind::FULL_JOIN: return get_trivial_left_join_indices(left_conditional, stream, mr); - // Inner joins return empty output because no matches can exist. - case join_kind::INNER_JOIN: - return std::pair(std::make_unique>(0, stream, mr), - std::make_unique>(0, stream, mr)); - default: CUDF_FAIL("Invalid join kind."); break; - } - } else if (left_num_rows == 0) { - switch (join_type) { - // Left and inner joins all return empty sets. - case join_kind::LEFT_JOIN: case join_kind::INNER_JOIN: - return std::pair(std::make_unique>(0, stream, mr), - std::make_unique>(0, stream, mr)); - // Full joins need to return the trivial complement. - case join_kind::FULL_JOIN: { - auto ret_flipped = get_trivial_left_join_indices(right_conditional, stream, mr); - return std::pair(std::move(ret_flipped.second), std::move(ret_flipped.first)); - } - default: CUDF_FAIL("Invalid join kind."); break; + return std::pair{std::make_unique>(0, stream, mr), + std::make_unique>(0, stream, mr)}; + default: CUDF_FAIL("Invalid join kind."); } } - auto setup = setup_mixed_join_common(left_equality, - right_equality, - left_conditional, - right_conditional, - binary_predicate, - compare_nulls, - join_type, - stream, - mr); - - bool const is_outer_join = - (join_type == join_kind::LEFT_JOIN || join_type == join_kind::FULL_JOIN); - - // If the join size data was not provided as an input, compute it here. - std::size_t join_size = 0; - // Using an optional because we only need to allocate a new vector if one was - // not passed as input, and rmm::device_uvector is not default constructible - std::optional> matches_per_row{}; - device_span matches_per_row_span{}; - - if (output_size_data.has_value()) { - join_size = output_size_data->first; - matches_per_row_span = output_size_data->second; - } else { - auto [size, matches] = compute_mixed_join_matches_per_row(setup.has_nulls, - *setup.left_conditional_view, - *setup.right_conditional_view, - is_outer_join, - setup.swap_tables, - setup.equality_probe, - setup.hash_table_storage, - setup.input_pairs.data(), - setup.hash_indices.data(), - setup.parser.device_expression_data, - setup.outer_num_rows, - setup.config, - setup.shmem_size_per_block, - stream, - mr); - join_size = size; - matches_per_row = std::move(*matches); - matches_per_row_span = cudf::device_span{ - matches_per_row->begin(), static_cast(setup.outer_num_rows)}; - } - - // Given the number of matches per row, we need to compute the offsets for insertion. - auto join_result_offsets = - rmm::device_uvector{static_cast(setup.outer_num_rows), stream, mr}; - thrust::exclusive_scan(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()), - matches_per_row_span.begin(), - matches_per_row_span.end(), - join_result_offsets.begin()); - - // Get total count from scan result: last offset + last matches_per_row - if (setup.outer_num_rows > 0 && !output_size_data.has_value()) { - auto const last_offset = join_result_offsets.element(setup.outer_num_rows - 1, stream); - auto const last_matches = matches_per_row->element(setup.outer_num_rows - 1, stream); - join_size = last_offset + last_matches; - } - - // The initial early exit clauses guarantee that we will not reach this point - // unless both the left and right tables are non-empty. Under that - // constraint, neither left nor full joins can return an empty result since - // at minimum we are guaranteed null matches for all non-matching rows. In - // all other cases (inner, left semi, and left anti joins) if we reach this - // point we can safely return an empty result. - if (join_size == 0) { - return std::pair(std::make_unique>(0, stream, mr), - std::make_unique>(0, stream, mr)); - } - - auto left_indices = std::make_unique>(join_size, stream, mr); - auto right_indices = std::make_unique>(join_size, stream, mr); - - auto const& join_output_l = left_indices->data(); - auto const& join_output_r = right_indices->data(); - - if (setup.has_nulls) { - launch_mixed_join(*setup.left_conditional_view, - *setup.right_conditional_view, - is_outer_join, - setup.swap_tables, - setup.equality_probe, - setup.hash_table_storage, - setup.input_pairs.data(), - setup.hash_indices.data(), - setup.parser.device_expression_data, - join_output_l, - join_output_r, - join_result_offsets.data(), - setup.config, - setup.shmem_size_per_block, - stream); - } else { - launch_mixed_join(*setup.left_conditional_view, - *setup.right_conditional_view, - is_outer_join, - setup.swap_tables, - setup.equality_probe, - setup.hash_table_storage, - setup.input_pairs.data(), - setup.hash_indices.data(), - setup.parser.device_expression_data, - join_output_l, - join_output_r, - join_result_offsets.data(), - setup.config, - setup.shmem_size_per_block, - stream); - } - - auto join_indices = std::pair(std::move(left_indices), std::move(right_indices)); - - // For full joins, get the indices in the right table that were not joined to - // by any row in the left table. + // A full join is a left join plus the unmatched-right complement. Build the left-outer result and + // append the complement with finalize_full_join rather than splitting failed pairs, which would + // emit spurious unmatched rows for keys that also match elsewhere. if (join_type == join_kind::FULL_JOIN) { - join_indices = detail::finalize_full_join( - std::move(join_indices), left_num_rows, right_num_rows, stream, mr); + auto left_outer = mixed_join(left_equality, + right_equality, + left_conditional, + right_conditional, + binary_predicate, + compare_nulls, + join_kind::LEFT_JOIN, + std::nullopt, + stream, + mr); + return finalize_full_join( + std::move(left_outer), left_conditional.num_rows(), right_conditional.num_rows(), stream, mr); } - return join_indices; + + auto const hash_joiner = cudf::hash_join{right_equality, compare_nulls, stream}; + auto const [left_indices, right_indices] = + equality_join_indices(hash_joiner, left_equality, join_type, stream, mr); + + auto const output_size = output_size_data.has_value() + ? std::optional{output_size_data->first} + : std::nullopt; + + return detail::filter_join_indices(left_conditional, + right_conditional, + *left_indices, + *right_indices, + binary_predicate, + join_type, + output_size, + stream, + mr); } std::pair>> @@ -542,67 +138,43 @@ compute_mixed_join_output_size(table_view const& left_equality, { CUDF_EXPECTS(join_type != join_kind::FULL_JOIN, "Size estimation is not available for full joins."); - CUDF_EXPECTS( (join_type != join_kind::LEFT_SEMI_JOIN) && (join_type != join_kind::LEFT_ANTI_JOIN), "Left semi and anti join size estimation should use compute_mixed_join_output_size_semi."); + CUDF_EXPECTS(left_conditional.num_rows() == left_equality.num_rows(), + "The left conditional and equality tables must have the same number of rows."); + CUDF_EXPECTS(right_conditional.num_rows() == right_equality.num_rows(), + "The right conditional and equality tables must have the same number of rows."); - auto const right_num_rows = right_conditional.num_rows(); - auto const left_num_rows = left_conditional.num_rows(); - - // Handle empty table cases early - if (right_num_rows == 0 || left_num_rows == 0) { - auto const outer_num_rows = - ((join_type == join_kind::INNER_JOIN) && (right_num_rows > left_num_rows)) ? right_num_rows - : left_num_rows; - auto matches_per_row = std::make_unique>( - static_cast(outer_num_rows), stream, mr); - auto matches_per_row_span = cudf::device_span{ - matches_per_row->begin(), static_cast(outer_num_rows)}; - - if (right_num_rows == 0 && join_type == join_kind::LEFT_JOIN) { - thrust::fill(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()), - matches_per_row_span.begin(), - matches_per_row_span.end(), - 1); - return {left_num_rows, std::move(matches_per_row)}; - } else { - thrust::fill(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()), - matches_per_row_span.begin(), - matches_per_row_span.end(), - 0); - return {0, std::move(matches_per_row)}; + // hash_join requires a non-empty build (right) table. + if (right_conditional.num_rows() == 0) { + auto const left_num_rows = left_conditional.num_rows(); + if (join_type == join_kind::LEFT_JOIN) { + auto counts = + rmm::device_uvector(static_cast(left_num_rows), stream, mr); + thrust::uninitialized_fill( + rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()), + counts.begin(), + counts.end(), + size_type{1}); + return {static_cast(left_num_rows), + std::make_unique>(std::move(counts))}; } + return {0, std::make_unique>(0, stream, mr)}; } - auto setup = setup_mixed_join_common(left_equality, - right_equality, - left_conditional, - right_conditional, - binary_predicate, - compare_nulls, - join_type, - stream, - mr); - - bool const is_outer_join = (join_type == join_kind::LEFT_JOIN); - - // Use the helper function to compute matches per row - return compute_mixed_join_matches_per_row(setup.has_nulls, - *setup.left_conditional_view, - *setup.right_conditional_view, - is_outer_join, - setup.swap_tables, - setup.equality_probe, - setup.hash_table_storage, - setup.input_pairs.data(), - setup.hash_indices.data(), - setup.parser.device_expression_data, - setup.outer_num_rows, - setup.config, - setup.shmem_size_per_block, - stream, - mr); + auto const hash_joiner = cudf::hash_join{right_equality, compare_nulls, stream}; + auto const [left_indices, right_indices] = + equality_join_indices(hash_joiner, left_equality, join_type, stream, mr); + + return cudf::filter_join_indices_output_size(left_conditional, + right_conditional, + *left_indices, + *right_indices, + binary_predicate, + join_type, + stream, + mr); } } // namespace detail diff --git a/cpp/src/join/mixed_join_common_utils.cuh b/cpp/src/join/mixed_join_common_utils.cuh index ecf9360c096d..195ef534d80a 100644 --- a/cpp/src/join/mixed_join_common_utils.cuh +++ b/cpp/src/join/mixed_join_common_utils.cuh @@ -1,85 +1,17 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ #pragma once #include -#include #include #include #include -#include #include -#include - -#include -#include -#include - -#include namespace cudf::detail { -using pair_type = cuco::pair; - -using hash_type = cuco::murmurhash3_32; - -/** - * @brief A custom comparator used for the mixed join multiset insertion - */ -struct mixed_join_always_not_equal { - __device__ constexpr bool operator()(pair_type const&, pair_type const&) const noexcept - { - return false; - } -}; - -/** - * @brief Hash functions for double hashing in mixed joins. - * - * These hashers implement a double hashing scheme for the mixed join multiset: - * - * - mixed_join_hasher1: Determines the initial probe slot for a given key. We simply use - * the precomputed row hash value, which is the first element of our (row_hash, row_index) pair. - * - * - mixed_join_hasher2: Determines the step size for the probing sequence. This allows keys - * with the same hash value to have different step sizes, helping to avoid secondary clustering. - * - * Note: Strictly speaking, this setup does not truly avoid secondary clustering because rows with - * the same hash value still receive the same step size. A true secondary clustering avoidance - * method would compute a different hash value for each row. However, based on performance testing, - * this current approach actually delivers better performance than computing row hashes with a - * different hasher. - */ -struct mixed_join_hasher1 { - __device__ constexpr hash_value_type operator()(pair_type const& key) const noexcept - { - return key.first; - } -}; - -struct mixed_join_hasher2 { - mixed_join_hasher2(hash_value_type seed) : _hash{seed} {} - - __device__ constexpr hash_value_type operator()(pair_type const& key) const noexcept - { - return _hash(key.first); - } - - private: - hash_type _hash; -}; - -using mixed_multiset_type = - cuco::static_multiset, - cuda::thread_scope_device, - mixed_join_always_not_equal, - cuco::double_hashing<1, mixed_join_hasher1, mixed_join_hasher2>, - rmm::mr::polymorphic_allocator, - cuco::storage<2>>; - using row_hash = cudf::detail::row::hash::device_row_hasher; @@ -116,141 +48,4 @@ struct expression_equality { row_equality const& equality_probe; }; -/** - * @brief Equality comparator for cuco::static_multiset queries. - * - * This equality comparator is designed for use with cuco::static_multiset's APIs. - * A probe hit indicates that the hashes of the keys are equal, at which point - * this comparator checks whether the keys themselves are equal (using the - * provided row_equality comparator) and then evaluates the conditional expression - */ -template -struct pair_expression_equality : public expression_equality { - using expression_equality::expression_equality; - -#ifndef NDEBUG - __attribute__((noinline)) -#else - __forceinline__ -#endif - __device__ bool - operator()(pair_type const& left_row, pair_type const& right_row) const noexcept - { - using cudf::detail::row::lhs_index_type; - using cudf::detail::row::rhs_index_type; - - auto output_dest = cudf::ast::detail::value_expression_result(); - // Three levels of checks: - // 1. Row hashes of the columns involved in the equality condition are equal. - // 2. The contents of the columns involved in the equality condition are equal. - // 3. The predicate evaluated on the relevant columns (already encoded in the evaluator) - // evaluates to true. - if ((left_row.first == right_row.first) && - this->equality_probe(lhs_index_type{left_row.second}, rhs_index_type{right_row.second})) { - auto const lrow_idx = this->swap_tables ? right_row.second : left_row.second; - auto const rrow_idx = this->swap_tables ? left_row.second : right_row.second; - this->evaluator.evaluate( - output_dest, lrow_idx, rrow_idx, 0, this->thread_intermediate_storage); - return (output_dest.is_valid() && output_dest.value()); - } - return false; - } -}; - -/** - * @brief Common utility for probing a hash table bucket and checking slot equality - * - * This encapsulates the common logic of reading bucket slots and checking for - * empty slots and key equality, used by both count and retrieve operations. - */ -template -struct hash_probe_result { - bool first_slot_is_empty_; - bool second_slot_is_empty_; - bool first_slot_equals_; - bool second_slot_equals_; - - __device__ __forceinline__ hash_probe_result( - pair_expression_equality const& key_equal, - cudf::device_span> hash_table_storage, - cuco::pair const& probe_key, - std::size_t probe_idx) - { - auto const* data = hash_table_storage.data(); - __builtin_assume_aligned(data, 2 * sizeof(cuco::pair)); - auto const first = *(data + probe_idx); - auto const second = *(data + probe_idx + 1); - - first_slot_is_empty_ = first.second == cudf::JoinNoMatch; - second_slot_is_empty_ = second.second == cudf::JoinNoMatch; - first_slot_equals_ = (not first_slot_is_empty_ and key_equal(probe_key, first)); - second_slot_equals_ = (not second_slot_is_empty_ and key_equal(probe_key, second)); - } - - __device__ __forceinline__ bool has_empty_slot() const noexcept - { - return first_slot_is_empty_ or second_slot_is_empty_; - } - - __device__ __forceinline__ cudf::size_type match_count() const noexcept - { - return static_cast(first_slot_equals_) + - static_cast(second_slot_equals_); - } - - __device__ __forceinline__ bool has_match() const noexcept - { - return first_slot_equals_ or second_slot_equals_; - } -}; - -/** - * @brief Iterator-style wrapper for probing through a hash table - * - * This encapsulates the common double hashing probe sequence used by both - * count and retrieve kernels. - */ -template -struct hash_table_prober { - cudf::device_span> hash_table_storage_; - pair_expression_equality const& key_equal_; - cuco::pair const& probe_key_; - std::size_t probe_idx_; - std::size_t step_; - std::size_t extent_; - - __device__ __forceinline__ hash_table_prober( - pair_expression_equality const& key_equal, - cudf::device_span> hash_table_storage, - cuco::pair const& probe_key, - cuda::std::pair const& hash_idx) - : hash_table_storage_{hash_table_storage}, - key_equal_{key_equal}, - probe_key_{probe_key}, - probe_idx_{static_cast(hash_idx.first)}, - step_{static_cast(hash_idx.second)}, - extent_{hash_table_storage.size()} - { - } - - __device__ __forceinline__ hash_probe_result probe_current_bucket() const - { - return hash_probe_result{key_equal_, hash_table_storage_, probe_key_, probe_idx_}; - } - - __device__ __forceinline__ void advance() noexcept - { - probe_idx_ = (probe_idx_ + step_) % extent_; - } - - __device__ __forceinline__ auto get_bucket_slots() const noexcept - { - auto const* data = hash_table_storage_.data(); - __builtin_assume_aligned(data, 2 * sizeof(cuco::pair)); - auto const first = *(data + probe_idx_); - auto const second = *(data + probe_idx_ + 1); - return cuda::std::pair{first, second}; - } -}; - } // namespace cudf::detail diff --git a/cpp/src/join/mixed_join_kernel.cu b/cpp/src/join/mixed_join_kernel.cu deleted file mode 100644 index 982ca557f5b6..000000000000 --- a/cpp/src/join/mixed_join_kernel.cu +++ /dev/null @@ -1,28 +0,0 @@ -/* - * SPDX-FileCopyrightText: Copyright (c) 2022-2025, NVIDIA CORPORATION. - * SPDX-License-Identifier: Apache-2.0 - */ - -#include "mixed_join_kernel.cuh" -#include "mixed_join_kernel.hpp" - -namespace cudf::detail { - -template void launch_mixed_join( - table_device_view left_table, - table_device_view right_table, - bool is_outer_join, - bool swap_tables, - row_equality equality_probe, - cudf::device_span> hash_table_storage, - cuco::pair const* input_pairs, - cuda::std::pair const* hash_indices, - cudf::ast::detail::expression_device_view device_expression_data, - size_type* join_output_l, - size_type* join_output_r, - cudf::size_type const* join_result_offsets, - detail::grid_1d config, - int64_t shmem_size_per_block, - rmm::cuda_stream_view stream); - -} // namespace cudf::detail diff --git a/cpp/src/join/mixed_join_kernel.cuh b/cpp/src/join/mixed_join_kernel.cuh deleted file mode 100644 index 4c865b44d3b4..000000000000 --- a/cpp/src/join/mixed_join_kernel.cuh +++ /dev/null @@ -1,182 +0,0 @@ -/* - * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION. - * SPDX-License-Identifier: Apache-2.0 - */ - -#pragma once - -#include "join_common_utils.hpp" -#include "mixed_join_common_utils.cuh" -#include "mixed_join_kernel.hpp" - -#include -#include -#include -#include -#include - -#include -#include - -namespace cudf { -namespace detail { - -/** - * @brief Optimized retrieve implementation using precomputed matches per row - * - * This implementation uses precomputed match counts to avoid expensive atomic - * operations and directly fills output arrays based on known match positions. - * - * @tparam is_outer Boolean flag indicating whether outer join semantics should be used - * @tparam has_nulls Whether the input tables may contain nulls - */ -template -__device__ __forceinline__ void retrieve_matches( - cudf::device_span> hash_table_storage, - pair_expression_equality const& key_equal, - cuco::pair const& probe_key, - cuda::std::pair const& hash_idx, - cudf::size_type* probe_output, - cudf::size_type* match_output) noexcept -{ - auto const probe_row_index = probe_key.second; - cudf::size_type output_idx = 0; - bool found_match = false; - auto prober = hash_table_prober{key_equal, hash_table_storage, probe_key, hash_idx}; - - while (true) { - auto const result = prober.probe_current_bucket(); - auto const bucket_slots = prober.get_bucket_slots(); - - if (result.first_slot_equals_) { - probe_output[output_idx] = probe_row_index; - match_output[output_idx] = bucket_slots.first.second; - output_idx++; - found_match = true; - } - - if (result.second_slot_equals_) { - probe_output[output_idx] = probe_row_index; - match_output[output_idx] = bucket_slots.second.second; - output_idx++; - found_match = true; - } - - // Exit if we find an empty slot - if (result.has_empty_slot()) { break; } - - prober.advance(); - } - - // Handle outer join logic for non-matching rows - if constexpr (is_outer) { - if (not found_match) { - probe_output[0] = probe_row_index; - match_output[0] = cudf::JoinNoMatch; - } - } -} - -template -CUDF_KERNEL void __launch_bounds__(DEFAULT_JOIN_BLOCK_SIZE) - mixed_join(table_device_view left_table, - table_device_view right_table, - bool is_outer_join, - bool swap_tables, - row_equality equality_probe, - cudf::device_span> hash_table_storage, - cuco::pair const* input_pairs, - cuda::std::pair const* hash_indices, - cudf::ast::detail::expression_device_view device_expression_data, - size_type* join_output_l, - size_type* join_output_r, - cudf::size_type const* join_result_offsets) -{ - // Normally the casting of a shared memory array is used to create multiple - // arrays of different types from the shared memory buffer, but here it is - // used to circumvent conflicts between arrays of different types between - // different template instantiations due to the extern specifier. - extern __shared__ char raw_intermediate_storage[]; - cudf::ast::detail::IntermediateDataType* intermediate_storage = - reinterpret_cast*>(raw_intermediate_storage); - auto thread_intermediate_storage = - &intermediate_storage[threadIdx.x * device_expression_data.num_intermediates]; - - cudf::size_type const left_num_rows = left_table.num_rows(); - cudf::size_type const right_num_rows = right_table.num_rows(); - auto const outer_num_rows = (swap_tables ? right_num_rows : left_num_rows); - - auto const start_idx = cudf::detail::grid_1d::global_thread_id(); - auto const stride = cudf::detail::grid_1d::grid_stride(); - - auto const evaluator = cudf::ast::detail::expression_evaluator{ - left_table, right_table, device_expression_data}; - - auto const equality = pair_expression_equality{ - evaluator, thread_intermediate_storage, swap_tables, equality_probe}; - - // Process each row and write matches to precomputed output positions - for (auto outer_row_index = start_idx; outer_row_index < outer_num_rows; - outer_row_index += stride) { - auto const& probe_key = input_pairs[outer_row_index]; - auto const& hash_idx = hash_indices[outer_row_index]; - auto const output_offset = join_result_offsets[outer_row_index]; - - if (is_outer_join) { - retrieve_matches( - hash_table_storage, - equality, - probe_key, - hash_idx, - swap_tables ? join_output_r + output_offset : join_output_l + output_offset, - swap_tables ? join_output_l + output_offset : join_output_r + output_offset); - } else { - retrieve_matches( - hash_table_storage, - equality, - probe_key, - hash_idx, - swap_tables ? join_output_r + output_offset : join_output_l + output_offset, - swap_tables ? join_output_l + output_offset : join_output_r + output_offset); - } - } -} - -template -void launch_mixed_join( - table_device_view left_table, - table_device_view right_table, - bool is_outer_join, - bool swap_tables, - row_equality equality_probe, - cudf::device_span> hash_table_storage, - cuco::pair const* input_pairs, - cuda::std::pair const* hash_indices, - cudf::ast::detail::expression_device_view device_expression_data, - size_type* join_output_l, - size_type* join_output_r, - cudf::size_type const* join_result_offsets, - detail::grid_1d config, - int64_t shmem_size_per_block, - rmm::cuda_stream_view stream) -{ - mixed_join - <<>>( - left_table, - right_table, - is_outer_join, - swap_tables, - equality_probe, - hash_table_storage, - input_pairs, - hash_indices, - device_expression_data, - join_output_l, - join_output_r, - join_result_offsets); - CUDF_CUDA_TRY(cudaGetLastError()); -} - -} // namespace detail - -} // namespace cudf diff --git a/cpp/src/join/mixed_join_kernel.hpp b/cpp/src/join/mixed_join_kernel.hpp deleted file mode 100644 index d465e8b21c60..000000000000 --- a/cpp/src/join/mixed_join_kernel.hpp +++ /dev/null @@ -1,73 +0,0 @@ -/* - * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION. - * SPDX-License-Identifier: Apache-2.0 - */ - -#pragma once - -#include "mixed_join_common_utils.cuh" - -#include -#include -#include -#include -#include - -#include - -#include -#include - -namespace CUDF_EXPORT cudf { -namespace detail { - -/** - * @brief Performs a join using the combination of a hash lookup to identify - * equal rows between one pair of tables and the evaluation of an expression - * containing an arbitrary expression. - * - * This method probes the hash table with each row in the probe table using a - * custom equality comparator that also checks that the conditional expression - * evaluates to true between the left/right tables when a match is found - * between probe and build rows. - * - * @tparam has_nulls Whether or not the inputs may contain nulls. - * - * @param[in] left_table The left table - * @param[in] right_table The right table - * @param[in] is_outer_join Whether this is an outer join - * @param[in] swap_tables If true, the kernel was launched with one thread per right row and - * the kernel needs to internally loop over left rows. Otherwise, loop over right rows. - * @param[in] equality_probe The equality comparator used when probing the hash table. - * @param[in] hash_table_storage Device span of the hash table storage - * @param[in] input_pairs Precomputed input pairs for probing - * @param[in] hash_indices Precomputed hash indices for efficient probing - * @param[in] device_expression_data Container of device data required to evaluate the desired - * expression. - * @param[out] join_output_l The left result of the join operation - * @param[out] join_output_r The right result of the join operation - * @param[in] join_result_offsets Prefix sum of matches_per_row to get output offsets - * @param[in] config Grid configuration for the kernel launch - * @param[in] shmem_size_per_block Shared memory size per block - * @param[in] stream CUDA stream to use - */ -template -void launch_mixed_join( - table_device_view left_table, - table_device_view right_table, - bool is_outer_join, - bool swap_tables, - row_equality equality_probe, - cudf::device_span> hash_table_storage, - cuco::pair const* input_pairs, - cuda::std::pair const* hash_indices, - cudf::ast::detail::expression_device_view device_expression_data, - size_type* join_output_l, - size_type* join_output_r, - cudf::size_type const* join_result_offsets, - detail::grid_1d config, - int64_t shmem_size_per_block, - rmm::cuda_stream_view stream); - -} // namespace detail -} // namespace CUDF_EXPORT cudf diff --git a/cpp/src/join/mixed_join_kernel_nulls.cu b/cpp/src/join/mixed_join_kernel_nulls.cu deleted file mode 100644 index d3b382331dc9..000000000000 --- a/cpp/src/join/mixed_join_kernel_nulls.cu +++ /dev/null @@ -1,28 +0,0 @@ -/* - * SPDX-FileCopyrightText: Copyright (c) 2022-2025, NVIDIA CORPORATION. - * SPDX-License-Identifier: Apache-2.0 - */ - -#include "mixed_join_kernel.cuh" -#include "mixed_join_kernel.hpp" - -namespace cudf::detail { - -template void launch_mixed_join( - table_device_view left_table, - table_device_view right_table, - bool is_outer_join, - bool swap_tables, - row_equality equality_probe, - cudf::device_span> hash_table_storage, - cuco::pair const* input_pairs, - cuda::std::pair const* hash_indices, - cudf::ast::detail::expression_device_view device_expression_data, - size_type* join_output_l, - size_type* join_output_r, - cudf::size_type const* join_result_offsets, - detail::grid_1d config, - int64_t shmem_size_per_block, - rmm::cuda_stream_view stream); - -} // namespace cudf::detail diff --git a/cpp/src/join/mixed_join_semi.cu b/cpp/src/join/mixed_join_semi.cu index aa7674fb86dc..8021cdab21bc 100644 --- a/cpp/src/join/mixed_join_semi.cu +++ b/cpp/src/join/mixed_join_semi.cu @@ -11,6 +11,7 @@ #include #include #include +#include #include #include #include diff --git a/cpp/src/join/mixed_join_size_kernel.cu b/cpp/src/join/mixed_join_size_kernel.cu deleted file mode 100644 index b594b8a1a334..000000000000 --- a/cpp/src/join/mixed_join_size_kernel.cu +++ /dev/null @@ -1,28 +0,0 @@ -/* - * SPDX-FileCopyrightText: Copyright (c) 2022-2025, NVIDIA CORPORATION. - * SPDX-License-Identifier: Apache-2.0 - */ - -#include "mixed_join_size_kernel.cuh" -#include "mixed_join_size_kernel.hpp" - -namespace cudf { -namespace detail { - -template void launch_mixed_join_count( - cudf::table_device_view left_table, - cudf::table_device_view right_table, - bool is_outer_join, - bool swap_tables, - row_equality equality_probe, - cudf::device_span> hash_table_storage, - cuco::pair const* input_pairs, - cuda::std::pair const* hash_indices, - ast::detail::expression_device_view device_expression_data, - cudf::device_span matches_per_row, - detail::grid_1d config, - int64_t shmem_size_per_block, - rmm::cuda_stream_view stream); - -} // namespace detail -} // namespace cudf diff --git a/cpp/src/join/mixed_join_size_kernel.cuh b/cpp/src/join/mixed_join_size_kernel.cuh deleted file mode 100644 index 9a719170d7a3..000000000000 --- a/cpp/src/join/mixed_join_size_kernel.cuh +++ /dev/null @@ -1,136 +0,0 @@ -/* - * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION. - * SPDX-License-Identifier: Apache-2.0 - */ - -#pragma once - -#include "join_common_utils.hpp" -#include "mixed_join_common_utils.cuh" -#include "mixed_join_size_kernel.hpp" - -#include -#include -#include -#include -#include -#include - -#include -#include - -namespace cudf::detail { - -/** - * @brief Standalone count implementation using precomputed hash indices - * - * This implementation provides essential count functionality for mixed joins - * using precomputed probe indices and step sizes. - */ -template -__device__ __forceinline__ auto standalone_count( - pair_expression_equality const& key_equal, - cudf::device_span> hash_table_storage, - cuco::pair const& probe_key, - cuda::std::pair const& hash_idx, - bool is_outer_join) noexcept -{ - cudf::size_type count = 0; - auto prober = hash_table_prober{key_equal, hash_table_storage, probe_key, hash_idx}; - - while (true) { - auto const result = prober.probe_current_bucket(); - count += result.match_count(); - - // Exit if we find an empty slot - if (result.has_empty_slot()) { - // Handle outer join logic: non-matching rows are counted as 1 match - if (is_outer_join && count == 0) { return 1; } - return count; - } - - prober.advance(); - } -} - -template -CUDF_KERNEL void __launch_bounds__(DEFAULT_JOIN_BLOCK_SIZE) mixed_join_count( - table_device_view left_table, - table_device_view right_table, - bool is_outer_join, - bool swap_tables, - row_equality equality_probe, - cudf::device_span> hash_table_storage, - cuco::pair const* input_pairs, - cuda::std::pair const* hash_indices, - ast::detail::expression_device_view device_expression_data, - cudf::device_span matches_per_row) -{ - // The (required) extern storage of the shared memory array leads to - // conflicting declarations between different templates. The easiest - // workaround is to declare an arbitrary (here char) array type then cast it - // after the fact to the appropriate type. - extern __shared__ char raw_intermediate_storage[]; - auto intermediate_storage = - reinterpret_cast*>(raw_intermediate_storage); - auto thread_intermediate_storage = - intermediate_storage + (threadIdx.x * device_expression_data.num_intermediates); - - auto const start_idx = cudf::detail::grid_1d::global_thread_id(); - auto const stride = cudf::detail::grid_1d::grid_stride(); - cudf::size_type const left_num_rows = left_table.num_rows(); - cudf::size_type const right_num_rows = right_table.num_rows(); - auto const outer_num_rows = (swap_tables ? right_num_rows : left_num_rows); - - auto const evaluator = cudf::ast::detail::expression_evaluator{ - left_table, right_table, device_expression_data}; - - // Figure out the number of elements for this key. - // TODO: Address asymmetry in operator. - auto count_equality = pair_expression_equality{ - evaluator, thread_intermediate_storage, swap_tables, equality_probe}; - - for (auto outer_row_index = start_idx; outer_row_index < outer_num_rows; - outer_row_index += stride) { - auto const& probe_key = input_pairs[outer_row_index]; - auto const& hash_idx = hash_indices[outer_row_index]; - - auto match_count = - standalone_count(count_equality, hash_table_storage, probe_key, hash_idx, is_outer_join); - - matches_per_row[outer_row_index] = match_count; - } -} - -template -void launch_mixed_join_count( - table_device_view left_table, - table_device_view right_table, - bool is_outer_join, - bool swap_tables, - row_equality equality_probe, - cudf::device_span> hash_table_storage, - cuco::pair const* input_pairs, - cuda::std::pair const* hash_indices, - ast::detail::expression_device_view device_expression_data, - cudf::device_span matches_per_row, - detail::grid_1d config, - int64_t shmem_size_per_block, - rmm::cuda_stream_view stream) -{ - mixed_join_count - <<>>( - left_table, - right_table, - is_outer_join, - swap_tables, - equality_probe, - hash_table_storage, - input_pairs, - hash_indices, - device_expression_data, - matches_per_row); - CUDF_CUDA_TRY(cudaGetLastError()); -} - -} // namespace cudf::detail diff --git a/cpp/src/join/mixed_join_size_kernel.hpp b/cpp/src/join/mixed_join_size_kernel.hpp deleted file mode 100644 index 16909cee1518..000000000000 --- a/cpp/src/join/mixed_join_size_kernel.hpp +++ /dev/null @@ -1,74 +0,0 @@ -/* - * SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION. - * SPDX-License-Identifier: Apache-2.0 - */ - -#pragma once - -#include "mixed_join_common_utils.cuh" - -#include -#include -#include -#include -#include -#include - -#include - -#include -#include - -namespace CUDF_EXPORT cudf { -namespace detail { - -/** - * @brief Computes the output size of joining the left table to the right table. - * - * This method probes the hash table with each row in the probe table using a - * custom equality comparator that also checks that the conditional expression - * evaluates to true between the left/right tables when a match is found - * between probe and build rows. - * - * @tparam has_nulls Whether or not the inputs may contain nulls. - * - * @param[in] left_table The left table - * @param[in] right_table The right table - * @param[in] is_outer_join Whether this is an outer join - * @param[in] swap_tables If true, the kernel was launched with one thread per right row and - * the kernel needs to internally loop over left rows. Otherwise, loop over right rows. - * @param[in] equality_probe The equality comparator used when probing the hash table. - * @param[in] hash_table_storage Device span of the hash table storage - * @param[in] input_pairs Precomputed input pairs for probing - * @param[in] hash_indices Precomputed hash indices for efficient probing - * @param[in] device_expression_data Container of device data required to evaluate the desired - * expression. - * @param[out] matches_per_row The number of matches in one pair of - * equality/conditional tables for each row in the other pair of tables. If - * swap_tables is true, matches_per_row corresponds to the right_table, - * otherwise it corresponds to the left_table. Note that corresponding swap of - * left/right tables to determine which is the build table and which is the - * probe table has already happened on the host. - * @param[in] config Grid configuration for the kernel launch - * @param[in] shmem_size_per_block Shared memory size per block - * @param[in] stream CUDA stream to use - */ - -template -void launch_mixed_join_count( - table_device_view left_table, - table_device_view right_table, - bool is_outer_join, - bool swap_tables, - row_equality equality_probe, - cudf::device_span> hash_table_storage, - cuco::pair const* input_pairs, - cuda::std::pair const* hash_indices, - ast::detail::expression_device_view device_expression_data, - cudf::device_span matches_per_row, - detail::grid_1d config, - int64_t shmem_size_per_block, - rmm::cuda_stream_view stream); - -} // namespace detail -} // namespace CUDF_EXPORT cudf diff --git a/cpp/src/join/mixed_join_size_kernel_nulls.cu b/cpp/src/join/mixed_join_size_kernel_nulls.cu deleted file mode 100644 index 2f94da3bfd03..000000000000 --- a/cpp/src/join/mixed_join_size_kernel_nulls.cu +++ /dev/null @@ -1,28 +0,0 @@ -/* - * SPDX-FileCopyrightText: Copyright (c) 2022-2025, NVIDIA CORPORATION. - * SPDX-License-Identifier: Apache-2.0 - */ - -#include "mixed_join_size_kernel.cuh" -#include "mixed_join_size_kernel.hpp" - -namespace cudf { -namespace detail { - -template void launch_mixed_join_count( - cudf::table_device_view left_table, - cudf::table_device_view right_table, - bool is_outer_join, - bool swap_tables, - row_equality equality_probe, - cudf::device_span> hash_table_storage, - cuco::pair const* input_pairs, - cuda::std::pair const* hash_indices, - ast::detail::expression_device_view device_expression_data, - cudf::device_span matches_per_row, - detail::grid_1d config, - int64_t shmem_size_per_block, - rmm::cuda_stream_view stream); - -} // namespace detail -} // namespace cudf diff --git a/cpp/tests/join/mixed_join_tests.cu b/cpp/tests/join/mixed_join_tests.cu index 3300e3b56f17..701800cd5726 100644 --- a/cpp/tests/join/mixed_join_tests.cu +++ b/cpp/tests/join/mixed_join_tests.cu @@ -18,6 +18,8 @@ #include #include +#include + #include #include #include @@ -269,15 +271,14 @@ struct MixedJoinPairReturnTest : public MixedJoinTest { left_equality, right_equality, left_conditional, right_conditional, predicate, compare_nulls); EXPECT_TRUE(result_size == expected_outputs.size()); - cudf::test::fixed_width_column_wrapper expected_counts_cw( - expected_counts.begin(), expected_counts.end()); - auto const actual_counts_view = - cudf::column_view(cudf::data_type{cudf::type_to_id()}, - actual_counts->size(), - actual_counts->data(), - nullptr, - 0); - CUDF_TEST_EXPECT_COLUMNS_EQUAL(expected_counts_cw, actual_counts_view); + auto const expected_total = + std::accumulate(expected_counts.begin(), expected_counts.end(), std::size_t{0}); + EXPECT_EQ(expected_total, result_size); + auto const actual_total = thrust::reduce(rmm::exec_policy_nosync(cudf::get_default_stream()), + actual_counts->begin(), + actual_counts->end(), + std::size_t{0}); + EXPECT_EQ(actual_total, result_size); auto result = this->join(left_equality, right_equality, @@ -434,14 +435,14 @@ struct MixedInnerJoinTest : public MixedJoinPairReturnTest { this->compare_join_results(mixed_result, ast_filter_result); // Verify filter_join_indices_output_size matches the materialized output size. - auto const fji_size = cudf::filter_join_indices_output_size( + auto const filter_output_size_result = cudf::filter_join_indices_output_size( left_conditional, right_conditional, cudf::device_span(*hash_join_result.first), cudf::device_span(*hash_join_result.second), predicate, cudf::join_kind::INNER_JOIN); - EXPECT_EQ(fji_size, ast_filter_result.first->size()); + EXPECT_EQ(filter_output_size_result.first, ast_filter_result.first->size()); // Verify JIT filter_join_indices if provided if (!jit_predicate.empty()) { @@ -1102,14 +1103,14 @@ struct MixedLeftJoinTest : public MixedJoinPairReturnTest { this->compare_join_results(mixed_result, ast_filter_result); // Verify filter_join_indices_output_size matches the materialized output size. - auto const fji_size = cudf::filter_join_indices_output_size( + auto const filter_output_size_result = cudf::filter_join_indices_output_size( left_conditional, right_conditional, cudf::device_span(*hash_join_result.first), cudf::device_span(*hash_join_result.second), predicate, cudf::join_kind::LEFT_JOIN); - EXPECT_EQ(fji_size, ast_filter_result.first->size()); + EXPECT_EQ(filter_output_size_result.first, ast_filter_result.first->size()); // Verify JIT filter_join_indices if provided if (!jit_predicate.empty()) { @@ -1369,60 +1370,8 @@ struct MixedFullJoinTest : public MixedJoinPairReturnTest { cudf::null_equality compare_nulls = cudf::null_equality::EQUAL, std::string const& jit_predicate = "") override { - // Test both approaches and verify they produce the same results - auto mixed_result = cudf::mixed_full_join( + return cudf::mixed_full_join( left_equality, right_equality, left_conditional, right_conditional, predicate, compare_nulls); - - // Alternative approach: hash_join + filter_join_indices - // Skip hash_join approach for empty tables (hash_join doesn't support empty tables) - if (left_equality.num_rows() > 0 && right_equality.num_rows() > 0) { - cudf::hash_join hash_joiner(right_equality, compare_nulls); - auto hash_join_result = hash_joiner.full_join(left_equality); - - // Verify AST filter_join_indices - auto ast_filter_result = cudf::filter_join_indices( - left_conditional, - right_conditional, - cudf::device_span(*hash_join_result.first), - cudf::device_span(*hash_join_result.second), - predicate, - cudf::join_kind::FULL_JOIN); - this->compare_join_results(mixed_result, ast_filter_result); - - // Verify filter_join_indices_output_size matches the materialized output size. - auto const fji_size = cudf::filter_join_indices_output_size( - left_conditional, - right_conditional, - cudf::device_span(*hash_join_result.first), - cudf::device_span(*hash_join_result.second), - predicate, - cudf::join_kind::FULL_JOIN); - EXPECT_EQ(fji_size, ast_filter_result.first->size()); - - // Verify JIT filter_join_indices if provided - if (!jit_predicate.empty()) { - auto jit_filter_result = cudf::filter_join_indices_jit( - left_conditional, - right_conditional, - cudf::device_span(*hash_join_result.first), - cudf::device_span(*hash_join_result.second), - jit_predicate, - cudf::join_kind::FULL_JOIN); - this->compare_join_results(mixed_result, jit_filter_result); - } - - // Verify AST-based JIT filter_join_indices - auto jit_ast_filter_result = cudf::filter_join_indices_jit( - left_conditional, - right_conditional, - cudf::device_span(*hash_join_result.first), - cudf::device_span(*hash_join_result.second), - predicate, - cudf::join_kind::FULL_JOIN); - this->compare_join_results(mixed_result, jit_ast_filter_result); - } - - return mixed_result; } std::pair>> join_size( @@ -1520,6 +1469,24 @@ TYPED_TEST(MixedFullJoinTest, Basic2) {cudf::JoinNoMatch, 2}}); } +TYPED_TEST(MixedFullJoinTest, MultiMatchUnmatchedDedup) +{ + auto const predicate = + cudf::ast::operation(cudf::ast::ast_operator::GREATER, col_ref_left_0, col_ref_right_0); + this->test({{5, 5, 7}, {10, 1, 10}}, + {{5, 5, 9}, {2, 20, 0}}, + {0}, + {1}, + predicate, + {}, + {{0, 0}, + {1, cudf::JoinNoMatch}, + {2, cudf::JoinNoMatch}, + {cudf::JoinNoMatch, 1}, + {cudf::JoinNoMatch, 2}}, + make_jit_comparison(1, 1, 0, 0, ">")); +} + using MixedFullJoinTest_int32 = MixedFullJoinTest; TEST_F(MixedFullJoinTest_int32, NullableColumnsWithModuloFilter) { diff --git a/cpp/tests/streams/join_test.cpp b/cpp/tests/streams/join_test.cpp index 3c48c825bf0b..b66223be1dde 100644 --- a/cpp/tests/streams/join_test.cpp +++ b/cpp/tests/streams/join_test.cpp @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ @@ -152,6 +152,7 @@ TEST_F(JoinTest, LeftJoinWithPostFilter) cudf::device_span(*hash_join_result.second), left_zero_eq_right_zero, cudf::join_kind::LEFT_JOIN, + std::nullopt, cudf::test::get_default_stream()); }