Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
35 commits
Select commit Hold shift + click to select a range
4dfc295
PERF/API: Specify datasource size to avoid HEAD requests for S3 endpo…
TomAugspurger Jun 1, 2026
ed52e1d
Fixups
TomAugspurger Jun 1, 2026
72c757e
Merge remote-tracking branch 'upstream/main' into tom/cudf-sourceinfo…
TomAugspurger Jun 1, 2026
27aadc0
const type
TomAugspurger Jun 1, 2026
a65f65d
test style
TomAugspurger Jun 1, 2026
3152205
Test fixup
TomAugspurger Jun 2, 2026
b087627
test fix
TomAugspurger Jun 2, 2026
2f34b6e
Merge remote-tracking branch 'upstream/main' into tom/cudf-sourceinfo…
TomAugspurger Jun 2, 2026
ddc0e31
Merge remote-tracking branch 'upstream/main' into tom/cudf-sourceinfo…
TomAugspurger Jun 5, 2026
7887199
Merge remote-tracking branch 'upstream/main' into tom/cudf-sourceinfo…
TomAugspurger Jun 9, 2026
62e8221
Build cudf_streaming against CI run's pylibcudf
TomAugspurger Jun 9, 2026
e0c4596
fix quoting
TomAugspurger Jun 9, 2026
56df08a
Merge remote-tracking branch 'upstream/main' into tom/cudf-sourceinfo…
TomAugspurger Jun 9, 2026
e76a02b
libcudf too
TomAugspurger Jun 9, 2026
a8afc20
Merge remote-tracking branch 'upstream/main' into tom/cudf-sourceinfo…
TomAugspurger Jun 15, 2026
e32eedd
Merge branch 'main' into tom/cudf-sourceinfo-size
TomAugspurger Jun 16, 2026
2c4a077
Merge remote-tracking branch 'upstream/main' into tom/cudf-sourceinfo…
TomAugspurger Jun 22, 2026
5b504da
CI
TomAugspurger Jun 22, 2026
e9ed5b2
Merge remote-tracking branch 'upstream/main' into tom/cudf-sourceinfo…
TomAugspurger Jun 22, 2026
dcda6e1
Match the siganture with remote_io off
TomAugspurger Jun 22, 2026
60a5ecd
reinclude memory
TomAugspurger Jun 22, 2026
b460be2
Remove unreachable public s3 endpoint type
TomAugspurger Jun 23, 2026
e20c06d
Merge remote-tracking branch 'upstream/main' into tom/cudf-sourceinfo…
TomAugspurger Jun 23, 2026
8c6f01a
Merge remote-tracking branch 'upstream/main' into tom/cudf-sourceinfo…
TomAugspurger Jun 23, 2026
4f89dc5
Merge remote-tracking branch 'upstream/main' into tom/cudf-sourceinfo…
TomAugspurger Jun 24, 2026
e65fb89
Use kvikio API for inferring the remote endpoint type
TomAugspurger Jun 24, 2026
82c2f19
Merge remote-tracking branch 'upstream/main' into tom/cudf-sourceinfo…
TomAugspurger Jun 24, 2026
c967399
Merge branch 'main' into tom/cudf-sourceinfo-size
TomAugspurger Jun 25, 2026
0bd481e
Merge branch 'main' into tom/cudf-sourceinfo-size
TomAugspurger Jun 25, 2026
b375237
Also pin libcudf in build_wheel_libcudf_streaming.sh
TomAugspurger Jun 25, 2026
c8a15b1
Merge remote-tracking branch 'upstream/main' into tom/cudf-sourceinfo…
TomAugspurger Jun 25, 2026
1bd862e
Merge branch 'main' into tom/cudf-sourceinfo-size
TomAugspurger Jun 26, 2026
307b589
Merge remote-tracking branch 'upstream/main' into tom/cudf-sourceinfo…
TomAugspurger Jun 26, 2026
7dbe50a
Added note about s3
TomAugspurger Jun 26, 2026
9a13ecd
Merge branch 'tom/cudf-sourceinfo-size' of https://github.com/TomAugs…
TomAugspurger Jun 26, 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
9 changes: 6 additions & 3 deletions cpp/include/cudf/io/datasource.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
#include <rmm/cuda_stream_view.hpp>

#include <future>
#include <memory>
Comment thread
TomAugspurger marked this conversation as resolved.
#include <optional>

namespace CUDF_EXPORT cudf {
//! IO interfaces
Expand Down Expand Up @@ -98,11 +98,14 @@ class datasource {
* @param[in] offset Starting byte offset from which data will be read (the default is zero)
* @param[in] max_size_estimate Upper estimate of the data range that will be read (the default is
* zero, which means the whole file after `offset`)
* @param[in] known_size Optional known file size in bytes. When set for remote URLs, KvikIO skips
Comment thread
TomAugspurger marked this conversation as resolved.
Outdated
* the size query HEAD request at open time.
* @return Constructed datasource object
*/
static std::unique_ptr<datasource> create(std::string const& filepath,
size_t offset = 0,
size_t max_size_estimate = 0);
size_t offset = 0,
size_t max_size_estimate = 0,
std::optional<std::size_t> known_size = std::nullopt);
Comment thread
TomAugspurger marked this conversation as resolved.

/**
* @brief Creates a source from a host memory buffer.
Expand Down
49 changes: 46 additions & 3 deletions cpp/include/cudf/io/types.hpp
Original file line number Diff line number Diff line change
@@ -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
*/

Expand Down Expand Up @@ -310,6 +310,17 @@ constexpr inline auto is_byte_like_type()
std::is_same_v<non_cv_T, std::byte>;
}

/**
* @brief A file path with an optional known size in bytes.
*
* When `size` is set for a remote URL, libcudf passes it to KvikIO at open time so the remote
* server is not queried for file size.
*/
Comment thread
TomAugspurger marked this conversation as resolved.
struct filepath_source {

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.

How about the name filepath_info?

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.

Hmm I was hoping for some kind of convention like a "source" is a thing that goes into a SourceInfo. Does that make sense?

If we change it to filepath_info, I'll want to change the python class to class FilepathInfo, which is a bit too close to SourceInfo for my taste.

std::string path; ///< Path or URL of the input file
std::optional<std::size_t> size{}; ///< Known file size; omit to query via KvikIO
};

/**
* @brief Source information for read interfaces
*/
Expand All @@ -325,8 +336,13 @@ struct source_info {
* @param file_paths Input files paths
*/
explicit source_info(std::vector<std::string> file_paths)
: _type(io_type::FILEPATH), _num_sources(file_paths.size()), _filepaths(std::move(file_paths))
: _type(io_type::FILEPATH), _num_sources(file_paths.size())
Comment thread
TomAugspurger marked this conversation as resolved.
{
_filepath_sources.reserve(file_paths.size());
for (auto& path : file_paths) {
_filepath_sources.push_back({std::move(path), std::nullopt});
}
rebuild_filepaths();
}

/**
Expand All @@ -335,10 +351,21 @@ struct source_info {
* @param file_path Single input file
*/
explicit source_info(std::string file_path)
: _type(io_type::FILEPATH), _num_sources(1), _filepaths({std::move(file_path)})
: source_info(std::vector<std::string>{std::move(file_path)})
{
}

/**
* @brief Construct a new source info object from filepath sources with optional known sizes
*
* @param sources Input filepath sources
*/
explicit source_info(std::vector<filepath_source> sources)
: _type(io_type::FILEPATH), _num_sources(sources.size()), _filepath_sources(std::move(sources))
{
rebuild_filepaths();
}

/**
* @brief Construct a new source info object for multiple buffers in host memory
*
Expand Down Expand Up @@ -424,6 +451,12 @@ struct source_info {
* @return The type of the input
*/
[[nodiscard]] auto type() const { return _type; }
/**
* @brief Get the filepath sources of the input
*
* @return The filepath sources of the input
*/
[[nodiscard]] auto const& filepath_sources() const { return _filepath_sources; }
/**
* @brief Get the filepaths of the input
*
Expand Down Expand Up @@ -457,8 +490,18 @@ struct source_info {
[[nodiscard]] auto num_sources() const { return _num_sources; }

private:
void rebuild_filepaths()
{
_filepaths.clear();
_filepaths.reserve(_filepath_sources.size());
for (auto const& source : _filepath_sources) {
_filepaths.push_back(source.path);
}
}

io_type _type = io_type::VOID;
size_t _num_sources = 0;
std::vector<filepath_source> _filepath_sources;
std::vector<std::string> _filepaths;
Comment thread
mhaseeb123 marked this conversation as resolved.
std::vector<cudf::host_span<std::byte const>> _host_buffers;
std::vector<cudf::device_span<std::byte const>> _device_buffers;
Expand Down
18 changes: 10 additions & 8 deletions cpp/src/io/functions.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -162,24 +162,26 @@ std::vector<std::unique_ptr<cudf::io::datasource>> make_datasources(source_info
switch (info.type()) {
case io_type::FILEPATH: {
std::vector<std::unique_ptr<cudf::io::datasource>> sources;
sources.reserve(info.filepaths().size());
sources.reserve(info.filepath_sources().size());
// Creating sources in a single thread is faster for a small number of sources
auto const pool_use_threshold =
cudf::detail::getenv_or("LIBCUDF_DATASOURCE_PARALLEL_CREATION_THRESHOLD", 8ul);
if (info.filepaths().size() >= pool_use_threshold) {
if (info.filepath_sources().size() >= pool_use_threshold) {
std::vector<std::future<std::unique_ptr<cudf::io::datasource>>> source_tasks;
source_tasks.reserve(info.filepaths().size());
for (auto const& path : info.filepaths()) {
source_tasks.emplace_back(cudf::detail::host_worker_pool().submit_task(
[=] { return cudf::io::datasource::create(path, offset, max_size_estimate); }));
source_tasks.reserve(info.filepath_sources().size());
for (auto const& fs : info.filepath_sources()) {
source_tasks.emplace_back(cudf::detail::host_worker_pool().submit_task([=] {
return cudf::io::datasource::create(fs.path, offset, max_size_estimate, fs.size);
}));
}
std::transform(
source_tasks.begin(), source_tasks.end(), std::back_inserter(sources), [](auto& task) {
return task.get();
});
} else {
for (auto const& filepath : info.filepaths()) {
sources.emplace_back(cudf::io::datasource::create(filepath, offset, max_size_estimate));
for (auto const& fs : info.filepath_sources()) {
sources.emplace_back(
cudf::io::datasource::create(fs.path, offset, max_size_estimate, fs.size));
}
}
return sources;
Expand Down
41 changes: 36 additions & 5 deletions cpp/src/io/utilities/datasource.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
#include <vector>

#ifdef CUDF_KVIKIO_REMOTE_IO
#include <kvikio/hdfs.hpp>
#include <kvikio/remote_handle.hpp>
#endif

Expand Down Expand Up @@ -353,14 +354,43 @@ class user_datasource_wrapper : public datasource {
datasource* const source; ///< A non-owning pointer to the user-implemented datasource
};

#ifdef CUDF_KVIKIO_REMOTE_IO
/**
* @brief Infer the KvikIO remote endpoint type from a URL (no network I/O).
*
* Mirrors the order used by `kvikio::RemoteHandle::open()` in AUTO mode.
*/
kvikio::RemoteEndpointType infer_remote_endpoint_type(std::string const& url)
Comment thread
TomAugspurger marked this conversation as resolved.
Outdated
{
if (kvikio::S3Endpoint::is_url_valid(url)) { return kvikio::RemoteEndpointType::S3; }
if (kvikio::S3PublicEndpoint::is_url_valid(url)) { return kvikio::RemoteEndpointType::S3_PUBLIC; }

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.

Here a URL from public S3 would still be inferred as private S3.
The challenge for S3 is that private and public endpoints follow the same URL pattern (S3PublicEndpoint::is_url_valid's implementation is the same with S3Endpoint::is_url_valid), and I think it is not quite possible to infer the type from URL alone without network I/O.
In KvikIO's open() method, given a S3-like URL (that is valid for both private and public S3), we first try to create a private S3 endpoint and access its file size, and in case of an exception, we fall back to the public S3 path.

@TomAugspurger TomAugspurger Jun 23, 2026 •

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.

Would it be fair to say that there aren't really public and private URLs? Just URLs that happen to be anonymously accessible, and URLs that happen to require authorization, so they can't be disambiguated just by looking at the URL?

I think something like rapidsai/kvikio#951 would solve this pretty cleanly: you define a chain of credentials to attempt when making some HTTP request. So long as anonymous access (no credentials) is included in the chain then things will succeed.

For now, I think I'll remove the S3PublicEndpoint case, since IIUC it should be unreachable right now.

if (kvikio::S3EndpointWithPresignedUrl::is_url_valid(url)) {
return kvikio::RemoteEndpointType::S3_PRESIGNED_URL;
}
if (kvikio::WebHdfsEndpoint::is_url_valid(url)) { return kvikio::RemoteEndpointType::WEBHDFS; }
if (kvikio::HttpEndpoint::is_url_valid(url)) { return kvikio::RemoteEndpointType::HTTP; }
return kvikio::RemoteEndpointType::HTTP;
}

kvikio::RemoteHandle open_remote_handle(char const* filepath, std::optional<std::size_t> known_size)
Comment thread
TomAugspurger marked this conversation as resolved.
{
if (known_size.has_value()) {
auto const endpoint_type = infer_remote_endpoint_type(filepath);
return kvikio::RemoteHandle::open(filepath, endpoint_type, std::nullopt, *known_size);
}
return kvikio::RemoteHandle::open(filepath);
}
#endif
Comment thread
TomAugspurger marked this conversation as resolved.
Outdated

#ifdef CUDF_KVIKIO_REMOTE_IO
/**
* @brief Remote file source backed by KvikIO, which handles S3 filepaths seamlessly.
*/
class remote_file_source : public kvikio_source<kvikio::RemoteHandle> {
public:
explicit remote_file_source(char const* filepath)
: kvikio_source{kvikio::RemoteHandle::open(filepath)}
explicit remote_file_source(char const* filepath,
std::optional<std::size_t> known_size = std::nullopt)
: kvikio_source{open_remote_handle(filepath, known_size)}
{
}

Expand Down Expand Up @@ -397,7 +427,8 @@ class remote_file_source : public file_source {

std::unique_ptr<datasource> datasource::create(std::string const& filepath,
size_t offset,
size_t max_size_estimate)
size_t max_size_estimate,
std::optional<std::size_t> known_size)
{
auto const use_memory_mapping = [] {
auto const policy = cudf::detail::getenv_or("LIBCUDF_MMAP_ENABLED", std::string{"OFF"});
Expand All @@ -410,7 +441,7 @@ std::unique_ptr<datasource> datasource::create(std::string const& filepath,

if (remote_file_source::could_be_remote_url(filepath)) {
try {
return std::make_unique<remote_file_source>(filepath.c_str());
return std::make_unique<remote_file_source>(filepath.c_str(), known_size);
} catch (std::exception const& ex) {
std::string redacted_msg;
try {
Expand Down Expand Up @@ -450,7 +481,7 @@ std::unique_ptr<datasource> datasource::create(std::string const& filepath,
// Create a remote file resource only when the pattern is found and replaced; otherwise, still
// create a local file resource
if (filepath != remote_file_path) {
return std::make_unique<remote_file_source>(remote_file_path.c_str());
return std::make_unique<remote_file_source>(remote_file_path.c_str(), known_size);
}
}

Expand Down
1 change: 1 addition & 0 deletions cpp/tests/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -320,6 +320,7 @@ ConfigureTest(
# * io tests --------------------------------------------------------------------------------------
ConfigureTest(COMPRESSION_TEST io/comp/comp_test.cpp)
ConfigureTest(ROW_SELECTION_TEST io/row_selection_test.cpp)
ConfigureTest(FILEPATH_SOURCE_TEST io/filepath_source_test.cpp)

ConfigureTest(
CSV_TEST io/csv_test.cpp
Expand Down
73 changes: 73 additions & 0 deletions cpp/tests/io/filepath_source_test.cpp
Comment thread
TomAugspurger marked this conversation as resolved.
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
/*
* SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION.
* SPDX-License-Identifier: Apache-2.0
*/

#include <cudf_test/base_fixture.hpp>
#include <cudf_test/column_utilities.hpp>
#include <cudf_test/column_wrapper.hpp>

#include <cudf/io/datasource.hpp>
#include <cudf/io/parquet.hpp>
#include <cudf/io/types.hpp>

#include <filesystem>

struct FilepathSourceTest : public cudf::test::BaseFixture {};

TEST_F(FilepathSourceTest, StringConstructorsPopulateFilepathSources)
{
auto const single = cudf::io::source_info{"test.parquet"};
ASSERT_EQ(single.filepath_sources().size(), 1);
EXPECT_EQ(single.filepath_sources().front().path, "test.parquet");
EXPECT_FALSE(single.filepath_sources().front().size.has_value());
EXPECT_EQ(single.filepaths(), std::vector<std::string>{"test.parquet"});

auto const multi = cudf::io::source_info{std::vector<std::string>{"a.parquet", "b.parquet"}};
ASSERT_EQ(multi.filepath_sources().size(), 2);
EXPECT_EQ(multi.filepaths().size(), 2);
EXPECT_FALSE(multi.filepath_sources()[1].size.has_value());
}

TEST_F(FilepathSourceTest, FilepathSourceConstructorPreservesSize)
{
std::vector<cudf::io::filepath_source> sources{
{"s3://bucket/object.parquet", 12345},
{"https://example.com/data.parquet", std::nullopt},
};

auto const info = cudf::io::source_info{std::move(sources)};
ASSERT_EQ(info.filepath_sources().size(), 2);
EXPECT_EQ(info.filepath_sources()[0].path, "s3://bucket/object.parquet");
ASSERT_TRUE(info.filepath_sources()[0].size.has_value());
EXPECT_EQ(info.filepath_sources()[0].size.value(), 12345);
EXPECT_FALSE(info.filepath_sources()[1].size.has_value());
EXPECT_EQ(info.filepaths()[0], "s3://bucket/object.parquet");
EXPECT_EQ(info.filepaths()[1], "https://example.com/data.parquet");
}

TEST_F(FilepathSourceTest, KnownSizePlumbsThroughMakeDatasources)
{
auto const filepath =
(std::filesystem::temp_directory_path() / "filepath_source_test.parquet").string();

auto col = cudf::test::fixed_width_column_wrapper<int32_t>{1, 2, 3};
cudf::table_view const table{{col}};

cudf::io::parquet_writer_options write_opts =
cudf::io::parquet_writer_options::builder(cudf::io::sink_info{filepath}, table);
cudf::io::write_parquet(write_opts);

auto const file_size = std::filesystem::file_size(filepath);
std::vector<cudf::io::filepath_source> sources{{filepath, file_size}};
auto const source_info = cudf::io::source_info{std::move(sources)};

auto datasources = cudf::io::make_datasources(source_info);
ASSERT_EQ(datasources.size(), 1);
EXPECT_EQ(datasources.front()->size(), file_size);

auto const read_opts =
cudf::io::parquet_reader_options::builder(source_info).column_names({"0"}).build();
auto const result = cudf::io::read_parquet(read_opts);
CUDF_TEST_EXPECT_COLUMNS_EQUAL(col, result.tbl->get_column(0).view());
}
23 changes: 23 additions & 0 deletions docs/cudf/source/cudf/io/io.md
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,29 @@ Note that:
For more information about error handling, compatibility mode, and
tuning parameters in KvikIO see: <https://github.com/rapidsai/kvikio>

### Remote file sizes and HEAD requests

When reading remote files (for example `s3://...` URLs) via `pylibcudf.io.SourceInfo`, KvikIO
may send HEAD requests at open time to probe connectivity and query file size. To skip those
requests when the file size is already known (for example from object-store metadata), pass a
`pylibcudf.io.FilepathSource` with the `size` argument set:

```python
import pylibcudf as plc

content_length = ... # from external metadata
sources = plc.io.SourceInfo([
plc.io.FilepathSource("s3://bucket/object.parquet", size=content_length),
])
table = plc.io.parquet.read_parquet(
plc.io.parquet.ParquetReaderOptions.builder(sources).build()
)
```

Providing an incorrect size avoids the extra HEAD requests but will break footer reads and other
operations that depend on the true file length. Plain string paths in `SourceInfo` preserve the
previous behavior (size queried via KvikIO).

Operations that support the use of GPUDirect Storage:

- {py:func}`cudf.read_avro`
Expand Down
3 changes: 2 additions & 1 deletion python/pylibcudf/pylibcudf/io/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,11 @@
types,
)
from .parquet_metadata import FileMetaData
from .types import SinkInfo, SourceInfo, TableWithMetadata
from .types import FilepathSource, SinkInfo, SourceInfo, TableWithMetadata

__all__ = [
"FileMetaData",
"FilepathSource",
"SinkInfo",
"SourceInfo",
"TableWithMetadata",
Expand Down
5 changes: 5 additions & 0 deletions python/pylibcudf/pylibcudf/io/types.pxd
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ from pylibcudf.libcudf.io.types cimport (
column_name_info,
compression_type,
dictionary_policy,
filepath_source,
io_type,
partition_info,
quote_style,
Expand Down Expand Up @@ -88,6 +89,10 @@ cdef class TableWithMetadata:
table_with_metadata& tbl, object stream, DeviceMemoryResource mr
)

cdef class FilepathSource:
cdef public object path
cdef public object size

cdef class SourceInfo:
cdef source_info c_obj
# Keep the bytes converted from stringio alive
Expand Down
Loading
Loading