feat(io): add ResolvingFileIO to resolve FileIO by location scheme and forward vended credentials (#828)
diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
index 26ff8e0..dec79ad 100644
--- a/src/iceberg/CMakeLists.txt
+++ b/src/iceberg/CMakeLists.txt
@@ -83,6 +83,7 @@
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 a7e9862..e311845 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/log_macros.h"
#include "iceberg/util/macros.h"
#include "iceberg/util/string_util.h"
@@ -90,9 +91,11 @@
return std::string(endpoint);
}
+// Location prefixes this FileIO can serve: must cover every scheme
+// ResolveFileIOName routes here, or such a credential would be dropped.
bool IsS3FileIOCredentialPrefix(std::string_view prefix) {
return prefix == "s3" || prefix.starts_with("s3://") || prefix.starts_with("s3a://") ||
- prefix.starts_with("s3n://");
+ prefix.starts_with("s3n://") || prefix.starts_with("oss://");
}
} // namespace
@@ -181,6 +184,7 @@
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://"}) {
if (location.starts_with(scheme)) {
@@ -235,10 +239,11 @@
// 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 +254,14 @@
CanonicalizeS3Scheme(credential.prefix),
std::make_unique<ArrowFileSystemFileIO>(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.
+ ICEBERG_LOG_WARN(
+ "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 fe8a2b1..4fc5122 100644
--- a/src/iceberg/catalog/rest/rest_file_io.cc
+++ b/src/iceberg/catalog/rest/rest_file_io.cc
@@ -21,7 +21,6 @@
#include <string>
#include <unordered_map>
-#include <utility>
#include <vector>
#include "iceberg/catalog/rest/types.h"
@@ -33,11 +32,6 @@
namespace {
-bool IsBuiltinImpl(std::string_view io_impl) {
- return io_impl == FileIORegistry::kArrowLocalFileIO ||
- io_impl == FileIORegistry::kArrowS3FileIO;
-}
-
std::unordered_map<std::string, std::string> MergeFileIOProperties(
const std::unordered_map<std::string, std::string>& catalog_config,
const std::unordered_map<std::string, std::string>& table_config) {
@@ -50,56 +44,13 @@
} // namespace
-Result<BuiltinFileIOKind> 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<std::unique_ptr<FileIO>> 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 @@
const std::unordered_map<std::string, std::string>& table_config,
const std::vector<StorageCredential>& 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 9bd0a82..e2316c3 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 <cstdint>
#include <memory>
-#include <string_view>
#include <unordered_map>
#include <vector>
@@ -37,16 +35,8 @@
namespace iceberg::rest {
-enum class BuiltinFileIOKind : uint8_t {
- kArrowLocal,
- kArrowS3,
-};
-
-ICEBERG_REST_EXPORT Result<BuiltinFileIOKind> 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<std::unique_ptr<FileIO>> MakeCatalogFileIO(
const RestCatalogProperties& config);
diff --git a/src/iceberg/file_io_registry.cc b/src/iceberg/file_io_registry.cc
index 77ff4a9..073e7f3 100644
--- a/src/iceberg/file_io_registry.cc
+++ b/src/iceberg/file_io_registry.cc
@@ -22,6 +22,8 @@
#include <mutex>
#include <utility>
+#include "iceberg/resolving_file_io.h"
+
namespace iceberg {
namespace {
@@ -29,6 +31,15 @@
struct RegistryState {
std::mutex mutex;
std::unordered_map<std::string, FileIORegistry::Factory> registry;
+
+ RegistryState() {
+ // Always available: the scheme-resolving FileIO lives in the core library.
+ registry[std::string(FileIORegistry::kResolvingFileIO)] =
+ [](const std::unordered_map<std::string, std::string>& properties)
+ -> Result<std::unique_ptr<FileIO>> {
+ return std::make_unique<ResolvingFileIO>(properties);
+ };
+ }
};
RegistryState& State() {
diff --git a/src/iceberg/file_io_registry.h b/src/iceberg/file_io_registry.h
index 1643bf4..e9b899f 100644
--- a/src/iceberg/file_io_registry.h
+++ b/src/iceberg/file_io_registry.h
@@ -43,6 +43,8 @@
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<Result<std::unique_ptr<FileIO>>(
diff --git a/src/iceberg/meson.build b/src/iceberg/meson.build
index 5e742f2..8f5680a 100644
--- a/src/iceberg/meson.build
+++ b/src/iceberg/meson.build
@@ -136,6 +136,7 @@
'partition_spec.cc',
'partition_summary.cc',
'puffin_dv_io.cc',
+ 'resolving_file_io.cc',
'row/arrow_array_wrapper.cc',
'row/manifest_wrapper.cc',
'row/partition_values.cc',
@@ -329,6 +330,7 @@
'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 0000000..ce91a52
--- /dev/null
+++ b/src/iceberg/resolving_file_io.cc
@@ -0,0 +1,124 @@
+/*
+ * 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/resolving_file_io_internal.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> 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) {
+ // 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_);
+ storage_credentials_ = storage_credentials;
+ io_by_name_.clear();
+ return {};
+}
+
+const std::vector<StorageCredential>& 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 0000000..d96b0ca
--- /dev/null
+++ b/src/iceberg/resolving_file_io.h
@@ -0,0 +1,89 @@
+/*
+ * 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.
+///
+/// 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.
+class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
+ public SupportsStorageCredentials {
+ public:
+ explicit ResolvingFileIO(std::unordered_map<std::string, std::string> properties);
+ ~ResolvingFileIO() override;
+
+ Result<std::unique_ptr<InputFile>> NewInputFile(std::string file_location) override;
+
+ Result<std::unique_ptr<InputFile>> NewInputFile(std::string file_location,
+ size_t length) override;
+
+ Result<std::unique_ptr<OutputFile>> NewOutputFile(std::string file_location) override;
+
+ Status DeleteFile(const std::string& file_location) override;
+
+ Status DeleteFiles(const std::vector<std::string>& file_locations) override;
+
+ Status SetStorageCredentials(
+ const std::vector<StorageCredential>& storage_credentials) override;
+
+ const std::vector<StorageCredential>& credentials() const override;
+
+ SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; }
+
+ private:
+ /// \brief Load (or return the cached) implementation serving `location`.
+ Result<FileIO*> FileIOForPath(std::string_view location);
+
+ std::unordered_map<std::string, std::string> properties_;
+ // Guards lazy resolution; set credentials before sharing across threads.
+ std::mutex mutex_;
+ std::vector<StorageCredential> storage_credentials_;
+ std::unordered_map<std::string, std::unique_ptr<FileIO>, StringHash, StringEqual>
+ io_by_name_;
+};
+
+} // namespace iceberg
diff --git a/src/iceberg/resolving_file_io_internal.h b/src/iceberg/resolving_file_io_internal.h
new file mode 100644
index 0000000..45c5a98
--- /dev/null
+++ b/src/iceberg/resolving_file_io_internal.h
@@ -0,0 +1,36 @@
+/*
+ * 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_internal.h
+/// \brief Internal helpers for ResolvingFileIO. Not part of the public API.
+
+#include <string_view>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.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<std::string_view> ResolveFileIOName(std::string_view location);
+
+} // namespace iceberg
diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt
index 7cf013c..1181a72 100644
--- a/src/iceberg/test/CMakeLists.txt
+++ b/src/iceberg/test/CMakeLists.txt
@@ -140,6 +140,7 @@
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
@@ -290,12 +291,22 @@
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 "$<IF:$<TARGET_EXISTS:iceberg_bundle_static>,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_s3_file_io_test.cc b/src/iceberg/test/arrow_s3_file_io_test.cc
index 701a8f0..aa949ab 100644
--- a/src/iceberg/test/arrow_s3_file_io_test.cc
+++ b/src/iceberg/test/arrow_s3_file_io_test.cc
@@ -17,8 +17,10 @@
* under the License.
*/
+#include <algorithm>
#include <cstdlib>
#include <iostream>
+#include <memory>
#include <optional>
#include <string>
#include <string_view>
@@ -36,8 +38,10 @@
#include "iceberg/arrow/arrow_io_util.h"
#include "iceberg/arrow/s3/s3_properties.h"
#include "iceberg/file_io.h"
+#include "iceberg/logging/logger.h"
#include "iceberg/result.h"
#include "iceberg/storage_credential.h"
+#include "iceberg/test/logging_test_helpers.h"
#include "iceberg/test/matchers.h"
#include "iceberg/util/macros.h"
@@ -126,6 +130,12 @@
std::optional<std::string> base_uri_;
};
+bool HasWarning(const CapturingLogger& logger) {
+ const auto records = logger.records();
+ return std::ranges::any_of(
+ records, [](const LogMessage& record) { return record.level == LogLevel::kWarn; });
+}
+
Status CheckReadWrite(FileIO& io, const std::string& object_uri,
std::string_view content) {
ICEBERG_RETURN_UNEXPECTED(io.WriteFile(object_uri, content));
@@ -156,18 +166,64 @@
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.
+ auto logger = std::make_shared<CapturingLogger>();
+ ScopedDefaultLogger scoped(logger);
+ std::vector<StorageCredential> credentials = {
+ {.prefix = "gs://bucket/table", .config = {{"k", "v"}}},
+ {.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);
+ // The whole list is retained, but only the S3 one is applied — and it must
+ // be, otherwise the skip silently degrades to "no credentials at all".
+ EXPECT_FALSE(HasWarning(*logger));
+}
+
+// Every prefix form this FileIO claims to serve must actually be applied: a
+// 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"}) {
+ SCOPED_TRACE(prefix);
+ auto result = MakeS3FileIO({});
+ ASSERT_THAT(result, IsOk());
+ auto* credentialed = result.value()->AsSupportsStorageCredentials();
+ ASSERT_NE(credentialed, nullptr);
+
+ auto logger = std::make_shared<CapturingLogger>();
+ ScopedDefaultLogger scoped(logger);
+ std::vector<StorageCredential> credentials = {
+ {.prefix = std::string(prefix),
+ .config = {{std::string(S3Properties::kAccessKeyId), "access-key"},
+ {std::string(S3Properties::kSecretAccessKey), "secret"}}}};
+ EXPECT_THAT(credentialed->SetStorageCredentials(credentials), IsOk());
+ EXPECT_FALSE(HasWarning(*logger));
+ }
+}
+
+TEST_F(ArrowS3FileIOTest, WarnsWhenNoCredentialApplies) {
+ auto result = MakeS3FileIO({});
+ ASSERT_THAT(result, IsOk());
+ auto* credentialed = result.value()->AsSupportsStorageCredentials();
+ ASSERT_NE(credentialed, nullptr);
+
+ // Succeeds (S3 falls back to the default credentials) but must not be silent.
+ auto logger = std::make_shared<CapturingLogger>();
+ ScopedDefaultLogger scoped(logger);
+ std::vector<StorageCredential> credentials = {
+ {.prefix = "gs://bucket/table", .config = {{"k", "v"}}}};
+ EXPECT_THAT(credentialed->SetStorageCredentials(credentials), IsOk());
+ EXPECT_EQ(credentialed->credentials(), credentials);
+ EXPECT_TRUE(HasWarning(*logger));
}
TEST_F(ArrowS3FileIOTest, RejectsIncompleteStaticCredentials) {
diff --git a/src/iceberg/test/meson.build b/src/iceberg/test/meson.build
index f9549ee..6dde45e 100644
--- a/src/iceberg/test/meson.build
+++ b/src/iceberg/test/meson.build
@@ -113,6 +113,7 @@
'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 0000000..97c1d78
--- /dev/null
+++ b/src/iceberg/test/resolving_file_io_test.cc
@@ -0,0 +1,183 @@
+/*
+ * 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 <memory>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include "iceberg/file_io_registry.h"
+#include "iceberg/resolving_file_io_internal.h"
+#include "iceberg/test/matchers.h"
+
+namespace iceberg {
+
+namespace {
+
+/// Records every NewInputFile location routed to it.
+class RecordingFileIO : public FileIO {
+ public:
+ Result<std::unique_ptr<InputFile>> NewInputFile(std::string file_location) override {
+ locations.push_back(std::move(file_location));
+ return NotImplemented("recording mock");
+ }
+
+ std::vector<std::string> locations;
+};
+
+class RecordingCredentialedFileIO : public RecordingFileIO,
+ public SupportsStorageCredentials {
+ public:
+ Status SetStorageCredentials(
+ const std::vector<StorageCredential>& storage_credentials) override {
+ credentials_ = storage_credentials;
+ return {};
+ }
+
+ const std::vector<StorageCredential>& credentials() const override {
+ return credentials_;
+ }
+
+ SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; }
+
+ private:
+ std::vector<StorageCredential> 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<std::string, std::string> 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<std::string, std::string>& properties)
+ -> Result<std::unique_ptr<FileIO>> {
+ ++s3_factory_calls;
+ s3_factory_properties = properties;
+ auto io = std::make_unique<RecordingCredentialedFileIO>();
+ last_s3_io = io.get();
+ return io;
+ });
+ FileIORegistry::Register(
+ std::string(FileIORegistry::kArrowLocalFileIO),
+ [](const std::unordered_map<std::string, std::string>& /*properties*/)
+ -> Result<std::unique_ptr<FileIO>> {
+ ++local_factory_calls;
+ auto io = std::make_unique<RecordingFileIO>();
+ last_local_io = io.get();
+ return io;
+ });
+}
+
+} // 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(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<StorageCredential> 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);
+ EXPECT_EQ(s3_factory_calls, 1);
+
+ // Delegates are rebuilt with the new credentials, not mutated in place.
+ std::vector<StorageCredential> refreshed = {{.prefix = "s3", .config = {{"k3", "v3"}}}};
+ EXPECT_THAT(io.SetStorageCredentials(refreshed), IsOk());
+ (void)io.NewInputFile("s3://bucket/db/table/data/other.parquet");
+ ASSERT_NE(last_s3_io, nullptr);
+ EXPECT_EQ(last_s3_io->credentials(), refreshed);
+ EXPECT_EQ(s3_factory_calls, 2);
+
+ // 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_arrow_file_io_test.cc b/src/iceberg/test/rest_arrow_file_io_test.cc
new file mode 100644
index 0000000..67cb820
--- /dev/null
+++ b/src/iceberg/test/rest_arrow_file_io_test.cc
@@ -0,0 +1,97 @@
+/*
+ * 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 <algorithm>
+#include <memory>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <tuple>
+#include <vector>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#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<CapturingLogger>();
+ 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_file_io_test.cc b/src/iceberg/test/rest_file_io_test.cc
index 7f41b0b..6e584ea 100644
--- a/src/iceberg/test/rest_file_io_test.cc
+++ b/src/iceberg/test/rest_file_io_test.cc
@@ -69,73 +69,18 @@
} // 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<std::string, std::string>& /*properties*/)
- -> Result<std::unique_ptr<FileIO>> { return std::make_unique<MockFileIO>(); });
-
- 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<std::string, std::string>& /*properties*/)
- -> Result<std::unique_ptr<FileIO>> { return std::make_unique<MockFileIO>(); });
-
- 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<std::string, std::string>& /*properties*/)
- -> Result<std::unique_ptr<FileIO>> { return std::make_unique<MockFileIO>(); });
-
- 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 @@
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");
+ EXPECT_EQ(captured_storage_credentials, credentials);
}
TEST(RestFileIOTest, TableFileIOMergesConfigAndCredentials) {
