This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 3c5715c6 refactor(io): hand out S3 delegates by shared_ptr and build
them off the lock (#898)
3c5715c6 is described below
commit 3c5715c66fb4d4306c54eec4aac327e2897f800b
Author: Jiajia Li <[email protected]>
AuthorDate: Tue Sep 29 11:19:32 2026 +0800
refactor(io): hand out S3 delegates by shared_ptr and build them off the
lock (#898)
ResolvingFileIO::FileIOForPath held its mutex across FileIORegistry::Load
and
the delegate's SetStorageCredentials, so building the first S3 client (which
can look up a bucket region) stalled every other operation on the resolver.
It now loads without the lock and caches the result only if no credential
install happened meanwhile; otherwise it loads again with the new
credentials. Delegates retired by an install are torn down outside the lock.
Tests now exercise real IO where the old ones stopped short: the
multi-prefix
DeleteFiles test failed URI parsing before reaching any delegate, and the
concurrency test only built input-file wrappers. The replacements write,
delete and read through the test object store, and the stress test waits
for every worker to run before installing.
---
src/iceberg/arrow/s3/arrow_s3_file_io.cc | 132 ++++++++++++++++++++-------
src/iceberg/file_io.h | 7 +-
src/iceberg/resolving_file_io.cc | 68 +++++++++-----
src/iceberg/resolving_file_io.h | 13 ++-
src/iceberg/test/arrow_s3_file_io_test.cc | 138 +++++++++++++++++++++++++++++
src/iceberg/test/resolving_file_io_test.cc | 80 +++++++++++++++--
src/iceberg/test/rest_file_io_test.cc | 2 +-
7 files changed, 373 insertions(+), 67 deletions(-)
diff --git a/src/iceberg/arrow/s3/arrow_s3_file_io.cc
b/src/iceberg/arrow/s3/arrow_s3_file_io.cc
index c981759c..4fbf3309 100644
--- a/src/iceberg/arrow/s3/arrow_s3_file_io.cc
+++ b/src/iceberg/arrow/s3/arrow_s3_file_io.cc
@@ -17,9 +17,12 @@
* under the License.
*/
+#include <algorithm>
#include <cstdlib>
#include <memory>
+#include <mutex>
#include <optional>
+#include <shared_mutex>
#include <string>
#include <string_view>
#include <unordered_map>
@@ -179,7 +182,7 @@ class ArrowS3FileIO final : public FileIO, public
SupportsStorageCredentials {
public:
ArrowS3FileIO(std::shared_ptr<::arrow::fs::FileSystem> arrow_fs,
std::unordered_map<std::string, std::string>
default_properties)
- : default_file_io_(std::move(arrow_fs)),
+ :
default_file_io_(std::make_shared<ArrowFileSystemFileIO>(std::move(arrow_fs))),
default_properties_(std::move(default_properties)) {}
Result<std::unique_ptr<InputFile>> NewInputFile(std::string file_location)
override;
@@ -196,27 +199,67 @@ class ArrowS3FileIO final : public FileIO, public
SupportsStorageCredentials {
Status SetStorageCredentials(
const std::vector<StorageCredential>& storage_credentials) override;
- const std::vector<StorageCredential>& credentials() const override {
+ std::vector<StorageCredential> credentials() const override {
+ std::shared_lock lock(mutex_);
return storage_credentials_;
}
SupportsStorageCredentials* AsSupportsStorageCredentials() override { return
this; }
private:
- ArrowFileSystemFileIO& FileIOForPath(std::string_view location);
-
- ArrowFileSystemFileIO default_file_io_;
+ /// \brief Delegate serving `location`, pinned by the caller against a
+ /// concurrent credential install.
+ std::shared_ptr<ArrowFileSystemFileIO> FileIOForPath(std::string_view
location);
+
+ using DelegatesByPrefix =
+ std::vector<std::pair<std::string,
std::shared_ptr<ArrowFileSystemFileIO>>>;
+
+ /// \brief Longest-prefix match against one consistent view of the delegates.
+ static std::shared_ptr<ArrowFileSystemFileIO> MatchDelegate(
+ const std::shared_ptr<ArrowFileSystemFileIO>& fallback,
+ const DelegatesByPrefix& by_prefix, std::string_view location);
+
+ /// \brief Build a delegate for each credential this FileIO can serve.
+ ///
+ /// Runs without holding `mutex_`: building an S3 client can block (without
+ /// static keys the AWS SDK may wait on the EC2 metadata service), which
would
+ /// stall every concurrent operation. Reads no mutable member state.
+ Result<DelegatesByPrefix> BuildDelegates(
+ const std::vector<StorageCredential>& storage_credentials) const;
+
+ /// \brief Swap in credentials and delegates, handing back the retired ones.
+ ///
+ /// Callers must hold `mutex_` exclusively and let the returned generation
+ /// destruct only after releasing it: tearing down an S3 client can block on
+ /// in-flight requests, which would stall every operation.
+ void InstallCredentials(std::vector<StorageCredential>& storage_credentials,
+ DelegatesByPrefix& delegates);
+
+ std::shared_ptr<ArrowFileSystemFileIO> default_file_io_;
std::unordered_map<std::string, std::string> default_properties_;
+ // Guards everything below; shared because reads happen per file operation.
+ mutable std::shared_mutex mutex_;
std::vector<StorageCredential> storage_credentials_;
- std::vector<std::pair<std::string, std::unique_ptr<ArrowFileSystemFileIO>>>
- file_io_by_prefix_;
+ DelegatesByPrefix file_io_by_prefix_;
};
Status ArrowS3FileIO::SetStorageCredentials(
const std::vector<StorageCredential>& storage_credentials) {
- std::vector<std::pair<std::string, std::unique_ptr<ArrowFileSystemFileIO>>>
- file_io_by_prefix;
- file_io_by_prefix.reserve(storage_credentials.size());
+ ICEBERG_ASSIGN_OR_RAISE(auto delegates, BuildDelegates(storage_credentials));
+ auto credentials = storage_credentials;
+ {
+ std::unique_lock lock(mutex_);
+ InstallCredentials(credentials, delegates);
+ }
+ // `credentials` and `delegates` now hold the retired generation and destruct
+ // here, outside the lock.
+ return {};
+}
+
+Result<ArrowS3FileIO::DelegatesByPrefix> ArrowS3FileIO::BuildDelegates(
+ const std::vector<StorageCredential>& storage_credentials) const {
+ DelegatesByPrefix delegates;
+ delegates.reserve(storage_credentials.size());
// TODO(gangwu): Refresh vended credentials via credentials.uri before
tokens expire.
for (const auto& credential : storage_credentials) {
ICEBERG_RETURN_UNEXPECTED(credential.Validate());
@@ -231,11 +274,10 @@ Status ArrowS3FileIO::SetStorageCredentials(
properties[key] = value;
}
ICEBERG_ASSIGN_OR_RAISE(auto fs, BuildArrowS3FileSystem(properties));
- file_io_by_prefix.emplace_back(
- CanonicalizeS3Scheme(credential.prefix),
- std::make_unique<ArrowFileSystemFileIO>(std::move(fs)));
+ delegates.emplace_back(CanonicalizeS3Scheme(credential.prefix),
+
std::make_shared<ArrowFileSystemFileIO>(std::move(fs)));
}
- if (file_io_by_prefix.empty() && !storage_credentials.empty()) {
+ if (delegates.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(
@@ -243,50 +285,80 @@ Status ArrowS3FileIO::SetStorageCredentials(
"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 {};
+ return delegates;
}
-ArrowFileSystemFileIO& ArrowS3FileIO::FileIOForPath(std::string_view location)
{
- if (file_io_by_prefix_.empty()) {
- return default_file_io_;
+void ArrowS3FileIO::InstallCredentials(
+ std::vector<StorageCredential>& storage_credentials, DelegatesByPrefix&
delegates) {
+ file_io_by_prefix_.swap(delegates);
+ storage_credentials_.swap(storage_credentials);
+}
+
+std::shared_ptr<ArrowFileSystemFileIO> ArrowS3FileIO::MatchDelegate(
+ const std::shared_ptr<ArrowFileSystemFileIO>& fallback,
+ const DelegatesByPrefix& by_prefix, std::string_view location) {
+ if (by_prefix.empty()) {
+ return fallback;
}
const std::string canonical = CanonicalizeS3Scheme(location);
- ArrowFileSystemFileIO* best = &default_file_io_;
+ auto best = fallback;
size_t best_len = 0;
- for (const auto& [prefix, file_io] : file_io_by_prefix_) {
+ for (const auto& [prefix, file_io] : by_prefix) {
if (prefix.size() > best_len && canonical.starts_with(prefix)) {
- best = file_io.get();
+ best = file_io;
best_len = prefix.size();
}
}
- return *best;
+ return best;
+}
+
+std::shared_ptr<ArrowFileSystemFileIO> ArrowS3FileIO::FileIOForPath(
+ std::string_view location) {
+ std::shared_lock lock(mutex_);
+ return MatchDelegate(default_file_io_, file_io_by_prefix_, location);
}
Result<std::unique_ptr<InputFile>> ArrowS3FileIO::NewInputFile(
std::string file_location) {
- return FileIOForPath(file_location).NewInputFile(std::move(file_location));
+ return FileIOForPath(file_location)->NewInputFile(std::move(file_location));
}
Result<std::unique_ptr<InputFile>> ArrowS3FileIO::NewInputFile(std::string
file_location,
size_t length) {
- return FileIOForPath(file_location).NewInputFile(std::move(file_location),
length);
+ return FileIOForPath(file_location)->NewInputFile(std::move(file_location),
length);
}
Result<std::unique_ptr<OutputFile>> ArrowS3FileIO::NewOutputFile(
std::string file_location) {
- return FileIOForPath(file_location).NewOutputFile(std::move(file_location));
+ return FileIOForPath(file_location)->NewOutputFile(std::move(file_location));
}
Status ArrowS3FileIO::DeleteFile(const std::string& file_location) {
- return FileIOForPath(file_location).DeleteFile(file_location);
+ return FileIOForPath(file_location)->DeleteFile(file_location);
}
Status ArrowS3FileIO::DeleteFiles(const std::vector<std::string>&
file_locations) {
- std::unordered_map<ArrowFileSystemFileIO*, std::vector<std::string>>
locations_by_io;
+ // One snapshot so the whole batch matches the same delegate generation; only
+ // ever a handful of delegates, so a linear scan beats hashing.
+ std::shared_ptr<ArrowFileSystemFileIO> fallback;
+ DelegatesByPrefix by_prefix;
+ {
+ std::shared_lock lock(mutex_);
+ fallback = default_file_io_;
+ by_prefix = file_io_by_prefix_;
+ }
+ std::vector<std::pair<std::shared_ptr<ArrowFileSystemFileIO>,
std::vector<std::string>>>
+ locations_by_io;
for (const auto& file_location : file_locations) {
- locations_by_io[&FileIOForPath(file_location)].push_back(file_location);
+ auto file_io = MatchDelegate(fallback, by_prefix, file_location);
+ auto it = std::ranges::find_if(
+ locations_by_io, [&](const auto& entry) { return entry.first ==
file_io; });
+ if (it == locations_by_io.end()) {
+ locations_by_io.emplace_back(std::move(file_io),
+ std::vector<std::string>{file_location});
+ } else {
+ it->second.push_back(file_location);
+ }
}
for (auto& [file_io, locations] : locations_by_io) {
ICEBERG_RETURN_UNEXPECTED(file_io->DeleteFiles(locations));
diff --git a/src/iceberg/file_io.h b/src/iceberg/file_io.h
index 3ea4afa4..e22e5cf2 100644
--- a/src/iceberg/file_io.h
+++ b/src/iceberg/file_io.h
@@ -193,8 +193,11 @@ class ICEBERG_EXPORT SupportsStorageCredentials {
virtual Status SetStorageCredentials(
const std::vector<StorageCredential>& storage_credentials) = 0;
- /// \brief Return currently installed storage credentials.
- virtual const std::vector<StorageCredential>& credentials() const = 0;
+ /// \brief Return the storage credentials this FileIO holds.
+ ///
+ /// By value because a concurrent install may replace them. An implementation
+ /// that delegates may report what was installed on it.
+ virtual std::vector<StorageCredential> credentials() const = 0;
};
} // namespace iceberg
diff --git a/src/iceberg/resolving_file_io.cc b/src/iceberg/resolving_file_io.cc
index 8a4e8113..91be67ea 100644
--- a/src/iceberg/resolving_file_io.cc
+++ b/src/iceberg/resolving_file_io.cc
@@ -40,28 +40,46 @@ Result<std::shared_ptr<FileIO>>
ResolvingFileIO::FileIOForPath(
const auto scheme =
StringUtils::ToLower(LocationUtil::ParseScheme(location));
ICEBERG_ASSIGN_OR_RAISE(const auto name, FileIORegistry::Resolve(scheme));
- {
- std::shared_lock lock(mutex_);
- if (const auto cached = io_by_name_.find(name); cached !=
io_by_name_.end()) {
- return cached->second;
- }
- }
-
- 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(name, properties_));
- // Forward all credentials; each implementation applies the prefixes it
- // understands.
- if (!storage_credentials_.empty()) {
+ // Loads without holding `mutex_`: building a client can block (an S3 client
+ // without static keys may wait on the EC2 metadata service), which would
+ // stall every other operation. Forwards all credentials; each implementation
+ // applies the prefixes it understands.
+ auto load = [&](const std::vector<StorageCredential>& credentials)
+ -> Result<std::shared_ptr<FileIO>> {
+ ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<FileIO> io,
+ FileIORegistry::Load(name, properties_));
+ if (!credentials.empty()) {
if (auto* credentialed = io->AsSupportsStorageCredentials()) {
- ICEBERG_RETURN_UNEXPECTED(
- credentialed->SetStorageCredentials(storage_credentials_));
+
ICEBERG_RETURN_UNEXPECTED(credentialed->SetStorageCredentials(credentials));
+ }
+ }
+ return io;
+ };
+
+ while (true) {
+ uint64_t generation = 0;
+ std::vector<StorageCredential> credentials;
+ {
+ std::shared_lock lock(mutex_);
+ if (const auto cached = io_by_name_.find(name); cached !=
io_by_name_.end()) {
+ return cached->second;
}
+ generation = credential_generation_;
+ credentials = storage_credentials_;
+ }
+ // Declared before the lock, so a delegate that is not cached is torn down
+ // only after the lock is released.
+ auto loaded = load(credentials);
+ std::unique_lock lock(mutex_);
+ if (generation != credential_generation_) {
+ continue; // Credentials were replaced mid-load; load again with them.
}
- it = io_by_name_.emplace(std::string(name), std::move(io)).first;
+ if (const auto cached = io_by_name_.find(name); cached !=
io_by_name_.end()) {
+ return cached->second;
+ }
+ ICEBERG_RETURN_UNEXPECTED(loaded);
+ return io_by_name_.try_emplace(name, *loaded).first->second;
}
- return it->second;
}
Result<std::unique_ptr<InputFile>> ResolvingFileIO::NewInputFile(
@@ -103,13 +121,19 @@ 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::unique_lock lock(mutex_);
- storage_credentials_ = storage_credentials;
- io_by_name_.clear();
+ // Retired outside the lock: tearing down a delegate can block.
+ decltype(io_by_name_) retired;
+ {
+ std::unique_lock lock(mutex_);
+ storage_credentials_ = storage_credentials;
+ ++credential_generation_;
+ retired.swap(io_by_name_);
+ }
return {};
}
-const std::vector<StorageCredential>& ResolvingFileIO::credentials() const {
+std::vector<StorageCredential> ResolvingFileIO::credentials() const {
+ std::shared_lock lock(mutex_);
return storage_credentials_;
}
diff --git a/src/iceberg/resolving_file_io.h b/src/iceberg/resolving_file_io.h
index 837afd60..9a5cf047 100644
--- a/src/iceberg/resolving_file_io.h
+++ b/src/iceberg/resolving_file_io.h
@@ -22,6 +22,7 @@
/// \file iceberg/resolving_file_io.h
/// \brief FileIO that resolves the concrete implementation per file-path
scheme.
+#include <cstdint>
#include <memory>
#include <shared_mutex>
#include <string>
@@ -38,6 +39,9 @@
namespace iceberg {
/// \brief FileIO that resolves and caches implementations by registry name.
+///
+/// Vended credentials are forwarded to every resolved implementation that
+/// supports them; each applies what it understands.
class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
public SupportsStorageCredentials
{
public:
@@ -58,7 +62,7 @@ class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
Status SetStorageCredentials(
const std::vector<StorageCredential>& storage_credentials) override;
- const std::vector<StorageCredential>& credentials() const override;
+ std::vector<StorageCredential> credentials() const override;
SupportsStorageCredentials* AsSupportsStorageCredentials() override { return
this; }
@@ -67,9 +71,12 @@ class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
Result<std::shared_ptr<FileIO>> FileIOForPath(std::string_view location);
std::unordered_map<std::string, std::string> properties_;
- // Guards lazy resolution and credential refresh.
- std::shared_mutex mutex_;
+ // Guards lazy resolution and credential state.
+ mutable std::shared_mutex mutex_;
std::vector<StorageCredential> storage_credentials_;
+ // Bumped by every credential install, so a delegate loaded from an older set
+ // never reaches the cache.
+ uint64_t credential_generation_ = 0;
std::unordered_map<std::string, std::shared_ptr<FileIO>, StringHash,
StringEqual>
io_by_name_;
};
diff --git a/src/iceberg/test/arrow_s3_file_io_test.cc
b/src/iceberg/test/arrow_s3_file_io_test.cc
index 7235ea36..24628949 100644
--- a/src/iceberg/test/arrow_s3_file_io_test.cc
+++ b/src/iceberg/test/arrow_s3_file_io_test.cc
@@ -18,12 +18,15 @@
*/
#include <algorithm>
+#include <atomic>
#include <cstdlib>
#include <iostream>
#include <memory>
#include <optional>
+#include <span>
#include <string>
#include <string_view>
+#include <thread>
#include <unordered_map>
#include <utility>
#include <vector>
@@ -237,6 +240,57 @@ TEST_F(ArrowS3FileIOTest, WarnsWhenNoCredentialApplies) {
EXPECT_TRUE(HasWarning(*logger));
}
+TEST_F(ArrowS3FileIOTest, OperationsSurviveConcurrentCredentialInstalls) {
+ auto result = MakeS3FileIO({});
+ ASSERT_THAT(result, IsOk());
+ auto* credentialed = result.value()->AsSupportsStorageCredentials();
+ ASSERT_NE(credentialed, nullptr);
+
+ auto credential = [](std::string_view access_key) {
+ return StorageCredential{
+ .prefix = "s3://bucket",
+ .config = {{std::string(S3Properties::kAccessKeyId),
std::string(access_key)},
+ {std::string(S3Properties::kSecretAccessKey), "secret"}}};
+ };
+ ASSERT_THAT(credentialed->SetStorageCredentials({credential("first")}),
IsOk());
+
+ std::atomic<bool> stop = false;
+ std::atomic<int> started = 0;
+ std::atomic<int> failures = 0;
+ std::vector<std::thread> operations;
+ operations.reserve(4);
+ for (int i = 0; i < 4; ++i) {
+ operations.emplace_back([&] {
+ auto open = [&] {
+ if (!result.value()->NewInputFile("s3://bucket/key").has_value()) {
+ ++failures;
+ }
+ };
+ open();
+ ++started;
+ while (!stop.load()) {
+ open();
+ }
+ });
+ }
+ // Every worker has run and is still looping before the first install.
+ while (started.load() < 4) {
+ std::this_thread::yield();
+ }
+ // No assertions until the threads are joined: a fatal assertion here would
+ // destroy joinable threads and terminate the binary, masking the failure.
+ Status install_status = {};
+ for (int round = 0; round < 3 && install_status.has_value(); ++round) {
+ install_status =
credentialed->SetStorageCredentials({credential("replacement")});
+ }
+ stop = true;
+ for (auto& operation : operations) {
+ operation.join();
+ }
+ ASSERT_THAT(install_status, IsOk());
+ EXPECT_EQ(failures, 0);
+}
+
TEST_F(ArrowS3FileIOTest, RejectsIncompleteStaticCredentials) {
auto result =
MakeS3FileIO({{std::string(S3Properties::kAccessKeyId),
"access-key-only"}});
@@ -341,6 +395,90 @@ TEST_F(ArrowS3FileIOTest,
AppliesOssCredentialInRealRoundTrip) {
EXPECT_THAT(CheckReadWrite(*io, s3_uri, "hello oss with vended
credentials"), IsOk());
}
+TEST_F(ArrowS3FileIOTest, DeleteFilesReachesEveryCredentialPrefix) {
+ if (!HasIntegrationEnv()) {
+ GTEST_SKIP() << "Set ICEBERG_TEST_S3_URI to enable S3 IO test";
+ }
+
+ auto properties = PropertiesFromEnv();
+ if (!properties.contains(std::string(S3Properties::kAccessKeyId)) ||
+ !properties.contains(std::string(S3Properties::kSecretAccessKey))) {
+ GTEST_SKIP() << "Set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY to enable
"
+ "credential routing test";
+ }
+
+ // Only the prefix delegates can authenticate, so a location that falls to
+ // the default one fails the batch instead of passing unnoticed.
+ auto bad_defaults = properties;
+ for (const auto& [key, value] : BadS3Credentials()) {
+ bad_defaults.insert_or_assign(key, value);
+ }
+ ICEBERG_UNWRAP_OR_FAIL(auto io, MakeS3FileIO(std::move(bad_defaults)));
+ auto* credentialed = io->AsSupportsStorageCredentials();
+ ASSERT_NE(credentialed, nullptr);
+
+ const auto a = ObjectUri("delete_files_a/");
+ const auto b = ObjectUri("delete_files_b/");
+ ASSERT_THAT(credentialed->SetStorageCredentials({{.prefix = a, .config =
properties},
+ {.prefix = b, .config =
properties}}),
+ IsOk());
+
+ const std::vector<std::string> paths = {a + "first", b + "only", a +
"second"};
+ for (const auto& path : paths) {
+ ASSERT_THAT(io->WriteFile(path, "payload"), IsOk());
+ }
+ ASSERT_THAT(io->DeleteFiles(paths), IsOk());
+ // The writes above authenticated on these paths, so a failed read means the
+ // object is gone.
+ for (const auto& path : paths) {
+ EXPECT_FALSE(io->ReadFile(path, std::nullopt).has_value()) << path;
+ }
+}
+
+TEST_F(ArrowS3FileIOTest, InputFileOutlivesCredentialInstall) {
+ if (!HasIntegrationEnv()) {
+ GTEST_SKIP() << "Set ICEBERG_TEST_S3_URI to enable S3 IO test";
+ }
+
+ auto properties = PropertiesFromEnv();
+ if (!properties.contains(std::string(S3Properties::kAccessKeyId)) ||
+ !properties.contains(std::string(S3Properties::kSecretAccessKey))) {
+ GTEST_SKIP() << "Set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY to enable
"
+ "credential routing test";
+ }
+
+ auto bad_defaults = properties;
+ for (const auto& [key, value] : BadS3Credentials()) {
+ bad_defaults.insert_or_assign(key, value);
+ }
+ ICEBERG_UNWRAP_OR_FAIL(auto io, MakeS3FileIO(std::move(bad_defaults)));
+ auto* credentialed = io->AsSupportsStorageCredentials();
+ ASSERT_NE(credentialed, nullptr);
+
+ const auto prefix = ObjectUri("retained_input/");
+ const std::vector<StorageCredential> credentials = {
+ {.prefix = prefix, .config = properties}};
+ ASSERT_THAT(credentialed->SetStorageCredentials(credentials), IsOk());
+
+ const auto uri = prefix + "object";
+ constexpr std::string_view kContent = "written before the credential
install";
+ ASSERT_THAT(io->WriteFile(uri, kContent), IsOk());
+ ICEBERG_UNWRAP_OR_FAIL(auto file, io->NewInputFile(uri));
+ ICEBERG_UNWRAP_OR_FAIL(auto opened_before, file->Open());
+
+ // Retires the delegate `file` came from; both handles must still read.
+ ASSERT_THAT(credentialed->SetStorageCredentials(credentials), IsOk());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto opened_after, file->Open());
+ for (auto* stream : {opened_before.get(), opened_after.get()}) {
+ std::string read(kContent.size(), '\0');
+ EXPECT_THAT(stream->ReadFully(0, std::as_writable_bytes(std::span(read))),
IsOk());
+ EXPECT_EQ(read, kContent);
+ EXPECT_THAT(stream->Close(), IsOk());
+ }
+ EXPECT_THAT(io->DeleteFile(uri), IsOk());
+}
+
#if ICEBERG_S3_ENABLED
TEST_F(ArrowS3FileIOTest, ClientRegion) {
auto result =
diff --git a/src/iceberg/test/resolving_file_io_test.cc
b/src/iceberg/test/resolving_file_io_test.cc
index 788b59be..2189b64a 100644
--- a/src/iceberg/test/resolving_file_io_test.cc
+++ b/src/iceberg/test/resolving_file_io_test.cc
@@ -19,8 +19,13 @@
#include "iceberg/resolving_file_io.h"
+#include <chrono>
+#include <condition_variable>
+#include <future>
#include <memory>
+#include <mutex>
#include <string>
+#include <thread>
#include <unordered_map>
#include <vector>
@@ -60,9 +65,7 @@ class RecordingCredentialedFileIO : public RecordingFileIO,
return {};
}
- const std::vector<StorageCredential>& credentials() const override {
- return credentials_;
- }
+ std::vector<StorageCredential> credentials() const override { return
credentials_; }
SupportsStorageCredentials* AsSupportsStorageCredentials() override { return
this; }
@@ -153,9 +156,9 @@ TEST(ResolvingFileIOTest,
RoutesPathsAndCachesResolvedImplementations) {
ResolvingFileIO io({{"k", "v"}});
// Errors come from the recording mock; routing is what is under test.
- (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");
+ std::ignore = io.NewInputFile("s3a://bucket/db/table/data/file.parquet");
+ std::ignore = io.NewInputFile("s3://bucket/db/table/data/file.parquet");
+ std::ignore = io.NewInputFile("/tmp/local/file.parquet");
ASSERT_NE(last_s3_io, nullptr);
ASSERT_NE(last_local_io, nullptr);
@@ -238,7 +241,7 @@ TEST(ResolvingFileIOTest,
ForwardsAllCredentialsToResolvedImplementations) {
EXPECT_THAT(io.SetStorageCredentials(credentials), IsOk());
EXPECT_EQ(io.credentials(), credentials);
- (void)io.NewInputFile("s3://bucket/db/table/data/file.parquet");
+ std::ignore = 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);
@@ -246,14 +249,73 @@ TEST(ResolvingFileIOTest,
ForwardsAllCredentialsToResolvedImplementations) {
// 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");
+ std::ignore = 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");
+ std::ignore = io.NewInputFile("/tmp/local/file.parquet");
ASSERT_NE(last_local_io, nullptr);
}
+TEST(ResolvingFileIOTest, LoadsWithoutTheLockAndDropsStaleDelegates) {
+ // The first load parks until new credentials are installed. Loading outside
+ // the lock lets that install proceed; what the parked load built then
carries
+ // stale credentials and must be redone rather than cached.
+ // Static because registry factories are process-global; reset for reruns.
+ static std::mutex gate;
+ static std::condition_variable cv;
+ static bool parked;
+ static bool resume;
+ static int calls;
+ static RecordingCredentialedFileIO* last;
+ parked = resume = false;
+ calls = 0;
+ last = nullptr;
+ FileIORegistry::Register(
+ "test.file-io.slow",
+ {.create =
+ [](const FileIORegistry::Properties&) ->
Result<std::unique_ptr<FileIO>> {
+ if (++calls == 1) {
+ std::unique_lock lock(gate);
+ parked = true;
+ cv.notify_all();
+ cv.wait(lock, [] { return resume; });
+ }
+ auto io = std::make_unique<RecordingCredentialedFileIO>();
+ last = io.get();
+ return io;
+ },
+ .accepts = [](std::string_view scheme) { return scheme == "slow"; }});
+
+ ResolvingFileIO io({});
+ const std::vector<StorageCredential> stale = {
+ {.prefix = "slow", .config = {{"k", "1"}}}};
+ const std::vector<StorageCredential> fresh = {
+ {.prefix = "slow", .config = {{"k", "2"}}}};
+ ASSERT_THAT(io.SetStorageCredentials(stale), IsOk());
+
+ std::thread reader([&] { std::ignore =
io.NewInputFile("slow://bucket/file"); });
+ {
+ std::unique_lock lock(gate);
+ cv.wait(lock, [] { return parked; });
+ }
+ auto install =
+ std::async(std::launch::async, [&] { return
io.SetStorageCredentials(fresh); });
+ const bool install_waited =
+ install.wait_for(std::chrono::seconds(10)) != std::future_status::ready;
+ {
+ std::lock_guard lock(gate);
+ resume = true;
+ }
+ cv.notify_all();
+ reader.join();
+
+ EXPECT_FALSE(install_waited) << "a credential install waited on an in-flight
load";
+ EXPECT_THAT(install.get(), IsOk());
+ ASSERT_EQ(calls, 2);
+ EXPECT_EQ(last->credentials(), fresh);
+}
+
} // namespace iceberg
diff --git a/src/iceberg/test/rest_file_io_test.cc
b/src/iceberg/test/rest_file_io_test.cc
index dde3238a..2fce9dac 100644
--- a/src/iceberg/test/rest_file_io_test.cc
+++ b/src/iceberg/test/rest_file_io_test.cc
@@ -61,7 +61,7 @@ class MockCredentialedFileIO : public MockFileIO, public
SupportsStorageCredenti
return {};
}
- const std::vector<StorageCredential>& credentials() const override {
+ std::vector<StorageCredential> credentials() const override {
return captured_storage_credentials;
}
