Parquet statistics, dictionary and page-index filter fixes

cpp/benchmarks/io/parquet/experimental/hybrid_scan/dict_page_filter.cpp

@@ -1,5 +1,5 @@
 /*
- * 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
  */
 
@@ -85,8 +85,8 @@ void BM_filter_string_row_groups_with_dicts_common(nvbench::state& state,
       timer.start();
 
       // Get dictionary page byte ranges
-      dict_page_byte_ranges =
-        std::get<1>(reader->secondary_filters_byte_ranges(input_row_group_indices, read_opts));
+      dict_page_byte_ranges = cudf::io::parquet::experimental::dictionary_page_byte_ranges_to_read(
+        std::get<1>(reader->secondary_filters_byte_ranges(input_row_group_indices, read_opts)));
       CUDF_EXPECTS(not dict_page_byte_ranges.empty(), "No dictionary page byte ranges found");
 
       // Fetch dictionary page data

cpp/benchmarks/io/parquet/experimental/hybrid_scan/hybrid_scan_composer.cpp

@@ -1,6 +1,6 @@
 
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -74,8 +74,11 @@ std::vector<cudf::size_type> apply_row_group_filters(
 
   if (filters.contains(hybrid_scan_filter_type::ROW_GROUPS_WITH_DICT_PAGES) or
       filters.contains(hybrid_scan_filter_type::ROW_GROUPS_WITH_BLOOM_FILTERS)) {
-    std::tie(bloom_filter_byte_ranges, dict_page_byte_ranges) =
+    auto dict_page_ranges = std::vector<cudf::io::parquet::experimental::dictionary_page_range>{};
+    std::tie(bloom_filter_byte_ranges, dict_page_ranges) =
       reader.secondary_filters_byte_ranges(current_row_group_indices, options);
+    dict_page_byte_ranges =
+      cudf::io::parquet::experimental::dictionary_page_byte_ranges_to_read(dict_page_ranges);
   } else {
     return std::vector<cudf::size_type>(current_row_group_indices.begin(),
                                         current_row_group_indices.end());

cpp/examples/hybrid_scan_io/hybrid_scan_composer.cpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -136,8 +136,11 @@ std::vector<cudf::size_type> apply_row_group_filters(
       filters.contains(hybrid_scan_filter_type::ROW_GROUPS_WITH_BLOOM_FILTERS)) {
     if (verbose) { std::cout << "READER: Get bloom filter and dictionary page byte ranges...\n"; }
     timer.reset();
-    std::tie(bloom_filter_byte_ranges, dict_page_byte_ranges) =
+    auto dict_page_ranges = std::vector<cudf::io::parquet::experimental::dictionary_page_range>{};
+    std::tie(bloom_filter_byte_ranges, dict_page_ranges) =
       reader.secondary_filters_byte_ranges(current_row_group_indices, options);
+    dict_page_byte_ranges =
+      cudf::io::parquet::experimental::dictionary_page_byte_ranges_to_read(dict_page_ranges);
     if (verbose) { timer.print_elapsed_millis(); }
   }
 

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

@@ -16,7 +16,9 @@
 
 #include <cuda/stream>
 
+#include <limits>
 #include <memory>
+#include <optional>
 #include <span>
 #include <utility>
 #include <vector>
@@ -53,6 +55,63 @@ enum class use_data_page_mask : bool {
   NO  = false  ///< Do not compute or use a data page mask
 };
 
+/**
+ * @brief How closely a dictionary page byte range describes the page it points at
+ *
+ * An `upper_bound_if_present` range begins at the dictionary page if the column chunk has one, and
+ * ends no earlier than that page does. A writer is allowed to leave out where the page ends, and to
+ * say that a chunk is dictionary encoded when it holds no dictionary page at all, so a range of
+ * this kind is a bound on a page that may not be there.
+ */
+enum class dictionary_page_extent : bool {
+  exact,                  ///< The range is exactly the dictionary page
+  upper_bound_if_present  ///< The range bounds a dictionary page that may not be there
+};
+
+/**
+ * @brief Byte range of a column chunk's dictionary page, and how closely it describes that page
+ *
+ * A caller is free to read less than an `upper_bound_if_present` range, which is how it caps what
+ * it spends looking for a page that may not be there. The reader still wants a span holding exactly
+ * one dictionary page, so a caller that reads such a range measures the page in it with
+ * `dictionary_page_length`, and passes an empty span for a chunk whose page is not there or does
+ * not fit in what was read.
+ */
+struct dictionary_page_range {
+  byte_range_info byte_range;     ///< Byte range to read from the file
+  dictionary_page_extent extent;  ///< How closely `byte_range` describes the dictionary page
+};
+
+/**
+ * @brief Byte ranges to read for the specified dictionary page ranges
+ *
+ * No more than `max_upper_bound_size` bytes are read of a range that only bounds its dictionary
+ * page, which is how a caller caps what it spends looking for a page that may not be there. By
+ * default the whole of every range is read. What is read of such a range still has to be trimmed to
+ * the dictionary page before it is handed to the reader, see `dictionary_page_range`.
+ *
+ * @param dictionary_page_ranges Dictionary page ranges from `secondary_filters_byte_ranges`
+ * @param max_upper_bound_size Most bytes to read of a range that only bounds its dictionary page
+ * @return Byte ranges to read, one per input dictionary page range
+ */
+[[nodiscard]] std::vector<byte_range_info> dictionary_page_byte_ranges_to_read(
+  cudf::host_span<dictionary_page_range const> dictionary_page_ranges,
+  int64_t max_upper_bound_size = std::numeric_limits<int64_t>::max());
+
+/**
+ * @brief Length of the dictionary page at the front of the specified bytes, header included
+ *
+ * What was read of a range that only bounds its dictionary page begins at that page and runs past
+ * it. The page's own header says how long the page is, so this reads that header to find where the
+ * page ends, which is what turns such a range into the one page the reader takes.
+ *
+ * @param page_bytes Bytes read for a dictionary page range, from the start of the range
+ * @return Length of the dictionary page, or `std::nullopt` if these bytes do not begin with a whole
+ *         dictionary page, which is the case for a column chunk that has none to prune with
+ */
+[[nodiscard]] std::optional<int64_t> dictionary_page_length(
+  cudf::host_span<uint8_t const> page_bytes);
+
 /**
  * @brief The experimental parquet reader class to optimally read parquet files subject to
  *        highly selective filters, called a Hybrid Scan operation
@@ -148,18 +207,29 @@ enum class use_data_page_mask : bool {
  * current_row_group_indices = stats_filtered_row_group_indices;
  *
  * // Get byte ranges of bloom filters and dictionaries for the current row groups
- * auto [bloom_filter_byte_ranges, dict_page_byte_ranges] =
+ * auto [bloom_filter_byte_ranges, dict_page_ranges] =
  *   reader->secondary_filters_byte_ranges(current_row_group_indices, options);
  *
  * // Optional: Prune row groups if we have valid dictionary pages
  * auto dict_filtered_row_group_indices = std::vector<size_type>{};
  *
- * if (dict_page_byte_ranges.size()) {
+ * if (dict_page_ranges.size()) {
+ *   // Decide how much of each range to read. A range that only bounds its dictionary page can be
+ *   // much larger than the page it bounds, so read no more of it than a dictionary page is worth.
+ *   auto const dict_page_byte_ranges =
+ *     dictionary_page_byte_ranges_to_read(dict_page_ranges, max_dict_page_size);
+ *
  *   // Fetch dictionary page byte ranges into device buffers and create spans
  *   auto [dict_page_buffers, dict_page_data, dict_page_tasks] =
  *     parquet::fetch_byte_ranges_to_device_async(datasource, dict_page_byte_ranges, stream, mr);
  *   dict_page_tasks.get();
  *
+ *   // The spans above are what the reader takes as long as every range was exactly a page. What
+ *   // was read of a range that only bounds its page runs past that page instead, and may hold no
+ *   // page at all, so such a range has to be fetched into host memory, measured with
+ *   // `dictionary_page_length`, and copied to the device cut down to its page. A column chunk
+ *   // left with an empty span is not pruned with.
+ *
  *   // Prune row groups using dictionaries
  *   dict_filtered_row_group_indices = reader->filter_row_groups_with_dictionary_pages(
  *     dict_page_data, current_row_group_indices, options, stream);
@@ -396,16 +466,20 @@ class hybrid_scan_reader {
    *
    * @param row_group_indices Input row groups indices
    * @param options Parquet reader options
-   * @return Pair of vectors of byte ranges of column chunk with bloom filters and dictionary
-   *         pages subject to filter predicate
+   * @return Pair of a vector of byte ranges of column chunks with bloom filters and a vector of
+   *         dictionary page ranges, subject to filter predicate
    */
-  [[nodiscard]] std::pair<std::vector<byte_range_info>, std::vector<byte_range_info>>
+  [[nodiscard]] std::pair<std::vector<byte_range_info>, std::vector<dictionary_page_range>>
   secondary_filters_byte_ranges(std::span<size_type const> row_group_indices,
                                 parquet_reader_options const& options) const;
 
   /**
    * @brief Filter the row groups using column chunk dictionary pages
    *
+   * Each span must hold exactly one dictionary page, or nothing at all for a column chunk that has
+   * no dictionary page to prune with. See `dictionary_page_range` for trimming a range that only
+   * bounds its page.
+   *
    * @param dictionary_page_data Device spans of dictionary page data of column chunks with an
    * (in)equality predicate, in the same order as the byte ranges returned by
    * `secondary_filters_byte_ranges` including empty spans against empty byte ranges

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

@@ -194,16 +194,20 @@ class hybrid_scan_multifile {
    *
    * @param row_group_indices Span of vectors of input row group indices, one per source
    * @param options Parquet reader options
-   * @return Pair of flattened byte ranges to column chunk dictionary pages subject to the filter
+   * @return Pair of flattened dictionary page ranges of column chunks subject to the filter
    *         predicate and their corresponding source indices
    */
-  [[nodiscard]] std::pair<std::vector<byte_range_info>, std::vector<size_type>>
+  [[nodiscard]] std::pair<std::vector<dictionary_page_range>, std::vector<size_type>>
   dictionary_pages_byte_ranges(cudf::host_span<std::vector<size_type> const> row_group_indices,
                                parquet_reader_options const& options) const;
 
   /**
    * @brief Filter the row groups using column chunk dictionary pages
    *
+   * Each span must hold exactly one dictionary page, or nothing at all for a column chunk that has
+   * no dictionary page to prune with. See `dictionary_page_range` for trimming a range that only
+   * bounds its page.
+   *
    * @param dictionary_page_data Device spans of dictionary page data of column chunks with an
    * (in)equality predicate, in the same order as the byte ranges returned by
    * `dictionary_pages_byte_ranges` including empty spans against empty byte ranges

cpp/src/io/parquet/experimental/dictionary_page_filter.cu

@@ -293,8 +293,14 @@ CUDF_KERNEL void query_dictionaries(cudf::device_span<T> decoded_data,
 
   // Evaluate the scalar against all cuco hash sets of this column
   for (auto set_idx = group.thread_rank(); set_idx < total_row_groups; set_idx += group.size()) {
-    // If the set is empty (no dictionary page data), then skip the dictionary page filter
-    if (set_offsets[set_idx + 1] - set_offsets[set_idx] == 0) {
+    // Number of values in this hash set
+    auto const num_set_values = value_offsets[set_idx + 1] - value_offsets[set_idx];
+
+    // Skip the dictionary page filter for a column chunk with no dictionary page. Emptiness must be
+    // read from the value count and not from the number of slots, because cuco rounds every
+    // capacity up to at least one bucket, so an empty dictionary still has slots. Its set was never
+    // built, so probing it would report the literal as absent and prune the row group.
+    if (num_set_values == 0) {
       result[set_idx] = operators[scalar_idx] == ast::ast_operator::EQUAL;
       continue;
     }
@@ -311,8 +317,6 @@ CUDF_KERNEL void query_dictionaries(cudf::device_span<T> decoded_data,
                                              storage_ref};
     auto set_find_ref = hash_set_ref.rebind_operators(cuco::contains);
 
-    // Number of values in this hash set
-    auto const num_set_values = value_offsets[set_idx + 1] - value_offsets[set_idx];
     // Literal value to find in this hash set
     auto const literal_value = scalar.value<T>();
 
@@ -901,6 +905,8 @@ CUDF_KERNEL void __launch_bounds__(DECODE_BLOCK_SIZE)
     results[i][row_group_idx] = false;
   }
 
+  group.sync();
+
   // Decode values from the current dictionary page with the current thread block
   for (auto value_idx = group.thread_rank(); value_idx < page.num_input_values;
        value_idx += group.num_threads()) {

cpp/src/io/parquet/experimental/hybrid_scan.cpp

@@ -3,6 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+#include "../compact_protocol_reader.hpp"
 #include "hybrid_scan_impl.hpp"
 
 #include <cudf/detail/nvtx/ranges.hpp>
@@ -11,8 +12,59 @@
 
 #include <thrust/host_vector.h>
 
+#include <algorithm>
+#include <iterator>
+#include <optional>
+#include <utility>
+
 namespace cudf::io::parquet::experimental {
 
+std::vector<text::byte_range_info> dictionary_page_byte_ranges_to_read(
+  cudf::host_span<dictionary_page_range const> dictionary_page_ranges, int64_t max_upper_bound_size)
+{
+  CUDF_EXPECTS(max_upper_bound_size >= 0, "Maximum bytes to read must not be negative");
+
+  auto byte_ranges = std::vector<text::byte_range_info>{};
+  byte_ranges.reserve(dictionary_page_ranges.size());
+  std::transform(dictionary_page_ranges.begin(),
+                 dictionary_page_ranges.end(),
+                 std::back_inserter(byte_ranges),
+                 [max_upper_bound_size](auto const& range) {
+                   if (range.extent == dictionary_page_extent::exact) { return range.byte_range; }
+                   return text::byte_range_info{
+                     range.byte_range.offset(),
+                     std::min(range.byte_range.size(), max_upper_bound_size)};
+                 });
+  return byte_ranges;
+}
+
+std::optional<int64_t> dictionary_page_length(cudf::host_span<uint8_t const> page_bytes)
+{
+  auto header = PageHeader{};
+  auto reader = parquet::detail::CompactProtocolReader{page_bytes.data(), page_bytes.size()};
+
+  // Nothing says these bytes are a page header at all, so a parse that gives up on them means there
+  // is no dictionary page here rather than that the file is corrupt.
+  try {
+    reader.read(&header);
+  } catch (std::exception const&) {
+    return std::nullopt;
+  }
+
+  // A chunk that claims dictionary encoding may have been written without a dictionary page, in
+  // which case these bytes are the chunk's first data page.
+  if (header.type != PageType::DICTIONARY_PAGE or header.compressed_page_size <= 0) {
+    return std::nullopt;
+  }
+
+  // A header cut off by the end of what was read stops parsing without complaint, and a page longer
+  // than what was read cannot be pruned with either way.
+  auto const page_length = static_cast<int64_t>(reader.bytecount()) + header.compressed_page_size;
+  if (std::cmp_greater(page_length, page_bytes.size())) { return std::nullopt; }
+
+  return page_length;
+}
+
 hybrid_scan_reader::hybrid_scan_reader(cudf::host_span<uint8_t const> footer_bytes,
                                        parquet_reader_options const& options)
   : _impl{std::make_unique<detail::hybrid_scan_reader_impl>(
@@ -95,7 +147,7 @@ std::vector<cudf::size_type> hybrid_scan_reader::filter_row_groups_with_stats(
   return _impl->filter_row_groups_with_stats(input_row_group_indices, options, stream).front();
 }
 
-std::pair<std::vector<text::byte_range_info>, std::vector<text::byte_range_info>>
+std::pair<std::vector<text::byte_range_info>, std::vector<dictionary_page_range>>
 hybrid_scan_reader::secondary_filters_byte_ranges(std::span<size_type const> row_group_indices,
                                                   parquet_reader_options const& options) const
 {