Forward-merge release/26.08 into main

ci/run_cudf_polars_polars_tests.sh

@@ -62,9 +62,9 @@ DESELECTED_TESTS_STR=$(printf -- " --deselect %s" "${DESELECTED_TESTS[@]}")
 # Don't quote the `DESELECTED_...` variable because `pytest` can't handle
 # multiple quoted arguments inline
 # shellcheck disable=SC2086
-# Fail fast (-x) because failed tests pollute the state
+# Fail fast (-x) rather than trying to continue because failed tests pollute the state
 echo "Run polars tests with injected in-memory GPU engine"
-python "${TIMEOUT_TOOL_PATH}" --enable-python 3600 \
+python "${TIMEOUT_TOOL_PATH}" --enable-python 5400 \
    python -m pytest \
        --import-mode=importlib \
        --cache-clear \
@@ -85,12 +85,11 @@ python "${TIMEOUT_TOOL_PATH}" --enable-python 3600 \
 echo "Run polars tests with injected SPMD GPU engine, small blocksize"
 CUDF_POLARS__EXECUTOR__TARGET_PARTITION_SIZE=805306368 \
 CUDF_POLARS__EXECUTOR__FALLBACK_MODE=silent \
-python "${TIMEOUT_TOOL_PATH}" --enable-python 3600 \
+python "${TIMEOUT_TOOL_PATH}" --enable-python 5400 \
    python -m pytest \
        --import-mode=importlib \
        --cache-clear \
        -x \
-       -v \
        -m "" \
        -p cudf_polars.testing.inject_gpu_engine \
        -W ignore::ResourceWarning \

ci/run_cudf_polars_pytests.sh

@@ -10,5 +10,5 @@ TIMEOUT_TOOL_PATH="$(dirname "$(realpath "${BASH_SOURCE[0]}")")"/timeout_with_st
 
 cd "$(dirname "$(realpath "${BASH_SOURCE[0]}")")"/../python/cudf_polars/
 
-python "${TIMEOUT_TOOL_PATH}" --enable-python 3600 \
+python "${TIMEOUT_TOOL_PATH}" --enable-python 5400 \
        python -m pytest --cache-clear "$@" tests

ci/test_python_other.sh

@@ -1,5 +1,5 @@
 #!/bin/bash
-# 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
 
 set -euo pipefail
@@ -42,8 +42,9 @@ timeout 30m ./ci/run_custreamz_pytests.sh \
   --cov-report=term
 
 rapids-logger "pytest cudf-polars"
+# Fail fast (-x) rather than trying to continue because failed tests pollute the state
 ./ci/run_cudf_polars_pytests.sh \
-  -vv \
+  -x \
   --junitxml="${RAPIDS_TESTS_DIR}/junit-cudf-polars.xml" \
   --numprocesses=4 \
   --dist=worksteal \

ci/test_wheel_cudf_polars.sh

@@ -91,7 +91,6 @@ for version in "${VERSIONS[@]}"; do
 
     # Fail fast (-x) rather than trying to continue because failed tests pollute the state
     ./ci/run_cudf_polars_pytests.sh \
-        -vv \
         "${COVERAGE_ARGS[@]}" \
         --numprocesses=4 \
         --dist=worksteal \

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
@@ -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/io/experimental/deletion_vectors.hpp

@@ -48,6 +48,9 @@ struct deletion_vector_info {
   std::vector<size_t> row_group_offsets;
   /// Number of rows in each row group to be read from the Parquet source(s)
   std::vector<size_type> row_group_num_rows;
+
+  /// Whether the roaring bitmaps represent retention vectors
+  bool are_retention_vectors = false;
 };
 
 /**
@@ -147,6 +150,7 @@ class chunked_parquet_reader {
   std::queue<size_type> _deletion_vector_row_counts;
   size_t _start_row;
   bool _is_unspecified_row_group_data;
+  bool _are_retentions;
   rmm::cuda_stream_view _stream;
   rmm::device_async_resource_ref _mr;
   rmm::device_async_resource_ref _table_mr;

cpp/include/cudf/io/experimental/variant.hpp

@@ -29,34 +29,33 @@ namespace io::parquet::experimental {
  */
 
 /**
- * @brief Extract the raw VARIANT-encoded bytes of a nested object field by JSONPath-like path.
- *
- * Walks `path` step by step, descending into object values (`basic_type == 2`) at each name step.
- * Returns a `list<uint8>` column containing the raw encoded bytes of the value at the end of
- * the path for each row.
- *
- * Null is produced when the struct row is null, a name step's key is absent from the dictionary,
- * or the current value is not an object (`basic_type != 2`).
+ * @brief Extract the raw VARIANT-encoded bytes of a nested field by JSONPath-like path.
  *
  * Path grammar:
- *   path  := "$"? first_step ("." name)*
- *   first := name | "." name
- *   name  := [^.\[]+   // any byte except '.' (step separator) and '[' (reserved)
+ *   path  := "$"? first_step step*
+ *   first := name | "." name | "[" index "]"
+ *   step  := "." name | "[" index "]"
+ *   name  := any sequence of bytes other than '.' or '['
+ *   index := non-negative base-10 integer (leading zeros are allowed, e.g. "[01]" == "[1]")
  *
  * Examples:
- *   "x"          -> top-level field "x" (leading $ optional)
- *   "$.foo"      -> top-level field "foo"
- *   "$.foo.bar"  -> object descent foo -> bar
+ *   "x"            -> top-level field "x" (leading $ optional)
+ *   "$.foo"        -> top-level field "foo"
+ *   "$.foo.bar"    -> object descent foo -> bar
+ *   "$[0]"         -> first element of a top-level array
+ *   "$.a[0].b"     -> object key "a" -> first array element -> object key "b"
  *
  * @param variant_column Struct column (VARIANT materialization) with `list<uint8>` children
  *                       (`metadata`, `value`), plus optional shredded siblings
- * @param path JSONPath-like path string identifying the target object field
+ * @param path JSONPath-like path string identifying the target field
  * @param stream CUDA stream
  * @param mr Device memory resource
- * @return `list<uint8>` column with the extracted field's encoded bytes
+ * @return `list<uint8>` column with the extracted value's encoded bytes. A row is null when the
+ *         input row is null, a name is absent, an index is out of bounds, or a step does not match
+ *         the current value.
  *
- * @throws std::invalid_argument on empty path or malformed syntax (including bracket steps,
- *         which require array-indexing support that is not yet implemented)
+ * @throws std::invalid_argument on empty path or malformed syntax (`[*]` wildcards, negative
+ *         indices, out-of-range indices, and quoted names inside `[...]` are not supported)
  */
 [[nodiscard]] std::unique_ptr<column> get_variant_field(
   column_view const& variant_column,

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/include/cudf/reduction/bloom_filter.cuh

@@ -0,0 +1,27 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <cuco/bloom_filter_policies.cuh>
+
+#include <cstdint>
+
+namespace cudf {
+
+/**
+ * @brief Policy describing the Apache Arrow Block-Split Bloom Filter layout.
+ *
+ * Uses cuco's `parametric_filter_policy` with the Apache Arrow layout: 256-bit blocks (8 x
+ * `uint32_t`), 8 fingerprint bits per key, fully horizontal add (Theta=8), and fully vertical
+ * contains (Phi=8). This layout is bit-compatible with Apache Arrow.
+ *
+ * @tparam Hash The hash function used to generate a hash for each key.
+ */
+template <typename Hash>
+using arrow_filter_policy =
+  cuco::parametric_filter_policy<Hash, std::uint32_t, 8, 8, 8, 1, 1, 8, false, false>;
+
+}  // namespace cudf

cpp/libcudf_streaming/benchmarks/streaming/ndsh/q03.cpp

@@ -357,8 +357,8 @@ int main(int argc, char** argv)
   int device;
   RAPIDSMPF_CUDA_TRY(cudaGetDevice(&device));
   RAPIDSMPF_CUDA_TRY(cudaDeviceGetAttribute(&l2size, cudaDevAttrL2CacheSize, device));
-  auto const num_filter_blocks =
-    cudf_streaming::bloom_filter::fitting_num_blocks(static_cast<std::size_t>(l2size));
+  auto const filter_size =
+    cudf_streaming::bloom_filter::aligned_size(static_cast<std::size_t>(l2size) * 2 / 3);
 
   for (int i = 0; i < arguments.num_iterations; i++) {
     int op_id{0};
@@ -406,7 +406,7 @@ int main(int argc, char** argv)
       actors.push_back(fanout_bounded(
         ctx, comm, customer_x_orders, bloom_filter_input, {0}, customer_x_orders_input));
       auto bloom_filter =
-        cudf_streaming::bloom_filter(ctx, comm, cudf::DEFAULT_HASH_SEED, num_filter_blocks);
+        cudf_streaming::bloom_filter(ctx, comm, cudf::DEFAULT_HASH_SEED, filter_size);
       actors.push_back(bloom_filter.build(
         bloom_filter_input, bloom_filter_output, static_cast<rapidsmpf::OpID>(10 * i + op_id++)));
       // Out: l_orderkey, l_extendedprice, l_discount

cpp/libcudf_streaming/benchmarks/streaming/ndsh/q04.cpp

@@ -256,8 +256,8 @@ int main(int argc, char** argv)
   int device;
   RAPIDSMPF_CUDA_TRY(cudaGetDevice(&device));
   RAPIDSMPF_CUDA_TRY(cudaDeviceGetAttribute(&l2size, cudaDevAttrL2CacheSize, device));
-  auto const num_filter_blocks =
-    cudf_streaming::bloom_filter::fitting_num_blocks(static_cast<std::size_t>(l2size));
+  auto const filter_size =
+    cudf_streaming::bloom_filter::aligned_size(static_cast<std::size_t>(l2size) * 2 / 3);
 
   for (int i = 0; i < arguments.num_iterations; i++) {
     rapidsmpf::OpID op_id{0};
@@ -302,7 +302,7 @@ int main(int argc, char** argv)
       // Build bloom filter from filtered orders' o_orderkey
       auto bloom_filter_output = ctx->create_channel();
       auto bloom_filter =
-        cudf_streaming::bloom_filter(ctx, comm, cudf::DEFAULT_HASH_SEED, num_filter_blocks);
+        cudf_streaming::bloom_filter(ctx, comm, cudf::DEFAULT_HASH_SEED, filter_size);
       actors.push_back(bloom_filter.build(
         bloom_filter_input, bloom_filter_output, static_cast<rapidsmpf::OpID>(10 * i + op_id++)));