diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index da820c26b..beec96217 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -78,6 +78,7 @@ set(ICEBERG_SOURCES partition_field.cc partition_spec.cc partition_summary.cc + resolving_file_io.cc row/arrow_array_wrapper.cc row/manifest_wrapper.cc row/partition_values.cc diff --git a/src/iceberg/arrow/s3/arrow_s3_file_io.cc b/src/iceberg/arrow/s3/arrow_s3_file_io.cc index a7e98620e..b093d2bfb 100644 --- a/src/iceberg/arrow/s3/arrow_s3_file_io.cc +++ b/src/iceberg/arrow/s3/arrow_s3_file_io.cc @@ -35,6 +35,7 @@ #include "iceberg/arrow/arrow_io_util.h" #include "iceberg/arrow/arrow_status_internal.h" #include "iceberg/arrow/s3/s3_properties.h" +#include "iceberg/logging/logger.h" #include "iceberg/util/macros.h" #include "iceberg/util/string_util.h" @@ -90,6 +91,9 @@ std::string SplitEndpointScheme(std::string_view endpoint, return std::string(endpoint); } +// Deliberately narrower than the schemes this FileIO serves: `oss`-prefixed +// credentials target native OSS connectors, while S3-compatible access is +// vended under an `s3` prefix (servers vend both side by side). bool IsS3FileIOCredentialPrefix(std::string_view prefix) { return prefix == "s3" || prefix.starts_with("s3://") || prefix.starts_with("s3a://") || prefix.starts_with("s3n://"); @@ -181,6 +185,7 @@ Result> BuildArrowS3FileSystem( return std::shared_ptr<::arrow::fs::FileSystem>(std::move(fs)); } +// Keep in sync with ResolvingFileIO::ResolveFileIOName's S3-compatible schemes. std::string CanonicalizeS3Scheme(std::string_view location) { for (std::string_view scheme : {"s3a://", "s3n://", "oss://"}) { if (location.starts_with(scheme)) { @@ -235,10 +240,11 @@ Status ArrowS3FileIO::SetStorageCredentials( // TODO(gangwu): Refresh vended credentials via credentials.uri before tokens expire. for (const auto& credential : storage_credentials) { ICEBERG_RETURN_UNEXPECTED(credential.Validate()); + // A server may vend credentials for several storage systems at once; + // non-S3 prefixes are skipped, not rejected (Java S3FileIO filters + // credentials by the "s3" prefix). if (!IsS3FileIOCredentialPrefix(credential.prefix)) { - return NotSupported( - "Storage credential prefix '{}' is unsupported by Arrow S3 FileIO", - credential.prefix); + continue; } auto properties = default_properties_; for (const auto& [key, value] : credential.config) { @@ -249,6 +255,14 @@ Status ArrowS3FileIO::SetStorageCredentials( CanonicalizeS3Scheme(credential.prefix), std::make_unique(std::move(fs))); } + if (file_io_by_prefix.empty() && !storage_credentials.empty()) { + // Silent skipping of every vended credential is hard to diagnose: S3 access + // would proceed with the default credentials and fail only at IO time. + Log(LogLevel::kWarn, + "None of the {} vended storage credential(s) has an S3-compatible prefix; " + "S3 access will use the default credentials", + storage_credentials.size()); + } file_io_by_prefix_ = std::move(file_io_by_prefix); storage_credentials_ = storage_credentials; return {}; diff --git a/src/iceberg/catalog/rest/rest_file_io.cc b/src/iceberg/catalog/rest/rest_file_io.cc index fe8a2b155..4fc5122fe 100644 --- a/src/iceberg/catalog/rest/rest_file_io.cc +++ b/src/iceberg/catalog/rest/rest_file_io.cc @@ -21,7 +21,6 @@ #include #include -#include #include #include "iceberg/catalog/rest/types.h" @@ -33,11 +32,6 @@ namespace iceberg::rest { namespace { -bool IsBuiltinImpl(std::string_view io_impl) { - return io_impl == FileIORegistry::kArrowLocalFileIO || - io_impl == FileIORegistry::kArrowS3FileIO; -} - std::unordered_map MergeFileIOProperties( const std::unordered_map& catalog_config, const std::unordered_map& table_config) { @@ -50,56 +44,13 @@ std::unordered_map MergeFileIOProperties( } // namespace -Result DetectBuiltinFileIO(std::string_view location) { - const auto pos = location.find("://"); - if (pos == std::string_view::npos) { - return BuiltinFileIOKind::kArrowLocal; - } - - const auto scheme = location.substr(0, pos); - if (scheme == "file") { - return BuiltinFileIOKind::kArrowLocal; - } - if (scheme == "s3" || scheme == "s3a" || scheme == "s3n") { - return BuiltinFileIOKind::kArrowS3; - } - - return NotSupported("URI scheme '{}' is not supported for automatic FileIO resolution", - scheme); -} - -std::string_view BuiltinFileIOName(BuiltinFileIOKind kind) { - switch (kind) { - case BuiltinFileIOKind::kArrowLocal: - return FileIORegistry::kArrowLocalFileIO; - case BuiltinFileIOKind::kArrowS3: - return FileIORegistry::kArrowS3FileIO; - } - std::unreachable(); -} - Result> MakeCatalogFileIO(const RestCatalogProperties& config) { std::string io_impl = config.Get(RestCatalogProperties::kIOImpl); - std::string warehouse = config.Get(RestCatalogProperties::kWarehouse); - if (io_impl.empty()) { - if (warehouse.empty()) { - return InvalidArgument(R"("{}" or "{}" property is required to create FileIO)", - RestCatalogProperties::kIOImpl.key(), - RestCatalogProperties::kWarehouse.key()); - } - ICEBERG_ASSIGN_OR_RAISE(const auto detected_kind, DetectBuiltinFileIO(warehouse)); - io_impl = std::string(BuiltinFileIOName(detected_kind)); - } - - if (!warehouse.empty() && IsBuiltinImpl(io_impl)) { - ICEBERG_ASSIGN_OR_RAISE(const auto detected_kind, DetectBuiltinFileIO(warehouse)); - const auto detected_name = BuiltinFileIOName(detected_kind); - if (io_impl != detected_name) { - return InvalidArgument( - R"("io-impl" value '{}' is incompatible with warehouse '{}')", io_impl, - warehouse); - } + // 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); } // TODO(gangwu): Support Java-style customized FileIO creation flows instead of @@ -112,19 +63,8 @@ Result> MakeTableFileIO( const std::unordered_map& table_config, const std::vector& storage_credentials) { const auto default_properties = MergeFileIOProperties(catalog_config, table_config); - const auto properties = RestCatalogProperties::FromMap(default_properties); - auto io_impl = properties.Get(RestCatalogProperties::kIOImpl); - if (io_impl.empty()) { - const auto warehouse = properties.Get(RestCatalogProperties::kWarehouse); - if (warehouse.empty()) { - return InvalidArgument(R"("{}" or "{}" property is required to create FileIO)", - RestCatalogProperties::kIOImpl.key(), - RestCatalogProperties::kWarehouse.key()); - } - ICEBERG_ASSIGN_OR_RAISE(const auto detected_kind, DetectBuiltinFileIO(warehouse)); - io_impl = std::string(BuiltinFileIOName(detected_kind)); - } - ICEBERG_ASSIGN_OR_RAISE(auto io, FileIORegistry::Load(io_impl, default_properties)); + ICEBERG_ASSIGN_OR_RAISE( + auto io, MakeCatalogFileIO(RestCatalogProperties::FromMap(default_properties))); if (storage_credentials.empty()) { return io; diff --git a/src/iceberg/catalog/rest/rest_file_io.h b/src/iceberg/catalog/rest/rest_file_io.h index 9bd0a826e..e2316c3e8 100644 --- a/src/iceberg/catalog/rest/rest_file_io.h +++ b/src/iceberg/catalog/rest/rest_file_io.h @@ -22,9 +22,7 @@ /// \file iceberg/catalog/rest/rest_file_io.h /// \brief Provide helpers to create FileIO instances for REST catalog responses. -#include #include -#include #include #include @@ -37,16 +35,8 @@ namespace iceberg::rest { -enum class BuiltinFileIOKind : uint8_t { - kArrowLocal, - kArrowS3, -}; - -ICEBERG_REST_EXPORT Result DetectBuiltinFileIO( - std::string_view location); - -ICEBERG_REST_EXPORT std::string_view BuiltinFileIOName(BuiltinFileIOKind kind); - +/// \brief Build the catalog FileIO: the configured `io-impl`, or the +/// scheme-resolving FileIO by default. ICEBERG_REST_EXPORT Result> MakeCatalogFileIO( const RestCatalogProperties& config); diff --git a/src/iceberg/file_io_registry.cc b/src/iceberg/file_io_registry.cc index 77ff4a9d7..073e7f382 100644 --- a/src/iceberg/file_io_registry.cc +++ b/src/iceberg/file_io_registry.cc @@ -22,6 +22,8 @@ #include #include +#include "iceberg/resolving_file_io.h" + namespace iceberg { namespace { @@ -29,6 +31,15 @@ namespace { struct RegistryState { std::mutex mutex; std::unordered_map registry; + + 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); + }; + } }; RegistryState& State() { diff --git a/src/iceberg/file_io_registry.h b/src/iceberg/file_io_registry.h index 1643bf457..e9b899fe2 100644 --- a/src/iceberg/file_io_registry.h +++ b/src/iceberg/file_io_registry.h @@ -43,6 +43,8 @@ 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>( diff --git a/src/iceberg/meson.build b/src/iceberg/meson.build index e5158a4b9..7a3532ce2 100644 --- a/src/iceberg/meson.build +++ b/src/iceberg/meson.build @@ -103,6 +103,7 @@ iceberg_sources = files( 'partition_field.cc', 'partition_spec.cc', 'partition_summary.cc', + 'resolving_file_io.cc', 'row/arrow_array_wrapper.cc', 'row/manifest_wrapper.cc', 'row/partition_values.cc', @@ -295,6 +296,7 @@ install_headers( 'name_mapping.h', 'partition_field.h', 'partition_spec.h', + 'resolving_file_io.h', 'result.h', 'schema_field.h', 'schema.h', diff --git a/src/iceberg/resolving_file_io.cc b/src/iceberg/resolving_file_io.cc new file mode 100644 index 000000000..1f148933a --- /dev/null +++ b/src/iceberg/resolving_file_io.cc @@ -0,0 +1,126 @@ +/* + * 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. + */ + +#include "iceberg/resolving_file_io.h" + +#include + +#include "iceberg/file_io_registry.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +ResolvingFileIO::ResolvingFileIO(std::unordered_map properties) + : properties_(std::move(properties)) {} + +ResolvingFileIO::~ResolvingFileIO() = default; + +Result ResolvingFileIO::ResolveFileIOName(std::string_view location) { + const auto pos = location.find("://"); + if (pos == std::string_view::npos) { + return FileIORegistry::kArrowLocalFileIO; + } + + 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; + } + + 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_); + 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_)); + // Forward all credentials; each implementation applies the prefixes it + // understands. + if (!storage_credentials_.empty()) { + if (auto* credentialed = io->AsSupportsStorageCredentials()) { + ICEBERG_RETURN_UNEXPECTED( + credentialed->SetStorageCredentials(storage_credentials_)); + } + } + it = io_by_name_.emplace(std::string(name), std::move(io)).first; + } + return it->second.get(); +} + +Result> ResolvingFileIO::NewInputFile( + std::string 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)); + return io->NewInputFile(std::move(file_location), length); +} + +Result> ResolvingFileIO::NewOutputFile( + std::string 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)); + return io->DeleteFile(file_location); +} + +Status ResolvingFileIO::DeleteFiles(const std::vector& file_locations) { + std::unordered_map> locations_by_io; + for (const auto& file_location : file_locations) { + ICEBERG_ASSIGN_OR_RAISE(auto* io, FileIOForPath(file_location)); + locations_by_io[io].push_back(file_location); + } + for (auto& [io, locations] : locations_by_io) { + ICEBERG_RETURN_UNEXPECTED(io->DeleteFiles(locations)); + } + return {}; +} + +Status ResolvingFileIO::SetStorageCredentials( + const std::vector& storage_credentials) { + std::lock_guard lock(mutex_); + storage_credentials_ = storage_credentials; + for (auto& [name, io] : io_by_name_) { + if (auto* credentialed = io->AsSupportsStorageCredentials()) { + ICEBERG_RETURN_UNEXPECTED( + credentialed->SetStorageCredentials(storage_credentials_)); + } + } + return {}; +} + +const std::vector& ResolvingFileIO::credentials() const { + return storage_credentials_; +} + +} // namespace iceberg diff --git a/src/iceberg/resolving_file_io.h b/src/iceberg/resolving_file_io.h new file mode 100644 index 000000000..e01578eb2 --- /dev/null +++ b/src/iceberg/resolving_file_io.h @@ -0,0 +1,87 @@ +/* + * 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. + */ + +#pragma once + +/// \file iceberg/resolving_file_io.h +/// \brief FileIO that resolves the concrete implementation per file-path scheme. + +#include +#include +#include +#include +#include +#include + +#include "iceberg/file_io.h" +#include "iceberg/iceberg_export.h" +#include "iceberg/result.h" +#include "iceberg/storage_credential.h" +#include "iceberg/util/string_util.h" + +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. +class ICEBERG_EXPORT ResolvingFileIO final : public FileIO, + public SupportsStorageCredentials { + public: + explicit ResolvingFileIO(std::unordered_map properties); + ~ResolvingFileIO() override; + + /// \brief The FileIORegistry name of the implementation serving `location`. + static Result ResolveFileIOName(std::string_view location); + + Result> NewInputFile(std::string file_location) override; + + Result> NewInputFile(std::string file_location, + size_t length) override; + + Result> NewOutputFile(std::string file_location) override; + + Status DeleteFile(const std::string& file_location) override; + + Status DeleteFiles(const std::vector& file_locations) override; + + Status SetStorageCredentials( + const std::vector& storage_credentials) override; + + const std::vector& credentials() const override; + + SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; } + + private: + /// \brief Load (or return the cached) implementation serving `location`. + Result FileIOForPath(std::string_view location); + + std::unordered_map properties_; + // Guards lazy resolution; set credentials before sharing across threads. + std::mutex mutex_; + std::vector storage_credentials_; + std::unordered_map, StringHash, StringEqual> + io_by_name_; +}; + +} // namespace iceberg diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 7a1b45b70..a1b53b40c 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -144,6 +144,7 @@ add_iceberg_test(util_test roaring_position_bitmap_test.cc position_delete_index_test.cc position_delete_range_consumer_test.cc + resolving_file_io_test.cc retry_util_test.cc string_util_test.cc struct_like_set_test.cc diff --git a/src/iceberg/test/arrow_s3_file_io_test.cc b/src/iceberg/test/arrow_s3_file_io_test.cc index 701a8f0d9..d5d1c06ce 100644 --- a/src/iceberg/test/arrow_s3_file_io_test.cc +++ b/src/iceberg/test/arrow_s3_file_io_test.cc @@ -156,18 +156,37 @@ TEST_F(ArrowS3FileIOTest, StoresCredentials) { EXPECT_EQ(credentialed->credentials(), credentials); } -TEST_F(ArrowS3FileIOTest, RejectsCredentialPrefix) { +TEST_F(ArrowS3FileIOTest, SkipsNonS3CredentialPrefix) { auto result = MakeS3FileIO({}); ASSERT_THAT(result, IsOk()); auto* credentialed = result.value()->AsSupportsStorageCredentials(); ASSERT_NE(credentialed, nullptr); - auto status = credentialed->SetStorageCredentials( - {{.prefix = "gs://bucket/table", - .config = {{std::string(S3Properties::kAccessKeyId), "access-key"}, - {std::string(S3Properties::kSecretAccessKey), "secret"}}}}); - EXPECT_THAT(status, IsError(ErrorKind::kNotSupported)); - EXPECT_THAT(status, HasErrorMessage("unsupported by Arrow S3 FileIO")); + // A server may vend credentials for several storage systems at once; + // non-S3 prefixes (here `oss`) are skipped rather than rejected. + std::vector credentials = { + {.prefix = "oss://bucket/table", + .config = {{std::string(S3Properties::kAccessKeyId), "oss-access-key"}, + {std::string(S3Properties::kSecretAccessKey), "oss-secret"}}}, + {.prefix = "s3://bucket/table", + .config = {{std::string(S3Properties::kAccessKeyId), "access-key"}, + {std::string(S3Properties::kSecretAccessKey), "secret"}}}}; + EXPECT_THAT(credentialed->SetStorageCredentials(credentials), IsOk()); + EXPECT_EQ(credentialed->credentials(), credentials); +} + +TEST_F(ArrowS3FileIOTest, AcceptsCredentialsWithoutS3Prefix) { + auto result = MakeS3FileIO({}); + ASSERT_THAT(result, IsOk()); + auto* credentialed = result.value()->AsSupportsStorageCredentials(); + ASSERT_NE(credentialed, nullptr); + + // Even when no vended credential is S3-family, setting them succeeds (a + // warning is logged) and S3 access falls back to the default credentials. + std::vector credentials = { + {.prefix = "gs://bucket/table", .config = {{"k", "v"}}}}; + EXPECT_THAT(credentialed->SetStorageCredentials(credentials), IsOk()); + EXPECT_EQ(credentialed->credentials(), credentials); } TEST_F(ArrowS3FileIOTest, RejectsIncompleteStaticCredentials) { diff --git a/src/iceberg/test/meson.build b/src/iceberg/test/meson.build index 12460237d..5363fbfd4 100644 --- a/src/iceberg/test/meson.build +++ b/src/iceberg/test/meson.build @@ -108,6 +108,7 @@ iceberg_tests = { 'math_util_internal_test.cc', 'position_delete_index_test.cc', 'position_delete_range_consumer_test.cc', + 'resolving_file_io_test.cc', 'retry_util_test.cc', 'roaring_position_bitmap_test.cc', 'string_util_test.cc', diff --git a/src/iceberg/test/resolving_file_io_test.cc b/src/iceberg/test/resolving_file_io_test.cc new file mode 100644 index 000000000..2cc266f46 --- /dev/null +++ b/src/iceberg/test/resolving_file_io_test.cc @@ -0,0 +1,178 @@ +/* + * 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. + */ + +#include "iceberg/resolving_file_io.h" + +#include +#include +#include +#include + +#include +#include + +#include "iceberg/file_io_registry.h" +#include "iceberg/test/matchers.h" + +namespace iceberg { + +namespace { + +/// Records every NewInputFile location routed to it. +class RecordingFileIO : public FileIO { + public: + Result> NewInputFile(std::string file_location) override { + locations.push_back(std::move(file_location)); + return NotImplemented("recording mock"); + } + + std::vector locations; +}; + +class RecordingCredentialedFileIO : public RecordingFileIO, + public SupportsStorageCredentials { + public: + Status SetStorageCredentials( + const std::vector& storage_credentials) override { + credentials_ = storage_credentials; + return {}; + } + + const std::vector& credentials() const override { + return credentials_; + } + + SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; } + + private: + std::vector credentials_; +}; + +// File-scope recording state: registry factories are process-global, so they +// must not capture test-local objects. +int s3_factory_calls = 0; +int local_factory_calls = 0; +std::unordered_map s3_factory_properties; +RecordingCredentialedFileIO* last_s3_io = nullptr; +RecordingFileIO* last_local_io = nullptr; + +/// Registers recording mocks for the builtin S3/local names and resets the +/// recording state. +void RegisterRecordingFileIOs() { + s3_factory_calls = 0; + local_factory_calls = 0; + s3_factory_properties.clear(); + last_s3_io = nullptr; + 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; + }); + 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; + }); +} + +} // namespace + +TEST(ResolvingFileIOTest, ResolvesImplementationNameFromScheme) { + EXPECT_THAT(ResolvingFileIO::ResolveFileIOName("s3://bucket/path"), + HasValue(::testing::Eq(FileIORegistry::kArrowS3FileIO))); + EXPECT_THAT(ResolvingFileIO::ResolveFileIOName("s3a://bucket/path"), + HasValue(::testing::Eq(FileIORegistry::kArrowS3FileIO))); + EXPECT_THAT(ResolvingFileIO::ResolveFileIOName("s3n://bucket/path"), + HasValue(::testing::Eq(FileIORegistry::kArrowS3FileIO))); + EXPECT_THAT(ResolvingFileIO::ResolveFileIOName("oss://bucket/path"), + HasValue(::testing::Eq(FileIORegistry::kArrowS3FileIO))); + EXPECT_THAT(ResolvingFileIO::ResolveFileIOName("file:///tmp/path"), + HasValue(::testing::Eq(FileIORegistry::kArrowLocalFileIO))); + EXPECT_THAT(ResolvingFileIO::ResolveFileIOName("/tmp/path"), + HasValue(::testing::Eq(FileIORegistry::kArrowLocalFileIO))); + + auto result = ResolvingFileIO::ResolveFileIOName("gs://bucket/path"); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); + EXPECT_THAT(result, HasErrorMessage("not supported for FileIO resolution")); +} + +TEST(ResolvingFileIOTest, RoutesPathsAndCachesResolvedImplementations) { + RegisterRecordingFileIOs(); + 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("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", + "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. + EXPECT_EQ(s3_factory_calls, 1); + EXPECT_EQ(local_factory_calls, 1); + EXPECT_THAT(s3_factory_properties, + ::testing::UnorderedElementsAre(::testing::Pair("k", "v"))); + + auto unsupported = io.NewInputFile("gs://bucket/file.parquet"); + EXPECT_THAT(unsupported, IsError(ErrorKind::kNotSupported)); +} + +TEST(ResolvingFileIOTest, ForwardsAllCredentialsToResolvedImplementations) { + RegisterRecordingFileIOs(); + ResolvingFileIO io({}); + + // The full credential list is forwarded; each implementation applies the + // prefixes it understands. + std::vector credentials = { + {.prefix = "oss", .config = {{"k1", "v1"}}}, + {.prefix = "s3", .config = {{"k2", "v2"}}}}; + EXPECT_THAT(io.SetStorageCredentials(credentials), IsOk()); + EXPECT_EQ(io.credentials(), credentials); + + (void)io.NewInputFile("s3://bucket/db/table/data/file.parquet"); + ASSERT_NE(last_s3_io, nullptr); + EXPECT_EQ(last_s3_io->credentials(), credentials); + + // Credentials set after an implementation was resolved reach it as well. + std::vector refreshed = {{.prefix = "s3", .config = {{"k3", "v3"}}}}; + EXPECT_THAT(io.SetStorageCredentials(refreshed), IsOk()); + EXPECT_EQ(last_s3_io->credentials(), refreshed); + + // The local FileIO does not support credentials; resolving it still works. + (void)io.NewInputFile("/tmp/local/file.parquet"); + ASSERT_NE(last_local_io, nullptr); +} + +} // namespace iceberg diff --git a/src/iceberg/test/rest_file_io_test.cc b/src/iceberg/test/rest_file_io_test.cc index 7f41b0b46..6e584ea22 100644 --- a/src/iceberg/test/rest_file_io_test.cc +++ b/src/iceberg/test/rest_file_io_test.cc @@ -69,73 +69,18 @@ class MockCredentialedFileIO : public MockFileIO, public SupportsStorageCredenti } // namespace -TEST(RestFileIOTest, DetectBuiltinKindFromScheme) { - EXPECT_THAT(DetectBuiltinFileIO("s3://bucket/path"), - HasValue(::testing::Eq(BuiltinFileIOKind::kArrowS3))); - EXPECT_THAT(DetectBuiltinFileIO("s3a://bucket/path"), - HasValue(::testing::Eq(BuiltinFileIOKind::kArrowS3))); - EXPECT_THAT(DetectBuiltinFileIO("s3n://bucket/path"), - HasValue(::testing::Eq(BuiltinFileIOKind::kArrowS3))); - EXPECT_THAT(DetectBuiltinFileIO("/tmp/warehouse"), - HasValue(::testing::Eq(BuiltinFileIOKind::kArrowLocal))); - EXPECT_THAT(DetectBuiltinFileIO("file:///tmp/warehouse"), - HasValue(::testing::Eq(BuiltinFileIOKind::kArrowLocal))); -} - -TEST(RestFileIOTest, DetectBuiltinKindRejectsUnsupportedScheme) { - auto result = DetectBuiltinFileIO("gs://bucket/warehouse"); - EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); - EXPECT_THAT(result, HasErrorMessage("not supported for automatic FileIO resolution")); -} - -TEST(RestFileIOTest, MakeCatalogFileIOMissingImplAndWarehouse) { - auto result = MakeCatalogFileIO(RestCatalogProperties::default_properties()); - EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument)); -} - -TEST(RestFileIOTest, MakeCatalogFileIORejectsIncompatibleWarehouse) { - FileIORegistry::Register( - std::string(FileIORegistry::kArrowS3FileIO), - [](const std::unordered_map& /*properties*/) - -> Result> { return std::make_unique(); }); - - auto config = RestCatalogProperties::FromMap( - {{"io-impl", std::string(FileIORegistry::kArrowS3FileIO)}, - {"warehouse", "/tmp/warehouse"}}); - auto result = MakeCatalogFileIO(config); - EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument)); - EXPECT_THAT(result, HasErrorMessage("incompatible")); -} - -TEST(RestFileIOTest, MakeCatalogFileIOAutoDetectsFromWarehouse) { - FileIORegistry::Register( - std::string(FileIORegistry::kArrowLocalFileIO), - [](const std::unordered_map& /*properties*/) - -> Result> { return std::make_unique(); }); - - auto config = RestCatalogProperties::FromMap({{"warehouse", "/tmp/warehouse"}}); - auto result = MakeCatalogFileIO(config); - ASSERT_THAT(result, IsOk()); -} - -TEST(RestFileIOTest, MakeCatalogFileIORejectsUnsupportedWarehouseScheme) { - auto config = RestCatalogProperties::FromMap({{"warehouse", "gs://bucket/warehouse"}}); - auto result = MakeCatalogFileIO(config); - EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); - EXPECT_THAT(result, HasErrorMessage("not supported for automatic FileIO resolution")); -} - -TEST(RestFileIOTest, MakeCatalogFileIOAllowsCompatibleWarehouse) { - FileIORegistry::Register( - std::string(FileIORegistry::kArrowS3FileIO), - [](const std::unordered_map& /*properties*/) - -> Result> { return std::make_unique(); }); - - auto config = RestCatalogProperties::FromMap( - {{"io-impl", std::string(FileIORegistry::kArrowS3FileIO)}, - {"warehouse", "s3://my-bucket/warehouse"}}); - auto result = MakeCatalogFileIO(config); - ASSERT_THAT(result, IsOk()); +TEST(RestFileIOTest, MakeCatalogFileIODefaultsToResolvingFileIO) { + // Without an explicit io-impl the scheme-resolving FileIO is used; no + // warehouse is required and its value never selects the implementation. + for (const auto& config : + {RestCatalogProperties::default_properties(), + RestCatalogProperties::FromMap({{"warehouse", "logical_warehouse_name"}}), + 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); + } } TEST(RestFileIOTest, MakeCatalogFileIOPassesThroughCustomImpl) { @@ -158,16 +103,29 @@ TEST(RestFileIOTest, MakeCatalogFileIOUnregisteredCustomImplReturnsNotFound) { EXPECT_THAT(result, IsError(ErrorKind::kNotFound)); } -TEST(RestFileIOTest, MakeCatalogFileIOSkipsCheckWhenWarehouseAbsent) { +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::kArrowLocalFileIO), + std::string(FileIORegistry::kArrowS3FileIO), [](const std::unordered_map& /*properties*/) - -> Result> { return std::make_unique(); }); + -> Result> { + return std::make_unique(); + }); - auto config = RestCatalogProperties::FromMap( - {{"io-impl", std::string(FileIORegistry::kArrowLocalFileIO)}}); - auto result = MakeCatalogFileIO(config); + 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) {