Reduce distinct_helpers build time

cpp/CMakeLists.txt

@@ -1004,6 +1004,14 @@ add_library(
   src/stream_compaction/apply_boolean_mask.cu
   src/stream_compaction/distinct.cu
   src/stream_compaction/distinct_helpers.cu
+  src/stream_compaction/distinct_helpers_flat_nan_equal_any.cu
+  src/stream_compaction/distinct_helpers_flat_nan_equal_ordered.cu
+  src/stream_compaction/distinct_helpers_flat_nan_unequal_any.cu
+  src/stream_compaction/distinct_helpers_flat_nan_unequal_ordered.cu
+  src/stream_compaction/distinct_helpers_nested_nan_equal_any.cu
+  src/stream_compaction/distinct_helpers_nested_nan_equal_ordered.cu
+  src/stream_compaction/distinct_helpers_nested_nan_unequal_any.cu
+  src/stream_compaction/distinct_helpers_nested_nan_unequal_ordered.cu
   src/stream_compaction/drop_nans.cu
   src/stream_compaction/drop_nulls.cu
   src/stream_compaction/filter/filter.cu

cpp/src/hash/murmurhash3_x86_32.cu

@@ -2,50 +2,86 @@
  * 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.hpp>
 #include <cudf/column/column_factories.hpp>
 #include <cudf/detail/nvtx/ranges.hpp>
 #include <cudf/detail/row_operator/hashing.cuh>
+#include <cudf/detail/row_operator/preprocessed_table.cuh>
+#include <cudf/hashing.hpp>
 #include <cudf/hashing/detail/hashing.hpp>
 #include <cudf/hashing/detail/murmurhash3_x86_32.cuh>
+#include <cudf/table/table_view.hpp>
+#include <cudf/types.hpp>
 #include <cudf/utilities/error.hpp>
 
 #include <rmm/cuda_stream_view.hpp>
+#include <rmm/resource_ref.hpp>
 
 #include <cub/device/device_for.cuh>
 
+#include <cstdint>
+#include <memory>
+
 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

cpp/src/stream_compaction/distinct.cu

@@ -1,27 +1,36 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2019-2025, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
 #include "distinct_helpers.hpp"
+#include "hash/murmurhash3_x86_32.cuh"
 
-#include <cudf/column/column_view.hpp>
+#include <cudf/column/column.hpp>
+#include <cudf/copying.hpp>
 #include <cudf/detail/cuco_helpers.hpp>
 #include <cudf/detail/gather.hpp>
-#include <cudf/detail/iterator.cuh>
 #include <cudf/detail/nvtx/ranges.hpp>
 #include <cudf/detail/row_operator/equality.cuh>
 #include <cudf/detail/row_operator/hashing.cuh>
 #include <cudf/detail/stream_compaction.hpp>
+#include <cudf/hashing.hpp>
+#include <cudf/stream_compaction.hpp>
 #include <cudf/table/table.hpp>
 #include <cudf/table/table_view.hpp>
 #include <cudf/types.hpp>
 #include <cudf/utilities/memory_resource.hpp>
 
 #include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_buffer.hpp>
 #include <rmm/device_uvector.hpp>
 #include <rmm/mr/polymorphic_allocator.hpp>
+#include <rmm/resource_ref.hpp>
 
+#include <cuco/types.cuh>
+
+#include <memory>
+#include <type_traits>
 #include <utility>
 #include <vector>
 
@@ -85,24 +94,54 @@ rmm::device_uvector<size_type> distinct_indices(table_view const& input,
   auto const row_hash  = cudf::detail::row::hash::row_hasher(preprocessed_input);
   auto const row_equal = cudf::detail::row::equality::self_comparator(preprocessed_input);
 
-  auto const helper_func = [&](auto const& d_equal) {
+  auto const helper_func = [&](auto const& d_equal, auto const& d_hash, auto const& reduce_func) {
     using RowEqual = std::decay_t<decltype(d_equal)>;
-    auto set       = distinct_set_t<RowEqual>{num_rows,
-                                              0.5,  // desired load factor
-                                              cuco::empty_key{cudf::detail::CUDF_SIZE_TYPE_SENTINEL},
-                                              d_equal,
-                                              {row_hash.device_hasher(has_nulls)},
-                                              {},
-                                              {},
-                                              rmm::mr::polymorphic_allocator<char>{},
-                                              stream.value()};
-    return detail::reduce_by_row(set, num_rows, keep, stream, mr);
+    using RowHash  = std::decay_t<decltype(d_hash)>;
+    auto set =
+      distinct_set_t<RowEqual, RowHash>{num_rows,
+                                        0.5,  // desired load factor
+                                        cuco::empty_key{cudf::detail::CUDF_SIZE_TYPE_SENTINEL},
+                                        d_equal,
+                                        d_hash,
+                                        {},
+                                        {},
+                                        rmm::mr::polymorphic_allocator<char>{},
+                                        stream.value()};
+    return reduce_func(set);
   };
 
-  if (cudf::detail::has_nested_columns(input)) {
-    return dispatch_row_equal<true>(nulls_equal, nans_equal, has_nulls, row_equal, helper_func);
+  if (has_nested_columns) {
+    if (keep == duplicate_keep_option::KEEP_ANY) {
+      auto const hashes =
+        cudf::hashing::detail::murmurhash3_x86_32(preprocessed_input,
+                                                  num_rows,
+                                                  cudf::DEFAULT_HASH_SEED,
+                                                  stream,
+                                                  cudf::get_current_device_resource_ref());
+      auto const d_hash = distinct_precomputed_hash{hashes->view().data<hash_value_type>()};
+      return dispatch_row_equal<true>(
+        nulls_equal, nans_equal, has_nulls, row_equal, [&](auto const& d_equal) {
+          return helper_func(d_equal, d_hash, [&](auto& set) {
+            return detail::reduce_by_row_keep_any(set, num_rows, stream, mr);
+          });
+        });
+    }
+
+    auto const d_hash = row_hash.device_hasher(has_nulls);
+    return dispatch_row_equal<true>(
+      nulls_equal, nans_equal, has_nulls, row_equal, [&](auto const& d_equal) {
+        return helper_func(d_equal, d_hash, [&](auto& set) {
+          return detail::reduce_by_row_keep_first_last_none(set, num_rows, keep, stream, mr);
+        });
+      });
   } else {
-    return dispatch_row_equal<false>(nulls_equal, nans_equal, has_nulls, row_equal, helper_func);
+    auto const d_hash = row_hash.device_hasher(has_nulls);
+    return dispatch_row_equal<false>(
+      nulls_equal, nans_equal, has_nulls, row_equal, [&](auto const& d_equal) {
+        return helper_func(d_equal, d_hash, [&](auto& set) {
+          return detail::reduce_by_row(set, num_rows, keep, stream, mr);
+        });
+      });
   }
 }