Avoid serializing I/O requests from Parquet reader

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
  */
 
@@ -92,7 +92,11 @@ void BM_filter_string_row_groups_with_dicts_common(nvbench::state& state,
       // Fetch dictionary page data
       auto [dictionary_page_buffers, dictionary_page_data, read_task] =
         cudf::io::parquet::fetch_byte_ranges_to_device_async(
-          datasource_ref, dict_page_byte_ranges, stream, cudf::get_current_device_resource_ref());
+          datasource_ref,
+          dict_page_byte_ranges,
+          cudf::io::parquet::io_submission_policy::SERIALIZE,
+          stream,
+          cudf::get_current_device_resource_ref());
       read_task.get();
 
       // Filter row groups with dictionary pages

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

@@ -89,7 +89,11 @@ std::vector<cudf::size_type> apply_row_group_filters(
       dict_page_byte_ranges.size()) {
     auto [dictionary_page_buffers, dictionary_page_data, dict_read_tasks] =
       cudf::io::parquet::fetch_byte_ranges_to_device_async(
-        datasource, dict_page_byte_ranges, stream, mr);
+        datasource,
+        dict_page_byte_ranges,
+        cudf::io::parquet::io_submission_policy::SERIALIZE,
+        stream,
+        mr);
     dict_read_tasks.get();
 
     dict_page_filtered_row_groups = reader.filter_row_groups_with_dictionary_pages(
@@ -109,7 +113,11 @@ std::vector<cudf::size_type> apply_row_group_filters(
     auto aligned_mr = rmm::mr::aligned_resource_adaptor(mr, bloom_filter_alignment);
     auto [bloom_filter_buffers, bloom_filter_data, bloom_read_tasks] =
       cudf::io::parquet::fetch_byte_ranges_to_device_async(
-        datasource, bloom_filter_byte_ranges, stream, aligned_mr);
+        datasource,
+        bloom_filter_byte_ranges,
+        cudf::io::parquet::io_submission_policy::SERIALIZE,
+        stream,
+        aligned_mr);
     bloom_read_tasks.get();
 
     bloom_filtered_row_groups = reader.filter_row_groups_with_bloom_filters(
@@ -134,7 +142,11 @@ std::unique_ptr<cudf::table> single_step_materialize(
     reader.all_column_chunks_byte_ranges(current_row_group_indices, options);
   auto [all_column_chunk_buffers, all_column_chunk_data, all_column_chunk_read_tasks] =
     cudf::io::parquet::fetch_byte_ranges_to_device_async(
-      datasource, all_column_chunk_byte_ranges, stream, mr);
+      datasource,
+      all_column_chunk_byte_ranges,
+      cudf::io::parquet::io_submission_policy::SERIALIZE,
+      stream,
+      mr);
   all_column_chunk_read_tasks.get();
 
   return reader

cpp/examples/hybrid_scan_io/io_utils.cpp

@@ -38,5 +38,6 @@ fetch_byte_ranges_async(cudf::io::datasource& datasource,
                         rmm::device_async_resource_ref mr)
 {
   // Using libcudf utility but may have custom implementation in the future
-  return cudf::io::parquet::fetch_byte_ranges_to_device_async(datasource, byte_ranges, stream, mr);
+  return cudf::io::parquet::fetch_byte_ranges_to_device_async(
+    datasource, byte_ranges, cudf::io::parquet::io_submission_policy::SERIALIZE, stream, mr);
 }

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

@@ -226,7 +226,8 @@ class hybrid_scan_metadata {
  * if (dict_page_byte_ranges.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);
+ *     parquet::fetch_byte_ranges_to_device_async(
+ *       datasource, dict_page_byte_ranges, parquet::io_submission_policy::SERIALIZE, stream, mr);
  *   dict_page_tasks.get();
  *
  *   // Prune row groups using dictionaries
@@ -243,8 +244,9 @@ class hybrid_scan_metadata {
  * if (bloom_filter_byte_ranges.size()) {
  *   // Fetch bloom filter byte ranges into device buffers and create spans
  *   auto [bloom_filter_buffers, bloom_filter_data, bloom_filter_tasks] =
- *     parquet::fetch_byte_ranges_to_device_async(datasource, bloom_filter_byte_ranges, stream, mr);
- *   bloom_filter_tasks.get();
+ *     parquet::fetch_byte_ranges_to_device_async(
+ *       datasource, bloom_filter_byte_ranges, parquet::io_submission_policy::SERIALIZE, stream,
+ * mr); bloom_filter_tasks.get();
  *
  *   // Prune row groups using bloom filters
  *   bloom_filtered_row_group_indices = reader->filter_row_groups_with_bloom_filters(
@@ -300,7 +302,8 @@ class hybrid_scan_metadata {
  *
  * // Fetch column chunk data into device buffers and create spans
  * auto [filter_col_buffers, filter_col_data, filter_col_tasks] =
- *   parquet::fetch_byte_ranges_to_device_async(datasource, filter_col_byte_ranges, stream, mr);
+ *   parquet::fetch_byte_ranges_to_device_async(
+ *     datasource, filter_col_byte_ranges, parquet::io_submission_policy::SERIALIZE, stream, mr);
  * filter_col_tasks.get();
  *
  * // Materialize the table with only the filter columns
@@ -327,7 +330,8 @@ class hybrid_scan_metadata {
  *
  * // Fetch column chunk data into device buffers and create spans
  * auto [payload_col_buffers, payload_col_data, payload_col_tasks] =
- *   parquet::fetch_byte_ranges_to_device_async(datasource, payload_col_byte_ranges, stream, mr);
+ *   parquet::fetch_byte_ranges_to_device_async(
+ *     datasource, payload_col_byte_ranges, parquet::io_submission_policy::SERIALIZE, stream, mr);
  * payload_col_tasks.get();
  *
  * // Materialize the table with only the payload columns

cpp/include/cudf/io/parquet_io_utils.hpp

@@ -7,6 +7,7 @@
 
 #include <cudf/io/datasource.hpp>
 #include <cudf/io/text/byte_range_info.hpp>
+#include <cudf/utilities/memory_resource.hpp>
 
 #include <rmm/device_buffer.hpp>
 #include <rmm/resource_ref.hpp>
@@ -37,6 +38,14 @@ namespace io::parquet {
 //! Using `byte_range_info` from cudf::io::text
 using cudf::io::text::byte_range_info;
 
+/**
+ * @brief Controls whether I/O submissions are serialized across callers.
+ */
+enum class io_submission_policy : bool {
+  SERIALIZE,  ///< Serialize submissions across callers.
+  INTERLEAVE  ///< Allow submissions from different callers to interleave.
+};
+
 /**
  * @brief Returns the Parquet reader's footer speculative read size in bytes.
  *
@@ -117,83 +126,190 @@ using cudf::io::text::byte_range_info;
  *
  * @param datasource Input datasource
  * @param byte_ranges Byte ranges to fetch
+ * @param policy Whether to serialize I/O submissions from callers
+ * @param stream CUDA stream
+ * @param mr Memory resources used to allocate the returned device buffers
+ *
+ * @return A tuple containing the device buffers, the device spans of the fetched data, and a future
+ * to wait on the read tasks
+ */
+std::tuple<std::vector<rmm::device_buffer>,
+           std::vector<cudf::device_span<uint8_t const>>,
+           std::future<void>>
+fetch_byte_ranges_to_device_async(
+  cudf::io::datasource& datasource,
+  std::span<byte_range_info const> byte_ranges,
+  io_submission_policy policy,
+  cuda::stream_ref stream   = cudf::get_default_stream(),
+  cudf::memory_resources mr = cudf::get_current_device_resource_ref());
+
+/**
+ * @brief Fetches lists of byte ranges from multiple datasources into device buffers
+ *
+ * @ingroup io_utils
+ *
+ * @param datasources Input datasources
+ * @param byte_ranges_per_source Vector of byte ranges to fetch, one per datasource
+ * @param policy Whether to serialize I/O submissions from callers
+ * @param stream CUDA stream
+ * @param mr Memory resources used to allocate the returned device buffers
+ *
+ * @return A tuple containing a vector of device buffers, a vector of vectors of device spans (one
+ * per byte range per datasource), and a future to wait on the read tasks
+ */
+std::tuple<std::vector<rmm::device_buffer>,
+           std::vector<std::vector<cudf::device_span<uint8_t const>>>,
+           std::future<void>>
+fetch_byte_ranges_to_device_async(
+  cudf::host_span<std::reference_wrapper<cudf::io::datasource> const> datasources,
+  cudf::host_span<std::vector<byte_range_info> const> byte_ranges_per_source,
+  io_submission_policy policy,
+  cuda::stream_ref stream   = cudf::get_default_stream(),
+  cudf::memory_resources mr = cudf::get_current_device_resource_ref());
+
+/**
+ * @brief Fetches Parquet bloom filter bitsets from a datasource into device buffers
+ *
+ * @ingroup io_utils
+ *
+ * @param datasource Input datasource
+ * @param bloom_filter_byte_ranges Byte ranges of complete bloom filters to fetch, must span a
+ * complete bloom filter
+ * @param policy Whether to serialize I/O submissions from callers
+ * @param stream CUDA stream
+ * @param mr Memory resources used to allocate the returned device buffers
+ *
+ * @return A pair containing buffers that own the fetched bitsets and one device span per input byte
+ * range
+ */
+std::pair<std::vector<rmm::device_buffer>, std::vector<cudf::device_span<uint8_t const>>>
+fetch_bloom_filters_to_device(cudf::io::datasource& datasource,
+                              cudf::host_span<byte_range_info const> bloom_filter_byte_ranges,
+                              io_submission_policy policy,
+                              cuda::stream_ref stream   = cudf::get_default_stream(),
+                              cudf::memory_resources mr = cudf::get_current_device_resource_ref());
+
+/**
+ * @brief Fetches Parquet bloom filter bitsets from multiple datasources into device buffers
+ *
+ * @ingroup io_utils
+ *
+ * @param datasources Input datasources
+ * @param bloom_filter_byte_ranges_per_source Byte ranges of complete bloom filters to fetch, one
+ * vector per datasource. Each byte range must span a complete bloom filter.
+ * @param policy Whether to serialize I/O submissions from callers
+ * @param stream CUDA stream
+ * @param mr Memory resources used to allocate the returned device buffers
+ *
+ * @return A pair containing buffers that own the fetched bitsets and per-source device spans, with
+ * one inner vector per datasource
+ */
+std::pair<std::vector<rmm::device_buffer>,
+          std::vector<std::vector<cudf::device_span<uint8_t const>>>>
+fetch_bloom_filters_to_device(
+  cudf::host_span<std::reference_wrapper<cudf::io::datasource> const> datasources,
+  cudf::host_span<std::vector<byte_range_info> const> bloom_filter_byte_ranges_per_source,
+  io_submission_policy policy,
+  cuda::stream_ref stream   = cudf::get_default_stream(),
+  cudf::memory_resources mr = cudf::get_current_device_resource_ref());
+
+/**
+ * @brief Fetches a list of byte ranges from a datasource into device buffers
+ *
+ * @ingroup io_utils
+ *
+ * @deprecated Use the overload that takes `io_submission_policy`.
+ *
+ * @param datasource Input datasource
+ * @param byte_ranges Byte ranges to fetch
  * @param stream CUDA stream
- * @param mr Device memory resource
+ * @param mr Memory resources used to allocate the returned device buffers
  *
  * @return A tuple containing the device buffers, the device spans of the fetched data, and a future
  * to wait on the read tasks
  */
+[[deprecated("Use the overload that takes io_submission_policy.")]]
 std::tuple<std::vector<rmm::device_buffer>,
            std::vector<cudf::device_span<uint8_t const>>,
            std::future<void>>
-fetch_byte_ranges_to_device_async(cudf::io::datasource& datasource,
-                                  std::span<byte_range_info const> byte_ranges,
-                                  cuda::stream_ref stream,
-                                  rmm::device_async_resource_ref mr);
+fetch_byte_ranges_to_device_async(
+  cudf::io::datasource& datasource,
+  std::span<byte_range_info const> byte_ranges,
+  cuda::stream_ref stream   = cudf::get_default_stream(),
+  cudf::memory_resources mr = cudf::get_current_device_resource_ref());
 
 /**
  * @brief Fetches lists of byte ranges from multiple datasources into device buffers
  *
  * @ingroup io_utils
  *
+ * @deprecated Use the overload that takes `io_submission_policy`.
+ *
  * @param datasources Input datasources
  * @param byte_ranges_per_source Vector of byte ranges to fetch, one per datasource
  * @param stream CUDA stream
- * @param mr Device memory resource
+ * @param mr Memory resources used to allocate the returned device buffers
  *
  * @return A tuple containing a vector of device buffers, a vector of vectors of device spans (one
  * per byte range per datasource), and a future to wait on the read tasks
  */
+[[deprecated("Use the overload that takes io_submission_policy.")]]
 std::tuple<std::vector<rmm::device_buffer>,
            std::vector<std::vector<cudf::device_span<uint8_t const>>>,
            std::future<void>>
 fetch_byte_ranges_to_device_async(
   cudf::host_span<std::reference_wrapper<cudf::io::datasource> const> datasources,
   cudf::host_span<std::vector<byte_range_info> const> byte_ranges_per_source,
-  cuda::stream_ref stream,
-  rmm::device_async_resource_ref mr);
+  cuda::stream_ref stream   = cudf::get_default_stream(),
+  cudf::memory_resources mr = cudf::get_current_device_resource_ref());
 
 /**
  * @brief Fetches Parquet bloom filter bitsets from a datasource into device buffers
  *
  * @ingroup io_utils
  *
+ * @deprecated Use the overload that takes `io_submission_policy`.
+ *
  * @param datasource Input datasource
  * @param bloom_filter_byte_ranges Byte ranges of complete bloom filters to fetch, must span a
  * complete bloom filter
  * @param stream CUDA stream
- * @param mr Device memory resource used to allocate the returned device buffers
+ * @param mr Memory resources used to allocate the returned device buffers
  *
  * @return A pair containing buffers that own the fetched bitsets and one device span per input byte
  * range
  */
+[[deprecated("Use the overload that takes io_submission_policy.")]]
 std::pair<std::vector<rmm::device_buffer>, std::vector<cudf::device_span<uint8_t const>>>
 fetch_bloom_filters_to_device(cudf::io::datasource& datasource,
                               cudf::host_span<byte_range_info const> bloom_filter_byte_ranges,
-                              cuda::stream_ref stream,
-                              rmm::device_async_resource_ref mr);
+                              cuda::stream_ref stream   = cudf::get_default_stream(),
+                              cudf::memory_resources mr = cudf::get_current_device_resource_ref());
 
 /**
  * @brief Fetches Parquet bloom filter bitsets from multiple datasources into device buffers
  *
  * @ingroup io_utils
  *
+ * @deprecated Use the overload that takes `io_submission_policy`.
+ *
  * @param datasources Input datasources
  * @param bloom_filter_byte_ranges_per_source Byte ranges of complete bloom filters to fetch, one
  * vector per datasource. Each byte range must span a complete bloom filter.
  * @param stream CUDA stream
- * @param mr Device memory resource used to allocate the returned device buffers
+ * @param mr Memory resources used to allocate the returned device buffers
  *
  * @return A pair containing buffers that own the fetched bitsets and per-source device spans, with
  * one inner vector per datasource
  */
+[[deprecated("Use the overload that takes io_submission_policy.")]]
 std::pair<std::vector<rmm::device_buffer>,
           std::vector<std::vector<cudf::device_span<uint8_t const>>>>
 fetch_bloom_filters_to_device(
   cudf::host_span<std::reference_wrapper<cudf::io::datasource> const> datasources,
   cudf::host_span<std::vector<byte_range_info> const> bloom_filter_byte_ranges_per_source,
-  cuda::stream_ref stream,
-  rmm::device_async_resource_ref mr);
+  cuda::stream_ref stream   = cudf::get_default_stream(),
+  cudf::memory_resources mr = cudf::get_current_device_resource_ref());
 
 /** @} */  // end of group
 }  // namespace io::parquet

cpp/src/io/parquet/bloom_filter_reader.cu

@@ -385,7 +385,11 @@ aggregate_reader_metadata::read_bloom_filters(
     });
 
   auto [bloom_filter_buffers, bitset_spans_per_source] =
-    fetch_bloom_filters_to_device(datasource_refs, bloom_filter_byte_ranges_per_source, stream, mr);
+    fetch_bloom_filters_to_device(datasource_refs,
+                                  bloom_filter_byte_ranges_per_source,
+                                  io_submission_policy::INTERLEAVE,
+                                  stream,
+                                  mr);
 
   // Flatten the per-source bitset spans into per-chunk order
   std::vector<cudf::device_span<cuda::std::byte const>> bloom_filter_data;

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

@@ -269,7 +269,7 @@ hybrid_scan_reader_impl::prepare_dictionaries(
                                           {},
                                           {},
                                           {},
-                                          _stream));
+                                          stream));
 
   CUDF_EXPECTS(
     row_groups_info.size() * dictionary_col_schemas.size() == dictionary_page_data.size(),