Reduce filtered join build time

cpp/CMakeLists.txt

@@ -855,7 +855,11 @@ add_library(
   src/join/filter_join_indices/filter_join_indices_output_size_kernel_null_complex.cu
   src/join/filter_join_indices/filter_join_indices_output_size_kernel_null_primitive.cu
   src/join/filter_join_indices/filter_join_indices_output_size_kernel_primitive.cu
-  src/join/filtered_join.cu
+  src/join/filtered_join/filtered_join.cu
+  src/join/filtered_join/filtered_join_flat.cu
+  src/join/filtered_join/filtered_join_nested.cu
+  src/join/filtered_join/filtered_join_nested_query.cu
+  src/join/filtered_join/filtered_join_primitive.cu
   src/join/hash_join/finalize_partitioned_full_join.cpp
   src/join/hash_join/full_join_match_context.cpp
   src/join/hash_join/full_join_retrieve.cu

cpp/include/cudf/detail/join/distinct_filtered_join.cuh

@@ -6,12 +6,20 @@
 
 #include <cudf/detail/join/filtered_join.cuh>
 #include <cudf/table/table_view.hpp>
-#include <cudf/utilities/default_stream.hpp>
-#include <cudf/utilities/memory_resource.hpp>
+#include <cudf/types.hpp>
+#include <cudf/utilities/span.hpp>
 
 #include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_uvector.hpp>
 #include <rmm/resource_ref.hpp>
 
+#include <cstdint>
+#include <memory>
+
+namespace cudf::detail::row::equality {
+struct preprocessed_table;
+}
+
 namespace cudf {
 
 // Forward declaration
@@ -44,30 +52,31 @@ class distinct_filtered_join : public filtered_join {
     rmm::cuda_stream_view stream,
     rmm::device_async_resource_ref mr);
 
-  /**
-   * @brief Core implementation for querying the hash table
-   *
-   * Performs the actual hash table query operation for both semi and anti joins
-   * using set semantics.
-   *
-   * @tparam CGSize CUDA cooperative group size
-   * @tparam Ref Reference type for the hash table
-   * @param left The left table to probe the hash table with
-   * @param preprocessed_left Preprocessed left table for row operators
-   * @param kind The kind of join to perform
-   * @param query_ref Reference to the hash table for querying
-   * @param stream CUDA stream on which to perform operations
-   * @param mr Memory resource for allocations
-   * @return Device vector of indices representing the join result
-   */
-  template <int32_t CGSize, typename Ref>
-  std::unique_ptr<rmm::device_uvector<cudf::size_type>> query_right_table(
+  // Queries the hash table for every left row and writes the matches to contains_map.
+  template <int32_t CGSize, typename Iterator, typename Ref>
+  void query_right_table(cudf::table_view const& left,
+                         Iterator left_iter,
+                         Ref query_ref,
+                         cudf::device_span<bool> contains_map,
+                         rmm::cuda_stream_view stream);
+
+  void query_right_table_primitive(
     cudf::table_view const& left,
-    std::shared_ptr<cudf::detail::row::equality::preprocessed_table> preprocessed_left,
-    join_kind kind,
-    Ref query_ref,
-    rmm::cuda_stream_view stream,
-    rmm::device_async_resource_ref mr);
+    std::shared_ptr<cudf::detail::row::equality::preprocessed_table> const& preprocessed_left,
+    cudf::device_span<bool> contains_map,
+    rmm::cuda_stream_view stream);
+
+  void query_right_table_flat(
+    cudf::table_view const& left,
+    std::shared_ptr<cudf::detail::row::equality::preprocessed_table> const& preprocessed_left,
+    cudf::device_span<bool> contains_map,
+    rmm::cuda_stream_view stream);
+
+  void query_right_table_nested(
+    cudf::table_view const& left,
+    std::shared_ptr<cudf::detail::row::equality::preprocessed_table> const& preprocessed_left,
+    cudf::device_span<bool> contains_map,
+    rmm::cuda_stream_view stream);
 
  public:
   /**

cpp/include/cudf/detail/join/filtered_join.cuh

@@ -4,25 +4,31 @@
  */
 #pragma once
 
-#include <cudf/detail/cuco_helpers.hpp>
-#include <cudf/detail/join/join.hpp>
-#include <cudf/detail/row_operator/equality.cuh>
-#include <cudf/detail/row_operator/hashing.cuh>
-#include <cudf/detail/row_operator/primitive_row_operators.cuh>
+#include <cudf/detail/row_operator/common_utils.cuh>
+#include <cudf/hashing.hpp>
 #include <cudf/join/join.hpp>
 #include <cudf/table/table_view.hpp>
-#include <cudf/utilities/default_stream.hpp>
-#include <cudf/utilities/memory_resource.hpp>
+#include <cudf/types.hpp>
 
 #include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_uvector.hpp>
 #include <rmm/mr/polymorphic_allocator.hpp>
 #include <rmm/resource_ref.hpp>
 
 #include <cuco/bucket_storage.cuh>
 #include <cuco/extent.cuh>
-#include <cuco/static_set_ref.cuh>
+#include <cuco/pair.cuh>
+#include <cuco/probing_scheme.cuh>
 #include <cuco/types.cuh>
-#include <cuda/std/type_traits>
+
+#include <cstddef>
+#include <cstdint>
+#include <limits>
+#include <memory>
+
+namespace cudf::detail::row::equality {
+struct preprocessed_table;
+}
 
 namespace cudf {
 namespace detail {
@@ -38,13 +44,6 @@ using cudf::detail::row::rhs_index_type;
  */
 class filtered_join {
  public:
-  /**
-   * @brief Properties of the right table used in the join operation
-   */
-  struct right_properties {
-    bool has_nested_columns;  ///< True if the right table contains nested columns
-  };
-
   /**
    * @brief Adapter for insertion operations in the hash table
    *
@@ -160,6 +159,8 @@ class filtered_join {
   virtual ~filtered_join() = default;
 
  protected:
+  enum class row_operator_mode : uint8_t { PRIMITIVE, FLAT, NESTED };
+
   // Key type used in the hash table
   using key = cuco::pair<hash_value_type, lhs_index_type>;
 
@@ -170,48 +171,30 @@ class filtered_join {
                          cuco::extent<std::size_t>,
                          rmm::mr::polymorphic_allocator<char>>;
 
-  // Hasher for primitive row types
-  using primitive_row_hasher =
-    cudf::detail::row::primitive::row_hasher<cudf::hashing::detail::default_hash>;
-  // Linear probing scheme with bucket size 1 for primitive types
-  using primitive_probing_scheme = cuco::linear_probing<1, hash_extract_fn>;
-  // Equality comparator for primitive rows
-  using primitive_row_comparator = cudf::detail::row::primitive::row_equality_comparator;
-
-  // Hasher for complex row types with compile-time null handling
-  using row_hasher =
-    cudf::detail::row::hash::device_row_hasher<cudf::hashing::detail::default_hash, nullate::YES>;
-  // Linear probing scheme with bucket size 4 for nested data structures
+  using single_probing_scheme = cuco::linear_probing<1, hash_extract_fn>;
+  // Nested rows use four cooperating threads per probe.
   using nested_probing_scheme = cuco::linear_probing<4, hash_extract_fn>;
-  // Linear probing scheme with bucket size 1 for simple data
-  using simple_probing_scheme = cuco::linear_probing<1, hash_extract_fn>;
-  // Equality comparator for complex rows with null handling and NaN comparison
-  using row_comparator = cudf::detail::row::equality::device_row_comparator<
-    true,
-    cudf::nullate::YES,
-    cudf::detail::row::equality::nan_equal_physical_equality_comparator>;
 
+  row_operator_mode const _right_mode;
   storage_type _bucket_storage;  ///< Storage for hash table buckets
 
   // Empty sentinel key used to mark empty slots in the hash table
   static constexpr auto empty_sentinel_key = cuco::empty_key{
     cuco::pair{std::numeric_limits<hash_value_type>::max(), lhs_index_type{cudf::JoinNoMatch}}};
-  right_properties _right_props;           ///< Properties of the right table
   cudf::table_view _right;                 ///< input table used to build the hash map
   cudf::null_equality const _nulls_equal;  ///< whether to consider nulls as equal
   std::shared_ptr<cudf::detail::row::equality::preprocessed_table>
     _preprocessed_right;  ///< input table preprocessed for row operators
 
-  /**
-   * @brief Populates the hash table with the right table
-   *
-   * @tparam CGSize CUDA cooperative group size
-   * @tparam Ref Reference type for the hash table
-   * @param insert_ref Reference to the hash table for insertion
-   * @param stream CUDA stream on which to perform operations
-   */
-  template <int32_t CGSize, typename Ref>
-  void insert_right_table(Ref const& insert_ref, rmm::cuda_stream_view stream);
+  // Build and probe row operators must use matching nullate modes. Since probe nullability is
+  // unknown at build time, primitive paths use DYNAMIC true and other paths use YES.
+  void insert_right_table_primitive(rmm::cuda_stream_view stream);
+  // Populates the hash table from the right-row iterator.
+  template <int32_t CGSize, typename Iterator, typename Ref>
+  void insert_right_table(Iterator right_iter, Ref const& insert_ref, rmm::cuda_stream_view stream);
+
+  void insert_right_table_flat(rmm::cuda_stream_view stream);
+  void insert_right_table_nested(rmm::cuda_stream_view stream);
 
  private:
   /**
@@ -220,11 +203,16 @@ class filtered_join {
    * Computes the appropriate size for the bucket storage based on the input
    * table size and desired load factor.
    *
-   * @param tbl Table for which to calculate storage
+   * @param num_rows Number of rows to store
    * @param load_factor Target load factor for the hash table
+   * @param mode Row-operator mode used by the hash table
    * @return Calculated bucket storage size
    */
-  auto compute_bucket_storage_size(cudf::table_view tbl, double load_factor);
+  static std::size_t compute_bucket_storage_size(cudf::size_type num_rows,
+                                                 double load_factor,
+                                                 row_operator_mode mode);
+
+  static row_operator_mode select_row_operator_mode(cudf::table_view const& table);
 };
 
 }  // namespace detail

cpp/src/hash/murmurhash3_x86_32.cu

@@ -2,6 +2,8 @@
  * SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
+#include "murmurhash3_x86_32.cuh"
+
 #include <cudf/column/column_factories.hpp>
 #include <cudf/detail/nvtx/ranges.hpp>
 #include <cudf/detail/row_operator/hashing.cuh>
@@ -17,35 +19,60 @@ namespace cudf {
 namespace hashing {
 namespace detail {
 
-std::unique_ptr<column> murmurhash3_x86_32(table_view const& input,
-                                           uint32_t seed,
-                                           rmm::cuda_stream_view stream,
-                                           rmm::device_async_resource_ref mr)
+namespace {
+
+template <typename Nullate>
+std::unique_ptr<column> murmurhash3_x86_32_impl(
+  std::shared_ptr<cudf::detail::row::equality::preprocessed_table> const& input,
+  size_type num_rows,
+  uint32_t seed,
+  Nullate nulls,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr)
 {
-  auto output = make_numeric_column(data_type(type_to_id<hash_value_type>()),
-                                    input.num_rows(),
-                                    mask_state::UNALLOCATED,
-                                    stream,
-                                    mr);
+  auto output = make_numeric_column(
+    data_type(type_to_id<hash_value_type>()), num_rows, mask_state::UNALLOCATED, stream, mr);
 
-  if (input.num_rows() == 0) { return output; }
+  if (num_rows == 0) { return output; }
 
-  bool const nullable   = has_nulls(input);
-  auto const row_hasher = cudf::detail::row::hash::row_hasher(input, stream);
+  auto const row_hasher = cudf::detail::row::hash::row_hasher(input);
   auto output_view      = output->mutable_view();
 
   // Compute the hash value for each row
   auto const output_begin = output_view.begin<hash_value_type>();
-  auto const hasher       = row_hasher.device_hasher<MurmurHash3_x86_32>(nullable, seed);
+  auto const hasher       = row_hasher.device_hasher<MurmurHash3_x86_32>(nulls, seed);
   // thrust::tabulate is slow here, see NVIDIA/cccl#9070
   CUDF_CUDA_TRY(cub::DeviceFor::Bulk(
-    input.num_rows(),
+    num_rows,
     [output_begin, hasher] __device__(size_type i) mutable { output_begin[i] = hasher(i); },
     stream.value()));
 
   return output;
 }
 
+}  // namespace
+
+std::unique_ptr<column> murmurhash3_x86_32(table_view const& input,
+                                           uint32_t seed,
+                                           rmm::cuda_stream_view stream,
+                                           rmm::device_async_resource_ref mr)
+{
+  auto const preprocessed_input =
+    cudf::detail::row::hash::preprocessed_table::create(input, stream);
+  return murmurhash3_x86_32_impl(
+    preprocessed_input, input.num_rows(), seed, nullate::DYNAMIC{has_nulls(input)}, stream, mr);
+}
+
+std::unique_ptr<column> murmurhash3_x86_32(
+  std::shared_ptr<cudf::detail::row::equality::preprocessed_table> const& input,
+  size_type num_rows,
+  uint32_t seed,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr)
+{
+  return murmurhash3_x86_32_impl(input, num_rows, seed, nullate::YES{}, stream, mr);
+}
+
 }  // namespace detail
 
 std::unique_ptr<column> murmurhash3_x86_32(table_view const& input,

cpp/src/hash/murmurhash3_x86_32.cuh

@@ -0,0 +1,33 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <cudf/types.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/resource_ref.hpp>
+
+#include <cstdint>
+#include <memory>
+
+namespace cudf {
+class column;
+
+namespace detail::row::equality {
+struct preprocessed_table;
+}
+
+namespace hashing::detail {
+
+std::unique_ptr<column> murmurhash3_x86_32(
+  std::shared_ptr<cudf::detail::row::equality::preprocessed_table> const& input,
+  size_type num_rows,
+  uint32_t seed,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr);
+
+}  // namespace hashing::detail
+}  // namespace cudf