wgtmac commented on code in PR #828: URL: https://github.com/apache/iceberg-cpp/pull/828#discussion_r3637203542
########## 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 <utility> + +#include "iceberg/file_io_registry.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +ResolvingFileIO::ResolvingFileIO(std::unordered_map<std::string, std::string> properties) + : properties_(std::move(properties)) {} + +ResolvingFileIO::~ResolvingFileIO() = default; + +Result<std::string_view> 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") { Review Comment: Please either remove `oss` from this mapping or make `ArrowS3FileIO` accept and canonicalize `oss://` credentials. REST credentials are scoped by storage-location prefix, but the S3 delegate currently drops `oss://` credentials and falls back to its defaults. A catalog that vends only `oss://bucket/table` credentials will therefore fail at I/O time. Java avoids this mismatch by not mapping `oss` to `S3FileIO`. ########## 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 <utility> + +#include "iceberg/file_io_registry.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +ResolvingFileIO::ResolvingFileIO(std::unordered_map<std::string, std::string> properties) + : properties_(std::move(properties)) {} + +ResolvingFileIO::~ResolvingFileIO() = default; + +Result<std::string_view> 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<FileIO*> 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<std::unique_ptr<InputFile>> ResolvingFileIO::NewInputFile( + std::string file_location) { + ICEBERG_ASSIGN_OR_RAISE(auto* io, FileIOForPath(file_location)); + return io->NewInputFile(std::move(file_location)); +} + +Result<std::unique_ptr<InputFile>> 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<std::unique_ptr<OutputFile>> 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<std::string>& file_locations) { + std::unordered_map<FileIO*, std::vector<std::string>> 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<StorageCredential>& storage_credentials) { + std::lock_guard lock(mutex_); + storage_credentials_ = storage_credentials; Review Comment: Please do not update `storage_credentials_` until the cached delegates have accepted the new credentials. If one delegate rejects the update, this method returns an error while `credentials()` reports the new values and the cached delegate keeps using the old ones; later cache hits never retry. Invalidating and rebuilding delegates on credential changes may be simpler than rolling back partial updates. ########## 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 <memory> +#include <mutex> +#include <string> +#include <string_view> +#include <unordered_map> +#include <vector> + +#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<std::string, std::string> properties); + ~ResolvingFileIO() override; + + /// \brief The FileIORegistry name of the implementation serving `location`. + static Result<std::string_view> ResolveFileIOName(std::string_view location); Review Comment: Can this be private or internal? The resolver and its tests are the only callers, while exposing registry implementation names from an installed header makes the current scheme mapping part of the public API. Java keeps the equivalent `implFromLocation` package-private. ########## src/iceberg/test/rest_file_io_test.cc: ########## @@ -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<std::string, std::string>& /*properties*/) - -> Result<std::unique_ptr<FileIO>> { return std::make_unique<MockFileIO>(); }); + -> Result<std::unique_ptr<FileIO>> { + return std::make_unique<MockCredentialedFileIO>(); + }); - auto config = RestCatalogProperties::FromMap( - {{"io-impl", std::string(FileIORegistry::kArrowLocalFileIO)}}); - auto result = MakeCatalogFileIO(config); + std::vector<StorageCredential> 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"); Review Comment: Could we cover this path with the real registered Arrow S3 FileIO? The mock only proves that the full credential list reaches the delegate; it cannot catch that production `ArrowS3FileIO` drops the `oss://` credential before opening this `oss://` location. A focused `REST -> ResolvingFileIO -> FileIORegistry -> ArrowS3FileIO` test would catch this regression. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
