Use pinned memory in PQ reader to avoid pageable copies (#20820)

Contributes to #20722

This PR replaces the use of small host vectors with pinned vectors to avoid pageable copies and improve pipeline performance when reading parquet files using multiple threads (each using a separate non-blocking stream)

Authors:
  - Muhammad Haseeb (https://github.com/mhaseeb123)

Approvers:
  - Bradley Dice (https://github.com/bdice)
  - Nghia Truong (https://github.com/ttnghia)
  - Vukasin Milovanovic (https://github.com/vuule)

URL: #20820

Author: mhaseeb123

cpp/benchmarks/io/parquet/experimental/parquet_dictionary_page_filter.cpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2022-2025, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -94,7 +94,6 @@ std::vector<rmm::device_buffer> fetch_byte_ranges(
       return buffer;
     });
 
-  stream.synchronize_no_throw();
   return buffers;
 }
 

cpp/src/bitmask/null_mask.cu

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2019-2025, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -205,7 +205,8 @@ void set_null_masks(cudf::host_span<bitmask_type*> bitmasks,
 
   size_t average_nullmask_words     = 0;
   size_t cumulative_null_mask_words = 0;
-  auto h_number_of_mask_words = cudf::detail::make_host_vector<size_type>(num_bitmasks, stream);
+  auto h_number_of_mask_words =
+    cudf::detail::make_pinned_vector_async<size_type>(num_bitmasks, stream);
   thrust::tabulate(
     thrust::host, h_number_of_mask_words.begin(), h_number_of_mask_words.end(), [&](auto i) {
       CUDF_EXPECTS(begin_bits[i] >= 0, "Invalid range.");

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

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -10,8 +10,6 @@
 
 #include <cudf/detail/iterator.cuh>
 #include <cudf/detail/nvtx/ranges.hpp>
-#include <cudf/detail/utilities/batched_memset.hpp>
-#include <cudf/detail/utilities/vector_factories.hpp>
 #include <cudf/io/parquet_schema.hpp>
 #include <cudf/types.hpp>
 #include <cudf/utilities/memory_resource.hpp>

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

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -1092,7 +1092,7 @@ thrust::host_vector<bool> aggregate_reader_metadata::compute_data_page_mask(
   auto tree_levels_data = rmm::device_uvector<bool>(tree_level_offsets.back(), stream, mr);
 
   // Pointers to each Fenwick tree level data
-  auto host_tree_level_ptrs = cudf::detail::make_host_vector<bool*>(num_levels, stream);
+  auto host_tree_level_ptrs = cudf::detail::make_pinned_vector_async<bool*>(num_levels, stream);
   // Zeroth level is just the row mask itself
   host_tree_level_ptrs[0] = const_cast<bool*>(row_mask.template begin<bool>()) + row_mask_offset;
   std::for_each(
@@ -1122,7 +1122,11 @@ thrust::host_vector<bool> aggregate_reader_metadata::compute_data_page_mask(
   //  Search the Fenwick tree to see if there's a surviving row in each page's row range
   auto const num_ranges = static_cast<cudf::size_type>(page_row_offsets.size() - 1);
   rmm::device_uvector<bool> device_data_page_mask(num_ranges, stream, mr);
-  auto page_offsets = cudf::detail::make_device_uvector_async(page_row_offsets, stream, mr);
+  // Use a pinned bounce buffer to avoid pageable h2d copy
+  auto host_page_offsets =
+    cudf::detail::make_pinned_vector_async<cudf::size_type>(page_row_offsets.size(), stream);
+  std::move(page_row_offsets.begin(), page_row_offsets.end(), host_page_offsets.begin());
+  auto page_offsets = cudf::detail::make_device_uvector_async(host_page_offsets, stream, mr);
   thrust::transform(
     rmm::exec_policy_nosync(stream),
     thrust::counting_iterator(0),
@@ -1131,7 +1135,12 @@ thrust::host_vector<bool> aggregate_reader_metadata::compute_data_page_mask(
     search_fenwick_tree_functor{fenwick_tree_level_ptrs.data(), page_offsets.data(), num_ranges});
 
   //  Copy over search results to host
-  auto host_results      = cudf::detail::make_host_vector_async(device_data_page_mask, stream);
+  auto host_results =
+    cudf::detail::make_pinned_vector_async<bool>(device_data_page_mask.size(), stream);
+  cudf::detail::cuda_memcpy_async(
+    cudf::host_span<bool>(host_results.data(), host_results.size()),
+    cudf::device_span<bool const>(device_data_page_mask.data(), device_data_page_mask.size()),
+    stream);
   auto const total_pages = page_row_offsets.size() - num_columns;
   auto data_page_mask    = thrust::host_vector<bool>(total_pages, stream);
   auto host_results_iter = host_results.begin();

cpp/src/io/parquet/predicate_pushdown.cpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2023-2025, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2023-2026, NVIDIA CORPORATION.
  * SPDX-License-Identifier: Apache-2.0
  */
 #include "reader_impl_helpers.hpp"
@@ -426,11 +426,15 @@ std::optional<std::vector<std::vector<size_type>>> collect_filtered_row_group_in
   CUDF_EXPECTS(predicate.type().id() == cudf::type_id::BOOL8,
                "Filter expression must return a boolean column");
 
-  auto const host_bitmask = [&] {
-    auto const num_bitmasks = num_bitmask_words(predicate.size());
+  auto host_bitmask = [&] {
+    std::size_t const num_bitmasks = num_bitmask_words(predicate.size());
     if (predicate.nullable()) {
-      return cudf::detail::make_host_vector(
-        device_span<bitmask_type const>(predicate.null_mask(), num_bitmasks), stream);
+      auto bitmask = cudf::detail::make_pinned_vector_async<bitmask_type>(num_bitmasks, stream);
+      cudf::detail::cuda_memcpy(
+        cudf::host_span<bitmask_type>{bitmask.data(), num_bitmasks},
+        cudf::device_span<bitmask_type const>{predicate.null_mask(), num_bitmasks},
+        stream);
+      return bitmask;
     } else {
       auto bitmask = cudf::detail::make_host_vector<bitmask_type>(num_bitmasks, stream);
       std::fill(bitmask.begin(), bitmask.end(), ~bitmask_type{0});
@@ -442,8 +446,12 @@ std::optional<std::vector<std::vector<size_type>>> collect_filtered_row_group_in
     0, [bitmask = host_bitmask.data()](auto bit_index) { return bit_is_set(bitmask, bit_index); });
 
   // Return only filtered row groups based on predicate
-  auto const is_row_group_required = cudf::detail::make_host_vector(
-    device_span<uint8_t const>(predicate.data<uint8_t>(), predicate.size()), stream);
+  auto is_row_group_required =
+    cudf::detail::make_pinned_vector_async<uint8_t>(predicate.size(), stream);
+  cudf::detail::cuda_memcpy(
+    cudf::host_span<uint8_t>(is_row_group_required.data(), predicate.size()),
+    cudf::device_span<uint8_t const>(predicate.data<uint8_t>(), predicate.size()),
+    stream);
 
   // Return if all are required, or all are nulls.
   if (predicate.null_count() == predicate.size() or std::all_of(is_row_group_required.cbegin(),

cpp/src/io/parquet/reader_impl.cpp

@@ -184,7 +184,7 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_
   if (has_strings) {
     // Host vector to initialize the initial string offsets
     auto host_offsets_vector =
-      cudf::detail::make_host_vector<size_t>(_input_columns.size(), _stream);
+      cudf::detail::make_pinned_vector_async<size_t>(_input_columns.size(), _stream);
     std::fill(
       host_offsets_vector.begin(), host_offsets_vector.end(), std::numeric_limits<size_t>::max());
     // Initialize the initial string offsets vector from the host vector
@@ -397,7 +397,12 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_
   update_output_nullmasks_for_pruned_pages(_subpass_page_mask, skip_rows, num_rows);
 
   // Copy over initial string offsets from device
-  auto h_initial_str_offsets = cudf::detail::make_host_vector_async(initial_str_offsets, _stream);
+  auto h_initial_str_offsets =
+    cudf::detail::make_pinned_vector_async<size_t>(initial_str_offsets.size(), _stream);
+  cudf::detail::cuda_memcpy_async(
+    cudf::host_span<size_t>(h_initial_str_offsets.data(), initial_str_offsets.size()),
+    cudf::device_span<size_t const>(initial_str_offsets.data(), initial_str_offsets.size()),
+    _stream);
 
   if (auto const error = error_code.value_sync(_stream); error != 0) {
     CUDF_FAIL("Parquet data decode failed with code(s) " + kernel_error::to_string(error));
@@ -427,14 +432,14 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_
 
         // the final offset for a list at level N is the size of it's child
         size_type const offset = child.type.id() == type_id::LIST ? child.size - 1 : child.size;
-        out_buffers.emplace_back(static_cast<size_type*>(out_buf.data()) + (out_buf.size - 1));
-        final_offsets.emplace_back(offset);
+        out_buffers.push_back(static_cast<size_type*>(out_buf.data()) + (out_buf.size - 1));
+        final_offsets.push_back(offset);
         out_buf.user_data |= PARQUET_COLUMN_BUFFER_FLAG_LIST_TERMINATED;
       } else if (out_buf.type.id() == type_id::STRING) {
         // only if it is not a large strings column
         if (std::cmp_less_equal(col_string_sizes[idx], strings::detail::get_offset64_threshold())) {
-          out_buffers.emplace_back(static_cast<size_type*>(out_buf.data()) + out_buf.size);
-          final_offsets.emplace_back(static_cast<size_type>(col_string_sizes[idx]));
+          out_buffers.push_back(static_cast<size_type*>(out_buf.data()) + out_buf.size);
+          final_offsets.push_back(static_cast<size_type>(col_string_sizes[idx]));
         }
         // Nested large strings column
         else if (input_col.nesting_depth() > 0) {
@@ -446,7 +451,13 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_
     }
   }
   // Write the final offsets for list and string columns in a batched manner
-  write_final_offsets(final_offsets, out_buffers, _stream);
+  auto pinned_final_offsets =
+    cudf::detail::make_pinned_vector_async<cudf::size_type>(final_offsets.size(), _stream);
+  auto pinned_out_buffers =
+    cudf::detail::make_pinned_vector_async<cudf::size_type*>(out_buffers.size(), _stream);
+  std::move(final_offsets.begin(), final_offsets.end(), pinned_final_offsets.begin());
+  std::move(out_buffers.begin(), out_buffers.end(), pinned_out_buffers.begin());
+  write_final_offsets(pinned_final_offsets, pinned_out_buffers, _stream);
 
   // update null counts in the final column buffers
   for (size_t idx = 0; idx < subpass.pages.size(); idx++) {
@@ -928,9 +939,9 @@ void reader_impl::update_output_nullmasks_for_pruned_pages(cudf::host_span<bool
   auto page_and_mask_begin =
     thrust::make_zip_iterator(cuda::std::make_tuple(pages.host_begin(), page_mask.begin()));
 
-  auto null_masks = std::vector<bitmask_type*>{};
-  auto begin_bits = std::vector<cudf::size_type>{};
-  auto end_bits   = std::vector<cudf::size_type>{};
+  auto host_null_masks = std::vector<bitmask_type*>{};
+  auto host_begin_bits = std::vector<cudf::size_type>{};
+  auto host_end_bits   = std::vector<cudf::size_type>{};
 
   std::for_each(
     page_and_mask_begin, page_and_mask_begin + pages.size(), [&](auto const& page_and_mask_pair) {
@@ -983,9 +994,9 @@ void reader_impl::update_output_nullmasks_for_pruned_pages(cudf::host_span<bool
         cols          = &out_buf.children;
         if (out_buf.user_data & PARQUET_COLUMN_BUFFER_FLAG_HAS_LIST_PARENT) { continue; }
         // Add the nullmask and bit bounds to corresponding lists
-        null_masks.emplace_back(out_buf.null_mask());
-        begin_bits.emplace_back(start_row);
-        end_bits.emplace_back(end_row);
+        host_null_masks.emplace_back(out_buf.null_mask());
+        host_begin_bits.emplace_back(start_row);
+        host_end_bits.emplace_back(end_row);
 
         // Increment the null count by the number of rows in this page
         out_buf.null_count() += page.num_rows;
@@ -995,9 +1006,20 @@ void reader_impl::update_output_nullmasks_for_pruned_pages(cudf::host_span<bool
   // Min number of nullmasks to use bulk update optimally
   constexpr auto min_nullmasks_for_bulk_update = 32;
 
+  // Use a bounce buffer to avoid pageable copies
+  auto null_masks =
+    cudf::detail::make_pinned_vector_async<bitmask_type*>(host_null_masks.size(), _stream);
+  auto begin_bits =
+    cudf::detail::make_pinned_vector_async<cudf::size_type>(host_begin_bits.size(), _stream);
+  auto end_bits =
+    cudf::detail::make_pinned_vector_async<cudf::size_type>(host_end_bits.size(), _stream);
+  std::move(host_null_masks.begin(), host_null_masks.end(), null_masks.begin());
+  std::move(host_begin_bits.begin(), host_begin_bits.end(), begin_bits.begin());
+  std::move(host_end_bits.begin(), host_end_bits.end(), end_bits.begin());
+
   // Bulk update the nullmasks if the number of pages is above the threshold
   if (null_masks.size() >= min_nullmasks_for_bulk_update) {
-    auto valids = cudf::detail::make_host_vector<bool>(null_masks.size(), _stream);
+    auto valids = cudf::detail::make_pinned_vector_async<bool>(null_masks.size(), _stream);
     std::fill(valids.begin(), valids.end(), false);
     cudf::set_null_masks_safe(null_masks, begin_bits, end_bits, valids, _stream);
   }

cpp/src/io/parquet/reader_impl_chunking.cu

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2023-2025, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2023-2026, NVIDIA CORPORATION.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -327,7 +327,11 @@ void reader_impl::setup_next_subpass(read_mode mode)
     subpass.pages = subpass.page_buf;
   }
 
-  auto const h_spans = cudf::detail::make_host_vector_async(page_indices, _stream);
+  auto h_spans = cudf::detail::make_pinned_vector_async<page_span>(page_indices.size(), _stream);
+  cudf::detail::cuda_memcpy_async(
+    cudf::host_span<page_span>{h_spans.data(), page_indices.size()},
+    cudf::device_span<page_span const>{page_indices.data(), page_indices.size()},
+    _stream);
   subpass.pages.device_to_host_async(_stream);
 
   _stream.synchronize();
@@ -690,7 +694,12 @@ void reader_impl::set_subpass_page_mask()
   }
 
   // Use the pass page index mask to gather the subpass page mask from the pass level page mask
-  auto const host_page_src_index = cudf::detail::make_host_vector(subpass->page_src_index, _stream);
+  auto host_page_src_index =
+    cudf::detail::make_pinned_vector_async<size_t>(subpass->page_src_index.size(), _stream);
+  cudf::detail::cuda_memcpy(
+    cudf::host_span<size_t>{host_page_src_index.data(), subpass->page_src_index.size()},
+    cudf::device_span<size_t const>{subpass->page_src_index.data(), subpass->page_src_index.size()},
+    _stream);
   thrust::gather(thrust::seq,
                  host_page_src_index.begin(),
                  host_page_src_index.end(),