From cc580e9055a79bba8064bd1fd321ce34950a4177 Mon Sep 17 00:00:00 2001 From: Gang Wu Date: Sun, 16 Aug 2026 23:52:39 +0800 Subject: [PATCH] feat(io): resolve FileIO by registered location schemes Add scheme-aware FileIO factories with deterministic registration precedence, route ResolvingFileIO by location scheme, and forward vended storage credentials through registered delegates. --- mkdocs/docs/file-io.md | 94 +++++++++++ mkdocs/mkdocs.yml | 1 + src/iceberg/CMakeLists.txt | 1 + src/iceberg/arrow/arrow_register.cc | 14 +- src/iceberg/arrow/s3/arrow_s3_file_io.cc | 8 +- src/iceberg/arrow/s3/s3_properties.h | 13 ++ src/iceberg/catalog/rest/rest_file_io.cc | 10 +- src/iceberg/file_io_registry.cc | 86 +++++++--- src/iceberg/file_io_registry.h | 30 ++-- src/iceberg/meson.build | 1 + src/iceberg/resolving_file_io.cc | 54 +++---- src/iceberg/resolving_file_io.h | 24 +-- src/iceberg/test/CMakeLists.txt | 13 +- src/iceberg/test/arrow_io_test.cc | 21 +++ src/iceberg/test/arrow_s3_file_io_test.cc | 12 +- src/iceberg/test/location_util_test.cc | 11 ++ src/iceberg/test/resolving_file_io_test.cc | 150 +++++++++++++----- src/iceberg/test/rest_arrow_file_io_test.cc | 97 ----------- .../test/rest_catalog_integration_test.cc | 6 +- src/iceberg/test/rest_file_io_test.cc | 65 ++++---- .../location_util.cc} | 20 +-- src/iceberg/util/location_util.h | 7 + 22 files changed, 437 insertions(+), 301 deletions(-) create mode 100644 mkdocs/docs/file-io.md delete mode 100644 src/iceberg/test/rest_arrow_file_io_test.cc rename src/iceberg/{resolving_file_io_internal.h => util/location_util.cc} (64%) diff --git a/mkdocs/docs/file-io.md b/mkdocs/docs/file-io.md new file mode 100644 index 000000000..ede624376 --- /dev/null +++ b/mkdocs/docs/file-io.md @@ -0,0 +1,94 @@ + + +# FileIO + +`FileIO` reads, writes, and deletes Iceberg data and metadata files. + +## Built-in implementations + +Call `iceberg::arrow::RegisterAll()` to register the Arrow-backed FileIO +implementations: + +| Registry name | Schemes | +|---|---| +| `arrow-fs-local` | paths without a scheme, `file` | +| `arrow-fs-s3` | `s3`, `s3a`, `s3n` | + +The S3 implementation requires Arrow S3 support. + +## Select an implementation + +Load a registered implementation directly: + +```cpp +#include "iceberg/arrow/arrow_register.h" +#include "iceberg/arrow/s3/s3_properties.h" +#include "iceberg/file_io_registry.h" +#include "iceberg/resolving_file_io.h" + +iceberg::arrow::RegisterAll(); + +auto file_io = iceberg::FileIORegistry::Load( + iceberg::FileIORegistry::kArrowS3FileIO, + {{std::string(iceberg::arrow::S3Properties::kEndpoint), + "https://s3.example.com"}, + {std::string(iceberg::arrow::S3Properties::kClientRegion), "us-east-1"}}); +``` + +For a REST catalog, set `io-impl` to the registry name. If it is omitted, the +REST catalog uses `ResolvingFileIO` and selects a registered implementation for +each file location's scheme. + +## Register a custom FileIO + +Register the factory before creating the catalog or resolver: + +```cpp +#include + +iceberg::FileIORegistry::Register( + "my-file-io", + {.create = [](const iceberg::FileIORegistry::Properties& properties) + -> iceberg::Result> { + return MakeMyFileIO(properties); + }, + .accepts = [](std::string_view scheme) { return scheme == "myfs"; }}); +``` + +`create` is required. Set `accepts` to enable automatic selection; it receives +the normalized lower-case scheme. Leave it empty for an implementation selected +only by `io-impl`. + +```cpp +iceberg::FileIORegistry::Load("my-file-io", {}); +auto file_io = std::make_unique( + iceberg::FileIORegistry::Properties{}); +file_io->NewInputFile("myfs://bucket/path/file.parquet"); +``` + +Later registrations take precedence for automatic scheme resolution. + +## Storage credentials + +When a REST catalog returns vended storage credentials for a table, it applies +them to the table's FileIO. A custom FileIO selected through `io-impl` must +implement `SupportsStorageCredentials`; otherwise table access with vended +credentials is unsupported. With automatic resolution, `ResolvingFileIO` +forwards credentials to registered delegates that support them. diff --git a/mkdocs/mkdocs.yml b/mkdocs/mkdocs.yml index 724c18930..000f7cd07 100644 --- a/mkdocs/mkdocs.yml +++ b/mkdocs/mkdocs.yml @@ -52,6 +52,7 @@ markdown_extensions: nav: - Home: index.md - Getting Started: getting-started.md + - FileIO: file-io.md - Contributing: contributing.md - Releases: - Release History: releases.md diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index dec79ada4..cabd93a4e 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -135,6 +135,7 @@ set(ICEBERG_SOURCES util/conversions.cc util/decimal.cc util/gzip_internal.cc + util/location_util.cc util/murmurhash3_internal.cc util/property_util.cc util/retry_util.cc diff --git a/src/iceberg/arrow/arrow_register.cc b/src/iceberg/arrow/arrow_register.cc index d2983ffae..4fedc914c 100644 --- a/src/iceberg/arrow/arrow_register.cc +++ b/src/iceberg/arrow/arrow_register.cc @@ -21,9 +21,10 @@ #include #include -#include +#include #include "iceberg/arrow/arrow_io_util.h" +#include "iceberg/arrow/s3/s3_properties.h" #include "iceberg/file_io_registry.h" namespace iceberg { @@ -37,16 +38,19 @@ namespace { void RegisterLocalFileIO() { FileIORegistry::Register( std::string(FileIORegistry::kArrowLocalFileIO), - [](const std::unordered_map& /*properties*/) - -> Result> { return MakeLocalFileIO(); }); + {.create = [](const FileIORegistry::Properties& /*properties*/) + -> Result> { return MakeLocalFileIO(); }, + .accepts = + [](std::string_view scheme) { return scheme.empty() || scheme == "file"; }}); } void RegisterS3FileIO() { #if ICEBERG_S3_ENABLED FileIORegistry::Register( std::string(FileIORegistry::kArrowS3FileIO), - [](const std::unordered_map& properties) - -> Result> { return MakeS3FileIO(properties); }); + {.create = [](const FileIORegistry::Properties& properties) + -> Result> { return MakeS3FileIO(properties); }, + .accepts = [](std::string_view scheme) { return IsS3Scheme(scheme); }}); #endif } diff --git a/src/iceberg/arrow/s3/arrow_s3_file_io.cc b/src/iceberg/arrow/s3/arrow_s3_file_io.cc index e3118453f..2a90006e4 100644 --- a/src/iceberg/arrow/s3/arrow_s3_file_io.cc +++ b/src/iceberg/arrow/s3/arrow_s3_file_io.cc @@ -91,11 +91,10 @@ std::string SplitEndpointScheme(std::string_view endpoint, return std::string(endpoint); } -// Location prefixes this FileIO can serve: must cover every scheme -// ResolveFileIOName routes here, or such a credential would be dropped. +// Location prefixes this FileIO can serve. bool IsS3FileIOCredentialPrefix(std::string_view prefix) { return prefix == "s3" || prefix.starts_with("s3://") || prefix.starts_with("s3a://") || - prefix.starts_with("s3n://") || prefix.starts_with("oss://"); + prefix.starts_with("s3n://"); } } // namespace @@ -184,9 +183,8 @@ Result> BuildArrowS3FileSystem( return std::shared_ptr<::arrow::fs::FileSystem>(std::move(fs)); } -// Keep in sync with ResolveFileIOName (resolving_file_io.cc). std::string CanonicalizeS3Scheme(std::string_view location) { - for (std::string_view scheme : {"s3a://", "s3n://", "oss://"}) { + for (std::string_view scheme : {"s3a://", "s3n://"}) { if (location.starts_with(scheme)) { return std::string("s3://").append(location.substr(scheme.size())); } diff --git a/src/iceberg/arrow/s3/s3_properties.h b/src/iceberg/arrow/s3/s3_properties.h index 03d1492ad..aeeb20690 100644 --- a/src/iceberg/arrow/s3/s3_properties.h +++ b/src/iceberg/arrow/s3/s3_properties.h @@ -22,6 +22,8 @@ /// \file iceberg/arrow/s3/s3_properties.h /// \brief Define S3 configuration property keys. +#include +#include #include namespace iceberg::arrow { @@ -54,4 +56,15 @@ struct S3Properties { static constexpr std::string_view kSocketTimeoutMs = "s3.socket-timeout-ms"; }; +/// \brief URI schemes served by the Arrow S3 FileIO, lower-case. +/// +/// Single source of truth: both the registry registration and IsS3Scheme derive +/// from this list, so a new alias only has to be added here. +inline constexpr std::array kS3Schemes = {"s3", "s3a", "s3n"}; + +/// \brief Return whether a normalized URI scheme is S3-compatible. +inline constexpr bool IsS3Scheme(std::string_view scheme) { + return std::ranges::contains(kS3Schemes, scheme); +} + } // namespace iceberg::arrow diff --git a/src/iceberg/catalog/rest/rest_file_io.cc b/src/iceberg/catalog/rest/rest_file_io.cc index 4fc5122fe..cdc1204f1 100644 --- a/src/iceberg/catalog/rest/rest_file_io.cc +++ b/src/iceberg/catalog/rest/rest_file_io.cc @@ -26,6 +26,7 @@ #include "iceberg/catalog/rest/types.h" #include "iceberg/file_io.h" #include "iceberg/file_io_registry.h" +#include "iceberg/resolving_file_io.h" #include "iceberg/util/macros.h" namespace iceberg::rest { @@ -45,16 +46,11 @@ std::unordered_map MergeFileIOProperties( } // namespace Result> MakeCatalogFileIO(const RestCatalogProperties& config) { - std::string io_impl = config.Get(RestCatalogProperties::kIOImpl); + const std::string io_impl = config.Get(RestCatalogProperties::kIOImpl); if (io_impl.empty()) { - // Resolve the FileIO per file-path scheme instead of guessing from - // `warehouse`, which is often a logical identifier rather than a storage - // URI (Java defaults to ResolvingFileIO likewise). - io_impl = std::string(FileIORegistry::kResolvingFileIO); + return std::make_unique(config.configs()); } - // TODO(gangwu): Support Java-style customized FileIO creation flows instead of - // resolving a single catalog-scoped FileIO instance only from properties. return FileIORegistry::Load(io_impl, config.configs()); } diff --git a/src/iceberg/file_io_registry.cc b/src/iceberg/file_io_registry.cc index 073e7f382..1a18d4627 100644 --- a/src/iceberg/file_io_registry.cc +++ b/src/iceberg/file_io_registry.cc @@ -19,27 +19,28 @@ #include "iceberg/file_io_registry.h" +#include #include +#include +#include #include +#include -#include "iceberg/resolving_file_io.h" +#include "iceberg/util/macros.h" +#include "iceberg/util/string_util.h" namespace iceberg { namespace { struct RegistryState { - std::mutex mutex; - std::unordered_map registry; + struct Entry { + std::string name; + FileIORegistry::Factory factory; + }; - RegistryState() { - // Always available: the scheme-resolving FileIO lives in the core library. - registry[std::string(FileIORegistry::kResolvingFileIO)] = - [](const std::unordered_map& properties) - -> Result> { - return std::make_unique(properties); - }; - } + std::mutex mutex; + std::vector registrations; }; RegistryState& State() { @@ -47,28 +48,69 @@ RegistryState& State() { return state; } +// Copy entries so user callbacks run outside the registry lock. +std::vector SnapshotEntries() { + auto& state = State(); + std::lock_guard lock(state.mutex); + return state.registrations; +} + +std::string FormatNames(const std::vector& entries) { + std::string result; + for (const auto& entry : entries) { + if (!result.empty()) { + result += ", "; + } + result += entry.name; + } + return result.empty() ? "(none registered)" : result; +} + } // namespace -void FileIORegistry::Register(const std::string& name, Factory factory) { +void FileIORegistry::Register(std::string name, Factory factory) { auto& state = State(); std::lock_guard lock(state.mutex); - state.registry[name] = std::move(factory); + std::erase_if(state.registrations, [&name](const RegistryState::Entry& entry) { + return entry.name == name; + }); + state.registrations.emplace_back(std::move(name), std::move(factory)); } -Result> FileIORegistry::Load( - const std::string& name, - const std::unordered_map& properties) { - Factory factory; +Result> FileIORegistry::Load(std::string_view name, + const Properties& properties) { + Factory::CreateFunction create; { auto& state = State(); std::lock_guard lock(state.mutex); - auto it = state.registry.find(name); - if (it == state.registry.end()) { - return NotFound("FileIO implementation not found: {}", name); + const auto entry = + std::ranges::find(state.registrations, name, &RegistryState::Entry::name); + if (entry == state.registrations.end()) { + return NotFound("FileIO not found: {}", name); + } + create = entry->factory.create; + } + if (!create) { + return InvalidArgument("FileIO '{}' has no create function", name); + } + ICEBERG_ASSIGN_OR_RAISE(auto io, create(properties)); + if (!io) { + return InvalidArgument("FileIO '{}' returned a null instance", name); + } + return std::move(io); +} + +Result FileIORegistry::Resolve(std::string_view scheme) { + const std::string normalized_scheme = StringUtils::ToLower(scheme); + const auto entries = SnapshotEntries(); + // Newest registrations take precedence. + for (const auto& entry : std::ranges::reverse_view(entries)) { + if (entry.factory.accepts && entry.factory.accepts(normalized_scheme)) { + return entry.name; } - factory = it->second; } - return factory(properties); + return NotSupported("No FileIO registered for URI scheme '{}'; registered: {}", + normalized_scheme, FormatNames(entries)); } } // namespace iceberg diff --git a/src/iceberg/file_io_registry.h b/src/iceberg/file_io_registry.h index e9b899fe2..527d19790 100644 --- a/src/iceberg/file_io_registry.h +++ b/src/iceberg/file_io_registry.h @@ -43,27 +43,39 @@ class ICEBERG_EXPORT FileIORegistry { public: static constexpr std::string_view kArrowLocalFileIO = "arrow-fs-local"; static constexpr std::string_view kArrowS3FileIO = "arrow-fs-s3"; - /// Always registered; resolves the concrete FileIO per file-path scheme. - static constexpr std::string_view kResolvingFileIO = "resolving-file-io"; - /// Factory function type for creating FileIO instances. - using Factory = std::function>( - const std::unordered_map& properties)>; + using Properties = std::unordered_map; + + /// Factory for explicit loading and optional scheme-based routing. + struct Factory { + using CreateFunction = + std::function>(const Properties& properties)>; + using AcceptsFunction = std::function; + + /// Required for explicit loading. + CreateFunction create; + /// Receives a lower-case scheme. Empty means explicit-only. + AcceptsFunction accepts; + }; /// \brief Register a FileIO factory under the given name. /// /// \param name The implementation name (e.g., "local", "s3") /// \param factory The factory function that creates the FileIO instance. - static void Register(const std::string& name, Factory factory); + static void Register(std::string name, Factory factory); /// \brief Load a FileIO implementation by name. /// /// \param name The implementation name to look up. /// \param properties Configuration properties to pass to the factory. /// \return A unique_ptr to the FileIO instance, or an error if not found. - static Result> Load( - const std::string& name, - const std::unordered_map& properties); + static Result> Load(std::string_view name, + const Properties& properties); + + /// \brief Returns the latest registered factory accepting `scheme`. + /// + /// Matching is case-insensitive; later registrations take precedence. + static Result Resolve(std::string_view scheme); }; } // namespace iceberg diff --git a/src/iceberg/meson.build b/src/iceberg/meson.build index 8f5680a99..89e6775ee 100644 --- a/src/iceberg/meson.build +++ b/src/iceberg/meson.build @@ -188,6 +188,7 @@ iceberg_sources = files( 'util/conversions.cc', 'util/decimal.cc', 'util/gzip_internal.cc', + 'util/location_util.cc', 'util/murmurhash3_internal.cc', 'util/property_util.cc', 'util/retry_util.cc', diff --git a/src/iceberg/resolving_file_io.cc b/src/iceberg/resolving_file_io.cc index ce91a52e8..8a4e81138 100644 --- a/src/iceberg/resolving_file_io.cc +++ b/src/iceberg/resolving_file_io.cc @@ -19,11 +19,14 @@ #include "iceberg/resolving_file_io.h" +#include +#include #include #include "iceberg/file_io_registry.h" -#include "iceberg/resolving_file_io_internal.h" +#include "iceberg/util/location_util.h" #include "iceberg/util/macros.h" +#include "iceberg/util/string_util.h" namespace iceberg { @@ -32,33 +35,22 @@ ResolvingFileIO::ResolvingFileIO(std::unordered_map pr ResolvingFileIO::~ResolvingFileIO() = default; -Result ResolveFileIOName(std::string_view location) { - const auto pos = location.find("://"); - if (pos == std::string_view::npos) { - return FileIORegistry::kArrowLocalFileIO; - } +Result> ResolvingFileIO::FileIOForPath( + std::string_view location) { + const auto scheme = StringUtils::ToLower(LocationUtil::ParseScheme(location)); + ICEBERG_ASSIGN_OR_RAISE(const auto name, FileIORegistry::Resolve(scheme)); - const auto scheme = location.substr(0, pos); - if (scheme == "file") { - return FileIORegistry::kArrowLocalFileIO; - } - // S3-compatible schemes served by the S3 FileIO (Java: SCHEME_TO_FILE_IO). - // Keep in sync with CanonicalizeS3Scheme in arrow_s3_file_io.cc. - if (scheme == "s3" || scheme == "s3a" || scheme == "s3n" || scheme == "oss") { - return FileIORegistry::kArrowS3FileIO; + { + std::shared_lock lock(mutex_); + if (const auto cached = io_by_name_.find(name); cached != io_by_name_.end()) { + return cached->second; + } } - return NotSupported("URI scheme '{}' is not supported for FileIO resolution", scheme); -} - -Result ResolvingFileIO::FileIOForPath(std::string_view location) { - ICEBERG_ASSIGN_OR_RAISE(const auto name, ResolveFileIOName(location)); - - std::lock_guard lock(mutex_); + std::unique_lock lock(mutex_); auto it = io_by_name_.find(name); if (it == io_by_name_.end()) { - ICEBERG_ASSIGN_OR_RAISE(auto io, - FileIORegistry::Load(std::string(name), properties_)); + ICEBERG_ASSIGN_OR_RAISE(auto io, FileIORegistry::Load(name, properties_)); // Forward all credentials; each implementation applies the prefixes it // understands. if (!storage_credentials_.empty()) { @@ -69,36 +61,36 @@ Result ResolvingFileIO::FileIOForPath(std::string_view location) { } it = io_by_name_.emplace(std::string(name), std::move(io)).first; } - return it->second.get(); + return it->second; } Result> ResolvingFileIO::NewInputFile( std::string file_location) { - ICEBERG_ASSIGN_OR_RAISE(auto* io, FileIOForPath(file_location)); + ICEBERG_ASSIGN_OR_RAISE(auto io, FileIOForPath(file_location)); return io->NewInputFile(std::move(file_location)); } Result> ResolvingFileIO::NewInputFile( std::string file_location, size_t length) { - ICEBERG_ASSIGN_OR_RAISE(auto* io, FileIOForPath(file_location)); + ICEBERG_ASSIGN_OR_RAISE(auto io, FileIOForPath(file_location)); return io->NewInputFile(std::move(file_location), length); } Result> ResolvingFileIO::NewOutputFile( std::string file_location) { - ICEBERG_ASSIGN_OR_RAISE(auto* io, FileIOForPath(file_location)); + ICEBERG_ASSIGN_OR_RAISE(auto io, FileIOForPath(file_location)); return io->NewOutputFile(std::move(file_location)); } Status ResolvingFileIO::DeleteFile(const std::string& file_location) { - ICEBERG_ASSIGN_OR_RAISE(auto* io, FileIOForPath(file_location)); + ICEBERG_ASSIGN_OR_RAISE(auto io, FileIOForPath(file_location)); return io->DeleteFile(file_location); } Status ResolvingFileIO::DeleteFiles(const std::vector& file_locations) { - std::unordered_map> locations_by_io; + std::unordered_map, std::vector> locations_by_io; for (const auto& file_location : file_locations) { - ICEBERG_ASSIGN_OR_RAISE(auto* io, FileIOForPath(file_location)); + ICEBERG_ASSIGN_OR_RAISE(auto io, FileIOForPath(file_location)); locations_by_io[io].push_back(file_location); } for (auto& [io, locations] : locations_by_io) { @@ -111,7 +103,7 @@ Status ResolvingFileIO::SetStorageCredentials( const std::vector& storage_credentials) { // Rebuild delegates lazily with the new credentials. Updating live delegates // instead would leave the resolver inconsistent if one of them rejected them. - std::lock_guard lock(mutex_); + std::unique_lock lock(mutex_); storage_credentials_ = storage_credentials; io_by_name_.clear(); return {}; diff --git a/src/iceberg/resolving_file_io.h b/src/iceberg/resolving_file_io.h index d96b0ca6e..61df2353d 100644 --- a/src/iceberg/resolving_file_io.h +++ b/src/iceberg/resolving_file_io.h @@ -23,7 +23,7 @@ /// \brief FileIO that resolves the concrete implementation per file-path scheme. #include -#include +#include #include #include #include @@ -37,19 +37,7 @@ namespace iceberg { -/// \brief FileIO that uses the location scheme to choose the concrete FileIO, -/// mirroring Java's ResolvingFileIO. -/// -/// Resolution is per file path and independent of `warehouse` (often a logical -/// identifier rather than a storage URI). Implementations are loaded lazily -/// from FileIORegistry with this FileIO's properties and cached. Vended -/// credentials are forwarded in full to every resolved FileIO that supports -/// them; each applies the prefixes it understands and ignores the rest. -/// -/// Lazy resolution is internally synchronized, so file operations may run -/// concurrently. Credentials are not: install them before sharing the instance, -/// since credentials() hands out a reference that SetStorageCredentials -/// replaces. +/// \brief FileIO that resolves each location to a registered implementation. class ICEBERG_EXPORT ResolvingFileIO final : public FileIO, public SupportsStorageCredentials { public: @@ -76,13 +64,13 @@ class ICEBERG_EXPORT ResolvingFileIO final : public FileIO, private: /// \brief Load (or return the cached) implementation serving `location`. - Result FileIOForPath(std::string_view location); + Result> FileIOForPath(std::string_view location); std::unordered_map properties_; - // Guards lazy resolution; set credentials before sharing across threads. - std::mutex mutex_; + // Guards lazy resolution and credential refresh. + std::shared_mutex mutex_; std::vector storage_credentials_; - std::unordered_map, StringHash, StringEqual> + std::unordered_map, StringHash, StringEqual> io_by_name_; }; diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 1181a722e..b3e1bdd56 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -278,11 +278,10 @@ endif() if(ICEBERG_BUILD_REST) function(add_rest_iceberg_test test_name) - set(options USE_BUNDLE) set(oneValueArgs) set(multiValueArgs SOURCES) cmake_parse_arguments(ARG - "${options}" + "" "${oneValueArgs}" "${multiValueArgs}" ${ARGN}) @@ -291,22 +290,12 @@ if(ICEBERG_BUILD_REST) target_include_directories(${test_name} PRIVATE "${CMAKE_BINARY_DIR}/iceberg/test/") target_sources(${test_name} PRIVATE ${ARG_SOURCES}) target_link_libraries(${test_name} PRIVATE GTest::gmock_main iceberg_rest_static) - if(ARG_USE_BUNDLE) - target_link_libraries(${test_name} - PRIVATE "$,iceberg_bundle_static,iceberg_bundle_shared>" - ) - endif() if(MSVC_TOOLCHAIN) target_compile_options(${test_name} PRIVATE /bigobj) endif() add_test(NAME ${test_name} COMMAND ${test_name}) endfunction() - if(ICEBERG_BUILD_BUNDLE) - add_rest_iceberg_test(rest_arrow_file_io_test USE_BUNDLE SOURCES - rest_arrow_file_io_test.cc) - endif() - add_rest_iceberg_test(rest_catalog_test SOURCES auth_manager_test.cc diff --git a/src/iceberg/test/arrow_io_test.cc b/src/iceberg/test/arrow_io_test.cc index a30e2da93..90a7ebee3 100644 --- a/src/iceberg/test/arrow_io_test.cc +++ b/src/iceberg/test/arrow_io_test.cc @@ -21,6 +21,7 @@ #include #include #include +#include #include #include @@ -29,6 +30,9 @@ #include #include "iceberg/arrow/arrow_io_internal.h" +#include "iceberg/arrow/arrow_register.h" +#include "iceberg/file_io_registry.h" +#include "iceberg/resolving_file_io.h" #include "iceberg/test/matchers.h" #include "iceberg/test/std_io.h" #include "iceberg/test/temp_file_test_base.h" @@ -309,6 +313,23 @@ class PermissiveOutputFileIO : public FileIO { } // namespace +TEST(ArrowRegisterTest, RegistersBuiltInFileIOs) { + arrow::RegisterAll(); + + ResolvingFileIO io({}); + EXPECT_THAT(io.NewInputFile("/tmp/file"), IsOk()); + EXPECT_THAT(io.NewInputFile("file:///tmp/file"), IsOk()); + +#if ICEBERG_S3_ENABLED + for (std::string_view scheme : {"s3", "s3a", "s3n"}) { + EXPECT_THAT(FileIORegistry::Resolve(scheme), + HasValue(::testing::Eq(FileIORegistry::kArrowS3FileIO))); + } +#else + EXPECT_THAT(FileIORegistry::Resolve("s3"), IsError(ErrorKind::kNotSupported)); +#endif +} + class LocalFileIOTest : public TempFileTestBase { protected: void SetUp() override { diff --git a/src/iceberg/test/arrow_s3_file_io_test.cc b/src/iceberg/test/arrow_s3_file_io_test.cc index aa949ab12..03b4b4858 100644 --- a/src/iceberg/test/arrow_s3_file_io_test.cc +++ b/src/iceberg/test/arrow_s3_file_io_test.cc @@ -191,8 +191,8 @@ TEST_F(ArrowS3FileIOTest, SkipsNonS3CredentialPrefix) { // credential that is silently skipped leaves S3 access on the default // credentials, which only surfaces much later as an auth error. TEST_F(ArrowS3FileIOTest, AppliesEveryS3CompatibleCredentialPrefix) { - for (std::string_view prefix : {"s3", "s3://bucket/table", "s3a://bucket/table", - "s3n://bucket/table", "oss://bucket/table"}) { + for (std::string_view prefix : + {"s3", "s3://bucket/table", "s3a://bucket/table", "s3n://bucket/table"}) { SCOPED_TRACE(prefix); auto result = MakeS3FileIO({}); ASSERT_THAT(result, IsOk()); @@ -308,10 +308,10 @@ TEST_F(ArrowS3FileIOTest, EndpointScheme) { std::string_view endpoint_override; std::string_view scheme; }; - const std::vector cases = {{"https://oss-cn-hangzhou.aliyuncs.com:443", - "oss-cn-hangzhou.aliyuncs.com:443", "https"}, - {"http://localhost:9000", "localhost:9000", "http"}, - {"localhost:9000", "localhost:9000", "https"}}; + const std::vector cases = { + {"https://s3.example.com:443", "s3.example.com:443", "https"}, + {"http://localhost:9000", "localhost:9000", "http"}, + {"localhost:9000", "localhost:9000", "https"}}; for (const auto& test_case : cases) { auto result = ConfigureS3Options( diff --git a/src/iceberg/test/location_util_test.cc b/src/iceberg/test/location_util_test.cc index 7098868a9..1aabe0ee0 100644 --- a/src/iceberg/test/location_util_test.cc +++ b/src/iceberg/test/location_util_test.cc @@ -61,4 +61,15 @@ TEST(LocationUtilTest, StripTrailingSlash) { ASSERT_EQ("/path///to/dir", LocationUtil::StripTrailingSlash("/path///to/dir/")); } +TEST(LocationUtilTest, ParseScheme) { + auto s3 = LocationUtil::ParseScheme("S3://bucket/path"); + EXPECT_EQ(s3, "S3"); + + auto no_scheme = LocationUtil::ParseScheme("/tmp/file.parquet"); + EXPECT_TRUE(no_scheme.empty()); + + auto empty_scheme = LocationUtil::ParseScheme("://bucket/path"); + EXPECT_TRUE(empty_scheme.empty()); +} + } // namespace iceberg diff --git a/src/iceberg/test/resolving_file_io_test.cc b/src/iceberg/test/resolving_file_io_test.cc index 97c1d7873..788b59beb 100644 --- a/src/iceberg/test/resolving_file_io_test.cc +++ b/src/iceberg/test/resolving_file_io_test.cc @@ -28,7 +28,6 @@ #include #include "iceberg/file_io_registry.h" -#include "iceberg/resolving_file_io_internal.h" #include "iceberg/test/matchers.h" namespace iceberg { @@ -43,7 +42,13 @@ class RecordingFileIO : public FileIO { return NotImplemented("recording mock"); } + Status DeleteFiles(const std::vector& file_locations) override { + deleted_batches.push_back(file_locations); + return {}; + } + std::vector locations; + std::vector> deleted_batches; }; class RecordingCredentialedFileIO : public RecordingFileIO, @@ -83,44 +88,64 @@ void RegisterRecordingFileIOs() { last_local_io = nullptr; FileIORegistry::Register( std::string(FileIORegistry::kArrowS3FileIO), - [](const std::unordered_map& properties) - -> Result> { - ++s3_factory_calls; - s3_factory_properties = properties; - auto io = std::make_unique(); - last_s3_io = io.get(); - return io; - }); + {.create = [](const std::unordered_map& properties) + -> Result> { + ++s3_factory_calls; + s3_factory_properties = properties; + auto io = std::make_unique(); + last_s3_io = io.get(); + return io; + }, + .accepts = + [](std::string_view scheme) { + return scheme == "s3" || scheme == "s3a" || scheme == "s3n"; + }}); FileIORegistry::Register( std::string(FileIORegistry::kArrowLocalFileIO), - [](const std::unordered_map& /*properties*/) - -> Result> { - ++local_factory_calls; - auto io = std::make_unique(); - last_local_io = io.get(); - return io; - }); + {.create = [](const std::unordered_map& /*properties*/) + -> Result> { + ++local_factory_calls; + auto io = std::make_unique(); + last_local_io = io.get(); + return io; + }, + .accepts = + [](std::string_view scheme) { return scheme.empty() || scheme == "file"; }}); } } // namespace -TEST(ResolvingFileIOTest, ResolvesImplementationNameFromScheme) { - EXPECT_THAT(ResolveFileIOName("s3://bucket/path"), - HasValue(::testing::Eq(FileIORegistry::kArrowS3FileIO))); - EXPECT_THAT(ResolveFileIOName("s3a://bucket/path"), - HasValue(::testing::Eq(FileIORegistry::kArrowS3FileIO))); - EXPECT_THAT(ResolveFileIOName("s3n://bucket/path"), - HasValue(::testing::Eq(FileIORegistry::kArrowS3FileIO))); - EXPECT_THAT(ResolveFileIOName("oss://bucket/path"), - HasValue(::testing::Eq(FileIORegistry::kArrowS3FileIO))); - EXPECT_THAT(ResolveFileIOName("file:///tmp/path"), - HasValue(::testing::Eq(FileIORegistry::kArrowLocalFileIO))); - EXPECT_THAT(ResolveFileIOName("/tmp/path"), - HasValue(::testing::Eq(FileIORegistry::kArrowLocalFileIO))); - - auto result = ResolveFileIOName("gs://bucket/path"); - EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); - EXPECT_THAT(result, HasErrorMessage("not supported for FileIO resolution")); +TEST(FileIORegistryTest, ResolvesLatestAndReplacedRegistration) { + FileIORegistry::Register( + "test.file-io.old", + {.create = + [](const FileIORegistry::Properties&) -> Result> { + return std::make_unique(); + }, + .accepts = [](std::string_view scheme) { return scheme == "custom"; }}); + FileIORegistry::Register( + "test.file-io.new", + {.create = + [](const FileIORegistry::Properties&) -> Result> { + return std::make_unique(); + }, + .accepts = [](std::string_view scheme) { return scheme == "custom"; }}); + + EXPECT_THAT(FileIORegistry::Resolve("CUSTOM"), + HasValue(::testing::Eq("test.file-io.new"))); + + FileIORegistry::Register( + "test.file-io.old", + {.create = + [](const FileIORegistry::Properties&) -> Result> { + return std::make_unique(); + }, + .accepts = [](std::string_view scheme) { return scheme == "replacement"; }}); + + EXPECT_THAT(FileIORegistry::Resolve("custom"), + HasValue(::testing::Eq("test.file-io.new"))); + EXPECT_THAT(FileIORegistry::Resolve("replacement"), + HasValue(::testing::Eq("test.file-io.old"))); } TEST(ResolvingFileIOTest, RoutesPathsAndCachesResolvedImplementations) { @@ -128,19 +153,20 @@ TEST(ResolvingFileIOTest, RoutesPathsAndCachesResolvedImplementations) { ResolvingFileIO io({{"k", "v"}}); // Errors come from the recording mock; routing is what is under test. - (void)io.NewInputFile("oss://bucket/db/table/data/file.parquet"); + (void)io.NewInputFile("s3a://bucket/db/table/data/file.parquet"); (void)io.NewInputFile("s3://bucket/db/table/data/file.parquet"); (void)io.NewInputFile("/tmp/local/file.parquet"); ASSERT_NE(last_s3_io, nullptr); ASSERT_NE(last_local_io, nullptr); EXPECT_THAT(last_s3_io->locations, - ::testing::ElementsAre("oss://bucket/db/table/data/file.parquet", + ::testing::ElementsAre("s3a://bucket/db/table/data/file.parquet", "s3://bucket/db/table/data/file.parquet")); EXPECT_THAT(last_local_io->locations, ::testing::ElementsAre("/tmp/local/file.parquet")); - // Resolved instances are cached; properties pass through to the factory. + // One instance per implementation despite two distinct S3 schemes routed to + // it; properties pass through to the factory. EXPECT_EQ(s3_factory_calls, 1); EXPECT_EQ(local_factory_calls, 1); EXPECT_THAT(s3_factory_properties, @@ -150,6 +176,56 @@ TEST(ResolvingFileIOTest, RoutesPathsAndCachesResolvedImplementations) { EXPECT_THAT(unsupported, IsError(ErrorKind::kNotSupported)); } +TEST(ResolvingFileIOTest, GroupsBulkDeletesByResolvedImplementation) { + RegisterRecordingFileIOs(); + ResolvingFileIO io({}); + + EXPECT_THAT(io.DeleteFiles({"s3://bucket/a", "/tmp/local-a", "s3a://bucket/b", + "file:///tmp/local-b"}), + IsOk()); + + ASSERT_NE(last_s3_io, nullptr); + ASSERT_NE(last_local_io, nullptr); + ASSERT_EQ(last_s3_io->deleted_batches.size(), 1); + EXPECT_THAT(last_s3_io->deleted_batches.front(), + ::testing::ElementsAre("s3://bucket/a", "s3a://bucket/b")); + ASSERT_EQ(last_local_io->deleted_batches.size(), 1); + EXPECT_THAT(last_local_io->deleted_batches.front(), + ::testing::ElementsAre("/tmp/local-a", "file:///tmp/local-b")); +} + +TEST(ResolvingFileIOTest, DoesNotDeleteWhenPathResolutionFails) { + RegisterRecordingFileIOs(); + ResolvingFileIO io({}); + + auto result = io.DeleteFiles({"s3://bucket/a", "gs://bucket/unknown"}); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); + ASSERT_NE(last_s3_io, nullptr); + EXPECT_TRUE(last_s3_io->deleted_batches.empty()); +} + +TEST(ResolvingFileIOTest, DoesNotFallbackAfterSelectedFactoryFails) { + FileIORegistry::Register( + "test.file-io.fallback", + {.create = + [](const FileIORegistry::Properties&) -> Result> { + return std::make_unique(); + }, + .accepts = [](std::string_view scheme) { return scheme == "failure"; }}); + FileIORegistry::Register( + "test.file-io.selected", + {.create = + [](const FileIORegistry::Properties&) -> Result> { + return InvalidArgument("selected factory failed"); + }, + .accepts = [](std::string_view scheme) { return scheme == "failure"; }}); + + ResolvingFileIO io({}); + auto result = io.NewInputFile("failure://bucket/file"); + EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(result, HasErrorMessage("selected factory failed")); +} + TEST(ResolvingFileIOTest, ForwardsAllCredentialsToResolvedImplementations) { RegisterRecordingFileIOs(); ResolvingFileIO io({}); @@ -157,7 +233,7 @@ TEST(ResolvingFileIOTest, ForwardsAllCredentialsToResolvedImplementations) { // The full credential list is forwarded; each implementation applies the // prefixes it understands. std::vector credentials = { - {.prefix = "oss", .config = {{"k1", "v1"}}}, + {.prefix = "s3a://bucket", .config = {{"k1", "v1"}}}, {.prefix = "s3", .config = {{"k2", "v2"}}}}; EXPECT_THAT(io.SetStorageCredentials(credentials), IsOk()); EXPECT_EQ(io.credentials(), credentials); diff --git a/src/iceberg/test/rest_arrow_file_io_test.cc b/src/iceberg/test/rest_arrow_file_io_test.cc deleted file mode 100644 index 67cb820c7..000000000 --- a/src/iceberg/test/rest_arrow_file_io_test.cc +++ /dev/null @@ -1,97 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -/// \file -/// \brief Covers REST -> ResolvingFileIO -> registry -> Arrow FileIO against the -/// real registered implementations, which mock delegates cannot exercise. - -#include -#include -#include -#include -#include -#include -#include - -#include -#include - -#include "iceberg/arrow/arrow_io_util.h" -#include "iceberg/arrow/arrow_register.h" -#include "iceberg/catalog/rest/rest_file_io.h" -#include "iceberg/logging/logger.h" -#include "iceberg/storage_credential.h" -#include "iceberg/test/logging_test_helpers.h" -#include "iceberg/test/matchers.h" -#include "iceberg/test/temp_file_test_base.h" - -namespace iceberg::rest { - -namespace { - -class RestArrowFileIOTest : public TempFileTestBase { - protected: - static void SetUpTestSuite() { iceberg::arrow::RegisterAll(); } - static void TearDownTestSuite() { std::ignore = iceberg::arrow::FinalizeS3(); } -}; - -TEST_F(RestArrowFileIOTest, ReadsBackWhatItWroteThroughRealLocalFileIO) { - auto io = MakeTableFileIO({{"warehouse", "logical_warehouse_name"}}, - /*table_config=*/{}, /*storage_credentials=*/{}); - ASSERT_THAT(io, IsOk()); - - const auto path = CreateNewTempFilePathWithSuffix(".txt"); - constexpr std::string_view kContent = "resolved through the real local FileIO"; - - ASSERT_THAT(io.value()->WriteFile(path, kContent), IsOk()); - EXPECT_THAT(io.value()->ReadFile(path, std::nullopt), - HasValue(::testing::Eq(std::string(kContent)))); - EXPECT_THAT(io.value()->DeleteFile(path), IsOk()); -} - -#if ICEBERG_S3_ENABLED - -bool HasWarning(const CapturingLogger& logger) { - const auto records = logger.records(); - return std::ranges::any_of( - records, [](const LogMessage& record) { return record.level == LogLevel::kWarn; }); -} - -TEST_F(RestArrowFileIOTest, AppliesOssCredentialThroughRealArrowS3FileIO) { - auto logger = std::make_shared(); - ScopedDefaultLogger scoped(logger); - - auto io = - MakeTableFileIO({{"warehouse", "logical_warehouse_name"}}, /*table_config=*/{}, - {{.prefix = "oss://bucket/table", .config = {{"k", "v"}}}}); - ASSERT_THAT(io, IsOk()); - - // Opening builds the delegate and applies the credential. The open itself hits - // the network, so only the failure modes before that are asserted: a routing - // break surfaces as kNotSupported, and a dropped credential as the warning. - auto input = io.value()->NewInputFile("oss://bucket/table/data/file.parquet"); - EXPECT_THAT(input, ::testing::Not(IsError(ErrorKind::kNotSupported))); - EXPECT_FALSE(HasWarning(*logger)); -} - -#endif // ICEBERG_S3_ENABLED - -} // namespace - -} // namespace iceberg::rest diff --git a/src/iceberg/test/rest_catalog_integration_test.cc b/src/iceberg/test/rest_catalog_integration_test.cc index b449bc50f..96f392533 100644 --- a/src/iceberg/test/rest_catalog_integration_test.cc +++ b/src/iceberg/test/rest_catalog_integration_test.cc @@ -109,10 +109,10 @@ class RestCatalogIntegrationTest : public ::testing::Test { static void SetUpTestSuite() { FileIORegistry::Register( std::string(kStdFileIOImpl), - [](const std::unordered_map& /*properties*/) - -> Result> { + {.create = [](const std::unordered_map& /*properties*/) + -> Result> { return std::make_unique(); - }); + }}); docker_compose_ = std::make_unique( std::string{kDockerProjectName}, GetResourcePath("iceberg-rest-fixture")); docker_compose_->Up(); diff --git a/src/iceberg/test/rest_file_io_test.cc b/src/iceberg/test/rest_file_io_test.cc index 6e584ea22..dde3238a8 100644 --- a/src/iceberg/test/rest_file_io_test.cc +++ b/src/iceberg/test/rest_file_io_test.cc @@ -28,6 +28,7 @@ #include "iceberg/catalog/rest/types.h" #include "iceberg/file_io_registry.h" +#include "iceberg/resolving_file_io.h" #include "iceberg/test/matchers.h" namespace iceberg::rest { @@ -78,17 +79,30 @@ TEST(RestFileIOTest, MakeCatalogFileIODefaultsToResolvingFileIO) { RestCatalogProperties::FromMap({{"warehouse", "s3://bucket/warehouse"}})}) { auto result = MakeCatalogFileIO(config); ASSERT_THAT(result, IsOk()); - // The resolving FileIO can carry vended storage credentials. - EXPECT_NE(result.value()->AsSupportsStorageCredentials(), nullptr); + EXPECT_NE(dynamic_cast(result.value().get()), nullptr); } } +TEST(RestFileIOTest, DefaultResolverDelegatesThroughRegistry) { + FileIORegistry::Register( + "test.rest.resolving-file-io", + {.create = [](const FileIORegistry::Properties&) + -> Result> { return std::make_unique(); }, + .accepts = [](std::string_view scheme) { return scheme == "rest-test"; }}); + + auto result = MakeTableFileIO({}, {}, {}); + ASSERT_THAT(result, IsOk()); + EXPECT_THAT(result.value()->DeleteFile("rest-test://file"), IsOk()); +} + TEST(RestFileIOTest, MakeCatalogFileIOPassesThroughCustomImpl) { const std::string custom_impl = "com.mycompany.CustomFileIO"; FileIORegistry::Register( custom_impl, - [](const std::unordered_map& /*properties*/) - -> Result> { return std::make_unique(); }); + {.create = [](const std::unordered_map& /*properties*/) + -> Result> { + return std::make_unique(); + }}); auto config = RestCatalogProperties::FromMap( {{"io-impl", custom_impl}, {"warehouse", "/tmp/warehouse"}}); @@ -103,42 +117,17 @@ TEST(RestFileIOTest, MakeCatalogFileIOUnregisteredCustomImplReturnsNotFound) { EXPECT_THAT(result, IsError(ErrorKind::kNotFound)); } -TEST(RestFileIOTest, TableFileIOBindsCredentialsWithLogicalWarehouseName) { - // Regression: credential-vending catalogs often use a logical warehouse name - // (bucket ARN / catalog name), not a storage URI; the S3 implementation must - // still be resolved per path scheme and receive the vended credentials, even - // when non-S3 credentials are vended alongside. - captured_storage_credentials.clear(); - FileIORegistry::Register( - std::string(FileIORegistry::kArrowS3FileIO), - [](const std::unordered_map& /*properties*/) - -> Result> { - return std::make_unique(); - }); - - std::vector credentials = { - {.prefix = "oss", .config = {{"k1", "v1"}}}, - {.prefix = "s3", .config = {{"k2", "v2"}}}}; - auto result = MakeTableFileIO({{"warehouse", "logical_warehouse_name"}}, - /*table_config=*/{}, credentials); - ASSERT_THAT(result, IsOk()); - - // Reaching a data file routes to the S3 FileIO with the full credential list. - (void)result.value()->NewInputFile("oss://bucket/db/table/data/file.parquet"); - EXPECT_EQ(captured_storage_credentials, credentials); -} - TEST(RestFileIOTest, TableFileIOMergesConfigAndCredentials) { const std::string custom_impl = "com.mycompany.CredentialedFileIO"; captured_file_io_properties.clear(); captured_storage_credentials.clear(); FileIORegistry::Register( custom_impl, - [](const std::unordered_map& properties) - -> Result> { + {.create = [](const std::unordered_map& properties) + -> Result> { captured_file_io_properties = properties; return std::make_unique(); - }); + }}); auto result = MakeTableFileIO( {{"warehouse", "s3://catalog/warehouse"}, @@ -170,11 +159,11 @@ TEST(RestFileIOTest, TableImplOverridesWarehouseScheme) { captured_file_io_properties.clear(); FileIORegistry::Register( std::string(FileIORegistry::kArrowS3FileIO), - [](const std::unordered_map& properties) - -> Result> { + {.create = [](const std::unordered_map& properties) + -> Result> { captured_file_io_properties = properties; return std::make_unique(); - }); + }}); auto result = MakeTableFileIO({{"warehouse", "/tmp/catalog-warehouse"}}, @@ -192,8 +181,10 @@ TEST(RestFileIOTest, TableFileIORejectsCredentials) { const std::string custom_impl = "com.mycompany.PlainFileIO"; FileIORegistry::Register( custom_impl, - [](const std::unordered_map& /*properties*/) - -> Result> { return std::make_unique(); }); + {.create = [](const std::unordered_map& /*properties*/) + -> Result> { + return std::make_unique(); + }}); auto result = MakeTableFileIO( {{"warehouse", "s3://catalog/warehouse"}}, {{"io-impl", custom_impl}}, diff --git a/src/iceberg/resolving_file_io_internal.h b/src/iceberg/util/location_util.cc similarity index 64% rename from src/iceberg/resolving_file_io_internal.h rename to src/iceberg/util/location_util.cc index 45c5a988b..3ae2decc1 100644 --- a/src/iceberg/resolving_file_io_internal.h +++ b/src/iceberg/util/location_util.cc @@ -17,20 +17,16 @@ * under the License. */ -#pragma once - -/// \file iceberg/resolving_file_io_internal.h -/// \brief Internal helpers for ResolvingFileIO. Not part of the public API. - -#include - -#include "iceberg/iceberg_export.h" -#include "iceberg/result.h" +#include "iceberg/util/location_util.h" namespace iceberg { -/// \brief The FileIORegistry name of the implementation serving `location`, -/// chosen by its URI scheme. Exported so tests can link it in shared builds. -ICEBERG_EXPORT Result ResolveFileIOName(std::string_view location); +std::string_view LocationUtil::ParseScheme(std::string_view location) { + const auto colon = location.find(':'); + if (colon == std::string_view::npos || colon == 0) { + return {}; + } + return location.substr(0, colon); +} } // namespace iceberg diff --git a/src/iceberg/util/location_util.h b/src/iceberg/util/location_util.h index c213a3bf7..176ea1230 100644 --- a/src/iceberg/util/location_util.h +++ b/src/iceberg/util/location_util.h @@ -30,6 +30,13 @@ namespace iceberg { class ICEBERG_EXPORT LocationUtil { public: + /// \brief Extract the URI scheme from a location. + /// + /// This follows Java's ResolvingFileIO: the text before the first colon is + /// the scheme; an empty result means that no scheme was found. It does not + /// validate the rest of the location. + static std::string_view ParseScheme(std::string_view location); + static std::string_view StripTrailingSlash(std::string_view path) { if (path.empty()) { return "";