From 1f5fb83a589c9959ab615d482780c190299f81e6 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Sat, 26 Sep 2026 02:22:57 -0400 Subject: [PATCH 1/4] fix(io): attempt every S3 delete and report how many failed ArrowS3FileIO::DeleteFiles grouped locations by credential prefix and returned at the first group that failed, so files in later groups were never attempted. Java's S3FileIO attempts every batch, logs each failed path and reports the failure count. Delete each file through its delegate, log each failure, and return "Failed to delete N of M files" at the end. This sends the same requests as before, since Arrow's S3 DeleteFiles also deletes one file at a time. ResolvingFileIO stays fail-fast across delegates, as it is in Java. --- src/iceberg/arrow/s3/arrow_s3_file_io.cc | 14 +++++--- src/iceberg/test/arrow_s3_file_io_test.cc | 39 +++++++++++++++++++++++ 2 files changed, 49 insertions(+), 4 deletions(-) diff --git a/src/iceberg/arrow/s3/arrow_s3_file_io.cc b/src/iceberg/arrow/s3/arrow_s3_file_io.cc index c981759ce..1beac9238 100644 --- a/src/iceberg/arrow/s3/arrow_s3_file_io.cc +++ b/src/iceberg/arrow/s3/arrow_s3_file_io.cc @@ -284,12 +284,18 @@ Status ArrowS3FileIO::DeleteFile(const std::string& file_location) { } Status ArrowS3FileIO::DeleteFiles(const std::vector& file_locations) { - std::unordered_map> locations_by_io; + // Like Java's S3FileIO, keep going after a failure and report the count. + // Arrow's S3 DeleteFiles deletes one file at a time too. + size_t failed = 0; for (const auto& file_location : file_locations) { - locations_by_io[&FileIOForPath(file_location)].push_back(file_location); + if (auto status = FileIOForPath(file_location).DeleteFile(file_location); + !status.has_value()) { + ICEBERG_LOG_WARN("Failed to delete {}: {}", file_location, status.error().message); + ++failed; + } } - for (auto& [file_io, locations] : locations_by_io) { - ICEBERG_RETURN_UNEXPECTED(file_io->DeleteFiles(locations)); + if (failed > 0) { + return IOError("Failed to delete {} of {} files", failed, file_locations.size()); } return {}; } diff --git a/src/iceberg/test/arrow_s3_file_io_test.cc b/src/iceberg/test/arrow_s3_file_io_test.cc index 7235ea367..8eb02206e 100644 --- a/src/iceberg/test/arrow_s3_file_io_test.cc +++ b/src/iceberg/test/arrow_s3_file_io_test.cc @@ -299,6 +299,45 @@ TEST_F(ArrowS3FileIOTest, LongestCredentialPrefix) { IsOk()); } +TEST_F(ArrowS3FileIOTest, DeleteFilesAttemptsEveryFile) { + if (!HasIntegrationEnv()) { + GTEST_SKIP() << "Set ICEBERG_TEST_S3_URI to enable S3 IO test"; + } + + auto properties = PropertiesFromEnv(); + if (properties.empty()) { + GTEST_SKIP() << "Set S3 properties to enable credential routing test"; + } + + auto io_res = MakeS3FileIO(properties); + ASSERT_THAT(io_res, IsOk()); + auto io = std::move(io_res).value(); + auto* credentialed = io->AsSupportsStorageCredentials(); + ASSERT_NE(credentialed, nullptr); + + const auto denied = ObjectUri("delete_denied/"); + const auto allowed = ObjectUri("delete_allowed/") + "only"; + const std::vector paths = {denied + "first", allowed, denied + "second"}; + for (const auto& path : paths) { + ASSERT_THAT(io->WriteFile(path, "payload"), IsOk()); + } + + // Deletes under `denied` fail; the one between them must still run. + auto bad_properties = properties; + for (const auto& [key, value] : BadS3Credentials()) { + bad_properties.insert_or_assign(key, value); + } + ASSERT_THAT(credentialed->SetStorageCredentials( + {{.prefix = denied, .config = std::move(bad_properties)}}), + IsOk()); + EXPECT_THAT(io->DeleteFiles(paths), HasErrorMessage("Failed to delete 2 of 3 files")); + EXPECT_FALSE(io->ReadFile(allowed, std::nullopt).has_value()); + + // The denied files remain: Arrow fails to delete a missing object. + ASSERT_THAT(credentialed->SetStorageCredentials({}), IsOk()); + EXPECT_THAT(io->DeleteFiles({paths[0], paths[2]}), IsOk()); +} + // The credential is vended under the oss spelling and the object addressed as // `s3://`, so they only meet through canonicalization — and every other path // to authentication is broken. (rest_arrow_file_io_test covers the mirrored From 0f6c2497f697dc18d8a48ad55fd8cb2e2e457aff Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Sat, 26 Sep 2026 06:45:25 -0400 Subject: [PATCH 2/4] perf(io): run S3 deletes on s3.delete.num-threads threads Java's S3FileIO runs its deletes on an executor sized by s3.delete.num-threads, which defaults to the number of processors. DeleteFiles now does the same, with the calling thread taking part. Java also packs keys into DeleteObjects batches. Arrow has no public batch delete for S3, so each thread still deletes one file at a time, as Arrow's own DeleteFiles does. --- src/iceberg/arrow/s3/arrow_s3_file_io.cc | 53 ++++++++++++++++++----- src/iceberg/arrow/s3/s3_properties.h | 2 + src/iceberg/test/arrow_s3_file_io_test.cc | 11 +++++ 3 files changed, 54 insertions(+), 12 deletions(-) diff --git a/src/iceberg/arrow/s3/arrow_s3_file_io.cc b/src/iceberg/arrow/s3/arrow_s3_file_io.cc index 1beac9238..7d1645121 100644 --- a/src/iceberg/arrow/s3/arrow_s3_file_io.cc +++ b/src/iceberg/arrow/s3/arrow_s3_file_io.cc @@ -17,11 +17,14 @@ * under the License. */ +#include +#include #include #include #include #include #include +#include #include #include #include @@ -178,9 +181,11 @@ std::string CanonicalizeS3Scheme(std::string_view location) { class ArrowS3FileIO final : public FileIO, public SupportsStorageCredentials { public: ArrowS3FileIO(std::shared_ptr<::arrow::fs::FileSystem> arrow_fs, - std::unordered_map default_properties) + std::unordered_map default_properties, + size_t delete_threads) : default_file_io_(std::move(arrow_fs)), - default_properties_(std::move(default_properties)) {} + default_properties_(std::move(default_properties)), + delete_threads_(delete_threads) {} Result> NewInputFile(std::string file_location) override; @@ -207,6 +212,7 @@ class ArrowS3FileIO final : public FileIO, public SupportsStorageCredentials { ArrowFileSystemFileIO default_file_io_; std::unordered_map default_properties_; + size_t delete_threads_; std::vector storage_credentials_; std::vector>> file_io_by_prefix_; @@ -284,18 +290,32 @@ Status ArrowS3FileIO::DeleteFile(const std::string& file_location) { } Status ArrowS3FileIO::DeleteFiles(const std::vector& file_locations) { - // Like Java's S3FileIO, keep going after a failure and report the count. - // Arrow's S3 DeleteFiles deletes one file at a time too. - size_t failed = 0; - for (const auto& file_location : file_locations) { - if (auto status = FileIOForPath(file_location).DeleteFile(file_location); - !status.has_value()) { - ICEBERG_LOG_WARN("Failed to delete {}: {}", file_location, status.error().message); - ++failed; + // Like Java's S3FileIO: delete concurrently, keep going after a failure and + // report the count. Arrow has no batch delete for S3. + std::atomic next = 0; + std::atomic failed = 0; + auto delete_remaining = [&] { + for (size_t i = next++; i < file_locations.size(); i = next++) { + const auto& file_location = file_locations[i]; + if (auto status = FileIOForPath(file_location).DeleteFile(file_location); + !status.has_value()) { + ICEBERG_LOG_WARN("Failed to delete {}: {}", file_location, + status.error().message); + ++failed; + } } + }; + { + // Plus the calling thread. + std::vector helpers; + for (size_t i = 1; i < std::min(delete_threads_, file_locations.size()); ++i) { + helpers.emplace_back(delete_remaining); + } + delete_remaining(); } if (failed > 0) { - return IOError("Failed to delete {} of {} files", failed, file_locations.size()); + return IOError("Failed to delete {} of {} files", failed.load(), + file_locations.size()); } return {}; } @@ -306,7 +326,16 @@ Result> MakeS3FileIO( const std::unordered_map& properties) { // Uses default credentials if properties are empty. ICEBERG_ASSIGN_OR_RAISE(auto fs, BuildArrowS3FileSystem(properties)); - return std::make_unique(std::move(fs), properties); + // Java defaults to one delete thread per processor. + size_t delete_threads = std::max(1u, std::thread::hardware_concurrency()); + if (const auto* value = FindProperty(properties, S3Properties::kDeleteNumThreads); + value != nullptr) { + ICEBERG_ASSIGN_OR_RAISE(delete_threads, StringUtils::ParseNumber(*value)); + if (delete_threads == 0) { + return InvalidArgument(R"("{}" must be positive)", S3Properties::kDeleteNumThreads); + } + } + return std::make_unique(std::move(fs), properties, delete_threads); } Status FinalizeS3() { diff --git a/src/iceberg/arrow/s3/s3_properties.h b/src/iceberg/arrow/s3/s3_properties.h index 50dafa56f..bcc186583 100644 --- a/src/iceberg/arrow/s3/s3_properties.h +++ b/src/iceberg/arrow/s3/s3_properties.h @@ -54,6 +54,8 @@ struct S3Properties { static constexpr std::string_view kConnectTimeoutMs = "s3.connect-timeout-ms"; /// Socket timeout in milliseconds static constexpr std::string_view kSocketTimeoutMs = "s3.socket-timeout-ms"; + /// Threads DeleteFiles uses; defaults to the hardware thread count, as in Java + static constexpr std::string_view kDeleteNumThreads = "s3.delete.num-threads"; }; /// \brief URI schemes served by the Arrow S3 FileIO, lower-case. diff --git a/src/iceberg/test/arrow_s3_file_io_test.cc b/src/iceberg/test/arrow_s3_file_io_test.cc index 8eb02206e..7606d1dcd 100644 --- a/src/iceberg/test/arrow_s3_file_io_test.cc +++ b/src/iceberg/test/arrow_s3_file_io_test.cc @@ -245,6 +245,15 @@ TEST_F(ArrowS3FileIOTest, RejectsIncompleteStaticCredentials) { "S3 client access key ID and secret access key must be set")); } +TEST_F(ArrowS3FileIOTest, RejectsInvalidDeleteThreads) { + for (std::string_view threads : {"0", "-1", "many"}) { + SCOPED_TRACE(threads); + EXPECT_THAT(MakeS3FileIO({{std::string(S3Properties::kDeleteNumThreads), + std::string(threads)}}), + IsError(ErrorKind::kInvalidArgument)); + } +} + TEST_F(ArrowS3FileIOTest, ReadWrite) { if (!HasIntegrationEnv()) { GTEST_SKIP() << "Set ICEBERG_TEST_S3_URI to enable S3 IO test"; @@ -308,6 +317,8 @@ TEST_F(ArrowS3FileIOTest, DeleteFilesAttemptsEveryFile) { if (properties.empty()) { GTEST_SKIP() << "Set S3 properties to enable credential routing test"; } + // A thread per file, so the deletes run concurrently. + properties[std::string(S3Properties::kDeleteNumThreads)] = "3"; auto io_res = MakeS3FileIO(properties); ASSERT_THAT(io_res, IsOk()); From 4341a22e2c5716987170a638d70ddc952370b241 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Sat, 26 Sep 2026 08:20:31 -0400 Subject: [PATCH 3/4] fix(io): run S3 delete helpers via std::async with the caller's logger std::jthread needs -fexperimental-library with libc++ 18 and 19, which the Clang 18+ requirement covers, so use std::async futures instead; they also wait for their thread if an exception unwinds. Helper threads now bind the caller's logger, as logger.h prescribes for thread pools. Without it their warnings went to the global logger while the returned error only carries the failure count. --- src/iceberg/arrow/s3/arrow_s3_file_io.cc | 21 ++++++++++++++------- src/iceberg/test/arrow_s3_file_io_test.cc | 11 ++++++++++- 2 files changed, 24 insertions(+), 8 deletions(-) diff --git a/src/iceberg/arrow/s3/arrow_s3_file_io.cc b/src/iceberg/arrow/s3/arrow_s3_file_io.cc index 7d1645121..217958104 100644 --- a/src/iceberg/arrow/s3/arrow_s3_file_io.cc +++ b/src/iceberg/arrow/s3/arrow_s3_file_io.cc @@ -20,6 +20,7 @@ #include #include #include +#include #include #include #include @@ -39,6 +40,7 @@ #include "iceberg/arrow/arrow_status_internal.h" #include "iceberg/arrow/s3/s3_properties.h" #include "iceberg/logging/log_macros.h" +#include "iceberg/logging/logger.h" #include "iceberg/util/macros.h" #include "iceberg/util/property_util.h" #include "iceberg/util/string_util.h" @@ -305,13 +307,18 @@ Status ArrowS3FileIO::DeleteFiles(const std::vector& file_locations } } }; - { - // Plus the calling thread. - std::vector helpers; - for (size_t i = 1; i < std::min(delete_threads_, file_locations.size()); ++i) { - helpers.emplace_back(delete_remaining); - } - delete_remaining(); + // Plus the calling thread. Helpers log where the caller does. + auto logger = GetCurrentLogger(); + std::vector> helpers; + for (size_t i = 1; i < std::min(delete_threads_, file_locations.size()); ++i) { + helpers.push_back(std::async(std::launch::async, [&] { + ScopedLogger bind(logger); + delete_remaining(); + })); + } + delete_remaining(); + for (auto& helper : helpers) { + helper.wait(); } if (failed > 0) { return IOError("Failed to delete {} of {} files", failed.load(), diff --git a/src/iceberg/test/arrow_s3_file_io_test.cc b/src/iceberg/test/arrow_s3_file_io_test.cc index 7606d1dcd..6c9c289bd 100644 --- a/src/iceberg/test/arrow_s3_file_io_test.cc +++ b/src/iceberg/test/arrow_s3_file_io_test.cc @@ -341,7 +341,16 @@ TEST_F(ArrowS3FileIOTest, DeleteFilesAttemptsEveryFile) { ASSERT_THAT(credentialed->SetStorageCredentials( {{.prefix = denied, .config = std::move(bad_properties)}}), IsOk()); - EXPECT_THAT(io->DeleteFiles(paths), HasErrorMessage("Failed to delete 2 of 3 files")); + // Each failure reaches the caller's logger, whichever thread hit it. + auto logger = std::make_shared(); + { + ScopedLogger bind(logger); + EXPECT_THAT(io->DeleteFiles(paths), HasErrorMessage("Failed to delete 2 of 3 files")); + } + EXPECT_EQ(std::ranges::count_if( + logger->records(), + [](const LogMessage& record) { return record.level == LogLevel::kWarn; }), + 2); EXPECT_FALSE(io->ReadFile(allowed, std::nullopt).has_value()); // The denied files remain: Arrow fails to delete a missing object. From 41ff77ebe390a9c6b125f3525ac2862a29ca11fb Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Sat, 26 Sep 2026 08:28:47 -0400 Subject: [PATCH 4/4] fix(io): rethrow exceptions from S3 delete helpers wait() leaves an exception stored in a helper's future, so a delete that threw was neither counted nor reported, and DeleteFiles could return success. get() rethrows it on the calling thread, as the sequential loop did. --- src/iceberg/arrow/s3/arrow_s3_file_io.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/iceberg/arrow/s3/arrow_s3_file_io.cc b/src/iceberg/arrow/s3/arrow_s3_file_io.cc index 217958104..77f8f4934 100644 --- a/src/iceberg/arrow/s3/arrow_s3_file_io.cc +++ b/src/iceberg/arrow/s3/arrow_s3_file_io.cc @@ -318,7 +318,7 @@ Status ArrowS3FileIO::DeleteFiles(const std::vector& file_locations } delete_remaining(); for (auto& helper : helpers) { - helper.wait(); + helper.get(); // Rethrows what a helper threw. } if (failed > 0) { return IOError("Failed to delete {} of {} files", failed.load(),