Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
17 changes: 16 additions & 1 deletion cpp/include/kvikio/remote_handle.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
#include <memory>
#include <optional>
#include <string>
#include <vector>

#include <kvikio/defaults.hpp>
#include <kvikio/error.hpp>
Expand Down Expand Up @@ -330,6 +331,20 @@ class S3EndpointWithPresignedUrl : public RemoteEndpoint {
static bool is_url_valid(std::string const& url) noexcept;
};

/**
* @brief Infer remote endpoint type from URL.
*
* This function follows the same endpoint-selection order as `RemoteHandle::open()` in
* `RemoteEndpointType::AUTO` mode, but only infers the endpoint type and does not create a handle.
* Note that this function will not return `RemoteEndpointType::S3_PUBLIC`, because disambiguating
* between a URL that's accessible only with authorization or only anonymously is not possible
* without making an HTTP request.
*
* @param url The URL of the remote file.
* @return The inferred endpoint type.
*/

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.

I think we need to document the difference between this utility function and the open function, that given a URL valid for both S3 private and public endpoints, this function will always return the private endpoint type, and that it is not possible for us to disambiguate S3 private and public endpoints from a URL alone.
Then in the cudf PR NVIDIA/cudf#22739, we may document that known file size + URL without S3 credential is not supported.

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.

I added a note in b752f5a.

But I'd like to get away from "public" vs. "private" URLs entirely. It'd much prefer to just deal with some URL and an authorization method (possibly a chained authorization method that tries multiple).

RemoteEndpointType infer_remote_endpoint_type(std::string const& url);

/**
* @brief Handle of remote file.
*/
Expand Down Expand Up @@ -416,7 +431,7 @@ class RemoteHandle {
* );
* @endcode
*/
static RemoteHandle open(std::string url,
static RemoteHandle open(std::string const& url,
RemoteEndpointType remote_endpoint_type = RemoteEndpointType::AUTO,
std::optional<std::vector<RemoteEndpointType>> allow_list = std::nullopt,
std::optional<std::size_t> nbytes = std::nullopt);
Expand Down
169 changes: 97 additions & 72 deletions cpp/src/remote_handle.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
*/

#include <algorithm>
#include <array>
Comment thread
TomAugspurger marked this conversation as resolved.
#include <cassert>
#include <cstddef>
#include <cstring>
Expand All @@ -13,6 +14,7 @@
#include <sstream>
#include <stdexcept>
#include <string>
#include <tuple>

#include <kvikio/bounce_buffer.hpp>
#include <kvikio/defaults.hpp>
Expand Down Expand Up @@ -231,6 +233,84 @@ std::string encode_special_chars_in_path(std::string const& url)
components.path = detail::UrlEncoder::encode_path(components.path.value());
return detail::UrlBuilder::build_manually(components);
}

std::vector<RemoteEndpointType> const& get_default_allow_list()
{
static std::vector const res{RemoteEndpointType::S3,
RemoteEndpointType::S3_PUBLIC,
RemoteEndpointType::S3_PRESIGNED_URL,
RemoteEndpointType::WEBHDFS,
RemoteEndpointType::HTTP};
return res;
}

std::unique_ptr<RemoteEndpoint> create_endpoint_from_type(std::string const& url,
std::string const& scheme,
RemoteEndpointType type)
{
switch (type) {
case RemoteEndpointType::S3:
if (!S3Endpoint::is_url_valid(url)) { return nullptr; }
if (scheme == "s3") {
auto const [bucket, object] = S3Endpoint::parse_s3_url(url);
return std::make_unique<S3Endpoint>(std::pair{bucket, object});
}
return std::make_unique<S3Endpoint>(url);

case RemoteEndpointType::S3_PUBLIC:
if (!S3PublicEndpoint::is_url_valid(url)) { return nullptr; }
return std::make_unique<S3PublicEndpoint>(url);

case RemoteEndpointType::S3_PRESIGNED_URL:
if (!S3EndpointWithPresignedUrl::is_url_valid(url)) { return nullptr; }
return std::make_unique<S3EndpointWithPresignedUrl>(url);

case RemoteEndpointType::WEBHDFS:
if (!WebHdfsEndpoint::is_url_valid(url)) { return nullptr; }
return std::make_unique<WebHdfsEndpoint>(url);

case RemoteEndpointType::HTTP:
if (!HttpEndpoint::is_url_valid(url)) { return nullptr; }
return std::make_unique<HttpEndpoint>(url);

default: return nullptr;
}
}

std::pair<std::unique_ptr<RemoteEndpoint>, std::optional<std::size_t>> infer_endpoint_impl(
std::string const& url,
std::vector<RemoteEndpointType> const& allow_list,
bool probe_s3_connectivity)
{
auto const scheme =
detail::UrlParser::extract_component(url, CURLUPART_SCHEME, CURLU_NON_SUPPORT_SCHEME);
KVIKIO_EXPECT(scheme.has_value(), "Missing scheme in URL.");

for (auto const& type : allow_list) {
try {
auto endpoint = create_endpoint_from_type(url, scheme.value(), type);
if (endpoint == nullptr) { continue; }

std::optional<std::size_t> probed_nbytes = std::nullopt;
if (probe_s3_connectivity && type == RemoteEndpointType::S3) {
// Check connectivity for the credential-based S3 endpoint and reuse this size in
// RemoteHandle::open to avoid a second HEAD request.
probed_nbytes = endpoint->get_file_size();

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.

Since the open() function is modified again, can we just cache the file size inside the endpoint as I suggested? That is, the first time the endpoint's file size gets called, HEAD (or whatever method necessary for the endpoint in question) is performed, result cached in a std::optional<std::size_t> _file_size, and subsequent file size query just reuses the cached value.

This way, this PR alone will reduce the cudf-polars per-file HEAD request to 1, if I'm not missing anything.

@TomAugspurger TomAugspurger Jun 22, 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.

Personally, I'm not comfortable making something like size part of the data model for RemoteHandle without also including something like an etag to ensure that we aren't operating on a stale file. By using size, cudf-polars is accepting that risk, but it's unavoidable for now; and it's at least no than what would currently happen if a file changes mid-query. But putting it into the kvikio API is a bit riskier I think.

And, at least from the cudf-polars side, we'll temporarily use kvikio to get the file size exactly once per file. But in the medium term we'll be able to get that information from cudf-polars; either way, we'll be able to provide it the size in all subsequent operations.

Finally, I don't think this would directly help cudf-polars since in our current design we don't store RemoteHandle instances. We'll just store the plc.io.types.SourceInfo and FileMetadata.

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.

Does this sound OK @kingcrimsontianyu? I'd like to merge this today if possible, so we can use it in NVIDIA/cudf#22739

}
return {std::move(endpoint), probed_nbytes};
} catch (...) {
// If the credential-based S3 endpoint cannot be used to access the URL, try using
// S3 public endpoint instead when it is in the allowlist.
if (type == RemoteEndpointType::S3 &&
std::find(allow_list.begin(), allow_list.end(), RemoteEndpointType::S3_PUBLIC) !=
allow_list.end()) {
return {std::make_unique<S3PublicEndpoint>(url), std::nullopt};
}
throw;
}
}
KVIKIO_FAIL("Unsupported endpoint URL.", std::runtime_error);
}
} // namespace

RemoteEndpoint::RemoteEndpoint(RemoteEndpointType remote_endpoint_type)
Expand Down Expand Up @@ -575,87 +655,29 @@ bool S3EndpointWithPresignedUrl::is_url_valid(std::string const& url) noexcept
}
}

RemoteHandle RemoteHandle::open(std::string url,
RemoteEndpointType infer_remote_endpoint_type(std::string const& url)
{
KVIKIO_NVTX_FUNC_RANGE();
auto [endpoint, _] = infer_endpoint_impl(url, get_default_allow_list(), false);
std::tie(endpoint, std::ignore) = infer_endpoint_impl(url, get_default_allow_list(), false);
Comment thread
TomAugspurger marked this conversation as resolved.
return endpoint->remote_endpoint_type();
}

RemoteHandle RemoteHandle::open(std::string const& url,
RemoteEndpointType remote_endpoint_type,
std::optional<std::vector<RemoteEndpointType>> allow_list,
std::optional<std::size_t> nbytes)
{
KVIKIO_NVTX_FUNC_RANGE();
if (!allow_list.has_value()) {
allow_list = {RemoteEndpointType::S3,
RemoteEndpointType::S3_PUBLIC,
RemoteEndpointType::S3_PRESIGNED_URL,
RemoteEndpointType::WEBHDFS,
RemoteEndpointType::HTTP};
}

auto const scheme =
detail::UrlParser::extract_component(url, CURLUPART_SCHEME, CURLU_NON_SUPPORT_SCHEME);
KVIKIO_EXPECT(scheme.has_value(), "Missing scheme in URL.");

// Helper to create endpoint based on type
auto create_endpoint =
[&url = url, &scheme = scheme](RemoteEndpointType type) -> std::unique_ptr<RemoteEndpoint> {
switch (type) {
case RemoteEndpointType::S3:
if (!S3Endpoint::is_url_valid(url)) { return nullptr; }
if (scheme.value() == "s3") {
auto const [bucket, object] = S3Endpoint::parse_s3_url(url);
return std::make_unique<S3Endpoint>(std::pair{bucket, object});
}
return std::make_unique<S3Endpoint>(url);

case RemoteEndpointType::S3_PUBLIC:
if (!S3PublicEndpoint::is_url_valid(url)) { return nullptr; }
return std::make_unique<S3PublicEndpoint>(url);

case RemoteEndpointType::S3_PRESIGNED_URL:
if (!S3EndpointWithPresignedUrl::is_url_valid(url)) { return nullptr; }
return std::make_unique<S3EndpointWithPresignedUrl>(url);

case RemoteEndpointType::WEBHDFS:
if (!WebHdfsEndpoint::is_url_valid(url)) { return nullptr; }
return std::make_unique<WebHdfsEndpoint>(url);

case RemoteEndpointType::HTTP:
if (!HttpEndpoint::is_url_valid(url)) { return nullptr; }
return std::make_unique<HttpEndpoint>(url);

default: return nullptr;
}
};
if (!allow_list.has_value()) { allow_list = get_default_allow_list(); }

std::unique_ptr<RemoteEndpoint> endpoint;
std::optional<std::size_t> probed_nbytes;

if (remote_endpoint_type == RemoteEndpointType::AUTO) {
// Try each allowed type in the order of allowlist
for (auto const& type : allow_list.value()) {
try {
endpoint = create_endpoint(type);
if (endpoint == nullptr) { continue; }
if (type == RemoteEndpointType::S3) {
// Check connectivity for the credential-based S3 endpoint, and throw an exception if
// failed. Reuse this size when constructing the handle to avoid a second HEAD request.
probed_nbytes = endpoint->get_file_size();
}
} catch (...) {
// If the credential-based S3 endpoint cannot be used to access the URL, try using S3 public
// endpoint instead if it is in the allowlist
if (type == RemoteEndpointType::S3 &&
std::find(allow_list->begin(), allow_list->end(), RemoteEndpointType::S3_PUBLIC) !=
allow_list->end()) {
endpoint = std::make_unique<S3PublicEndpoint>(url);
probed_nbytes = std::nullopt;
} else {
throw;
}
}

// At this point, a matching endpoint has been found
break;
}
KVIKIO_EXPECT(endpoint.get() != nullptr, "Unsupported endpoint URL.", std::runtime_error);
auto inferred = infer_endpoint_impl(url, allow_list.value(), true);
endpoint = std::move(inferred.first);
probed_nbytes = inferred.second;
} else {
// Validate it is in the allow list
KVIKIO_EXPECT(
Expand All @@ -665,7 +687,10 @@ RemoteHandle RemoteHandle::open(std::string url,
std::runtime_error);

// Create the specific type
endpoint = create_endpoint(remote_endpoint_type);
auto const scheme =
detail::UrlParser::extract_component(url, CURLUPART_SCHEME, CURLU_NON_SUPPORT_SCHEME);
KVIKIO_EXPECT(scheme.has_value(), "Missing scheme in URL.");
endpoint = create_endpoint_from_type(url, scheme.value(), remote_endpoint_type);
KVIKIO_EXPECT(endpoint.get() != nullptr,
std::string{"Invalid URL for "} +
get_remote_endpoint_type_name(remote_endpoint_type) + " endpoint",
Expand Down
24 changes: 24 additions & 0 deletions cpp/tests/test_remote_handle.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -312,3 +312,27 @@ TEST_F(RemoteHandleTest, test_open)
}
}
}

TEST_F(RemoteHandleTest, test_infer_remote_endpoint_type)
{
kvikio::test::EnvVarContext env_var_ctx{{"AWS_DEFAULT_REGION", "my_aws_default_region"},
{"AWS_ACCESS_KEY_ID", "my_aws_access_key_id"},
{"AWS_SECRET_ACCESS_KEY", "my_aws_secrete_access_key"}};

EXPECT_EQ(kvikio::infer_remote_endpoint_type("s3://bucket-name/object-key-name"),
kvikio::RemoteEndpointType::S3);
EXPECT_EQ(kvikio::infer_remote_endpoint_type("https://host:1234/webhdfs/v1/data.bin"),
kvikio::RemoteEndpointType::WEBHDFS);
EXPECT_EQ(kvikio::infer_remote_endpoint_type("https://example.com/path/file.bin"),
kvikio::RemoteEndpointType::HTTP);
EXPECT_EQ(kvikio::infer_remote_endpoint_type(
"https://bucket-name.s3.region-code.amazonaws.com/"
"object-key-name?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Signature=sig&"
"X-Amz-Credential=cred&X-Amz-SignedHeaders=host"),
kvikio::RemoteEndpointType::S3_PRESIGNED_URL);

EXPECT_THAT([&] { kvikio::infer_remote_endpoint_type("unsupported://example.com/path"); },
ThrowsMessage<std::runtime_error>(HasSubstr("Unsupported endpoint URL")));
EXPECT_THAT([&] { kvikio::infer_remote_endpoint_type("example.com/path"); },
ThrowsMessage<std::runtime_error>(HasSubstr("Bad scheme")));
}
2 changes: 2 additions & 0 deletions docs/source/api.rst
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,8 @@ RemoteFile
----------
.. currentmodule:: kvikio.remote_file

.. autofunction:: infer_remote_endpoint_type

.. autoclass:: RemoteEndpointType

.. autoclass:: RemoteFile
Expand Down
8 changes: 7 additions & 1 deletion python/kvikio/kvikio/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,12 @@
get_page_cache_info,
)
from kvikio.mmap import Mmap
from kvikio.remote_file import RemoteEndpointType, RemoteFile, is_remote_file_available
from kvikio.remote_file import (
RemoteEndpointType,
RemoteFile,
infer_remote_endpoint_type,
is_remote_file_available,
)
from kvikio.stream import stream_deregister, stream_register
from kvikio.utils import kvikio_deprecation_notice

Expand All @@ -36,6 +41,7 @@
"drop_system_page_cache",
"Mmap",
"get_page_cache_info",
"infer_remote_endpoint_type",
"is_remote_file_available",
"kvikio_deprecation_notice",
"RemoteEndpointType",
Expand Down
14 changes: 13 additions & 1 deletion python/kvikio/kvikio/_lib/remote_handle.pyx
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
# SPDX-FileCopyrightText: Copyright (c) 2024-2025, NVIDIA CORPORATION. All rights reserved.
# SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

# distutils: language = c++
Expand Down Expand Up @@ -60,6 +60,10 @@ cdef extern from "<kvikio/remote_handle.hpp>" namespace "kvikio" nogil:
(cpp_RemoteEndpoint):
cpp_S3EndpointWithPresignedUrl(string presigned_url) except +

RemoteEndpointType cpp_infer_remote_endpoint_type "kvikio::infer_remote_endpoint_type"(
string url
) except +

cdef cppclass cpp_RemoteHandle "kvikio::RemoteHandle":
cpp_RemoteHandle(
unique_ptr[cpp_RemoteEndpoint] endpoint, size_t nbytes
Expand Down Expand Up @@ -442,3 +446,11 @@ cdef class RemoteFile:
)

return _wrap_io_future(fut)


def infer_remote_endpoint_type(url: str) -> RemoteEndpointType:
cdef string cpp_url = _to_string(url)
cdef RemoteEndpointType result
with nogil:
result = cpp_infer_remote_endpoint_type(cpp_url)
return result
8 changes: 7 additions & 1 deletion python/kvikio/kvikio/remote_file.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
# SPDX-FileCopyrightText: Copyright (c) 2024-2025, NVIDIA CORPORATION. All rights reserved.
# SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

from __future__ import annotations
Expand Down Expand Up @@ -84,6 +84,12 @@ def _get_remote_module():
return kvikio._lib.remote_handle


def infer_remote_endpoint_type(url: str) -> RemoteEndpointType:
"""Infer endpoint type from URL using AUTO endpoint resolution rules."""
result = _get_remote_module().infer_remote_endpoint_type(url)
return RemoteEndpointType[result.name]


class RemoteFile:
"""File handle of a remote file."""

Expand Down
Loading