Rewrite mixed inner/left/full join with post-filtering

cpp/CMakeLists.txt

@@ -879,12 +879,8 @@ add_library(
   src/join/key_remapping.cu
   src/join/mark_join.cu
   src/join/mixed_join.cu
-  src/join/mixed_join_kernel.cu
-  src/join/mixed_join_kernel_nulls.cu
   src/join/mixed_join_kernels_semi.cu
   src/join/mixed_join_semi.cu
-  src/join/mixed_join_size_kernel.cu
-  src/join/mixed_join_size_kernel_nulls.cu
   src/join/sort_merge_join.cu
   src/json/json_path.cu
   src/lists/contains.cu

cpp/include/cudf/detail/join/join.hpp

@@ -1,15 +1,51 @@
 /*
- * 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/ast/expressions.hpp>
+#include <cudf/join/join.hpp>
+#include <cudf/table/table_view.hpp>
 #include <cudf/types.hpp>
+#include <cudf/utilities/span.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_uvector.hpp>
+#include <rmm/resource_ref.hpp>
+
+#include <cstddef>
+#include <memory>
+#include <optional>
+#include <utility>
 
 namespace cudf {
 namespace detail {
 
 constexpr int DEFAULT_JOIN_CG_SIZE = 2;
 
+/**
+ * @brief Internal `filter_join_indices` accepting a precomputed output size.
+ *
+ * Same semantics as `cudf::filter_join_indices`. When `output_size` is provided it is used directly
+ * to size the output, skipping the internal size-counting pass. The value must equal the size that
+ * the function would otherwise compute (for example the result of `filter_join_indices_output_size`
+ * for the same inputs); behavior is undefined otherwise.
+ *
+ * @param output_size Optional precomputed number of output rows; computed internally if not
+ * provided
+ */
+std::pair<std::unique_ptr<rmm::device_uvector<size_type>>,
+          std::unique_ptr<rmm::device_uvector<size_type>>>
+filter_join_indices(table_view const& left,
+                    table_view const& right,
+                    device_span<size_type const> left_indices,
+                    device_span<size_type const> right_indices,
+                    ast::expression const& predicate,
+                    join_kind join_kind,
+                    std::optional<std::size_t> output_size,
+                    rmm::cuda_stream_view stream,
+                    rmm::device_async_resource_ref mr);
+
 }  // namespace detail
 }  // namespace cudf

cpp/include/cudf/join/join.hpp

@@ -17,7 +17,11 @@
 
 #include <cuda/std/limits>
 
+#include <cstddef>
 #include <cstdint>
+#include <memory>
+#include <optional>
+#include <utility>
 
 /**
  * @file
@@ -340,6 +344,9 @@ std::unique_ptr<cudf::table> cross_join(
  * @param right_indices Device span of row indices in the right table from hash join.
  * @param predicate An AST expression that returns a boolean for each pair of rows.
  * @param join_kind The type of join operation. Must be INNER_JOIN, LEFT_JOIN, or FULL_JOIN.
+ * @param output_size Optional precomputed number of output rows. When provided, skips the internal
+ *        size-counting pass. Behavior is undefined if it differs from the size the function would
+ *        otherwise produce for the same inputs.
  * @param stream CUDA stream used for kernel launches and memory operations.
  * @param mr Device memory resource used to allocate output indices.
  *
@@ -354,23 +361,57 @@ filter_join_indices(cudf::table_view const& left,
                     cudf::device_span<size_type const> right_indices,
                     cudf::ast::expression const& predicate,
                     cudf::join_kind join_kind,
-                    rmm::cuda_stream_view stream      = cudf::get_default_stream(),
+                    std::optional<std::size_t> output_size = std::nullopt,
+                    rmm::cuda_stream_view stream           = cudf::get_default_stream(),
+                    rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref());
+
+/**
+ * @brief Filters join result indices based on a conditional predicate and join type.
+ *
+ * @deprecated Use the overload that accepts an optional output size instead.
+ *
+ * @param left The left table for predicate evaluation (conditional columns only).
+ * @param right The right table for predicate evaluation (conditional columns only).
+ * @param left_indices Device span of row indices in the left table from hash join.
+ * @param right_indices Device span of row indices in the right table from hash join.
+ * @param predicate An AST expression that returns a boolean for each pair of rows.
+ * @param join_kind The type of join operation. Must be INNER_JOIN, LEFT_JOIN, or FULL_JOIN.
+ * @param stream CUDA stream used for kernel launches and memory operations.
+ * @param mr Device memory resource used to allocate output indices.
+ *
+ * @return A pair of device vectors [filtered_left_indices, filtered_right_indices]
+ *         corresponding to rows that satisfy the join semantics and predicate.
+ */
+[[deprecated("Use the overload that takes an optional output_size parameter.")]]
+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,
+                    cudf::table_view const& right,
+                    cudf::device_span<size_type const> left_indices,
+                    cudf::device_span<size_type const> right_indices,
+                    cudf::ast::expression const& predicate,
+                    cudf::join_kind join_kind,
+                    rmm::cuda_stream_view stream,
                     rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref());
 
 /**
  * @brief Returns the exact output size of `filter_join_indices` without materializing
  *        the filtered index vectors.
  *
  * Runs the same predicate evaluation as `filter_join_indices` but skips the index
- * materialization step, returning only the total number of pairs that would be
- * emitted. The semantics per `join_kind` match `filter_join_indices`:
- * - INNER_JOIN: number of pairs where the predicate evaluates to true.
- * - LEFT_JOIN: predicate-passing pairs plus one entry per left row with no passing match.
- * - FULL_JOIN: input pairs plus one extra entry per pair whose predicate failed
- *   (because failed matches split into `(left, JoinNoMatch)` and `(JoinNoMatch, right)`).
- *
- * The returned size may be passed as a precomputed hint to APIs that compose
- * `filter_join_indices` (for example, the mixed join APIs).
+ * materialization step, returning the total number of pairs that would be emitted along with the
+ * 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)`).
+ * - 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.
  *
  * @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.
@@ -383,17 +424,21 @@ filter_join_indices(cudf::table_view const& left,
  * @param predicate An AST expression that returns a boolean for each pair of rows.
  * @param join_kind The type of join operation. Must be INNER_JOIN, LEFT_JOIN, or FULL_JOIN.
  * @param stream CUDA stream used for kernel launches and memory operations.
+ * @param mr Device memory resource used to allocate the returned contribution counts.
  *
- * @return The exact number of pairs that `filter_join_indices` would produce.
+ * @return A pair containing the exact number of pairs that `filter_join_indices` would produce
+ *         and the per-output contribution counts that sum to that number.
  */
-[[nodiscard]] std::size_t filter_join_indices_output_size(
+[[nodiscard]] std::pair<std::size_t, std::unique_ptr<rmm::device_uvector<size_type>>>
+filter_join_indices_output_size(
   cudf::table_view const& left,
   cudf::table_view const& right,
   cudf::device_span<size_type const> left_indices,
   cudf::device_span<size_type const> right_indices,
   cudf::ast::expression const& predicate,
   cudf::join_kind join_kind,
-  rmm::cuda_stream_view stream = cudf::get_default_stream());
+  rmm::cuda_stream_view stream      = cudf::get_default_stream(),
+  rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref());
 
 /**
  * @brief JIT-based filtering of join result indices using string predicate.