diff --git a/cpp/include/cudf/detail/join/distinct_hash_join.cuh b/cpp/include/cudf/detail/join/distinct_hash_join.cuh
index b6b6ea889ae0..3da903dc4158 100644
--- a/cpp/include/cudf/detail/join/distinct_hash_join.cuh
+++ b/cpp/include/cudf/detail/join/distinct_hash_join.cuh
@@ -49,7 +49,7 @@ struct always_not_equal {
 };
 
 /**
- * @brief An comparator adapter wrapping the two table comparator
+ * @brief A comparator adapter wrapping the two table comparator
  */
 template <typename Equal>
 struct comparator_adapter {
@@ -67,6 +67,25 @@ struct comparator_adapter {
   Equal _d_equal;
 };
 
+/**
+ * @brief A comparator adapter wrapping the two table comparator
+ */
+template <typename Equal>
+struct primitive_comparator_adapter {
+  primitive_comparator_adapter(Equal const& d_equal) : _d_equal{d_equal} {}
+
+  __device__ constexpr auto operator()(
+    cuco::pair<hash_value_type, lhs_index_type> const& lhs,
+    cuco::pair<hash_value_type, rhs_index_type> const& rhs) const noexcept
+  {
+    if (lhs.first != rhs.first) { return false; }
+    return _d_equal(static_cast<size_type>(lhs.second), static_cast<size_type>(rhs.second));
+  }
+
+ private:
+  Equal _d_equal;
+};
+
 /**
  * @brief Distinct hash join that builds hash table in creation and probes results in subsequent
  * `*_join` member functions.
diff --git a/cpp/include/cudf/table/experimental/row_operators.cuh b/cpp/include/cudf/table/experimental/row_operators.cuh
index 7c200aa9859f..33ea7fffbe55 100644
--- a/cpp/include/cudf/table/experimental/row_operators.cuh
+++ b/cpp/include/cudf/table/experimental/row_operators.cuh
@@ -56,6 +56,13 @@
 
 namespace CUDF_EXPORT cudf {
 
+namespace row::primitive {
+class row_equality_comparator;  // Forward declaration
+
+template <template <typename> class Hash>
+class row_hasher;  // Forward declaration
+}  // namespace row::primitive
+
 namespace experimental {
 
 /**
@@ -1577,6 +1584,11 @@ struct preprocessed_table {
   friend class self_comparator;       ///< Allow self_comparator to access private members
   friend class two_table_comparator;  ///< Allow two_table_comparator to access private members
   friend class hash::row_hasher;      ///< Allow row_hasher to access private members
+  /// Allow primitive equality comparator to access private members
+  friend class ::cudf::row::primitive::row_equality_comparator;
+
+  template <template <typename> class Hash>
+  friend class ::cudf::row::primitive::row_hasher;
 
   using table_device_view_owner =
     std::invoke_result_t<decltype(table_device_view::create), table_view, rmm::cuda_stream_view>;
diff --git a/cpp/include/cudf/table/primitive_row_operators.cuh b/cpp/include/cudf/table/primitive_row_operators.cuh
new file mode 100644
index 000000000000..cea12d7498c3
--- /dev/null
+++ b/cpp/include/cudf/table/primitive_row_operators.cuh
@@ -0,0 +1,233 @@
+/*
+ * Copyright (c) 2025, NVIDIA CORPORATION.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <cudf/column/column_device_view.cuh>
+#include <cudf/detail/utilities/assert.cuh>
+#include <cudf/hashing/detail/hash_functions.cuh>
+#include <cudf/hashing/detail/hashing.hpp>
+#include <cudf/table/experimental/row_operators.cuh>
+#include <cudf/table/row_operators.cuh>
+#include <cudf/table/table_device_view.cuh>
+#include <cudf/utilities/traits.hpp>
+#include <cudf/utilities/type_dispatcher.hpp>
+
+#include <cuda/std/limits>
+#include <cuda/std/type_traits>
+
+namespace CUDF_EXPORT cudf {
+namespace row::primitive {
+
+/**
+ * @brief Returns `void` if it's not a primitive type
+ */
+template <typename T>
+using primitive_type_t = cuda::std::conditional_t<cudf::is_numeric<T>(), T, void>;
+
+/**
+ * @brief Custom dispatcher for primitive types
+ */
+template <cudf::type_id Id>
+struct dispatch_primitive_type {
+  using type = primitive_type_t<id_to_type<Id>>;  ///< The underlying type
+};
+
+/**
+ * @brief Performs an equality comparison between two elements in two columns.
+ */
+class element_equality_comparator {
+ public:
+  /**
+   * @brief Compares the specified elements for equality.
+   *
+   * @param lhs The first column
+   * @param rhs The second column
+   * @param lhs_element_index The index of the first element
+   * @param rhs_element_index The index of the second element
+   * @return True if lhs and rhs element are equal
+   */
+  template <typename Element, CUDF_ENABLE_IF(cudf::is_equality_comparable<Element, Element>())>
+  __device__ bool operator()(column_device_view const& lhs,
+                             column_device_view const& rhs,
+                             size_type lhs_element_index,
+                             size_type rhs_element_index) const
+  {
+    return cudf::equality_compare(lhs.element<Element>(lhs_element_index),
+                                  rhs.element<Element>(rhs_element_index));
+  }
+
+  // @cond
+  template <typename Element, CUDF_ENABLE_IF(not cudf::is_equality_comparable<Element, Element>())>
+  __device__ bool operator()(column_device_view const&,
+                             column_device_view const&,
+                             size_type,
+                             size_type) const
+  {
+    CUDF_UNREACHABLE("Attempted to compare elements of uncomparable types.");
+  }
+  // @endcond
+};
+
+/**
+ * @brief Performs a relational comparison between two elements in two tables.
+ */
+class row_equality_comparator {
+ public:
+  /**
+   * @brief Construct a new row equality comparator object
+   *
+   * @param has_nulls Indicates if either input column contains nulls
+   * @param lhs Preprocessed table containing the first element
+   * @param rhs Preprocessed table containing the second element (may be the same as lhs)
+   * @param nulls_are_equal Indicates if two null elements are treated as equivalent
+   */
+  row_equality_comparator(
+    cudf::nullate::DYNAMIC const& has_nulls,
+    std::shared_ptr<cudf::experimental::row::equality::preprocessed_table> lhs,
+    std::shared_ptr<cudf::experimental::row::equality::preprocessed_table> rhs,
+    null_equality nulls_are_equal)
+    : _has_nulls{has_nulls}, _lhs{*lhs}, _rhs{*rhs}, _nulls_are_equal{nulls_are_equal}
+  {
+    CUDF_EXPECTS(_lhs.num_columns() == _rhs.num_columns(), "Mismatched number of columns.");
+  }
+
+  /**
+   * @brief Compares the specified rows for equality.
+   *
+   * @param lhs_row_index The index of the first row to compare (in the lhs table)
+   * @param rhs_row_index The index of the second row to compare (in the rhs table)
+   * @return true if both rows are equal, otherwise false
+   */
+  __device__ bool operator()(size_type lhs_row_index, size_type rhs_row_index) const
+  {
+    if (_has_nulls) {
+      bool const lhs_is_null{_lhs.column(0).is_null(lhs_row_index)};
+      bool const rhs_is_null{_rhs.column(0).is_null(rhs_row_index)};
+      if (lhs_is_null and rhs_is_null) {
+        return _nulls_are_equal == null_equality::EQUAL;
+      } else if (lhs_is_null != rhs_is_null) {
+        return false;
+      }
+    }
+    return cudf::type_dispatcher<dispatch_primitive_type>(_lhs.begin()->type(),
+                                                          element_equality_comparator{},
+                                                          _lhs.column(0),
+                                                          _rhs.column(0),
+                                                          lhs_row_index,
+                                                          rhs_row_index);
+  }
+
+ private:
+  cudf::nullate::DYNAMIC _has_nulls;
+  table_device_view _lhs;
+  table_device_view _rhs;
+  null_equality _nulls_are_equal;
+};
+
+/**
+ * @brief Function object for computing the hash value of a row in a column.
+ *
+ * @tparam Hash Hash functor to use for hashing elements
+ */
+template <template <typename> class Hash>
+class element_hasher {
+ public:
+  /**
+   * @brief Returns the hash value of the given element in the given column.
+   *
+   * @tparam T The type of the element to hash
+   * @param seed The seed value to use for hashing
+   * @param col The column to hash
+   * @param row_index The index of the row to hash
+   * @return The hash value of the given element
+   */
+  template <typename T, CUDF_ENABLE_IF(column_device_view::has_element_accessor<T>())>
+  __device__ hash_value_type operator()(hash_value_type seed,
+                                        column_device_view const& col,
+                                        size_type row_index) const
+  {
+    return Hash<T>{seed}(col.element<T>(row_index));
+  }
+
+  // @cond
+  template <typename T, CUDF_ENABLE_IF(not column_device_view::has_element_accessor<T>())>
+  __device__ hash_value_type operator()(hash_value_type, column_device_view const&, size_type) const
+  {
+    CUDF_UNREACHABLE("Unsupported type in hash.");
+  }
+  // @endcond
+};
+
+/**
+ * @brief Computes the hash value of a row in the given table.
+ *
+ * @tparam Hash Hash functor to use for hashing elements.
+ */
+template <template <typename> class Hash>
+class row_hasher {
+ public:
+  row_hasher() = delete;
+
+  /**
+   * @brief Constructs a row_hasher object with a seed value.
+   *
+   * @param has_nulls Indicates if the input column contains nulls
+   * @param t A table_device_view to hash
+   * @param seed A seed value to use for hashing
+   */
+  row_hasher(cudf::nullate::DYNAMIC const& has_nulls, table_device_view t, hash_value_type seed)
+    : _has_nulls{has_nulls}, _table{t}, _seed{seed}
+  {
+  }
+
+  /**
+   * @brief Constructs a row_hasher object with a seed value.
+   *
+   * @param has_nulls Indicates if the input column contains nulls
+   * @param t Preprocessed table to hash
+   * @param seed A seed value to use for hashing
+   */
+  row_hasher(cudf::nullate::DYNAMIC const& has_nulls,
+             std::shared_ptr<cudf::experimental::row::equality::preprocessed_table> t,
+             hash_value_type seed)
+    : _has_nulls{has_nulls}, _table{*t}, _seed{seed}
+  {
+  }
+
+  /**
+   * @brief Computes the hash value of the row at `row_index` in the `table`
+   *
+   * @param row_index The index of the row in the `table` to hash
+   * @return The hash value of the row at `row_index` in the `table`
+   */
+  __device__ auto operator()(size_type row_index) const
+  {
+    if (_has_nulls && _table.column(0).is_null(row_index)) {
+      return cuda::std::numeric_limits<hash_value_type>::max();
+    }
+    return cudf::type_dispatcher<dispatch_primitive_type>(
+      _table.column(0).type(), element_hasher<Hash>{}, _seed, _table.column(0), row_index);
+  }
+
+ private:
+  cudf::nullate::DYNAMIC _has_nulls;
+  table_device_view _table;
+  hash_value_type _seed;
+};
+
+}  // namespace row::primitive
+}  // namespace CUDF_EXPORT cudf
diff --git a/cpp/src/join/distinct_hash_join.cu b/cpp/src/join/distinct_hash_join.cu
index 73134bbf90de..2780ef049859 100644
--- a/cpp/src/join/distinct_hash_join.cu
+++ b/cpp/src/join/distinct_hash_join.cu
@@ -21,6 +21,7 @@
 #include <cudf/detail/utilities/cuda.cuh>
 #include <cudf/join/distinct_hash_join.hpp>
 #include <cudf/table/experimental/row_operators.cuh>
+#include <cudf/table/primitive_row_operators.cuh>
 #include <cudf/table/table_view.hpp>
 #include <cudf/types.hpp>
 #include <cudf/utilities/memory_resource.hpp>
@@ -47,6 +48,25 @@ namespace {
 
 bool constexpr has_nulls = true;  ///< Always has nulls
 
+/**
+ * @brief Device functor to create a pair of {hash_value, row_index} for a given row.
+ */
+template <typename T>
+class primitive_keys_fn {
+  using hasher = cudf::row::primitive::row_hasher<cudf::hashing::detail::default_hash>;
+
+ public:
+  CUDF_HOST_DEVICE constexpr primitive_keys_fn(hasher const& hash) : _hash{hash} {}
+
+  __device__ __forceinline__ auto operator()(size_type i) const noexcept
+  {
+    return cuco::pair{_hash(i), T{i}};
+  }
+
+ private:
+  hasher _hash;
+};
+
 /**
  * @brief Device functor to create a pair of {hash_value, row_index} for a given row.
  */
@@ -79,6 +99,59 @@ struct output_fn {
     return static_cast<cudf::size_type>(x.second);
   }
 };
+
+/**
+ * @brief Find matching rows in the hash table
+ *
+ * @tparam IterType Type of the iterator over hash values
+ * @tparam EqualType Type of the equality comparator
+ * @param hash_table The hash table to search in
+ * @param iter Iterator over hash values
+ * @param d_equal Equality comparator
+ * @param probe The probe table
+ * @param hasher Hash function
+ * @param nulls_equal Null equality setting
+ * @param found_begin Output iterator for found indices
+ * @param stream CUDA stream
+ */
+template <typename HashTableType,
+          typename IterType,
+          typename EqualType,
+          typename Hasher,
+          typename FoundIterator>
+void find_matches_in_hash_table(HashTableType const& hash_table,
+                                IterType iter,
+                                EqualType const& d_equal,
+                                cudf::table_view const& probe,
+                                Hasher hasher,
+                                cudf::null_equality nulls_equal,
+                                FoundIterator found_begin,
+                                rmm::cuda_stream_view stream)
+{
+  auto const probe_table_num_rows = probe.num_rows();
+  // If `idx` is within the range `[0, probe_table_num_rows)` and `found_indices[idx]` is not
+  // equal to `JoinNoneValue`, then `idx` has a match in the hash set.
+  if (nulls_equal == cudf::null_equality::EQUAL or (not cudf::nullable(probe))) {
+    hash_table.find_async(
+      iter, iter + probe_table_num_rows, d_equal, hasher, found_begin, stream.value());
+  } else {
+    auto stencil = thrust::counting_iterator<size_type>{0};
+    auto const row_bitmask =
+      cudf::detail::bitmask_and(probe, stream, cudf::get_current_device_resource_ref()).first;
+    auto const pred =
+      cudf::detail::row_is_valid{reinterpret_cast<bitmask_type const*>(row_bitmask.data())};
+
+    hash_table.find_if_async(iter,
+                             iter + probe_table_num_rows,
+                             stencil,
+                             pred,
+                             d_equal,
+                             hasher,
+                             found_begin,
+                             stream.value());
+  }
+}
+
 }  // namespace
 
 distinct_hash_join::distinct_hash_join(cudf::table_view const& build,
@@ -114,28 +187,42 @@ distinct_hash_join::distinct_hash_join(cudf::table_view const& build,
                "Invalid load factor: must be greater than 0 and less than or equal to 1.",
                std::invalid_argument);
 
-  if (this->_build.num_rows() == 0) { return; }
+  size_type const build_table_num_rows{_build.num_rows()};
 
-  auto const row_hasher = experimental::row::hash::row_hasher{this->_preprocessed_build};
-  auto const d_hasher   = row_hasher.device_hasher(nullate::DYNAMIC{has_nulls});
+  if (build_table_num_rows == 0) { return; }
 
-  auto const iter =
-    cudf::detail::make_counting_transform_iterator(0, build_keys_fn<rhs_index_type>{d_hasher});
+  auto const build_hash_table = [&](auto iter) {
+    if (this->_nulls_equal == cudf::null_equality::EQUAL or (not cudf::nullable(build))) {
+      this->_hash_table.insert_async(iter, iter + build_table_num_rows, stream.value());
+    } else {
+      auto stencil = thrust::counting_iterator<size_type>{0};
+      auto const row_bitmask =
+        cudf::detail::bitmask_and(_build, stream, cudf::get_current_device_resource_ref()).first;
+      auto const pred =
+        cudf::detail::row_is_valid{reinterpret_cast<bitmask_type const*>(row_bitmask.data())};
+
+      // insert valid rows
+      this->_hash_table.insert_if_async(
+        iter, iter + build_table_num_rows, stencil, pred, stream.value());
+    }
+  };
 
-  size_type const build_table_num_rows{build.num_rows()};
-  if (this->_nulls_equal == cudf::null_equality::EQUAL or (not cudf::nullable(this->_build))) {
-    this->_hash_table.insert_async(iter, iter + build_table_num_rows, stream.value());
+  if (_build.num_columns() == 1 and cudf::is_numeric(_build.column(0).type())) {
+    auto const d_hasher = cudf::row::primitive::row_hasher<cudf::hashing::detail::default_hash>{
+      nullate::DYNAMIC{has_nulls}, this->_preprocessed_build, DEFAULT_HASH_SEED};
+
+    auto const iter = cudf::detail::make_counting_transform_iterator(
+      0, primitive_keys_fn<rhs_index_type>{d_hasher});
+
+    build_hash_table(iter);
   } else {
-    auto stencil = thrust::counting_iterator<size_type>{0};
-    auto const row_bitmask =
-      cudf::detail::bitmask_and(this->_build, stream, cudf::get_current_device_resource_ref())
-        .first;
-    auto const pred =
-      cudf::detail::row_is_valid{reinterpret_cast<bitmask_type const*>(row_bitmask.data())};
+    auto const row_hasher = experimental::row::hash::row_hasher{this->_preprocessed_build};
+    auto const d_hasher   = row_hasher.device_hasher(nullate::DYNAMIC{has_nulls});
 
-    // insert valid rows
-    this->_hash_table.insert_if_async(
-      iter, iter + build_table_num_rows, stencil, pred, stream.value());
+    auto const iter =
+      cudf::detail::make_counting_transform_iterator(0, build_keys_fn<rhs_index_type>{d_hasher});
+
+    build_hash_table(iter);
   }
 }
 
@@ -155,61 +242,65 @@ distinct_hash_join::inner_join(cudf::table_view const& probe,
                      std::make_unique<rmm::device_uvector<size_type>>(0, stream, mr));
   }
 
-  auto preprocessed_probe =
-    cudf::experimental::row::equality::preprocessed_table::create(probe, stream);
-  auto const two_table_equal = cudf::experimental::row::equality::two_table_comparator(
-    preprocessed_probe, _preprocessed_build);
-
   auto build_indices =
     std::make_unique<rmm::device_uvector<size_type>>(probe_table_num_rows, stream, mr);
   auto probe_indices =
     std::make_unique<rmm::device_uvector<size_type>>(probe_table_num_rows, stream, mr);
 
-  auto const probe_row_hasher = cudf::experimental::row::hash::row_hasher{preprocessed_probe};
-  auto const d_probe_hasher   = probe_row_hasher.device_hasher(nullate::DYNAMIC{has_nulls});
-  auto const iter             = cudf::detail::make_counting_transform_iterator(
-    0, build_keys_fn<lhs_index_type>{d_probe_hasher});
-
   auto found_indices = rmm::device_uvector<size_type>(probe_table_num_rows, stream);
   auto const found_begin =
     thrust::make_transform_output_iterator(found_indices.begin(), output_fn{});
 
-  auto const comparator_helper = [&](auto device_comparator) {
-    // If `idx` is within the range `[0, probe_table_num_rows)` and `found_indices[idx]` is not
-    // equal to `JoinNoneValue`, then `idx` has a match in the hash set.
-    if (this->_nulls_equal == cudf::null_equality::EQUAL or (not cudf::nullable(probe))) {
-      this->_hash_table.find_async(iter,
-                                   iter + probe_table_num_rows,
-                                   comparator_adapter{device_comparator},
-                                   hasher{},
-                                   found_begin,
-                                   stream.value());
-    } else {
-      auto stencil = thrust::counting_iterator<size_type>{0};
-      auto const row_bitmask =
-        cudf::detail::bitmask_and(probe, stream, cudf::get_current_device_resource_ref()).first;
-      auto const pred =
-        cudf::detail::row_is_valid{reinterpret_cast<bitmask_type const*>(row_bitmask.data())};
+  auto preprocessed_probe =
+    cudf::experimental::row::equality::preprocessed_table::create(probe, stream);
+  if (_build.num_columns() == 1 and cudf::is_numeric(_build.column(0).type())) {
+    auto const d_hasher = cudf::row::primitive::row_hasher<cudf::hashing::detail::default_hash>{
+      nullate::DYNAMIC{has_nulls}, preprocessed_probe, DEFAULT_HASH_SEED};
+    auto const d_equal = cudf::row::primitive::row_equality_comparator{
+      nullate::DYNAMIC{has_nulls}, preprocessed_probe, _preprocessed_build, _nulls_equal};
+    auto const iter = cudf::detail::make_counting_transform_iterator(
+      0, primitive_keys_fn<lhs_index_type>{d_hasher});
+
+    find_matches_in_hash_table(this->_hash_table,
+                               iter,
+                               primitive_comparator_adapter{d_equal},
+                               probe,
+                               hasher{},
+                               _nulls_equal,
+                               found_begin,
+                               stream);
+  } else {
+    auto const two_table_equal = cudf::experimental::row::equality::two_table_comparator(
+      preprocessed_probe, _preprocessed_build);
 
-      this->_hash_table.find_if_async(iter,
-                                      iter + probe_table_num_rows,
-                                      stencil,
-                                      pred,
-                                      comparator_adapter{device_comparator},
-                                      hasher{},
-                                      found_begin,
-                                      stream.value());
-    }
-  };
+    auto const probe_row_hasher = cudf::experimental::row::hash::row_hasher{preprocessed_probe};
+    auto const d_probe_hasher   = probe_row_hasher.device_hasher(nullate::DYNAMIC{has_nulls});
+    auto const iter             = cudf::detail::make_counting_transform_iterator(
+      0, build_keys_fn<lhs_index_type>{d_probe_hasher});
 
-  if (_has_nested_columns) {
-    auto const device_comparator =
-      two_table_equal.equal_to<true>(nullate::DYNAMIC{has_nulls}, _nulls_equal);
-    comparator_helper(device_comparator);
-  } else {
-    auto const device_comparator =
-      two_table_equal.equal_to<false>(nullate::DYNAMIC{has_nulls}, _nulls_equal);
-    comparator_helper(device_comparator);
+    if (_has_nested_columns) {
+      auto const device_comparator =
+        two_table_equal.equal_to<true>(nullate::DYNAMIC{has_nulls}, _nulls_equal);
+      find_matches_in_hash_table(this->_hash_table,
+                                 iter,
+                                 comparator_adapter{device_comparator},
+                                 probe,
+                                 hasher{},
+                                 _nulls_equal,
+                                 found_begin,
+                                 stream);
+    } else {
+      auto const device_comparator =
+        two_table_equal.equal_to<false>(nullate::DYNAMIC{has_nulls}, _nulls_equal);
+      find_matches_in_hash_table(this->_hash_table,
+                                 iter,
+                                 comparator_adapter{device_comparator},
+                                 probe,
+                                 hasher{},
+                                 _nulls_equal,
+                                 found_begin,
+                                 stream);
+    }
   }
 
   auto const tuple_iter = cudf::detail::make_counting_transform_iterator(
@@ -252,61 +343,70 @@ std::unique_ptr<rmm::device_uvector<size_type>> distinct_hash_join::left_join(
 
   auto build_indices =
     std::make_unique<rmm::device_uvector<size_type>>(probe_table_num_rows, stream, mr);
+  auto const output_begin =
+    thrust::make_transform_output_iterator(build_indices->begin(), output_fn{});
 
-  // If build table is empty, return probe table
-  if (this->_build.num_rows() == 0) {
-    thrust::fill(
-      rmm::exec_policy_nosync(stream), build_indices->begin(), build_indices->end(), JoinNoneValue);
-  } else {
-    auto preprocessed_probe =
-      cudf::experimental::row::equality::preprocessed_table::create(probe, stream);
-    auto const two_table_equal = cudf::experimental::row::equality::two_table_comparator(
-      preprocessed_probe, _preprocessed_build);
-
-    auto const probe_row_hasher = cudf::experimental::row::hash::row_hasher{preprocessed_probe};
-    auto const d_probe_hasher   = probe_row_hasher.device_hasher(nullate::DYNAMIC{has_nulls});
-    auto const iter             = cudf::detail::make_counting_transform_iterator(
-      0, build_keys_fn<lhs_index_type>{d_probe_hasher});
+  auto preprocessed_probe =
+    cudf::experimental::row::equality::preprocessed_table::create(probe, stream);
 
-    auto const output_begin =
-      thrust::make_transform_output_iterator(build_indices->begin(), output_fn{});
-    auto const comparator_helper = [&](auto device_comparator) {
-      if (this->_nulls_equal == cudf::null_equality::EQUAL or (not cudf::nullable(probe))) {
-        this->_hash_table.find_async(iter,
-                                     iter + probe_table_num_rows,
-                                     comparator_adapter{device_comparator},
-                                     hasher{},
-                                     output_begin,
-                                     stream.value());
+  if (_build.num_columns() == 1 and cudf::is_numeric(_build.column(0).type())) {
+    auto const d_hasher = cudf::row::primitive::row_hasher<cudf::hashing::detail::default_hash>{
+      nullate::DYNAMIC{has_nulls}, preprocessed_probe, DEFAULT_HASH_SEED};
+    auto const d_equal = cudf::row::primitive::row_equality_comparator{
+      nullate::DYNAMIC{has_nulls}, preprocessed_probe, _preprocessed_build, _nulls_equal};
+
+    auto const iter = cudf::detail::make_counting_transform_iterator(
+      0, primitive_keys_fn<lhs_index_type>{d_hasher});
+
+    find_matches_in_hash_table(this->_hash_table,
+                               iter,
+                               primitive_comparator_adapter{d_equal},
+                               probe,
+                               hasher{},
+                               _nulls_equal,
+                               output_begin,
+                               stream);
+  } else {
+    // If build table is empty, return probe table
+    if (this->_build.num_rows() == 0) {
+      thrust::fill(rmm::exec_policy_nosync(stream),
+                   build_indices->begin(),
+                   build_indices->end(),
+                   JoinNoneValue);
+    } else {
+      auto const two_table_equal = cudf::experimental::row::equality::two_table_comparator(
+        preprocessed_probe, _preprocessed_build);
+
+      auto const probe_row_hasher = cudf::experimental::row::hash::row_hasher{preprocessed_probe};
+      auto const d_probe_hasher   = probe_row_hasher.device_hasher(nullate::DYNAMIC{has_nulls});
+      auto const iter             = cudf::detail::make_counting_transform_iterator(
+        0, build_keys_fn<lhs_index_type>{d_probe_hasher});
+
+      if (_has_nested_columns) {
+        auto const device_comparator =
+          two_table_equal.equal_to<true>(nullate::DYNAMIC{has_nulls}, _nulls_equal);
+        find_matches_in_hash_table(this->_hash_table,
+                                   iter,
+                                   comparator_adapter{device_comparator},
+                                   probe,
+                                   hasher{},
+                                   _nulls_equal,
+                                   output_begin,
+                                   stream);
       } else {
-        auto stencil = thrust::counting_iterator<size_type>{0};
-        auto const row_bitmask =
-          cudf::detail::bitmask_and(probe, stream, cudf::get_current_device_resource_ref()).first;
-        auto const pred =
-          cudf::detail::row_is_valid{reinterpret_cast<bitmask_type const*>(row_bitmask.data())};
-
-        this->_hash_table.find_if_async(iter,
-                                        iter + probe_table_num_rows,
-                                        stencil,
-                                        pred,
-                                        comparator_adapter{device_comparator},
-                                        hasher{},
-                                        output_begin,
-                                        stream.value());
+        auto const device_comparator =
+          two_table_equal.equal_to<false>(nullate::DYNAMIC{has_nulls}, _nulls_equal);
+        find_matches_in_hash_table(this->_hash_table,
+                                   iter,
+                                   comparator_adapter{device_comparator},
+                                   probe,
+                                   hasher{},
+                                   _nulls_equal,
+                                   output_begin,
+                                   stream);
       }
-    };
-
-    if (_has_nested_columns) {
-      auto const device_comparator =
-        two_table_equal.equal_to<true>(nullate::DYNAMIC{has_nulls}, _nulls_equal);
-      comparator_helper(device_comparator);
-    } else {
-      auto const device_comparator =
-        two_table_equal.equal_to<false>(nullate::DYNAMIC{has_nulls}, _nulls_equal);
-      comparator_helper(device_comparator);
     }
   }
-
   return build_indices;
 }
 }  // namespace detail