Fix FULL_JOIN filtering through internal finalization

cpp/include/cudf/join/join.hpp

@@ -298,14 +298,16 @@ std::unique_ptr<cudf::table> cross_join(
  *
  * The behavior depends on the join type:
  * - INNER_JOIN: Only pairs that satisfy the predicate and have valid indices are kept.
- * - LEFT_JOIN: All left rows are preserved. Failed predicates nullify right indices.
- * - FULL_JOIN: All rows from both sides are preserved. Failed predicates create separate pairs.
+ * - LEFT_JOIN: Keeps passing pairs and adds exactly one unmatched pair for each left row
+ *   with no passing match.
+ * - FULL_JOIN: Keeps passing pairs and adds exactly one unmatched pair for each row on either
+ *   side with no passing match.
  *
- * Note on JoinNoMatch pairs: If an input pair already contains `JoinNoMatch` in either
- * position, the predicate cannot be evaluated and the pair passes through unchanged. The
- * "separate pairs" splitting only occurs when both indices are valid but the predicate fails.
- * For example, a FULL_JOIN pair `(5, 10)` that fails the predicate becomes two pairs:
- * `(5, JoinNoMatch)` and `(JoinNoMatch, 10)`, ensuring both rows appear in the output.
+ * The input maps must come from an equality join of the requested kind, with row indices
+ * corresponding to the supplied conditional tables. Null predicate results are nonmatches.
+ * Unmatched rows in valid outer-join maps are preserved without evaluating the predicate.
+ * A failed candidate does not create an unmatched row if another candidate for that row passes.
+ * Empty input maps produce empty output maps, without completing an outer join.
  *
  * ## Usage Pattern
  *
@@ -382,15 +384,17 @@ filter_join_indices(cudf::table_view const& left,
  * per-output contribution counts whose sum is that total. The counts are laid out per `join_kind`
  * so that each entry records how many output rows the corresponding input contributes:
  * - INNER_JOIN: indexed per input pair; entry `i` is `1` if the predicate passes and `0` otherwise.
- * - FULL_JOIN: indexed per input pair; entry `i` is `1` for a preserved pair (predicate passes or
- *   the pair already contains a `JoinNoMatch`) and `2` for a failed valid pair (which splits into
- *   `(left, JoinNoMatch)` and `(JoinNoMatch, right)`).
+ * - FULL_JOIN: one entry per left row followed by one per right row. Left entries count passing
+ *   valid pairs, floored to `1`; right entries are `1` for rows with no passing match and `0`
+ *   otherwise. Empty input maps return empty counts.
  * - LEFT_JOIN: indexed per left row; each entry holds the number of passing pairs for that left
  *   row, floored to `1` to account for the synthetic `(left, JoinNoMatch)` entry.
  *
  * The returned size and contribution counts may be passed as a precomputed hint to APIs that
  * compose `filter_join_indices` (for example, the mixed join APIs). The layout above is an
  * implementation detail that callers should treat as opaque rather than rely upon.
+ * The input-map contract is the same as `filter_join_indices`. Empty maps return zero size and
+ * empty counts.
  *
  * @throw std::invalid_argument if `join_kind` is not INNER_JOIN, LEFT_JOIN, or FULL_JOIN.
  * @throw std::invalid_argument if `left_indices` and `right_indices` have different sizes.
@@ -424,11 +428,14 @@ filter_join_indices_output_size(
  *
  * This function provides a JIT-compiled alternative to filter_join_indices(),
  * taking a string-based predicate that gets compiled to optimized GPU code.
+ * The input-map contract is the same as `filter_join_indices`.
  *
  * The behavior depends on the join type (same as filter_join_indices):
  * - INNER_JOIN: Only pairs that satisfy the predicate and have valid indices are kept.
- * - LEFT_JOIN: All left rows are preserved. Failed predicates nullify right indices.
- * - FULL_JOIN: All rows from both sides are preserved. Failed predicates create separate pairs.
+ * - LEFT_JOIN: Keeps passing pairs and adds exactly one unmatched pair for each left row
+ *   with no passing match.
+ * - FULL_JOIN: Keeps passing pairs and adds exactly one unmatched pair for each row on either
+ *   side with no passing match.
  *
  * ## Usage Pattern
  *
@@ -499,6 +506,9 @@ filter_join_indices_jit(
  *
  * This overload converts an AST expression referencing columns from both left and right
  * tables into JIT-compiled CUDA code and uses it to filter the join index pairs.
+ * The input-map contract and join semantics are the same as `filter_join_indices`.
+ *
+ * @throw std::invalid_argument if join_kind is not INNER_JOIN, LEFT_JOIN, or FULL_JOIN.
  *
  * @param left The left table for predicate evaluation
  * @param right The right table for predicate evaluation

cpp/src/join/filter_join_indices/filter_join_indices.cu

@@ -5,6 +5,7 @@
 
 #include "join/filter_join_indices/filter_join_indices_kernel.hpp"
 #include "join/filter_join_indices/filter_join_indices_output_size_kernel.hpp"
+#include "join/filter_join_indices/full_join.hpp"
 #include "join/join_common_utils.hpp"
 
 #include <cudf/ast/detail/expression_parser.hpp>
@@ -42,13 +43,36 @@
 #include <thrust/reduce.h>
 #include <thrust/transform.h>
 
+#include <limits>
 #include <memory>
 #include <optional>
 #include <utility>
 
 namespace cudf {
 namespace detail {
 
+VectorPair full_to_left_join_indices(device_span<size_type const> left_indices,
+                                     device_span<size_type const> right_indices,
+                                     cuda::stream_ref stream,
+                                     rmm::device_async_resource_ref mr)
+{
+  auto const keep = [left = left_indices.data()] __device__(std::size_t i) {
+    return left[i] != JoinNoMatch;
+  };
+  auto const begin  = cuda::counting_iterator<std::size_t>{0};
+  auto const size   = cudf::detail::count_if(begin, begin + left_indices.size(), keep, stream);
+  auto left_result  = std::make_unique<rmm::device_uvector<size_type>>(size, stream, mr);
+  auto right_result = std::make_unique<rmm::device_uvector<size_type>>(size, stream, mr);
+  if (size > 0) {
+    auto const input =
+      cuda::make_zip_iterator(cuda::std::tuple{left_indices.begin(), right_indices.begin()});
+    auto const output =
+      cuda::make_zip_iterator(cuda::std::tuple{left_result->begin(), right_result->begin()});
+    cudf::detail::copy_if_async(input, input + left_indices.size(), begin, output, keep, stream);
+  }
+  return {std::move(left_result), std::move(right_result)};
+}
+
 std::pair<std::unique_ptr<rmm::device_uvector<size_type>>,
           std::unique_ptr<rmm::device_uvector<size_type>>>
 filter_join_indices(cudf::table_view const& left,
@@ -78,6 +102,39 @@ filter_join_indices(cudf::table_view const& left,
 
   if (left_indices.empty()) { return make_empty_result(); }
 
+  if (join_kind == join_kind::FULL_JOIN) {
+    auto left_maps = full_to_left_join_indices(
+      left_indices, right_indices, stream, cudf::get_current_device_resource_ref());
+    auto filtered_left =
+      cudf::detail::filter_join_indices(left,
+                                        right,
+                                        device_span<size_type const>{*left_maps.first},
+                                        device_span<size_type const>{*left_maps.second},
+                                        predicate,
+                                        join_kind::LEFT_JOIN,
+                                        std::nullopt,
+                                        stream,
+                                        mr);
+    auto const unmatched_right_count = [&]() -> std::optional<size_type> {
+      if (!output_size.has_value()) { return std::nullopt; }
+      CUDF_EXPECTS(*output_size >= filtered_left.first->size(),
+                   "Precomputed FULL_JOIN output size is smaller than its LEFT_JOIN result",
+                   std::invalid_argument);
+      auto const count = *output_size - filtered_left.first->size();
+      CUDF_EXPECTS(count <= static_cast<std::size_t>(std::numeric_limits<size_type>::max()),
+                   "Precomputed FULL_JOIN output size exceeds the column size limit",
+                   std::overflow_error);
+      return static_cast<size_type>(count);
+    }();
+    return finalize_full_join(std::move(filtered_left),
+                              left.num_rows(),
+                              right.num_rows(),
+                              std::nullopt,
+                              stream,
+                              mr,
+                              unmatched_right_count);
+  }
+
   // Check if predicate may evaluate to null
   auto const has_nulls = predicate.may_evaluate_null(left, right, stream);
 
@@ -160,7 +217,6 @@ filter_join_indices(cudf::table_view const& left,
 
   auto predicate_results_ptr = predicate_results.data();
   auto left_ptr              = left_indices.data();
-  auto right_ptr             = right_indices.data();
 
   auto make_result_vectors = [&](std::size_t size) {
     return std::pair{std::make_unique<rmm::device_uvector<size_type>>(size, stream, mr),
@@ -304,69 +360,6 @@ filter_join_indices(cudf::table_view const& left,
 
     return std::pair{std::move(filtered_left_indices), std::move(filtered_right_indices)};
 
-  } else if (join_kind == join_kind::FULL_JOIN) {
-    // FULL_JOIN: Optimized implementation using stream compaction
-    // Strategy: Use a single scan to identify failed matches, then use stream compaction
-
-    // First, identify failed matched pairs
-    auto is_failed_matched_pair = [=] __device__(size_type i) -> bool {
-      return !predicate_results_ptr[i] && left_ptr[i] != JoinNoMatch && right_ptr[i] != JoinNoMatch;
-    };
-
-    // Count failed matches for output sizing
-    auto const failed_matched_count =
-      output_size.has_value()
-        ? *output_size - left_indices.size()
-        : cudf::detail::count_if(
-            cuda::counting_iterator<cudf::size_type>{0},
-            cuda::counting_iterator{static_cast<size_type>(left_indices.size())},
-            is_failed_matched_pair,
-            stream);
-    auto const result_size = left_indices.size() + failed_matched_count;
-
-    if (result_size == 0) { return make_empty_result(); }
-
-    auto [filtered_left_indices, filtered_right_indices] = make_result_vectors(result_size);
-
-    // Use two-step approach with optimized memory management
-    // Step 1: Handle primary pairs
-    thrust::transform(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()),
-                      cuda::counting_iterator<cudf::size_type>{0},
-                      cuda::counting_iterator{static_cast<size_type>(left_indices.size())},
-                      cuda::make_zip_iterator(cuda::std::tuple{filtered_left_indices->begin(),
-                                                               filtered_right_indices->begin()}),
-                      [=] __device__(size_type i) -> cuda::std::tuple<size_type, size_type> {
-                        auto const left_idx  = left_ptr[i];
-                        auto const right_idx = right_ptr[i];
-                        // For FULL JOIN: preserve original unmatched rows, nullify right side of
-                        // failed matches
-                        auto const output_right_idx =
-                          (predicate_results_ptr[i] || left_idx == JoinNoMatch) ? right_idx
-                                                                                : JoinNoMatch;
-
-                        return cuda::std::tuple{left_idx, output_right_idx};
-                      });
-
-    // Step 2: Add secondary pairs for failed matches using stream compaction
-    if (failed_matched_count > 0) {
-      auto secondary_iter = cuda::make_zip_iterator(
-        cuda::std::tuple{filtered_left_indices->begin() + left_indices.size(),
-                         filtered_right_indices->begin() + left_indices.size()});
-
-      auto failed_match_iter = cudf::detail::make_counting_transform_iterator(
-        0, [=] __device__(size_type i) -> cuda::std::tuple<size_type, size_type> {
-          return cuda::std::tuple{JoinNoMatch, right_ptr[i]};
-        });
-      cudf::detail::copy_if_async(failed_match_iter,
-                                  failed_match_iter + left_indices.size(),
-                                  cuda::counting_iterator<cudf::size_type>{0},
-                                  secondary_iter,
-                                  is_failed_matched_pair,
-                                  stream);
-    }
-
-    return std::pair{std::move(filtered_left_indices), std::move(filtered_right_indices)};
-
   } else {
     CUDF_FAIL("Unsupported join kind for filter_join_indices");
   }
@@ -416,11 +409,13 @@ filter_join_indices_output_size(cudf::table_view const& left,
   detail::grid_1d const config(left_indices.size(), DEFAULT_JOIN_BLOCK_SIZE);
   auto const shmem_per_block = parser.shmem_per_thread * DEFAULT_JOIN_BLOCK_SIZE;
 
-  auto const counts_size = join_kind == join_kind::LEFT_JOIN
-                             ? static_cast<std::size_t>(left.num_rows())
-                             : left_indices.size();
+  auto const counts_size =
+    join_kind == join_kind::FULL_JOIN
+      ? static_cast<std::size_t>(left.num_rows()) + right.num_rows()
+      : (join_kind == join_kind::LEFT_JOIN ? static_cast<std::size_t>(left.num_rows())
+                                           : left_indices.size());
   auto output_counts =
-    join_kind == join_kind::LEFT_JOIN
+    join_kind != join_kind::INNER_JOIN
       ? cudf::detail::make_zeroed_device_uvector_async<size_type>(counts_size, stream, mr)
       : rmm::device_uvector<size_type>(counts_size, stream, mr);
 
@@ -450,6 +445,19 @@ filter_join_indices_output_size(cudf::table_view const& left,
                         [] __device__(size_type count) { return count > 0 ? count : 1; }));
   }
 
+  if (join_kind == join_kind::FULL_JOIN) {
+    auto const counts    = output_counts.data();
+    auto const left_size = static_cast<std::size_t>(left.num_rows());
+    thrust::transform(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()),
+                      cuda::counting_iterator<std::size_t>{0},
+                      cuda::counting_iterator<std::size_t>{counts_size},
+                      output_counts.begin(),
+                      [=] __device__(std::size_t i) -> size_type {
+                        return i < left_size ? (counts[i] > 0 ? counts[i] : 1)
+                                             : (counts[i] == 0 ? 1 : 0);
+                      });
+  }
+
   std::size_t const total =
     thrust::reduce(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()),
                    output_counts.begin(),