diff --git a/cpp/benchmarks/io/parquet/experimental/hybrid_scan/dict_page_filter.cpp b/cpp/benchmarks/io/parquet/experimental/hybrid_scan/dict_page_filter.cpp
index 3c93989ad825..a3f974898cb1 100644
--- a/cpp/benchmarks/io/parquet/experimental/hybrid_scan/dict_page_filter.cpp
+++ b/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
diff --git a/cpp/benchmarks/io/parquet/experimental/hybrid_scan/hybrid_scan_composer.cpp b/cpp/benchmarks/io/parquet/experimental/hybrid_scan/hybrid_scan_composer.cpp
index 2f85cb0fe43f..2ddf1f9361e9 100644
--- a/cpp/benchmarks/io/parquet/experimental/hybrid_scan/hybrid_scan_composer.cpp
+++ b/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
diff --git a/cpp/examples/hybrid_scan_io/io_utils.cpp b/cpp/examples/hybrid_scan_io/io_utils.cpp
index f9c0d63d124a..316c20ab7835 100644
--- a/cpp/examples/hybrid_scan_io/io_utils.cpp
+++ b/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);
 }
diff --git a/cpp/include/cudf/io/experimental/hybrid_scan.hpp b/cpp/include/cudf/io/experimental/hybrid_scan.hpp
index 772b65e62fc9..e5a82091cc6e 100644
--- a/cpp/include/cudf/io/experimental/hybrid_scan.hpp
+++ b/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
diff --git a/cpp/include/cudf/io/parquet_io_utils.hpp b/cpp/include/cudf/io/parquet_io_utils.hpp
index c2892f5c80ec..433a0c6f4826 100644
--- a/cpp/include/cudf/io/parquet_io_utils.hpp
+++ b/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
diff --git a/cpp/src/io/parquet/bloom_filter_reader.cu b/cpp/src/io/parquet/bloom_filter_reader.cu
index ac8b2a8a7755..d758c6b71590 100644
--- a/cpp/src/io/parquet/bloom_filter_reader.cu
+++ b/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;
diff --git a/cpp/src/io/parquet/io_utils/parquet_io_utils.cpp b/cpp/src/io/parquet/io_utils/parquet_io_utils.cpp
index 12f853d67c61..6fa801df8526 100644
--- a/cpp/src/io/parquet/io_utils/parquet_io_utils.cpp
+++ b/cpp/src/io/parquet/io_utils/parquet_io_utils.cpp
@@ -233,7 +233,8 @@ void read_ranges_to_host(
   OffsetIterator offsets,
   SizeIterator sizes,
   std::size_t count,
-  cudf::host_span<uint8_t> dst)
+  cudf::host_span<uint8_t> dst,
+  bool serialize_submissions)
 {
   std::vector<std::future<std::size_t>> host_read_tasks;
   std::vector<std::size_t> expected_sizes;
@@ -246,7 +247,8 @@ void read_ranges_to_host(
   // Schedule host reads holding the `host_read_mutex` so that all reads for a caller thread
   // are scheduled without interleaving with reads from other threads yielding better pipelining
   {
-    std::scoped_lock<std::mutex> lock(host_read_mutex());
+    std::unique_lock<std::mutex> lock(host_read_mutex(), std::defer_lock);
+    if (serialize_submissions) { lock.lock(); }
 
     std::for_each(iter, iter + count, [&](auto const& tuple) {
       auto const src_idx    = cuda::std::get<0>(tuple);
@@ -285,7 +287,8 @@ fetch_byte_ranges_to_device_async_impl(
   cudf::host_span<cudf::host_span<cudf::io::text::byte_range_info const> const>
     byte_ranges_per_source,
   cuda::stream_ref stream,
-  rmm::device_async_resource_ref mr)
+  rmm::device_async_resource_ref mr,
+  bool serialize_submissions)
 {
   auto const num_sources = datasources.size();
 
@@ -395,7 +398,8 @@ fetch_byte_ranges_to_device_async_impl(
   // Schedule host reads holding the `host_read_mutex` so that all reads for a caller thread
   // are scheduled without interleaving with reads from other threads yielding better pipelining
   {
-    std::scoped_lock<std::mutex> lock(host_read_mutex());
+    std::unique_lock<std::mutex> lock(host_read_mutex(), std::defer_lock);
+    if (serialize_submissions) { lock.lock(); }
 
     std::for_each(iter, iter + io_offsets.size(), [&](auto const& tuple) {
       auto const src_idx   = cuda::std::get<0>(tuple);
@@ -433,7 +437,8 @@ fetch_byte_ranges_to_device_async_impl(
   // Schedule device reads holding the `device_read_mutex` so that all reads for a caller thread
   // are scheduled without interleaving with reads from other threads yielding better pipelining
   {
-    std::scoped_lock<std::mutex> lock(device_read_mutex());
+    std::unique_lock<std::mutex> lock(device_read_mutex(), std::defer_lock);
+    if (serialize_submissions) { lock.lock(); }
 
     std::for_each(iter, iter + io_offsets.size(), [&](auto const& tuple) {
       auto const src_idx   = cuda::std::get<0>(tuple);
@@ -475,7 +480,8 @@ fetch_bloom_filters_to_device_impl(
   cudf::host_span<cudf::host_span<cudf::io::text::byte_range_info const> const>
     bloom_filter_byte_ranges_per_source,
   cuda::stream_ref stream,
-  rmm::device_async_resource_ref mr)
+  rmm::device_async_resource_ref mr,
+  bool serialize_submissions)
 {
   auto const num_sources = datasources.size();
   CUDF_EXPECTS(num_sources == bloom_filter_byte_ranges_per_source.size(),
@@ -526,7 +532,8 @@ fetch_bloom_filters_to_device_impl(
                       initial_offsets.cbegin(),
                       initial_sizes.cbegin(),
                       total_filters,
-                      initial_buffer);
+                      initial_buffer,
+                      serialize_submissions);
 
   // Phase 2: Parse headers, organize bitset slots, and record deferred bitset reads
   std::vector<void const*> copy_srcs;
@@ -601,7 +608,8 @@ fetch_bloom_filters_to_device_impl(
                         deferred_offsets.cbegin(),
                         deferred_sizes,
                         deferred_filter_indices.size(),
-                        deferred_buffer);
+                        deferred_buffer,
+                        serialize_submissions);
     std::size_t deferred_dst_offset = 0;
     std::for_each(
       deferred_filter_indices.begin(), deferred_filter_indices.end(), [&](auto const filter_idx) {
@@ -638,7 +646,8 @@ fetch_bloom_filters_to_device_impl(
   // One batched copy (entries with a null source or zero size are ignored by the batch API)
   if (total_device_size != 0) {
     {
-      std::scoped_lock<std::mutex> lock(device_read_mutex());
+      std::unique_lock<std::mutex> lock(device_read_mutex(), std::defer_lock);
+      if (serialize_submissions) { lock.lock(); }
       CUDF_CUDA_TRY(cudf::detail::memcpy_batch_async(
         copy_dsts.data(), copy_srcs.data(), copy_sizes.data(), total_filters, stream));
     }
@@ -702,8 +711,9 @@ std::tuple<std::vector<rmm::device_buffer>,
            std::future<void>>
 fetch_byte_ranges_to_device_async(cudf::io::datasource& datasource,
                                   std::span<cudf::io::text::byte_range_info const> byte_ranges,
+                                  io_submission_policy policy,
                                   cuda::stream_ref stream,
-                                  rmm::device_async_resource_ref mr)
+                                  cudf::memory_resources mr)
 {
   CUDF_FUNC_RANGE();
 
@@ -716,7 +726,8 @@ fetch_byte_ranges_to_device_async(cudf::io::datasource& datasource,
     {datasources.data(), datasources.size()},
     {byte_ranges_per_source.data(), byte_ranges_per_source.size()},
     stream,
-    mr);
+    mr.get_output_mr(),
+    policy == io_submission_policy::SERIALIZE);
 
   return {std::move(buffers), std::move(fetched_byte_ranges.front()), std::move(fut)};
 }
@@ -727,8 +738,9 @@ std::tuple<std::vector<rmm::device_buffer>,
 fetch_byte_ranges_to_device_async(
   cudf::host_span<std::reference_wrapper<cudf::io::datasource> const> datasources,
   cudf::host_span<std::vector<cudf::io::text::byte_range_info> const> byte_ranges_per_source,
+  io_submission_policy policy,
   cuda::stream_ref stream,
-  rmm::device_async_resource_ref mr)
+  cudf::memory_resources mr)
 {
   CUDF_FUNC_RANGE();
 
@@ -742,15 +754,17 @@ fetch_byte_ranges_to_device_async(
     datasources,
     {byte_range_spans_per_source.data(), byte_range_spans_per_source.size()},
     stream,
-    mr);
+    mr.get_output_mr(),
+    policy == io_submission_policy::SERIALIZE);
 }
 
 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<cudf::io::text::byte_range_info const> bloom_filter_byte_ranges,
+  io_submission_policy policy,
   cuda::stream_ref stream,
-  rmm::device_async_resource_ref mr)
+  cudf::memory_resources mr)
 {
   CUDF_FUNC_RANGE();
 
@@ -763,7 +777,8 @@ fetch_bloom_filters_to_device(
     {datasources.data(), datasources.size()},
     {bloom_filter_byte_ranges_per_source.data(), bloom_filter_byte_ranges_per_source.size()},
     stream,
-    mr);
+    mr.get_output_mr(),
+    policy == io_submission_policy::SERIALIZE);
 
   return {std::move(buffers), std::move(fetched_byte_ranges.front())};
 }
@@ -774,8 +789,9 @@ fetch_bloom_filters_to_device(
   cudf::host_span<std::reference_wrapper<cudf::io::datasource> const> datasources,
   cudf::host_span<std::vector<cudf::io::text::byte_range_info> const>
     bloom_filter_byte_ranges_per_source,
+  io_submission_policy policy,
   cuda::stream_ref stream,
-  rmm::device_async_resource_ref mr)
+  cudf::memory_resources mr)
 {
   CUDF_FUNC_RANGE();
 
@@ -790,7 +806,57 @@ fetch_bloom_filters_to_device(
                                             {bloom_filter_byte_range_spans_per_source.data(),
                                              bloom_filter_byte_range_spans_per_source.size()},
                                             stream,
-                                            mr);
+                                            mr.get_output_mr(),
+                                            policy == io_submission_policy::SERIALIZE);
+}
+
+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<cudf::io::text::byte_range_info const> byte_ranges,
+                                  cuda::stream_ref stream,
+                                  cudf::memory_resources mr)
+{
+  return fetch_byte_ranges_to_device_async(
+    datasource, byte_ranges, io_submission_policy::SERIALIZE, stream, mr);
+}
+
+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<cudf::io::text::byte_range_info> const> byte_ranges_per_source,
+  cuda::stream_ref stream,
+  cudf::memory_resources mr)
+{
+  return fetch_byte_ranges_to_device_async(
+    datasources, byte_ranges_per_source, io_submission_policy::SERIALIZE, stream, mr);
+}
+
+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<cudf::io::text::byte_range_info const> bloom_filter_byte_ranges,
+  cuda::stream_ref stream,
+  cudf::memory_resources mr)
+{
+  return fetch_bloom_filters_to_device(
+    datasource, bloom_filter_byte_ranges, io_submission_policy::SERIALIZE, stream, mr);
+}
+
+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<cudf::io::text::byte_range_info> const>
+    bloom_filter_byte_ranges_per_source,
+  cuda::stream_ref stream,
+  cudf::memory_resources mr)
+{
+  return fetch_bloom_filters_to_device(
+    datasources, bloom_filter_byte_ranges_per_source, io_submission_policy::SERIALIZE, stream, mr);
 }
 
 }  // namespace cudf::io::parquet
diff --git a/cpp/src/io/parquet/reader_impl_preprocess_utils.cu b/cpp/src/io/parquet/reader_impl_preprocess_utils.cu
index aaba5c12219e..6f113baba99b 100644
--- a/cpp/src/io/parquet/reader_impl_preprocess_utils.cu
+++ b/cpp/src/io/parquet/reader_impl_preprocess_utils.cu
@@ -199,6 +199,7 @@ void generate_depth_remappings(
   auto [buffers, data_per_source, read_task] = cudf::io::parquet::fetch_byte_ranges_to_device_async(
     {datasource_refs.data(), datasource_refs.size()},
     {source_byte_ranges.data(), source_byte_ranges.size()},
+    cudf::io::parquet::io_submission_policy::INTERLEAVE,
     stream,
     mr);
 
diff --git a/cpp/tests/io/experimental/hybrid_scan_common.cpp b/cpp/tests/io/experimental/hybrid_scan_common.cpp
index 2c9498408df2..ff53ed4b2ec5 100644
--- a/cpp/tests/io/experimental/hybrid_scan_common.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_common.cpp
@@ -202,6 +202,7 @@ multisource_device_data fetch_multisource_device_data(
   auto [buffers, per_source_spans, tasks] = cudf::io::parquet::fetch_byte_ranges_to_device_async(
     inputs.datasource_refs,
     cudf::host_span<std::vector<cudf::io::text::byte_range_info> const>{byte_ranges_per_source},
+    cudf::io::parquet::io_submission_policy::SERIALIZE,
     stream,
     mr);
   tasks.get();
@@ -252,7 +253,11 @@ auto filter_row_groups_with_dictionaries_impl(InputType& inputs,
       group_byte_ranges_by_source(dict_pages, inputs.datasources.size());
     [[maybe_unused]] auto [dict_page_buffers, dict_page_data_per_source, task] =
       cudf::io::parquet::fetch_byte_ranges_to_device_async(
-        inputs.datasource_refs, dict_page_ranges_per_source, stream, mr);
+        inputs.datasource_refs,
+        dict_page_ranges_per_source,
+        cudf::io::parquet::io_submission_policy::SERIALIZE,
+        stream,
+        mr);
     task.get();
 
     std::vector<cudf::device_span<uint8_t const>> dict_page_data;
@@ -270,7 +275,11 @@ auto filter_row_groups_with_dictionaries_impl(InputType& inputs,
 
     [[maybe_unused]] auto [dict_page_buffers, dict_page_data, dict_page_tasks] =
       cudf::io::parquet::fetch_byte_ranges_to_device_async(
-        inputs, dict_page_byte_ranges, stream, mr);
+        inputs,
+        dict_page_byte_ranges,
+        cudf::io::parquet::io_submission_policy::SERIALIZE,
+        stream,
+        mr);
     dict_page_tasks.get();
 
     return reader.filter_row_groups_with_dictionary_pages(
diff --git a/cpp/tests/io/experimental/hybrid_scan_composer.cpp b/cpp/tests/io/experimental/hybrid_scan_composer.cpp
index 7f39bc7121c7..af49b0bd2512 100644
--- a/cpp/tests/io/experimental/hybrid_scan_composer.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_composer.cpp
@@ -90,7 +90,11 @@ auto apply_hybrid_scan_filters(cudf::io::datasource& datasource,
     // Fetch dictionary page buffers from the input file buffer
     auto [dict_page_buffers, dict_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();
 
     // Filter row groups with dictionary pages
@@ -111,7 +115,11 @@ auto apply_hybrid_scan_filters(cudf::io::datasource& datasource,
 
     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();
 
     // Filter row groups with bloom filters
@@ -167,7 +175,11 @@ std::tuple<std::unique_ptr<cudf::table>, std::unique_ptr<cudf::table>> hybrid_sc
   // Fetch column chunk device buffers and spans from the input buffer
   auto [filter_col_buffers, filter_col_data, filter_col_tasks] =
     cudf::io::parquet::fetch_byte_ranges_to_device_async(
-      datasource, filter_column_chunk_byte_ranges, stream, mr);
+      datasource,
+      filter_column_chunk_byte_ranges,
+      cudf::io::parquet::io_submission_policy::SERIALIZE,
+      stream,
+      mr);
   filter_col_tasks.get();
 
   // Materialize the table with only the filter columns
@@ -188,7 +200,11 @@ std::tuple<std::unique_ptr<cudf::table>, std::unique_ptr<cudf::table>> hybrid_sc
   // Fetch column chunk device buffers and spans from the input buffer
   auto [payload_col_buffers, payload_col_data, payload_col_tasks] =
     cudf::io::parquet::fetch_byte_ranges_to_device_async(
-      datasource, payload_column_chunk_byte_ranges, stream, mr);
+      datasource,
+      payload_column_chunk_byte_ranges,
+      cudf::io::parquet::io_submission_policy::SERIALIZE,
+      stream,
+      mr);
   payload_col_tasks.get();
 
   // Materialize the table with only the payload columns
@@ -259,7 +275,11 @@ std::tuple<std::unique_ptr<cudf::table>, std::unique_ptr<cudf::table>> chunked_h
       reader->filter_column_chunks_byte_ranges(row_group_indices, options);
     auto [filter_buffers, filter_data, filter_tasks] =
       cudf::io::parquet::fetch_byte_ranges_to_device_async(
-        datasource, filter_byte_ranges, stream, mr);
+        datasource,
+        filter_byte_ranges,
+        cudf::io::parquet::io_submission_policy::SERIALIZE,
+        stream,
+        mr);
     filter_tasks.get();
 
     reader->setup_chunking_for_filter_columns(
@@ -282,7 +302,11 @@ std::tuple<std::unique_ptr<cudf::table>, std::unique_ptr<cudf::table>> chunked_h
       reader->payload_column_chunks_byte_ranges(row_group_indices, options);
     auto [payload_buffers, payload_data, payload_tasks] =
       cudf::io::parquet::fetch_byte_ranges_to_device_async(
-        datasource, payload_byte_ranges, stream, mr);
+        datasource,
+        payload_byte_ranges,
+        cudf::io::parquet::io_submission_policy::SERIALIZE,
+        stream,
+        mr);
     payload_tasks.get();
 
     reader->setup_chunking_for_payload_columns(
@@ -351,7 +375,11 @@ std::unique_ptr<cudf::table> hybrid_scan_single_step(
   // Fetch column chunk device buffers and spans from the input buffer
   auto [all_col_buffers, all_col_data, all_col_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_col_tasks.get();
 
   // Materialize the table with all columns
@@ -394,7 +422,11 @@ std::unique_ptr<cudf::table> chunked_hybrid_scan_single_step(
         reader->all_column_chunks_byte_ranges(row_group_indices, options);
       auto [all_column_chunk_buffers, all_column_chunk_data, all_column_chunk_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_tasks.get();
 
       // Setup chunking for all columns and materialize the columns
diff --git a/cpp/tests/io/experimental/hybrid_scan_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_filters_test.cpp
index aa57288646c5..3be1a8222c1f 100644
--- a/cpp/tests/io/experimental/hybrid_scan_filters_test.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_filters_test.cpp
@@ -892,7 +892,8 @@ TEST_F(HybridScanFiltersTest, OffsetIndexOnlyDataPageMask)
   auto const byte_ranges =
     offset_only_reader.payload_column_chunks_byte_ranges(selected_row_groups, options);
   auto [column_buffers, column_data, read_tasks] =
-    cudf::io::parquet::fetch_byte_ranges_to_device_async(*datasource, byte_ranges, stream, mr);
+    cudf::io::parquet::fetch_byte_ranges_to_device_async(
+      *datasource, byte_ranges, cudf::io::parquet::io_submission_policy::SERIALIZE, stream, mr);
   read_tasks.get();
 
   // Materialization maps the row mask to pages using only offset index, then applies the row mask.
@@ -918,7 +919,8 @@ TEST_F(HybridScanFiltersTest, OffsetIndexOnlyDataPageMask)
   auto const no_index_ranges =
     no_index_reader.payload_column_chunks_byte_ranges(no_index_row_groups, options);
   auto [no_index_buffers, no_index_data, no_index_tasks] =
-    cudf::io::parquet::fetch_byte_ranges_to_device_async(*datasource, no_index_ranges, stream, mr);
+    cudf::io::parquet::fetch_byte_ranges_to_device_async(
+      *datasource, no_index_ranges, cudf::io::parquet::io_submission_policy::SERIALIZE, stream, mr);
   no_index_tasks.get();
   auto const no_index_result = no_index_reader.materialize_payload_columns(
     no_index_row_groups,
@@ -1828,6 +1830,7 @@ TEST_F(HybridScanFiltersTest, FetchByteRangesInvalidRanges)
     cudf::io::parquet::fetch_byte_ranges_to_device_async(
       *datasource,
       std::vector<cudf::io::text::byte_range_info>{cudf::io::text::byte_range_info{-1, 16}},
+      cudf::io::parquet::io_submission_policy::SERIALIZE,
       stream,
       mr),
     cudf::logic_error);
@@ -1836,6 +1839,7 @@ TEST_F(HybridScanFiltersTest, FetchByteRangesInvalidRanges)
     cudf::io::parquet::fetch_byte_ranges_to_device_async(
       *datasource,
       std::vector<cudf::io::text::byte_range_info>{cudf::io::text::byte_range_info{512, 1024}},
+      cudf::io::parquet::io_submission_policy::SERIALIZE,
       stream,
       mr),
     cudf::logic_error);
@@ -1844,6 +1848,7 @@ TEST_F(HybridScanFiltersTest, FetchByteRangesInvalidRanges)
     cudf::io::parquet::fetch_byte_ranges_to_device_async(
       *datasource,
       std::vector<cudf::io::text::byte_range_info>{cudf::io::text::byte_range_info{0, -1}},
+      cudf::io::parquet::io_submission_policy::SERIALIZE,
       stream,
       mr),
     cudf::logic_error);
@@ -1851,6 +1856,7 @@ TEST_F(HybridScanFiltersTest, FetchByteRangesInvalidRanges)
   EXPECT_NO_THROW(cudf::io::parquet::fetch_byte_ranges_to_device_async(
     *datasource,
     std::vector<cudf::io::text::byte_range_info>{cudf::io::text::byte_range_info{1023, 1}},
+    cudf::io::parquet::io_submission_policy::SERIALIZE,
     stream,
     mr));
 }
diff --git a/cpp/tests/io/experimental/hybrid_scan_test.cpp b/cpp/tests/io/experimental/hybrid_scan_test.cpp
index b4b8c212f7f0..fd6dbac11fab 100644
--- a/cpp/tests/io/experimental/hybrid_scan_test.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_test.cpp
@@ -908,10 +908,10 @@ TEST_F(HybridScanTest, DecimalTypeOption)
     auto reader = std::make_unique<cudf::io::parquet::experimental::hybrid_scan_reader>(
       *footer_buffer, options);
 
-    auto const row_groups   = reader->all_row_groups(options);
-    auto const chunk_ranges = reader->all_column_chunks_byte_ranges(row_groups, options);
-    auto [buffers, col_data, tasks] =
-      cudf::io::parquet::fetch_byte_ranges_to_device_async(*datasource, chunk_ranges, stream, mr);
+    auto const row_groups           = reader->all_row_groups(options);
+    auto const chunk_ranges         = reader->all_column_chunks_byte_ranges(row_groups, options);
+    auto [buffers, col_data, tasks] = cudf::io::parquet::fetch_byte_ranges_to_device_async(
+      *datasource, chunk_ranges, cudf::io::parquet::io_submission_policy::SERIALIZE, stream, mr);
     tasks.get();
 
     return reader->materialize_all_columns(row_groups, col_data, options, stream, mr);
@@ -974,7 +974,11 @@ TEST_F(HybridScanTest, StructChildFilterColumn)
   auto const filter_byte_ranges = reader->filter_column_chunks_byte_ranges(row_groups, options);
   auto [filter_bufs, filter_data, filter_tasks] =
     cudf::io::parquet::fetch_byte_ranges_to_device_async(
-      *datasource, filter_byte_ranges, stream, mr);
+      *datasource,
+      filter_byte_ranges,
+      cudf::io::parquet::io_submission_policy::SERIALIZE,
+      stream,
+      mr);
   filter_tasks.get();
 
   using cudf::io::parquet::experimental::use_data_page_mask;
@@ -1007,10 +1011,10 @@ TEST_F(HybridScanTest, SharedMetadataReaderMatchesReadParquet)
   auto const read_all_columns = [&] {
     auto const reader =
       std::make_unique<cudf::io::parquet::experimental::hybrid_scan_reader>(metadata);
-    auto const row_groups   = reader->all_row_groups(options);
-    auto const chunk_ranges = reader->all_column_chunks_byte_ranges(row_groups, options);
-    auto [buffers, data, tasks] =
-      cudf::io::parquet::fetch_byte_ranges_to_device_async(*datasource, chunk_ranges, stream, mr);
+    auto const row_groups       = reader->all_row_groups(options);
+    auto const chunk_ranges     = reader->all_column_chunks_byte_ranges(row_groups, options);
+    auto [buffers, data, tasks] = cudf::io::parquet::fetch_byte_ranges_to_device_async(
+      *datasource, chunk_ranges, cudf::io::parquet::io_submission_policy::SERIALIZE, stream, mr);
     tasks.get();
     return reader->materialize_all_columns(row_groups, data, options, stream, mr).tbl;
   };
@@ -1057,10 +1061,10 @@ TEST_F(HybridScanTest, SharedMetadataFromFileMetaDataMatchesReadParquet)
   auto const read_all_columns = [&] {
     auto const reader =
       std::make_unique<cudf::io::parquet::experimental::hybrid_scan_reader>(metadata);
-    auto const row_groups   = reader->all_row_groups(options);
-    auto const chunk_ranges = reader->all_column_chunks_byte_ranges(row_groups, options);
-    auto [buffers, data, tasks] =
-      cudf::io::parquet::fetch_byte_ranges_to_device_async(*datasource, chunk_ranges, stream, mr);
+    auto const row_groups       = reader->all_row_groups(options);
+    auto const chunk_ranges     = reader->all_column_chunks_byte_ranges(row_groups, options);
+    auto [buffers, data, tasks] = cudf::io::parquet::fetch_byte_ranges_to_device_async(
+      *datasource, chunk_ranges, cudf::io::parquet::io_submission_policy::SERIALIZE, stream, mr);
     tasks.get();
     return reader->materialize_all_columns(row_groups, data, options, stream, mr).tbl;
   };
@@ -1131,9 +1135,9 @@ TEST_F(HybridScanTest, SharedMetadataConcurrentReadersMatchReadParquet)
     std::vector<cudf::size_type>(all_row_groups.begin() + split, all_row_groups.end());
 
   auto const materialize = [&](auto const& reader, auto const& row_group_indices) {
-    auto const chunk_ranges = reader->all_column_chunks_byte_ranges(row_group_indices, options);
-    auto [buffers, data, tasks] =
-      cudf::io::parquet::fetch_byte_ranges_to_device_async(*datasource, chunk_ranges, stream, mr);
+    auto const chunk_ranges     = reader->all_column_chunks_byte_ranges(row_group_indices, options);
+    auto [buffers, data, tasks] = cudf::io::parquet::fetch_byte_ranges_to_device_async(
+      *datasource, chunk_ranges, cudf::io::parquet::io_submission_policy::SERIALIZE, stream, mr);
     tasks.get();
     return reader->materialize_all_columns(row_group_indices, data, options, stream, mr).tbl;
   };
@@ -1201,7 +1205,11 @@ TEST_F(HybridScanTest, AllRowsPrunedReportsInputRowGroups)
   auto const filter_byte_ranges = reader->filter_column_chunks_byte_ranges(row_groups, options);
   auto [filter_buffers, filter_data, filter_tasks] =
     cudf::io::parquet::fetch_byte_ranges_to_device_async(
-      *datasource, filter_byte_ranges, stream, mr);
+      *datasource,
+      filter_byte_ranges,
+      cudf::io::parquet::io_submission_policy::SERIALIZE,
+      stream,
+      mr);
   filter_tasks.get();
 
   auto const filter_result = reader->materialize_filter_columns(
@@ -1214,7 +1222,11 @@ TEST_F(HybridScanTest, AllRowsPrunedReportsInputRowGroups)
   auto const payload_byte_ranges = reader->payload_column_chunks_byte_ranges(row_groups, options);
   auto [payload_buffers, payload_data, payload_tasks] =
     cudf::io::parquet::fetch_byte_ranges_to_device_async(
-      *datasource, payload_byte_ranges, stream, mr);
+      *datasource,
+      payload_byte_ranges,
+      cudf::io::parquet::io_submission_policy::SERIALIZE,
+      stream,
+      mr);
   payload_tasks.get();
 
   auto const payload_result = reader->materialize_payload_columns(
@@ -1274,7 +1286,11 @@ TEST_F(HybridScanTest, ChunkedReadRowMaskPerPass)
       reader->filter_column_chunks_byte_ranges(row_group_indices, options);
     auto [filter_buffers, filter_data, filter_tasks] =
       cudf::io::parquet::fetch_byte_ranges_to_device_async(
-        *datasource, filter_byte_ranges, stream, mr);
+        *datasource,
+        filter_byte_ranges,
+        cudf::io::parquet::io_submission_policy::SERIALIZE,
+        stream,
+        mr);
     filter_tasks.get();
 
     filter_tables.push_back(
@@ -1293,7 +1309,11 @@ TEST_F(HybridScanTest, ChunkedReadRowMaskPerPass)
       reader->payload_column_chunks_byte_ranges(row_group_indices, options);
     auto [payload_buffers, payload_data, payload_tasks] =
       cudf::io::parquet::fetch_byte_ranges_to_device_async(
-        *datasource, payload_byte_ranges, stream, mr);
+        *datasource,
+        payload_byte_ranges,
+        cudf::io::parquet::io_submission_policy::SERIALIZE,
+        stream,
+        mr);
     payload_tasks.get();
 
     payload_tables.push_back(
@@ -1381,7 +1401,11 @@ TEST_F(HybridScanTest, RowGroupPassesMatchesChunkedReader)
       auto const chunk_byte_ranges =
         reader->all_column_chunks_byte_ranges(pass_row_groups, options);
       auto [buffers, col_data, tasks] = cudf::io::parquet::fetch_byte_ranges_to_device_async(
-        *datasource, chunk_byte_ranges, stream, mr);
+        *datasource,
+        chunk_byte_ranges,
+        cudf::io::parquet::io_submission_policy::SERIALIZE,
+        stream,
+        mr);
       tasks.get();
 
       reader->setup_chunking_for_all_columns(
diff --git a/cpp/tests/streams/io/experimental/hybrid_scan_test.cpp b/cpp/tests/streams/io/experimental/hybrid_scan_test.cpp
index 3447b36ca254..5c1ef232109b 100644
--- a/cpp/tests/streams/io/experimental/hybrid_scan_test.cpp
+++ b/cpp/tests/streams/io/experimental/hybrid_scan_test.cpp
@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -119,10 +119,8 @@ TEST_F(HybridScanTest, DictionaryPageFiltering)
   auto const dict_byte_ranges =
     std::get<1>(reader->secondary_filters_byte_ranges(input_row_group_indices, in_opts));
   auto [dict_page_buffers, dict_page_data, dict_page_tasks] =
-    cudf::io::parquet::fetch_byte_ranges_to_device_async(datasource_ref,
-                                                         dict_byte_ranges,
-                                                         cudf::test::get_default_stream(),
-                                                         cudf::get_current_device_resource_ref());
+    cudf::io::parquet::fetch_byte_ranges_to_device_async(
+      datasource_ref, dict_byte_ranges, cudf::io::parquet::io_submission_policy::SERIALIZE);
   dict_page_tasks.get();
 
   auto result = reader->filter_row_groups_with_dictionary_pages(
diff --git a/python/cudf_polars/cudf_polars/streaming/io.py b/python/cudf_polars/cudf_polars/streaming/io.py
index b2dd79727af2..029e2eb0b678 100644
--- a/python/cudf_polars/cudf_polars/streaming/io.py
+++ b/python/cudf_polars/cudf_polars/streaming/io.py
@@ -262,7 +262,10 @@ def _read_with_hybrid_scan(
                 )
                 if bloom_ranges:
                     bloom_chunks = plc.io.parquet_io_utils.fetch_byte_ranges_to_device(
-                        source_info, bloom_ranges, stream=stream
+                        source_info,
+                        bloom_ranges,
+                        plc.io.parquet_io_utils.IOSubmissionPolicy.INTERLEAVE,
+                        stream=stream,
                     )
                     row_group_indices = reader.filter_row_groups_with_bloom_filters(
                         bloom_chunks, row_group_indices, options, stream=stream
@@ -293,6 +296,7 @@ def _read_with_hybrid_scan(
         filter_chunks = plc.io.parquet_io_utils.fetch_byte_ranges_to_device(
             source_info,
             reader.filter_column_chunks_byte_ranges(row_group_indices, options),
+            plc.io.parquet_io_utils.IOSubmissionPolicy.INTERLEAVE,
             stream=stream,
         )
         filter_tbl_w_meta = reader.materialize_filter_columns(
@@ -318,6 +322,7 @@ def _read_with_hybrid_scan(
             payload_chunks = plc.io.parquet_io_utils.fetch_byte_ranges_to_device(
                 source_info,
                 reader.payload_column_chunks_byte_ranges(row_group_indices, options),
+                plc.io.parquet_io_utils.IOSubmissionPolicy.INTERLEAVE,
                 stream=stream,
             )
             payload_tbl_w_meta = reader.materialize_payload_columns(
diff --git a/python/pylibcudf/pylibcudf/io/parquet_io_utils.pxd b/python/pylibcudf/pylibcudf/io/parquet_io_utils.pxd
index e8d872e08f61..ac775fd61b8b 100644
--- a/python/pylibcudf/pylibcudf/io/parquet_io_utils.pxd
+++ b/python/pylibcudf/pylibcudf/io/parquet_io_utils.pxd
@@ -3,11 +3,13 @@
 
 from pylibcudf.io.text cimport ByteRangeInfo
 from pylibcudf.io.types cimport SourceInfo
+from pylibcudf.libcudf.io.parquet_io_utils cimport io_submission_policy
 from rmm.pylibrmm.memory_resource cimport DeviceMemoryResource
 
 cpdef list fetch_byte_ranges_to_device(
     SourceInfo source_info,
     list byte_ranges,
+    io_submission_policy policy,
     object stream=*,
     DeviceMemoryResource mr=*,
 )
diff --git a/python/pylibcudf/pylibcudf/io/parquet_io_utils.pyi b/python/pylibcudf/pylibcudf/io/parquet_io_utils.pyi
index 1a18ab72f5d5..f7d2fb56876e 100644
--- a/python/pylibcudf/pylibcudf/io/parquet_io_utils.pyi
+++ b/python/pylibcudf/pylibcudf/io/parquet_io_utils.pyi
@@ -1,6 +1,8 @@
 # SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
+from enum import IntEnum
+
 from rmm.pylibrmm.memory_resource import DeviceMemoryResource
 
 from pylibcudf.gpumemoryview import gpumemoryview
@@ -8,11 +10,20 @@ from pylibcudf.io.text import ByteRangeInfo
 from pylibcudf.io.types import SourceInfo
 from pylibcudf.utils import CudaStreamLike
 
-__all__ = ["fetch_byte_ranges_to_device", "fetch_page_index_to_host"]
+__all__ = [
+    "IOSubmissionPolicy",
+    "fetch_byte_ranges_to_device",
+    "fetch_page_index_to_host",
+]
+
+class IOSubmissionPolicy(IntEnum):
+    SERIALIZE = 0
+    INTERLEAVE = 1
 
 def fetch_byte_ranges_to_device(
     source_info: SourceInfo,
     byte_ranges: list[ByteRangeInfo],
+    policy: IOSubmissionPolicy,
     stream: CudaStreamLike | None = None,
     mr: DeviceMemoryResource | None = None,
 ) -> list[gpumemoryview]: ...
diff --git a/python/pylibcudf/pylibcudf/io/parquet_io_utils.pyx b/python/pylibcudf/pylibcudf/io/parquet_io_utils.pyx
index febbd0541c8f..f546508527f2 100644
--- a/python/pylibcudf/pylibcudf/io/parquet_io_utils.pyx
+++ b/python/pylibcudf/pylibcudf/io/parquet_io_utils.pyx
@@ -24,6 +24,7 @@ from pylibcudf.libcudf.io.parquet_io_utils cimport (
     const_uint8_t,
     cpp_fetch_byte_ranges_to_device,
     fetch_page_index_to_host as cpp_fetch_page_index_to_host,
+    io_submission_policy as cpp_io_submission_policy,
 )
 
 from pylibcudf.libcudf.io.text cimport byte_range_info
@@ -34,12 +35,21 @@ from typing import TYPE_CHECKING
 if TYPE_CHECKING:
     from pylibcudf.typing import CudaStreamLike
 
-__all__ = ["fetch_byte_ranges_to_device", "fetch_page_index_to_host"]
+import pylibcudf.libcudf.io.parquet_io_utils
+
+IOSubmissionPolicy = pylibcudf.libcudf.io.parquet_io_utils.io_submission_policy
+
+__all__ = [
+    "IOSubmissionPolicy",
+    "fetch_byte_ranges_to_device",
+    "fetch_page_index_to_host",
+]
 
 
 cpdef list fetch_byte_ranges_to_device(
     SourceInfo source_info,
     list byte_ranges,
+    cpp_io_submission_policy policy,
     object stream: CudaStreamLike | None = None,
     DeviceMemoryResource mr=None,
 ):
@@ -55,6 +65,8 @@ cpdef list fetch_byte_ranges_to_device(
         :meth:`~pylibcudf.io.experimental.HybridScanReader.payload_column_chunks_byte_ranges`,
         or
         :meth:`~pylibcudf.io.experimental.HybridScanReader.all_column_chunks_byte_ranges`.
+    policy : IOSubmissionPolicy
+        Whether to serialize I/O submissions from callers.
     stream : Stream, optional
         CUDA stream.
     mr : DeviceMemoryResource, optional
@@ -91,6 +103,7 @@ cpdef list fetch_byte_ranges_to_device(
         fetched = cpp_fetch_byte_ranges_to_device(
             dereference(sources[0]),
             host_span[const_byte_range_info](ranges_vec.data(), ranges_vec.size()),
+            policy,
             _stream.view(),
             _mr.get_mr(),
         )
diff --git a/python/pylibcudf/pylibcudf/libcudf/io/CMakeLists.txt b/python/pylibcudf/pylibcudf/libcudf/io/CMakeLists.txt
index c04d78107251..8071828544fb 100644
--- a/python/pylibcudf/pylibcudf/libcudf/io/CMakeLists.txt
+++ b/python/pylibcudf/pylibcudf/libcudf/io/CMakeLists.txt
@@ -1,11 +1,11 @@
 # =============================================================================
 # cmake-format: off
-# SPDX-FileCopyrightText: Copyright (c) 2024-2025, NVIDIA CORPORATION.
+# SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 # cmake-format: on
 # =============================================================================
 
-set(cython_sources hybrid_scan.pyx json.pyx types.pyx)
+set(cython_sources hybrid_scan.pyx json.pyx parquet_io_utils.pyx types.pyx)
 
 set(linked_libraries cudf::cudf)
 
diff --git a/python/pylibcudf/pylibcudf/libcudf/io/parquet_io_utils.pxd b/python/pylibcudf/pylibcudf/libcudf/io/parquet_io_utils.pxd
index 47f871439b99..b5061dce9937 100644
--- a/python/pylibcudf/pylibcudf/libcudf/io/parquet_io_utils.pxd
+++ b/python/pylibcudf/pylibcudf/libcudf/io/parquet_io_utils.pxd
@@ -18,6 +18,13 @@ from pylibcudf.libcudf.utilities.span cimport device_span, host_span
 ctypedef const uint8_t const_uint8_t
 ctypedef const byte_range_info const_byte_range_info
 
+cdef extern from "cudf/io/parquet_io_utils.hpp" \
+        namespace "cudf::io::parquet" nogil:
+
+    cpdef enum class io_submission_policy:
+        SERIALIZE
+        INTERLEAVE
+
 cdef extern from * nogil:
     """
     #include <future>
@@ -28,12 +35,17 @@ cdef extern from * nogil:
     cpp_fetch_byte_ranges_to_device(
         cudf::io::datasource& datasource,
         cudf::host_span<cudf::io::text::byte_range_info const> byte_ranges,
+        cudf::io::parquet::io_submission_policy policy,
         rmm::cuda_stream_view stream,
         rmm::device_async_resource_ref mr)
     {
         auto [buffers, spans, fut] =
             cudf::io::parquet::fetch_byte_ranges_to_device_async(
-                datasource, byte_ranges, stream, mr);
+                datasource,
+                byte_ranges,
+                policy,
+                stream,
+                mr);
         // Block until the async fetch completes so the returned buffers/spans
         // are fully populated. This also avoids exposing std::future to Cython.
         fut.get();
@@ -44,6 +56,7 @@ cdef extern from * nogil:
         cpp_fetch_byte_ranges_to_device(
             datasource& source,
             host_span[const_byte_range_info] byte_ranges,
+            io_submission_policy policy,
             cuda_stream_view stream,
             device_async_resource_ref mr,
         ) except +libcudf_exception_handler
diff --git a/python/pylibcudf/pylibcudf/libcudf/io/parquet_io_utils.pyx b/python/pylibcudf/pylibcudf/libcudf/io/parquet_io_utils.pyx
new file mode 100644
index 000000000000..d51c4fe1e089
--- /dev/null
+++ b/python/pylibcudf/pylibcudf/libcudf/io/parquet_io_utils.pyx
@@ -0,0 +1,2 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0