Merge release/26.08 into main

cpp/CMakeLists.txt

@@ -881,12 +881,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
@@ -1420,7 +1416,7 @@ if(CUDF_BUILD_TESTUTIL)
   )
 
   target_link_libraries(
-    cudftestutil INTERFACE $<BUILD_LOCAL_INTERFACE:cuco::cuco> Threads::Threads cudf
+    cudftestutil INTERFACE $<BUILD_LOCAL_INTERFACE:cuco::cuco> Threads::Threads cudf rmm::rmm
                            cudftest_default_stream $<TARGET_NAME_IF_EXISTS:conda_env>
   )
 

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.

cpp/libcudf_streaming/include/cudf_streaming/detail/device_bloom_filter.hpp

@@ -19,6 +19,10 @@ namespace cudf_streaming::detail {
 
 /**
  * @brief A bloom filter, used for approximate set membership queries.
+ *
+ * @note All methods of this class launch work on the streams provided. It is the caller's
+ * responsibility to ensure that data are valid to read/write on that stream as
+ * appropriate.
  */
 struct device_bloom_filter {
   /**
@@ -28,27 +32,21 @@ struct device_bloom_filter {
    * @param seed Seed used for hashing each value.
    * @param storage Storage to view as a bloom filter, must be appropriately
    * initialized.
-   * @param stream CUDA stream for device operations.
    */
-  device_bloom_filter(std::size_t num_blocks,
-                      std::uint64_t seed,
-                      void* storage,
-                      rmm::cuda_stream_view stream);
+  device_bloom_filter(std::size_t num_blocks, std::uint64_t seed, void* storage);
 
   /**
    * @brief Create a read-only filter.
    *
    * @param num_blocks Number of blocks in the filter.
    * @param seed Seed used for hashing each value.
    * @param storage View of storage, must be appropriately initialized.
-   * @param stream CUDA stream for device operations.
    *
    * @return A const-qualified bloom filter viewing the underlying storage.
    */
   static device_bloom_filter const view(std::size_t num_blocks,
                                         std::uint64_t seed,
-                                        void const* storage,
-                                        rmm::cuda_stream_view stream);
+                                        void const* storage);
 
   /**
    * @brief Create uninitialized storage for a filter.
@@ -98,11 +96,6 @@ struct device_bloom_filter {
                                                    rmm::cuda_stream_view stream,
                                                    rmm::device_async_resource_ref mr) const;
 
-  /**
-   * @brief @return The stream the underlying storage is valid on.
-   */
-  [[nodiscard]] rmm::cuda_stream_view stream() const noexcept;
-
   /**
    * @brief @return Pointer to the underlying storage.
    */
@@ -127,10 +120,9 @@ struct device_bloom_filter {
   [[nodiscard]] static std::size_t fitting_num_blocks(std::size_t l2size) noexcept;
 
  private:
-  std::size_t num_blocks_;        ///< Number of blocks used in the filter.
-  std::uint64_t seed_;            ///< Seed used when hashing values.
-  void* storage_;                 ///< Backing storage.
-  rmm::cuda_stream_view stream_;  ///< Stream storage is valid on.
+  std::size_t num_blocks_;  ///< Number of blocks used in the filter.
+  std::uint64_t seed_;      ///< Seed used when hashing values.
+  void* storage_;           ///< Backing storage.
 };
 
 }  // namespace cudf_streaming::detail

cpp/libcudf_streaming/src/bloom_filter.cpp

@@ -37,8 +37,8 @@ rapidsmpf::streaming::Actor bloom_filter::build(
   auto storage =
     cudf_streaming::detail::device_bloom_filter::storage(num_filter_blocks_, filter_stream, mr);
   RAPIDSMPF_CUDA_TRY(cudaMemsetAsync(storage->data(), 0, storage->size(), filter_stream));
-  auto filter = cudf_streaming::detail::device_bloom_filter(
-    num_filter_blocks_, seed_, storage->data(), filter_stream);
+  auto filter =
+    cudf_streaming::detail::device_bloom_filter(num_filter_blocks_, seed_, storage->data());
   rapidsmpf::CudaEvent build_event;
   build_event.record(filter_stream);
   while (!ch_out->is_shutdown()) {
@@ -48,6 +48,14 @@ rapidsmpf::streaming::Actor bloom_filter::build(
     chunk      = co_await chunk.make_available(
       ctx_,
       -rapidsmpf::safe_cast<std::int64_t>(chunk.data_alloc_size(rapidsmpf::MemoryType::DEVICE)));
+
+    // Reservation for the hash values in add.
+    auto res = co_await ctx_->memory(rapidsmpf::MemoryType::DEVICE)
+                 ->reserve_or_wait(rapidsmpf::safe_cast<std::size_t>(chunk.table_view().num_rows())
+                                     // TODO: no magic numbers: the hashing algorithm in
+                                     // `add` below returns an int64 column.
+                                     * sizeof(std::int64_t),
+                                   0);
     // Filter is allocated on `filter_stream`, but we run the additions on the chunk's
     // stream. The addition modifies global memory but we can safely launch two
     // kernels doing that concurrently because the updates are atomic.
@@ -67,9 +75,9 @@ rapidsmpf::streaming::Actor bloom_filter::build(
       [num_blocks = num_filter_blocks_, seed = seed_](rapidsmpf::Buffer const* left,
                                                       rapidsmpf::Buffer* right) {
         right->write_access([&](std::byte* out_bytes, rmm::cuda_stream_view stream) {
-          auto const in = cudf_streaming::detail::device_bloom_filter::view(
-            num_blocks, seed, left->data(), stream);
-          cudf_streaming::detail::device_bloom_filter(num_blocks, seed, out_bytes, stream)
+          auto const in =
+            cudf_streaming::detail::device_bloom_filter::view(num_blocks, seed, left->data());
+          cudf_streaming::detail::device_bloom_filter(num_blocks, seed, out_bytes)
             .merge(in, stream);
         });
       });
@@ -95,7 +103,7 @@ rapidsmpf::streaming::Actor bloom_filter::apply(
   auto stream = storage.stream();
   rapidsmpf::CudaEvent event;
   auto filter =
-    cudf_streaming::detail::device_bloom_filter(num_filter_blocks_, seed_, storage.data(), stream);
+    cudf_streaming::detail::device_bloom_filter(num_filter_blocks_, seed_, storage.data());
   auto meta = co_await ch_in->receive_metadata();
   if (!meta.empty()) { co_await ch_out->send_metadata(std::move(meta)); }
   while (!ch_out->is_shutdown()) {

cpp/libcudf_streaming/src/detail/device_bloom_filter.cu

@@ -30,7 +30,6 @@
 
 #include <cudf/hashing.hpp>
 #include <cudf/table/table_view.hpp>
-#include <cudf/utilities/memory_resource.hpp>
 
 #include <cudf_streaming/detail/device_bloom_filter.hpp>
 
@@ -72,35 +71,29 @@ using StorageType = BloomFilterRefType::filter_block_type;
 
 }  // namespace
 
-device_bloom_filter::device_bloom_filter(std::size_t num_blocks,
-                                         std::uint64_t seed,
-                                         void* storage,
-                                         rmm::cuda_stream_view stream)
-  : num_blocks_{num_blocks}, seed_{seed}, storage_{storage}, stream_{stream}
+device_bloom_filter::device_bloom_filter(std::size_t num_blocks, std::uint64_t seed, void* storage)
+  : num_blocks_{num_blocks}, seed_{seed}, storage_{storage}
 {
-  // TODO: use an aligned allocator adaptor to ensure this holds.
-  // Today all RMM device allocators guarantee at least 256 byte alignment, but that is
-  // an implementation detail.
   RAPIDSMPF_EXPECTS(
     reinterpret_cast<std::uintptr_t>(storage_) % std::alignment_of_v<StorageType> == 0,
     "Allocation for bloom filter is not aligned.");
 }
 
 device_bloom_filter const device_bloom_filter::view(std::size_t num_blocks,
                                                     std::uint64_t seed,
-                                                    void const* storage,
-                                                    rmm::cuda_stream_view stream)
+                                                    void const* storage)
 {
   // const-cast is safe because the returned object is also const and therefore can't
   // call methods that throw away constness.
-  return device_bloom_filter(num_blocks, seed, const_cast<void*>(storage), stream);
+  return device_bloom_filter(num_blocks, seed, const_cast<void*>(storage));
 }
 
 std::unique_ptr<rmm::device_buffer> device_bloom_filter::storage(std::size_t num_blocks,
                                                                  rmm::cuda_stream_view stream,
                                                                  rmm::device_async_resource_ref mr)
 {
-  return std::make_unique<rmm::device_buffer>(num_blocks * sizeof(StorageType), stream, mr);
+  return std::make_unique<rmm::device_buffer>(
+    num_blocks * sizeof(StorageType), std::alignment_of_v<StorageType>, stream, mr);
 }
 
 void device_bloom_filter::add(cudf::table_view const& values_to_hash,
@@ -110,8 +103,7 @@ void device_bloom_filter::add(cudf::table_view const& values_to_hash,
   RAPIDSMPF_NVTX_FUNC_RANGE();
   auto filter_ref = BloomFilterRefType{
     static_cast<StorageType*>(storage_), num_blocks_, cuco::thread_scope_device, {}};
-  auto hashes = cudf::hashing::xxhash_64(
-    values_to_hash, seed_, stream, cudf::get_current_device_resource_ref());
+  auto hashes    = cudf::hashing::xxhash_64(values_to_hash, seed_, stream, mr);
   auto hash_view = hashes->view();
   RAPIDSMPF_EXPECTS(hash_view.type().id() == cudf::type_to_id<KeyType>(),
                     "Hash values do not have correct type");
@@ -136,9 +128,8 @@ rmm::device_uvector<bool> device_bloom_filter::contains(cudf::table_view const&
   RAPIDSMPF_NVTX_FUNC_RANGE();
   auto filter_ref = BloomFilterRefType{
     static_cast<StorageType*>(storage_), num_blocks_, cuco::thread_scope_device, {}};
-  auto hashes =
-    cudf::hashing::xxhash_64(values, seed_, stream, cudf::get_current_device_resource_ref());
-  auto view = hashes->view();
+  auto hashes = cudf::hashing::xxhash_64(values, seed_, stream, mr);
+  auto view   = hashes->view();
   rmm::device_uvector<bool> result{static_cast<std::size_t>(view.size()), stream, mr};
   filter_ref.contains_async(view.begin<KeyType>(), view.end<KeyType>(), result.begin(), stream);
   return result;
@@ -149,8 +140,6 @@ std::size_t device_bloom_filter::fitting_num_blocks(std::size_t l2size) noexcept
   return (l2size * 2) / (3 * sizeof(StorageType));
 }
 
-rmm::cuda_stream_view device_bloom_filter::stream() const noexcept { return stream_; }
-
 void* device_bloom_filter::data() noexcept { return storage_; }
 
 void const* device_bloom_filter::data() const noexcept { return storage_; }