Skip to content

Avoid serializing I/O requests from Parquet reader #23823

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mhaseeb123 wants to merge 13 commits into
base: main
Choose a base branch
from
+424 −82
Open

Avoid serializing I/O requests from Parquet reader #23823

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
0d9349d
Don't serialize IO submissions from Parquet reader
mhaseeb123 Aug 26, 2026
1c0a46a
Merge branch 'main' into fix/pq-reader-io-not-serialized
mhaseeb123 Aug 26, 2026
a83d197
Merge branch 'main' into fix/pq-reader-io-not-serialized
mhaseeb123 Aug 28, 2026
8113b6a
Add new Parquet IO utils to allow interleaving
mhaseeb123 Aug 28, 2026
bc4e5f0
style style style
mhaseeb123 Aug 28, 2026
579b6f4
Minor fix
mhaseeb123 Aug 28, 2026
0f0472b
Merge branch 'main' into fix/pq-reader-io-not-serialized
mhaseeb123 Aug 28, 2026
a712b58
Merge branch 'main' into fix/pq-reader-io-not-serialized
mhaseeb123 Aug 31, 2026
2b1c73f
Fix
mhaseeb123 Aug 31, 2026
87340d0
docs update
mhaseeb123 Aug 31, 2026
4447645
formatting
mhaseeb123 Sep 1, 2026
b467710
Address comments from @vuule
mhaseeb123 Sep 1, 2026
5abf786
doc update
mhaseeb123 Sep 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 12 additions & 4 deletions cpp/include/cudf/io/parquet_io_utils.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,7 @@ using cudf::io::text::byte_range_info;
* @param byte_ranges Byte ranges to fetch
* @param stream CUDA stream
* @param mr Device memory resource
* @param serialize_submissions Whether to serialize I/O submissions from callers
*
* @return A tuple containing the device buffers, the device spans of the fetched data, and a future
* to wait on the read tasks
Expand All @@ -129,7 +130,8 @@ std::tuple<std::vector<rmm::device_buffer>,
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);
rmm::device_async_resource_ref mr,
bool serialize_submissions = true);
Comment thread
vuule marked this conversation as resolved.
Outdated

/**
* @brief Fetches lists of byte ranges from multiple datasources into device buffers
Expand All @@ -140,6 +142,7 @@ fetch_byte_ranges_to_device_async(cudf::io::datasource& datasource,
* @param byte_ranges_per_source Vector of byte ranges to fetch, one per datasource
* @param stream CUDA stream
* @param mr Device memory resource
* @param serialize_submissions Whether to serialize I/O submissions from callers
*
* @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
Expand All @@ -151,7 +154,8 @@ 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);
rmm::device_async_resource_ref mr,
bool serialize_submissions = true);

/**
* @brief Fetches Parquet bloom filter bitsets from a datasource into device buffers
Expand All @@ -163,6 +167,7 @@ fetch_byte_ranges_to_device_async(
* complete bloom filter
* @param stream CUDA stream
* @param mr Device memory resource used to allocate the returned device buffers
* @param serialize_submissions Whether to serialize I/O submissions from callers
*
* @return A pair containing buffers that own the fetched bitsets and one device span per input byte
* range
Expand All @@ -171,7 +176,8 @@ std::pair<std::vector<rmm::device_buffer>, std::vector<cudf::device_span<uint8_t
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);
rmm::device_async_resource_ref mr,
bool serialize_submissions = true);

/**
* @brief Fetches Parquet bloom filter bitsets from multiple datasources into device buffers
Expand All @@ -183,6 +189,7 @@ fetch_bloom_filters_to_device(cudf::io::datasource& datasource,
* 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 serialize_submissions Whether to serialize I/O submissions from callers
*
* @return A pair containing buffers that own the fetched bitsets and per-source device spans, with
* one inner vector per datasource
Expand All @@ -193,7 +200,8 @@ 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);
rmm::device_async_resource_ref mr,
bool serialize_submissions = true);

/** @} */ // end of group
} // namespace io::parquet
Expand Down
4 changes: 2 additions & 2 deletions cpp/src/io/parquet/bloom_filter_reader.cu
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -384,8 +384,8 @@ aggregate_reader_metadata::read_bloom_filters(
return std::ref(*source);
});

auto [bloom_filter_buffers, bitset_spans_per_source] =
fetch_bloom_filters_to_device(datasource_refs, bloom_filter_byte_ranges_per_source, stream, mr);
auto [bloom_filter_buffers, bitset_spans_per_source] = fetch_bloom_filters_to_device(
datasource_refs, bloom_filter_byte_ranges_per_source, stream, mr, false);

// Flatten the per-source bitset spans into per-chunk order
std::vector<cudf::device_span<cuda::std::byte const>> bloom_filter_data;
Expand Down
51 changes: 34 additions & 17 deletions cpp/src/io/parquet/io_utils/parquet_io_utils.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Comment thread
mhaseeb123 marked this conversation as resolved.
{
std::vector<std::future<std::size_t>> host_read_tasks;
std::vector<std::size_t> expected_sizes;
Expand All @@ -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(); }
Comment thread
coderabbitai[bot] marked this conversation as resolved.

std::for_each(iter, iter + count, [&](auto const& tuple) {
auto const src_idx = cuda::std::get<0>(tuple);
Expand Down Expand Up @@ -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)
Comment thread
mhaseeb123 marked this conversation as resolved.
Outdated
{
auto const num_sources = datasources.size();

Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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

@bdice bdice Aug 28, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is surprising to me that serialization would yield better E2E throughput. Sure, pipelining might look better with nonoverlapping regions, but is there really enough contention for I/O or copy bandwidth to reduce E2E query throughput for many concurrent threads/streams? Do we have hard numbers on this that would justify serialization in certain cases? Is it true for both cloud and NVMe reads? If we do see reductions in throughput, are those influenced by I/O parameters like read sizes, thread pool sizes, etc? Inserting a mutex seems like the last thing we’d want to do, my naive expectation with no data is that we should rip out mutexes that are not required for thread safety / correctness, and instead we should optimize default parameters (in libcudf or engines using it) for concurrency instead.

@pmattione-nvidia pmattione-nvidia Aug 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It might depend entirely on the datasource, whether it's a local disk drive (SSDs are designed to be performant with high queue depths) or a resource across the network or on the cloud (bottlenecked, probably want serial so can decode the first file while you wait for the next read). It seems like this kind of decision belongs in the datasource itself, not the reader. @mhaseeb123

@vyasr vyasr Aug 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, this optimization was based on data, see #22550 (comment) and #22529 (comment). I doubt we tested the full range of cases that you're describing though. On the other hand, in the long run this discussion may be moot since hybrid scan will totally change who is managing the byte reads anyway.

@mhaseeb123 mhaseeb123 Aug 28, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In addition to what Vyas mentioned, both readers have been using these locking utils for a few months (even in Velox) for multithreaded cases and have seen mild speedups.

In hindsight, we should have made the regular reader use them without locking to begin with (unmodified behavior) but our microbenchmarks showed improvements with locking.

Why these APIs exist: Added strictly for libcudf-use across hybrid scan tests, benchmarks, examples as hybrid scan asks us to bring our own data feeder.

Unintended purpose: Velox's data feeder just uses these APis and the benchmarks show modest improvement with locking vs without, in multi-driver case in facebookincubator/velox#18602

{
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);
Expand Down Expand Up @@ -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)
Comment thread
mhaseeb123 marked this conversation as resolved.
Outdated
{
auto const num_sources = datasources.size();
CUDF_EXPECTS(num_sources == bloom_filter_byte_ranges_per_source.size(),
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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));
}
Expand Down Expand Up @@ -703,7 +712,8 @@ std::tuple<std::vector<rmm::device_buffer>,
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,
rmm::device_async_resource_ref mr)
rmm::device_async_resource_ref mr,
bool serialize_submissions)
{
CUDF_FUNC_RANGE();

Expand All @@ -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,
serialize_submissions);

return {std::move(buffers), std::move(fetched_byte_ranges.front()), std::move(fut)};
}
Expand All @@ -728,7 +739,8 @@ 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,
rmm::device_async_resource_ref mr)
rmm::device_async_resource_ref mr,
bool serialize_submissions)
{
CUDF_FUNC_RANGE();

Expand All @@ -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,
serialize_submissions);
}

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,
rmm::device_async_resource_ref mr)
rmm::device_async_resource_ref mr,
bool serialize_submissions)
{
CUDF_FUNC_RANGE();

Expand All @@ -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,
serialize_submissions);

return {std::move(buffers), std::move(fetched_byte_ranges.front())};
}
Expand All @@ -775,7 +790,8 @@ fetch_bloom_filters_to_device(
cudf::host_span<std::vector<cudf::io::text::byte_range_info> 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)
{
CUDF_FUNC_RANGE();

Expand All @@ -790,7 +806,8 @@ fetch_bloom_filters_to_device(
{bloom_filter_byte_range_spans_per_source.data(),
bloom_filter_byte_range_spans_per_source.size()},
stream,
mr);
mr,
serialize_submissions);
}

} // namespace cudf::io::parquet
3 changes: 2 additions & 1 deletion cpp/src/io/parquet/reader_impl_preprocess_utils.cu
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,7 +200,8 @@ void generate_depth_remappings(
{datasource_refs.data(), datasource_refs.size()},
{source_byte_ranges.data(), source_byte_ranges.size()},
stream,
mr);
mr,
false);

// Extract data pointers from returned spans
size_t range_idx = 0;
Expand Down
Loading