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

@@ -1417,7 +1417,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/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/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++)));
 

cpp/libcudf_streaming/benchmarks/streaming/ndsh/q21.cpp

@@ -499,8 +499,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};
     std::vector<rapidsmpf::streaming::Actor> actors;
@@ -670,7 +670,7 @@ int main(int argc, char** argv)
                                             rapidsmpf::streaming::actor::FanoutPolicy::UNBOUNDED));
       auto bloom_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);
       // Select the relevant key column(s) and build filter.
       actors.push_back(populate_bloom_filter(ctx,
                                              comm,

cpp/libcudf_streaming/include/cudf_streaming/bloom_filter.hpp

@@ -43,18 +43,25 @@ struct bloom_filter {
    * @param ctx Streaming context.
    * @param comm Communicator for the collective operation.
    * @param seed Hash seed used when hashing values into the filter.
-   * @param num_filter_blocks Number of blocks in the filter.
+   * @param filter_size Filter storage size in bytes. Must be positive and satisfy
+   * `aligned_size(filter_size) == filter_size`, and must not exceed the maximum size supported by
+   * the filter policy.
+   *
+   * @throws std::logic_error If `filter_size` is zero, incorrectly aligned, or exceeds the policy
+   * maximum.
    */
   explicit bloom_filter(std::shared_ptr<rapidsmpf::streaming::Context> ctx,
                         std::shared_ptr<rapidsmpf::Communicator> comm,
                         std::uint64_t seed,
-                        std::size_t num_filter_blocks) noexcept
-    : ctx_{std::move(ctx)},
-      comm_{std::move(comm)},
-      seed_{seed},
-      num_filter_blocks_{num_filter_blocks}
-  {
-  }
+                        std::size_t filter_size);
+
+  /**
+   * @brief Find the largest valid filter size no greater than a byte count.
+   *
+   * @param size Byte count to align.
+   * @return Largest valid filter size less than or equal to `size`.
+   */
+  [[nodiscard]] static std::size_t aligned_size(std::size_t size) noexcept;
 
   /**
    * @brief Gets the communicator associated with this bloom_filter.
@@ -100,22 +107,10 @@ struct bloom_filter {
     std::shared_ptr<rapidsmpf::streaming::Channel> ch_out,
     std::vector<cudf::size_type> keys);
 
-  /**
-   * @brief Compute number of filter blocks that fit in the given L2 cache size.
-   *
-   * @param l2size L2 cache size in bytes.
-   * @return Number of filter blocks that fit.
-   */
-  [[nodiscard]] static std::size_t fitting_num_blocks(std::size_t l2size) noexcept
-  {
-    using StorageType = std::uint32_t;
-    return (l2size * 2) / (3 * sizeof(StorageType));
-  }
-
  private:
   std::shared_ptr<rapidsmpf::streaming::Context> ctx_{};
   std::shared_ptr<rapidsmpf::Communicator> comm_{};
   std::uint64_t seed_{};
-  std::size_t num_filter_blocks_{};
+  std::size_t filter_size_{};
 };
 }  // namespace cudf_streaming

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

@@ -28,40 +28,58 @@ struct device_bloom_filter {
   /**
    * @brief Create a filter.
    *
-   * @param num_blocks Number of blocks in the filter.
+   * @param filter_size Filter storage size in bytes. Must be a positive multiple of the filter
+   * block size and no greater than the maximum supported by the filter policy.
    * @param seed Seed used for hashing each value.
    * @param storage Storage to view as a bloom filter, must be appropriately
    * initialized.
    */
-  device_bloom_filter(std::size_t num_blocks, std::uint64_t seed, void* storage);
+  device_bloom_filter(std::size_t filter_size, std::uint64_t seed, void* storage);
 
   /**
    * @brief Create a read-only filter.
    *
-   * @param num_blocks Number of blocks in the filter.
+   * @param filter_size Filter storage size in bytes. Must be a positive multiple of the filter
+   * block size and no greater than the maximum supported by the filter policy.
    * @param seed Seed used for hashing each value.
    * @param storage View of storage, must be appropriately initialized.
    *
    * @return A const-qualified bloom filter viewing the underlying storage.
    */
-  static device_bloom_filter const view(std::size_t num_blocks,
+  static device_bloom_filter const view(std::size_t filter_size,
                                         std::uint64_t seed,
                                         void const* storage);
 
   /**
    * @brief Create uninitialized storage for a filter.
    *
-   * @param num_blocks Number of blocks in the filter.
+   * @param filter_size Filter storage size in bytes. Must be a positive multiple of the filter
+   * block size and no greater than the maximum supported by the filter policy.
    * @param stream CUDA stream for device operations.
    * @param mr Memory resource for allocations.
    *
    * @return Unique pointer to a device buffer containing storage for the requested
-   * number of filter blocks.
+   * filter size.
    */
-  static std::unique_ptr<rmm::device_buffer> storage(std::size_t num_blocks,
+  static std::unique_ptr<rmm::device_buffer> storage(std::size_t filter_size,
                                                      rmm::cuda_stream_view stream,
                                                      rmm::device_async_resource_ref mr);
 
+  /**
+   * @brief Find the largest valid filter size no greater than a byte count.
+   *
+   * @param size Byte count to align.
+   * @return Largest valid filter size less than or equal to `size`.
+   */
+  [[nodiscard]] static std::size_t aligned_size(std::size_t size) noexcept;
+
+  /**
+   *  Return the largest storage size supported by the filter policy.
+   *
+   *  Maximum valid filter size in bytes.
+   */
+  [[nodiscard]] static std::size_t max_size() noexcept;
+
   /**
    * @brief Add values to the filter.
    *
@@ -111,14 +129,6 @@ struct device_bloom_filter {
    */
   [[nodiscard]] std::size_t size() const noexcept;
 
-  /**
-   * @brief @return Number of blocks to use if the filter should fit in a given L2 cache
-   * size.
-   *
-   * @param l2size Size of the L2 cache in bytes.
-   */
-  [[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.