Optimize hash full join

cpp/src/join/conditional_join.cu

@@ -277,7 +277,7 @@ conditional_join(table_view const& left,
   // by any row in the left table.
   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);
+      std::move(join_indices), left.num_rows(), right.num_rows(), std::nullopt, stream, mr);
   }
   return join_indices;
 }

cpp/src/join/hash_join/dispatch.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
@@ -52,6 +52,25 @@ struct output_fn {
   }
 };
 
+/**
+ * @brief Extracts a right-side row index and marks it as matched.
+ *
+ * This is used while retrieving a full join to build the right-side match set without a
+ * subsequent pass over the join output.
+ */
+struct mark_matched_output_fn {
+  size_type* right_matches;
+  size_type right_table_num_rows;
+
+  __device__ cudf::size_type operator()(
+    cuco::pair<hash_value_type, cudf::size_type> const& slot) const
+  {
+    auto const index = slot.second;
+    if (index >= 0 && index < right_table_num_rows) { right_matches[index] = 1; }
+    return index;
+  }
+};
+
 /**
  * @brief Equality comparator for cuco hash table probing with primitive row equality.
  */

cpp/src/join/hash_join/full_join_size_impl.cu

@@ -6,48 +6,22 @@
 #include "retrieve_impl.cuh"
 
 #include <cudf/detail/algorithms/reduce.cuh>
+#include <cudf/detail/utilities/vector_factories.hpp>
 #include <cudf/utilities/span.hpp>
 
-#include <rmm/device_uvector.hpp>
-#include <rmm/exec_policy.hpp>
-
 #include <cuda/iterator>
-#include <cuda/std/functional>
-#include <thrust/scatter.h>
-#include <thrust/uninitialized_fill.h>
-
-#include <memory>
 
 namespace cudf::detail {
 
 namespace {
-std::size_t compute_left_join_complement_size(cudf::device_span<size_type const> right_indices,
-                                              size_type left_table_row_count,
-                                              size_type right_table_row_count,
+std::size_t compute_left_join_complement_size(cudf::device_span<size_type const> right_matches,
                                               cuda::stream_ref stream)
 {
-  if (left_table_row_count == 0) { return right_table_row_count; }
-
-  auto invalid_index_map =
-    std::make_unique<rmm::device_uvector<size_type>>(right_table_row_count, stream);
-  thrust::uninitialized_fill(
-    rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()),
-    invalid_index_map->begin(),
-    invalid_index_map->end(),
-    int32_t{1});
-
-  valid_range<size_type> valid(0, right_table_row_count);
-
-  thrust::scatter_if(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()),
-                     cuda::make_constant_iterator(0),
-                     cuda::make_constant_iterator(0) + right_indices.size(),
-                     right_indices.begin(),
-                     right_indices.begin(),
-                     invalid_index_map->begin(),
-                     valid);
-
   return cudf::detail::count_if(
-    invalid_index_map->begin(), invalid_index_map->end(), cuda::std::identity{}, stream);
+    right_matches.begin(),
+    right_matches.end(),
+    [] __device__(size_type is_matched) -> bool { return is_matched == 0; },
+    stream);
 }
 }  // namespace
 
@@ -71,12 +45,13 @@ std::size_t get_full_join_size(
                                                                          compare_nulls,
                                                                          stream);
 
-  if (join_size == 0) { return join_size; }
+  if (join_size == 0) { return right_table.num_rows(); }
 
-  auto right_indices = std::make_unique<rmm::device_uvector<size_type>>(join_size, stream, mr);
-
-  auto const out_build_begin =
-    cuda::make_transform_output_iterator(right_indices->begin(), output_fn{});
+  auto right_matches =
+    cudf::detail::make_zeroed_device_uvector_async<size_type>(right_table.num_rows(), stream, mr);
+  auto const out_build_begin = cuda::make_transform_output_iterator(
+    cuda::make_discard_iterator(),
+    mark_matched_output_fn{right_matches.data(), right_table.num_rows()});
 
   retrieve_left_join_build_indices(right_table,
                                    left_table,
@@ -88,11 +63,7 @@ std::size_t get_full_join_size(
                                    out_build_begin,
                                    stream);
 
-  auto const left_table_row_count  = left_table.num_rows();
-  auto const right_table_row_count = right_table.num_rows();
-
-  return join_size + compute_left_join_complement_size(
-                       *right_indices, left_table_row_count, right_table_row_count, stream);
+  return join_size + compute_left_join_complement_size(right_matches, stream);
 }
 
 }  // namespace cudf::detail

cpp/src/join/hash_join/retrieve_impl.cuh

@@ -12,6 +12,7 @@
 
 #include <cudf/copying.hpp>
 #include <cudf/detail/nvtx/ranges.hpp>
+#include <cudf/detail/utilities/vector_factories.hpp>
 #include <cudf/join/join.hpp>
 #include <cudf/table/table_view.hpp>
 #include <cudf/utilities/memory_resource.hpp>
@@ -37,6 +38,7 @@ probe_join_hash_table(
   bool has_nulls,
   null_equality compare_nulls,
   std::optional<std::size_t> output_size,
+  cudf::device_span<size_type> right_matches,
   cuda::stream_ref stream,
   rmm::device_async_resource_ref mr)
 {
@@ -61,16 +63,34 @@ probe_join_hash_table(
                      std::make_unique<rmm::device_uvector<size_type>>(0, stream, mr));
   }
 
-  auto left_indices  = std::make_unique<rmm::device_uvector<size_type>>(join_size, stream, mr);
-  auto right_indices = std::make_unique<rmm::device_uvector<size_type>>(join_size, stream, mr);
+  // Without a supplied full-join size, reserve room for the largest possible right complement.
+  // This prevents finalization from reallocating and copying both left-join output vectors.
+  auto const allocation_size = Join == join_kind::FULL_JOIN && !output_size
+                                 ? join_size + static_cast<std::size_t>(right_table.num_rows())
+                                 : join_size;
+  auto left_indices = std::make_unique<rmm::device_uvector<size_type>>(allocation_size, stream, mr);
+  auto right_indices =
+    std::make_unique<rmm::device_uvector<size_type>>(allocation_size, stream, mr);
+  left_indices->resize(join_size, stream);
+  right_indices->resize(join_size, stream);
   cudf::prefetch::detail::prefetch(*left_indices, stream);
   cudf::prefetch::detail::prefetch(*right_indices, stream);
 
   auto const left_table_num_rows = left_table.num_rows();
   auto const out_probe_begin =
     cuda::make_transform_output_iterator(left_indices->begin(), output_fn{});
-  auto const out_build_begin =
-    cuda::make_transform_output_iterator(right_indices->begin(), output_fn{});
+  auto const out_build_begin = [&] {
+    if constexpr (Join == join_kind::FULL_JOIN) {
+      CUDF_EXPECTS(right_matches.size() == static_cast<std::size_t>(right_table.num_rows()),
+                   "full join requires one match flag per right row",
+                   std::invalid_argument);
+      return cuda::make_transform_output_iterator(
+        right_indices->begin(),
+        mark_matched_output_fn{right_matches.data(), right_table.num_rows()});
+    } else {
+      return cuda::make_transform_output_iterator(right_indices->begin(), output_fn{});
+    }
+  }();
 
   auto retrieve_results = [&](auto equality, auto d_hasher) {
     auto const iter = cudf::detail::make_counting_transform_iterator(0, pair_fn{d_hasher});
@@ -176,6 +196,11 @@ hash_join<Hasher>::join_retrieve(cudf::table_view const& left,
   auto const preprocessed_left =
     cudf::detail::row::equality::preprocessed_table::create(left, stream);
 
+  auto right_matches = cudf::detail::make_zeroed_device_uvector_async<size_type>(
+    Join == join_kind::FULL_JOIN ? _right.num_rows() : 0,
+    stream,
+    cudf::get_current_device_resource_ref());
+
   auto join_indices = cudf::detail::probe_join_hash_table<Join>(_right,
                                                                 left,
                                                                 _preprocessed_right,
@@ -184,12 +209,13 @@ hash_join<Hasher>::join_retrieve(cudf::table_view const& left,
                                                                 _has_nulls,
                                                                 _nulls_equal,
                                                                 output_size,
+                                                                right_matches,
                                                                 stream,
                                                                 mr);
 
   if constexpr (Join == join_kind::FULL_JOIN) {
     return detail::finalize_full_join(
-      std::move(join_indices), left.num_rows(), _right.num_rows(), stream, mr);
+      std::move(join_indices), left.num_rows(), _right.num_rows(), right_matches, stream, mr);
   } else {
     return join_indices;
   }

cpp/src/join/join_common_utils.hpp

@@ -15,6 +15,7 @@
 #include <cuda/stream_ref>
 
 #include <memory>
+#include <optional>
 #include <utility>
 
 namespace cudf::detail {
@@ -66,6 +67,8 @@ VectorPair get_trivial_left_join_indices(table_view const& left,
  * @param left_table_num_rows Number of rows in the left table (0 → every right row is
  *                            unmatched, fast path).
  * @param right_table_num_rows Number of rows in the right table.
+ * @param right_matches Optional precomputed flags indicating which right rows matched. When absent,
+ *                      the flags are derived from `indices.second`.
  * @param stream CUDA stream used for device memory operations and kernel launches.
  * @param mr Device memory resource used to allocate working storage.
  *
@@ -74,6 +77,7 @@ VectorPair get_trivial_left_join_indices(table_view const& left,
 VectorPair finalize_full_join(VectorPair&& indices,
                               size_type left_table_num_rows,
                               size_type right_table_num_rows,
+                              std::optional<cudf::device_span<size_type const>> right_matches,
                               cuda::stream_ref stream,
                               rmm::device_async_resource_ref mr);
 

cpp/src/join/join_utils.cu

@@ -86,6 +86,7 @@ struct to_no_match_pair {
 VectorPair finalize_full_join(VectorPair&& indices,
                               size_type left_table_num_rows,
                               size_type right_table_num_rows,
+                              std::optional<cudf::device_span<size_type const>> right_matches,
                               cuda::stream_ref stream,
                               rmm::device_async_resource_ref mr)
 {
@@ -121,20 +122,26 @@ VectorPair finalize_full_join(VectorPair&& indices,
   left_out->resize(upper, stream);
   right_out->resize(upper, stream);
 
-  // Mark matched right rows in an int32 flag array (one word per right row). Redundant stores
-  // of the same value are idempotent, so no atomics are needed. Word-sized stores coalesce into
-  // full 128-byte transactions per warp; byte-sized flags cost ~2–3× here because partial-word
-  // stores from dense scatters serialize within each 32-bit sector.
-  auto flags = cudf::detail::make_zeroed_device_uvector_async<size_type>(
-    right_table_num_rows, stream, cudf::get_current_device_resource_ref());
-
-  thrust::scatter_if(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()),
-                     cuda::make_constant_iterator(size_type{1}),
-                     cuda::make_constant_iterator(size_type{1}) + match_total,
-                     right_out->begin(),
-                     right_out->begin(),
-                     flags.begin(),
-                     valid_range<size_type>{0, right_table_num_rows});
+  CUDF_EXPECTS(
+    !right_matches || right_matches->size() == static_cast<std::size_t>(right_table_num_rows),
+    "right match flags must be absent or have one entry per right row",
+    std::invalid_argument);
+
+  // Hash joins mark right rows as part of retrieval and pass those flags here, eliminating an
+  // output-sized scatter. Other join implementations use this fallback to derive the same flags
+  // from their materialized right indices.
+  auto computed_matches = cudf::detail::make_zeroed_device_uvector_async<size_type>(
+    right_matches ? 0 : right_table_num_rows, stream, cudf::get_current_device_resource_ref());
+  if (!right_matches) {
+    thrust::scatter_if(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()),
+                       cuda::make_constant_iterator(size_type{1}),
+                       cuda::make_constant_iterator(size_type{1}) + match_total,
+                       right_out->begin(),
+                       right_out->begin(),
+                       computed_matches.begin(),
+                       valid_range<size_type>{0, right_table_num_rows});
+  }
+  auto const match_flags = right_matches ? right_matches->data() : computed_matches.data();
 
   // Fused compaction: for each unmatched right row, emit (JoinNoMatch, right_idx) into
   // (left_out_tail, right_out_tail) in a single CUB DeviceSelect pass.
@@ -146,7 +153,7 @@ VectorPair finalize_full_join(VectorPair&& indices,
     cudf::detail::copy_if(cuda::counting_iterator<size_type>{0},
                           cuda::counting_iterator<size_type>{right_table_num_rows},
                           out_iter,
-                          unmatched_flag{flags.data()},
+                          unmatched_flag{match_flags},
                           stream);
 
   auto const comp_size = cuda::std::distance(out_iter, new_end);
@@ -214,6 +221,7 @@ VectorPair finalize_full_join(
   return finalize_full_join(std::pair(std::move(left_out), std::move(right_out)),
                             left_table_num_rows,
                             right_table_num_rows,
+                            std::nullopt,
                             stream,
                             mr);
 }

cpp/src/join/mixed_join.cu

@@ -102,8 +102,12 @@ mixed_join(table_view const& left_equality,
                                  std::nullopt,
                                  stream,
                                  mr);
-    return finalize_full_join(
-      std::move(left_outer), left_conditional.num_rows(), right_conditional.num_rows(), stream, mr);
+    return finalize_full_join(std::move(left_outer),
+                              left_conditional.num_rows(),
+                              right_conditional.num_rows(),
+                              std::nullopt,
+                              stream,
+                              mr);
   }
 
   auto const hash_joiner = cudf::hash_join{right_equality, compare_nulls, stream};

cpp/tests/join/join_tests.cpp

@@ -1775,6 +1775,24 @@ TEST_F(JoinTest, EmptyLeftTableFullJoin)
   auto sorted_gold     = cudf::gather(gold.view(), *gold_sort_order);
 
   CUDF_TEST_EXPECT_TABLES_EQUIVALENT(*sorted_gold, *sorted_result);
+
+  auto hash_joiner       = cudf::hash_join(rhs, cudf::null_equality::EQUAL);
+  auto const output_size = hash_joiner.full_join_size(lhs);
+  EXPECT_EQ(output_size, rhs.num_rows());
+
+  auto const [left_indices, right_indices] = hash_joiner.full_join(lhs, output_size);
+  EXPECT_EQ(left_indices->size(), output_size);
+  EXPECT_EQ(right_indices->size(), output_size);
+
+  column_wrapper<cudf::size_type> expected_left_indices{
+    {NoneValue, NoneValue, NoneValue, NoneValue, NoneValue}};
+  column_wrapper<cudf::size_type> expected_right_indices{{0, 1, 2, 3, 4}};
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(
+    expected_left_indices,
+    cudf::column_view{cudf::device_span<cudf::size_type const>{*left_indices}});
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(
+    expected_right_indices,
+    cudf::column_view{cudf::device_span<cudf::size_type const>{*right_indices}});
 }
 
 // Empty Right Table