diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt
index 9edbb51e8820..66766953e798 100644
--- a/cpp/CMakeLists.txt
+++ b/cpp/CMakeLists.txt
@@ -1008,6 +1008,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
diff --git a/cpp/src/hash/murmurhash3_x86_32.cu b/cpp/src/hash/murmurhash3_x86_32.cu
index 848cb71b3e5f..f82d552456d7 100644
--- a/cpp/src/hash/murmurhash3_x86_32.cu
+++ b/cpp/src/hash/murmurhash3_x86_32.cu
@@ -4,17 +4,26 @@
  */
 #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 {
diff --git a/cpp/src/stream_compaction/distinct.cu b/cpp/src/stream_compaction/distinct.cu
index e371ea016b22..9db6bbd76c8c 100644
--- a/cpp/src/stream_compaction/distinct.cu
+++ b/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);
+        });
+      });
   }
 }
 
diff --git a/cpp/src/stream_compaction/distinct_helpers.cu b/cpp/src/stream_compaction/distinct_helpers.cu
index 7474c39997c7..2596496c8774 100644
--- a/cpp/src/stream_compaction/distinct_helpers.cu
+++ b/cpp/src/stream_compaction/distinct_helpers.cu
@@ -1,136 +1,67 @@
 /*
- * 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 "distinct_helpers.hpp"
 
 #include <cudf/detail/algorithms/copy_if.cuh>
+#include <cudf/stream_compaction.hpp>
+#include <cudf/types.hpp>
+#include <cudf/utilities/memory_resource.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/exec_policy.hpp>
+#include <rmm/resource_ref.hpp>
 
 #include <cuda/functional>
 #include <cuda/iterator>
-#include <cuda/std/atomic>
 #include <cuda/std/iterator>
+#include <thrust/uninitialized_fill.h>
 
 namespace cudf::detail {
 
-template <typename RowEqual>
-rmm::device_uvector<size_type> reduce_by_row(distinct_set_t<RowEqual>& set,
-                                             size_type num_rows,
-                                             duplicate_keep_option keep,
-                                             rmm::cuda_stream_view stream,
-                                             rmm::device_async_resource_ref mr)
+void initialize_reduction_results(size_type* results,
+                                  size_type num_rows,
+                                  duplicate_keep_option keep,
+                                  rmm::cuda_stream_view stream)
 {
-  auto output_indices = rmm::device_uvector<size_type>(num_rows, stream, mr);
-
-  // If we don't care about order, just gather indices of distinct keys taken from set.
-  if (keep == duplicate_keep_option::KEEP_ANY) {
-    auto const iter = cuda::counting_iterator<cudf::size_type>{0};
-    set.insert_async(iter, iter + num_rows, stream.value());
-    auto const output_end = set.retrieve_all(output_indices.begin(), stream.value());
-    output_indices.resize(cuda::std::distance(output_indices.begin(), output_end), stream);
-    return output_indices;
-  }
-
-  auto reduction_results = rmm::device_uvector<size_type>(num_rows, stream, mr);
   thrust::uninitialized_fill(
     rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()),
-    reduction_results.begin(),
-    reduction_results.end(),
+    results,
+    results + num_rows,
     reduction_init_value(keep));
+}
 
-  auto set_ref = set.ref(cuco::op::insert_and_find);
-
-  thrust::for_each(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()),
-                   cuda::counting_iterator<cudf::size_type>{0},
-                   cuda::counting_iterator{num_rows},
-                   [set_ref, keep, reduction_results = reduction_results.begin()] __device__(
-                     size_type const idx) mutable {
-                     auto const [inserted_idx_ptr, _] = set_ref.insert_and_find(idx);
-
-                     auto ref = cuda::atomic_ref<size_type, cuda::thread_scope_device>{
-                       reduction_results[*inserted_idx_ptr]};
-                     if (keep == duplicate_keep_option::KEEP_FIRST) {
-                       // Store the smallest index of all rows that are equal.
-                       ref.fetch_min(idx, cuda::memory_order_relaxed);
-                     } else if (keep == duplicate_keep_option::KEEP_LAST) {
-                       // Store the greatest index of all rows that are equal.
-                       ref.fetch_max(idx, cuda::memory_order_relaxed);
-                     } else {
-                       // Count the number of rows in each group of rows that are compared equal.
-                       ref.fetch_add(size_type{1}, cuda::memory_order_relaxed);
-                     }
-                   });
-
-  auto const map_end = [&] {
+size_type copy_reduction_results(size_type const* results,
+                                 size_type num_rows,
+                                 size_type* output,
+                                 duplicate_keep_option keep,
+                                 rmm::cuda_stream_view stream)
+{
+  auto const output_end = [&] {
     if (keep == duplicate_keep_option::KEEP_NONE) {
-      // Reduction results with `KEEP_NONE` are either group sizes of equal rows, or `0`.
-      // Thus, we only output index of the rows in the groups having group size of `1`.
+      // KEEP_NONE stores group sizes; retain only singleton groups.
       return cudf::detail::copy_if(
         cuda::counting_iterator<size_type>{0},
         cuda::counting_iterator<size_type>{num_rows},
-        output_indices.begin(),
+        output,
         cuda::proclaim_return_type<bool>(
-          [reduction_results = reduction_results.begin()] __device__(auto const idx) {
-            return reduction_results[idx] == size_type{1};
-          }),
+          [results] __device__(auto const idx) { return results[idx] == size_type{1}; }),
         stream);
     }
 
-    // Reduction results with `KEEP_FIRST` and `KEEP_LAST` are row indices of the first/last row in
-    // each group of equal rows (which are the desired output indices), or the value given by
-    // `reduction_init_value()`.
+    // KEEP_FIRST and KEEP_LAST store desired row indices or the mode's initial marker.
     return cudf::detail::copy_if(
-      reduction_results.begin(),
-      reduction_results.end(),
-      output_indices.begin(),
+      results,
+      results + num_rows,
+      output,
       cuda::proclaim_return_type<bool>([init_value = reduction_init_value(keep)] __device__(
                                          auto const idx) { return idx != init_value; }),
       stream);
   }();
 
-  output_indices.resize(cuda::std::distance(output_indices.begin(), map_end), stream);
-  return output_indices;
+  return cuda::std::distance(output, output_end);
 }
 
-template rmm::device_uvector<size_type> reduce_by_row(
-  distinct_set_t<cudf::detail::row::equality::device_row_comparator<
-    false,
-    cudf::nullate::DYNAMIC,
-    cudf::detail::row::equality::nan_equal_physical_equality_comparator>>& set,
-  size_type num_rows,
-  duplicate_keep_option keep,
-  rmm::cuda_stream_view stream,
-  rmm::device_async_resource_ref mr);
-
-template rmm::device_uvector<size_type> reduce_by_row(
-  distinct_set_t<cudf::detail::row::equality::device_row_comparator<
-    true,
-    cudf::nullate::DYNAMIC,
-    cudf::detail::row::equality::nan_equal_physical_equality_comparator>>& set,
-  size_type num_rows,
-  duplicate_keep_option keep,
-  rmm::cuda_stream_view stream,
-  rmm::device_async_resource_ref mr);
-
-template rmm::device_uvector<size_type> reduce_by_row(
-  distinct_set_t<cudf::detail::row::equality::device_row_comparator<
-    false,
-    cudf::nullate::DYNAMIC,
-    cudf::detail::row::equality::physical_equality_comparator>>& set,
-  size_type num_rows,
-  duplicate_keep_option keep,
-  rmm::cuda_stream_view stream,
-  rmm::device_async_resource_ref mr);
-
-template rmm::device_uvector<size_type> reduce_by_row(
-  distinct_set_t<cudf::detail::row::equality::device_row_comparator<
-    true,
-    cudf::nullate::DYNAMIC,
-    cudf::detail::row::equality::physical_equality_comparator>>& set,
-  size_type num_rows,
-  duplicate_keep_option keep,
-  rmm::cuda_stream_view stream,
-  rmm::device_async_resource_ref mr);
-
 }  // namespace cudf::detail
diff --git a/cpp/src/stream_compaction/distinct_helpers.cuh b/cpp/src/stream_compaction/distinct_helpers.cuh
new file mode 100644
index 000000000000..7649ea081380
--- /dev/null
+++ b/cpp/src/stream_compaction/distinct_helpers.cuh
@@ -0,0 +1,83 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "distinct_helpers.hpp"
+
+#include <cudf/stream_compaction.hpp>
+#include <cudf/types.hpp>
+#include <cudf/utilities/memory_resource.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_uvector.hpp>
+#include <rmm/exec_policy.hpp>
+#include <rmm/resource_ref.hpp>
+
+#include <cuco/operator.hpp>
+#include <cuda/iterator>
+#include <cuda/std/atomic>
+#include <cuda/std/iterator>
+#include <thrust/for_each.h>
+
+namespace cudf::detail {
+
+template <typename Set>
+rmm::device_uvector<size_type> reduce_by_row_keep_any(Set& set,
+                                                      size_type num_rows,
+                                                      rmm::cuda_stream_view stream,
+                                                      rmm::device_async_resource_ref mr)
+{
+  auto output_indices = rmm::device_uvector<size_type>(num_rows, stream, mr);
+
+  auto const iter = cuda::counting_iterator<cudf::size_type>{0};
+  set.insert_async(iter, iter + num_rows, stream.value());
+  auto const output_end = set.retrieve_all(output_indices.begin(), stream.value());
+  output_indices.resize(cuda::std::distance(output_indices.begin(), output_end), stream);
+  return output_indices;
+}
+
+template <typename Set>
+rmm::device_uvector<size_type> reduce_by_row_keep_first_last_none(Set& set,
+                                                                  size_type num_rows,
+                                                                  duplicate_keep_option keep,
+                                                                  rmm::cuda_stream_view stream,
+                                                                  rmm::device_async_resource_ref mr)
+{
+  auto output_indices = rmm::device_uvector<size_type>(num_rows, stream, mr);
+  auto reduction_results =
+    rmm::device_uvector<size_type>(num_rows, stream, cudf::get_current_device_resource_ref());
+  initialize_reduction_results(reduction_results.data(), num_rows, keep, stream);
+
+  auto set_ref = set.ref(cuco::op::insert_and_find);
+
+  thrust::for_each(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()),
+                   cuda::counting_iterator<cudf::size_type>{0},
+                   cuda::counting_iterator{num_rows},
+                   [set_ref, keep, reduction_results = reduction_results.begin()] __device__(
+                     size_type const idx) mutable -> void {
+                     auto const [inserted_idx_ptr, _] = set_ref.insert_and_find(idx);
+
+                     auto ref = cuda::atomic_ref<size_type, cuda::thread_scope_device>{
+                       reduction_results[*inserted_idx_ptr]};
+                     if (keep == duplicate_keep_option::KEEP_FIRST) {
+                       // Store the smallest index of all rows that are equal.
+                       ref.fetch_min(idx, cuda::memory_order_relaxed);
+                     } else if (keep == duplicate_keep_option::KEEP_LAST) {
+                       // Store the greatest index of all rows that are equal.
+                       ref.fetch_max(idx, cuda::memory_order_relaxed);
+                     } else {
+                       // Count the number of rows in each group of rows that are compared equal.
+                       ref.fetch_add(size_type{1}, cuda::memory_order_relaxed);
+                     }
+                   });
+
+  auto const output_size =
+    copy_reduction_results(reduction_results.data(), num_rows, output_indices.data(), keep, stream);
+  output_indices.resize(output_size, stream);
+  return output_indices;
+}
+
+}  // namespace cudf::detail
diff --git a/cpp/src/stream_compaction/distinct_helpers.hpp b/cpp/src/stream_compaction/distinct_helpers.hpp
index 57cb4d5ff3d1..c7903aae7672 100644
--- a/cpp/src/stream_compaction/distinct_helpers.hpp
+++ b/cpp/src/stream_compaction/distinct_helpers.hpp
@@ -1,24 +1,32 @@
 /*
- * 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 <cudf/detail/cuco_helpers.hpp>
-#include <cudf/detail/row_operator/equality.cuh>
 #include <cudf/detail/row_operator/hashing.cuh>
+#include <cudf/hashing.hpp>
+#include <cudf/hashing/detail/default_hash.cuh>
 #include <cudf/stream_compaction.hpp>
 #include <cudf/types.hpp>
+#include <cudf/utilities/error.hpp>
+#include <cudf/utilities/export.hpp>
 #include <cudf/utilities/memory_resource.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/extent.cuh>
+#include <cuco/probing_scheme.cuh>
 #include <cuco/static_set.cuh>
-#include <cuda/functional>
-#include <cuda/std/iterator>
+#include <cuco/storage.cuh>
+#include <cuda/atomic>
+
+#include <cstdint>
+#include <limits>
 
 namespace cudf::detail {
 
@@ -38,51 +46,102 @@ auto constexpr reduction_init_value(duplicate_keep_option keep)
   }
 }
 
-template <typename RowEqual>
-using distinct_set_t =
-  cuco::static_set<size_type,
-                   cuco::extent<int64_t>,
-                   cuda::thread_scope_device,
-                   RowEqual,
-                   cuco::linear_probing<
-                     1,
-                     cudf::detail::row::hash::device_row_hasher<cudf::hashing::detail::default_hash,
-                                                                cudf::nullate::DYNAMIC>>,
-                   rmm::mr::polymorphic_allocator<char>,
-                   cuco::storage<1>>;
+CUDF_HIDDEN void initialize_reduction_results(size_type* results,
+                                              size_type num_rows,
+                                              duplicate_keep_option keep,
+                                              rmm::cuda_stream_view stream);
+
+CUDF_HIDDEN size_type copy_reduction_results(size_type const* results,
+                                             size_type num_rows,
+                                             size_type* output,
+                                             duplicate_keep_option keep,
+                                             rmm::cuda_stream_view stream);
+
+struct distinct_precomputed_hash {
+  CUDF_HOST_DEVICE constexpr distinct_precomputed_hash(hash_value_type const* hashes)
+    : _hashes{hashes}
+  {
+  }
+
+  __device__ __forceinline__ hash_value_type operator()(size_type i) const noexcept
+  {
+    return _hashes[i];
+  }
+
+ private:
+  hash_value_type const* _hashes;
+};
+
+template <typename RowEqual,
+          typename RowHash = cudf::detail::row::hash::
+            device_row_hasher<cudf::hashing::detail::default_hash, cudf::nullate::DYNAMIC>>
+using distinct_set_t = cuco::static_set<size_type,
+                                        cuco::extent<int64_t>,
+                                        cuda::thread_scope_device,
+                                        RowEqual,
+                                        cuco::linear_probing<1, RowHash>,
+                                        rmm::mr::polymorphic_allocator<char>,
+                                        cuco::storage<1>>;
 
 /**
- * @brief Perform a reduction on groups of rows that are compared equal and returns output indices
- * of the occurrences of the distinct elements based on `keep` parameter.
+ * @brief Returns one unspecified row index from each group of equal rows.
  *
- * This is essentially a reduce-by-key operation with keys are non-contiguous rows and are compared
- * equal. A hash set is used to find groups of equal rows.
- *
- * Depending on the `keep` parameter, the reduction operation for each row group is:
- * - If `keep == KEEP_ANY` : order does not matter.
- * - If `keep == KEEP_FIRST`: min of row indices in the group.
- * - If `keep == KEEP_LAST`: max of row indices in the group.
- * - If `keep == KEEP_NONE`: count of equivalent rows (group size).
- *
- * Note that this function is not needed when `keep == KEEP_NONE`.
+ * @tparam Set The type of the auxiliary set
+ * @param set The auxiliary set used to identify groups of equal rows
+ * @param num_rows The number of input rows
+ * @param stream CUDA stream used for device memory operations and kernel launches
+ * @param mr Device memory resource used to allocate the returned vector
+ * @return A device vector containing one row index from each group
+ */
+template <typename Set>
+rmm::device_uvector<size_type> reduce_by_row_keep_any(Set& set,
+                                                      size_type num_rows,
+                                                      rmm::cuda_stream_view stream,
+                                                      rmm::device_async_resource_ref mr);
+
+/**
+ * @brief Returns row indices selected from groups of equal rows according to `keep`.
  *
- * At the beginning of the operation, the entire output array is filled with a value given by
- * the `reduction_init_value()` function. Then, the reduction result for each row group is written
- * into the output array at the index of an unspecified row in the group.
+ * `KEEP_FIRST` returns the smallest row index in each group, `KEEP_LAST` returns the largest, and
+ * `KEEP_NONE` returns indices only for singleton groups.
  *
- * @tparam RowEqual The type of row equality comparator
+ * @tparam Set The type of the auxiliary set
+ * @param set The auxiliary set used to identify groups of equal rows
+ * @param num_rows The number of input rows
+ * @param keep The duplicate selection mode; must not be `KEEP_ANY`
+ * @param stream CUDA stream used for device memory operations and kernel launches
+ * @param mr Device memory resource used to allocate the returned vector
+ * @return A device vector containing the selected row indices
+ */
+template <typename Set>
+rmm::device_uvector<size_type> reduce_by_row_keep_first_last_none(
+  Set& set,
+  size_type num_rows,
+  duplicate_keep_option keep,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr);
+
+/**
+ * @brief Returns row indices selected from groups of equal rows according to `keep`.
  *
- * @param set The auxiliary set to perform reduction
- * @param num_rows The number of all input rows
- * @param keep The parameter to determine what type of reduction to perform
+ * @tparam Set The type of the auxiliary set
+ * @param set The auxiliary set used to identify groups of equal rows
+ * @param num_rows The number of input rows
+ * @param keep The duplicate selection mode
  * @param stream CUDA stream used for device memory operations and kernel launches
  * @param mr Device memory resource used to allocate the returned vector
- * @return A device_uvector containing the output indices
+ * @return A device vector containing the selected row indices
  */
-template <typename RowEqual>
-rmm::device_uvector<size_type> reduce_by_row(distinct_set_t<RowEqual>& set,
+template <typename Set>
+rmm::device_uvector<size_type> reduce_by_row(Set& set,
                                              size_type num_rows,
                                              duplicate_keep_option keep,
                                              rmm::cuda_stream_view stream,
-                                             rmm::device_async_resource_ref mr);
+                                             rmm::device_async_resource_ref mr)
+{
+  if (keep == duplicate_keep_option::KEEP_ANY) {
+    return reduce_by_row_keep_any(set, num_rows, stream, mr);
+  }
+  return reduce_by_row_keep_first_last_none(set, num_rows, keep, stream, mr);
+}
 }  // namespace cudf::detail
diff --git a/cpp/src/stream_compaction/distinct_helpers_flat_nan_equal_any.cu b/cpp/src/stream_compaction/distinct_helpers_flat_nan_equal_any.cu
new file mode 100644
index 000000000000..91c112e03595
--- /dev/null
+++ b/cpp/src/stream_compaction/distinct_helpers_flat_nan_equal_any.cu
@@ -0,0 +1,27 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "distinct_helpers.cuh"
+#include "distinct_helpers.hpp"
+
+#include <cudf/detail/row_operator/equality.cuh>
+#include <cudf/types.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_uvector.hpp>
+#include <rmm/resource_ref.hpp>
+
+namespace cudf::detail {
+
+template rmm::device_uvector<size_type> reduce_by_row_keep_any(
+  distinct_set_t<cudf::detail::row::equality::device_row_comparator<
+    false,
+    cudf::nullate::DYNAMIC,
+    cudf::detail::row::equality::nan_equal_physical_equality_comparator>>& set,
+  size_type num_rows,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr);
+
+}  // namespace cudf::detail
diff --git a/cpp/src/stream_compaction/distinct_helpers_flat_nan_equal_ordered.cu b/cpp/src/stream_compaction/distinct_helpers_flat_nan_equal_ordered.cu
new file mode 100644
index 000000000000..f266862dcce6
--- /dev/null
+++ b/cpp/src/stream_compaction/distinct_helpers_flat_nan_equal_ordered.cu
@@ -0,0 +1,29 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "distinct_helpers.cuh"
+#include "distinct_helpers.hpp"
+
+#include <cudf/detail/row_operator/equality.cuh>
+#include <cudf/stream_compaction.hpp>
+#include <cudf/types.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_uvector.hpp>
+#include <rmm/resource_ref.hpp>
+
+namespace cudf::detail {
+
+template rmm::device_uvector<size_type> reduce_by_row_keep_first_last_none(
+  distinct_set_t<cudf::detail::row::equality::device_row_comparator<
+    false,
+    cudf::nullate::DYNAMIC,
+    cudf::detail::row::equality::nan_equal_physical_equality_comparator>>& set,
+  size_type num_rows,
+  duplicate_keep_option keep,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr);
+
+}  // namespace cudf::detail
diff --git a/cpp/src/stream_compaction/distinct_helpers_flat_nan_unequal_any.cu b/cpp/src/stream_compaction/distinct_helpers_flat_nan_unequal_any.cu
new file mode 100644
index 000000000000..ce224e6d91d0
--- /dev/null
+++ b/cpp/src/stream_compaction/distinct_helpers_flat_nan_unequal_any.cu
@@ -0,0 +1,27 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "distinct_helpers.cuh"
+#include "distinct_helpers.hpp"
+
+#include <cudf/detail/row_operator/equality.cuh>
+#include <cudf/types.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_uvector.hpp>
+#include <rmm/resource_ref.hpp>
+
+namespace cudf::detail {
+
+template rmm::device_uvector<size_type> reduce_by_row_keep_any(
+  distinct_set_t<cudf::detail::row::equality::device_row_comparator<
+    false,
+    cudf::nullate::DYNAMIC,
+    cudf::detail::row::equality::physical_equality_comparator>>& set,
+  size_type num_rows,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr);
+
+}  // namespace cudf::detail
diff --git a/cpp/src/stream_compaction/distinct_helpers_flat_nan_unequal_ordered.cu b/cpp/src/stream_compaction/distinct_helpers_flat_nan_unequal_ordered.cu
new file mode 100644
index 000000000000..40a9a327e6c4
--- /dev/null
+++ b/cpp/src/stream_compaction/distinct_helpers_flat_nan_unequal_ordered.cu
@@ -0,0 +1,29 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "distinct_helpers.cuh"
+#include "distinct_helpers.hpp"
+
+#include <cudf/detail/row_operator/equality.cuh>
+#include <cudf/stream_compaction.hpp>
+#include <cudf/types.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_uvector.hpp>
+#include <rmm/resource_ref.hpp>
+
+namespace cudf::detail {
+
+template rmm::device_uvector<size_type> reduce_by_row_keep_first_last_none(
+  distinct_set_t<cudf::detail::row::equality::device_row_comparator<
+    false,
+    cudf::nullate::DYNAMIC,
+    cudf::detail::row::equality::physical_equality_comparator>>& set,
+  size_type num_rows,
+  duplicate_keep_option keep,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr);
+
+}  // namespace cudf::detail
diff --git a/cpp/src/stream_compaction/distinct_helpers_nested_nan_equal_any.cu b/cpp/src/stream_compaction/distinct_helpers_nested_nan_equal_any.cu
new file mode 100644
index 000000000000..e60ad6a2547e
--- /dev/null
+++ b/cpp/src/stream_compaction/distinct_helpers_nested_nan_equal_any.cu
@@ -0,0 +1,28 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "distinct_helpers.cuh"
+#include "distinct_helpers.hpp"
+
+#include <cudf/detail/row_operator/equality.cuh>
+#include <cudf/types.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_uvector.hpp>
+#include <rmm/resource_ref.hpp>
+
+namespace cudf::detail {
+
+template rmm::device_uvector<size_type> reduce_by_row_keep_any(
+  distinct_set_t<cudf::detail::row::equality::device_row_comparator<
+                   true,
+                   cudf::nullate::DYNAMIC,
+                   cudf::detail::row::equality::nan_equal_physical_equality_comparator>,
+                 distinct_precomputed_hash>& set,
+  size_type num_rows,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr);
+
+}  // namespace cudf::detail
diff --git a/cpp/src/stream_compaction/distinct_helpers_nested_nan_equal_ordered.cu b/cpp/src/stream_compaction/distinct_helpers_nested_nan_equal_ordered.cu
new file mode 100644
index 000000000000..3030a4c63e99
--- /dev/null
+++ b/cpp/src/stream_compaction/distinct_helpers_nested_nan_equal_ordered.cu
@@ -0,0 +1,29 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "distinct_helpers.cuh"
+#include "distinct_helpers.hpp"
+
+#include <cudf/detail/row_operator/equality.cuh>
+#include <cudf/stream_compaction.hpp>
+#include <cudf/types.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_uvector.hpp>
+#include <rmm/resource_ref.hpp>
+
+namespace cudf::detail {
+
+template rmm::device_uvector<size_type> reduce_by_row_keep_first_last_none(
+  distinct_set_t<cudf::detail::row::equality::device_row_comparator<
+    true,
+    cudf::nullate::DYNAMIC,
+    cudf::detail::row::equality::nan_equal_physical_equality_comparator>>& set,
+  size_type num_rows,
+  duplicate_keep_option keep,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr);
+
+}  // namespace cudf::detail
diff --git a/cpp/src/stream_compaction/distinct_helpers_nested_nan_unequal_any.cu b/cpp/src/stream_compaction/distinct_helpers_nested_nan_unequal_any.cu
new file mode 100644
index 000000000000..111ee665bd53
--- /dev/null
+++ b/cpp/src/stream_compaction/distinct_helpers_nested_nan_unequal_any.cu
@@ -0,0 +1,28 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "distinct_helpers.cuh"
+#include "distinct_helpers.hpp"
+
+#include <cudf/detail/row_operator/equality.cuh>
+#include <cudf/types.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_uvector.hpp>
+#include <rmm/resource_ref.hpp>
+
+namespace cudf::detail {
+
+template rmm::device_uvector<size_type> reduce_by_row_keep_any(
+  distinct_set_t<cudf::detail::row::equality::device_row_comparator<
+                   true,
+                   cudf::nullate::DYNAMIC,
+                   cudf::detail::row::equality::physical_equality_comparator>,
+                 distinct_precomputed_hash>& set,
+  size_type num_rows,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr);
+
+}  // namespace cudf::detail
diff --git a/cpp/src/stream_compaction/distinct_helpers_nested_nan_unequal_ordered.cu b/cpp/src/stream_compaction/distinct_helpers_nested_nan_unequal_ordered.cu
new file mode 100644
index 000000000000..0a1b98929e81
--- /dev/null
+++ b/cpp/src/stream_compaction/distinct_helpers_nested_nan_unequal_ordered.cu
@@ -0,0 +1,29 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "distinct_helpers.cuh"
+#include "distinct_helpers.hpp"
+
+#include <cudf/detail/row_operator/equality.cuh>
+#include <cudf/stream_compaction.hpp>
+#include <cudf/types.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_uvector.hpp>
+#include <rmm/resource_ref.hpp>
+
+namespace cudf::detail {
+
+template rmm::device_uvector<size_type> reduce_by_row_keep_first_last_none(
+  distinct_set_t<cudf::detail::row::equality::device_row_comparator<
+    true,
+    cudf::nullate::DYNAMIC,
+    cudf::detail::row::equality::physical_equality_comparator>>& set,
+  size_type num_rows,
+  duplicate_keep_option keep,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr);
+
+}  // namespace cudf::detail