From 1c0b5bb2158d71f2a07708e0dc2fec165715b27b Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Sat, 26 Sep 2026 02:22:57 -0400 Subject: [PATCH 1/5] 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 | 25 +++++++-------- src/iceberg/test/arrow_s3_file_io_test.cc | 39 +++++++++++++++++++++++ 2 files changed, 50 insertions(+), 14 deletions(-) diff --git a/src/iceberg/arrow/s3/arrow_s3_file_io.cc b/src/iceberg/arrow/s3/arrow_s3_file_io.cc index 4fbf33098..552f67de6 100644 --- a/src/iceberg/arrow/s3/arrow_s3_file_io.cc +++ b/src/iceberg/arrow/s3/arrow_s3_file_io.cc @@ -338,8 +338,9 @@ Status ArrowS3FileIO::DeleteFile(const std::string& file_location) { } Status ArrowS3FileIO::DeleteFiles(const std::vector& file_locations) { - // One snapshot so the whole batch matches the same delegate generation; only - // ever a handful of delegates, so a linear scan beats hashing. + // Like Java's S3FileIO, keep going after a failure and report the count. + // Arrow's S3 DeleteFiles deletes one file at a time too. One snapshot so the + // whole batch matches the same delegate generation. std::shared_ptr fallback; DelegatesByPrefix by_prefix; { @@ -347,21 +348,17 @@ Status ArrowS3FileIO::DeleteFiles(const std::vector& file_locations fallback = default_file_io_; by_prefix = file_io_by_prefix_; } - std::vector, std::vector>> - locations_by_io; + size_t failed = 0; for (const auto& file_location : file_locations) { - 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{file_location}); - } else { - it->second.push_back(file_location); + if (auto status = + MatchDelegate(fallback, by_prefix, 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 24628949a..d8c015061 100644 --- a/src/iceberg/test/arrow_s3_file_io_test.cc +++ b/src/iceberg/test/arrow_s3_file_io_test.cc @@ -353,6 +353,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 bef73edc86aa993a2f2016d87ac6195e4d207991 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Sat, 26 Sep 2026 06:45:25 -0400 Subject: [PATCH 2/5] 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 | 54 +++++++++++++++++------ src/iceberg/arrow/s3/s3_properties.h | 2 + src/iceberg/test/arrow_s3_file_io_test.cc | 11 +++++ 3 files changed, 54 insertions(+), 13 deletions(-) diff --git a/src/iceberg/arrow/s3/arrow_s3_file_io.cc b/src/iceberg/arrow/s3/arrow_s3_file_io.cc index 552f67de6..87011d7f0 100644 --- a/src/iceberg/arrow/s3/arrow_s3_file_io.cc +++ b/src/iceberg/arrow/s3/arrow_s3_file_io.cc @@ -18,6 +18,7 @@ */ #include +#include #include #include #include @@ -25,6 +26,7 @@ #include #include #include +#include #include #include #include @@ -181,9 +183,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::make_shared(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; @@ -237,6 +241,7 @@ class ArrowS3FileIO final : public FileIO, public SupportsStorageCredentials { std::shared_ptr default_file_io_; std::unordered_map default_properties_; + size_t delete_threads_; // Guards everything below; shared because reads happen per file operation. mutable std::shared_mutex mutex_; std::vector storage_credentials_; @@ -338,8 +343,8 @@ 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. One snapshot so the + // Like Java's S3FileIO: delete concurrently, keep going after a failure and + // report the count. Arrow has no batch delete for S3. One snapshot so the // whole batch matches the same delegate generation. std::shared_ptr fallback; DelegatesByPrefix by_prefix; @@ -348,17 +353,31 @@ Status ArrowS3FileIO::DeleteFiles(const std::vector& file_locations fallback = default_file_io_; by_prefix = file_io_by_prefix_; } - size_t failed = 0; - for (const auto& file_location : file_locations) { - if (auto status = - MatchDelegate(fallback, by_prefix, file_location)->DeleteFile(file_location); - !status.has_value()) { - ICEBERG_LOG_WARN("Failed to delete {}: {}", file_location, status.error().message); - ++failed; + 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 = MatchDelegate(fallback, by_prefix, 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 {}; } @@ -369,7 +388,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 d8c015061..46e181d06 100644 --- a/src/iceberg/test/arrow_s3_file_io_test.cc +++ b/src/iceberg/test/arrow_s3_file_io_test.cc @@ -299,6 +299,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"; @@ -362,6 +371,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 aee06c29cebe9e03488e35490f6e877bf538ff80 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Sat, 26 Sep 2026 08:20:31 -0400 Subject: [PATCH 3/5] 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 87011d7f0..ba44715d3 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 @@ -41,6 +42,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" @@ -367,13 +369,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 46e181d06..f75958c70 100644 --- a/src/iceberg/test/arrow_s3_file_io_test.cc +++ b/src/iceberg/test/arrow_s3_file_io_test.cc @@ -395,7 +395,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 492c6cb34751110ff95da57c6a19a71b91b24fe1 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Sat, 26 Sep 2026 08:28:47 -0400 Subject: [PATCH 4/5] 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 ba44715d3..00afb46d9 100644 --- a/src/iceberg/arrow/s3/arrow_s3_file_io.cc +++ b/src/iceberg/arrow/s3/arrow_s3_file_io.cc @@ -380,7 +380,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(), From dd0da21ec72e7e1ff40e49a7c0aa38b90fce7cf7 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Tue, 29 Sep 2026 02:13:24 -0400 Subject: [PATCH 5/5] docs(io): document s3.delete.num-threads --- mkdocs/docs/file-io.md | 1 + 1 file changed, 1 insertion(+) diff --git a/mkdocs/docs/file-io.md b/mkdocs/docs/file-io.md index 5b123addf..6a5842ca7 100644 --- a/mkdocs/docs/file-io.md +++ b/mkdocs/docs/file-io.md @@ -66,6 +66,7 @@ each file location's scheme. | `client.region` | `us-east-1` | Region to sign requests for | | `s3.endpoint` | `https://127.0.0.1:9000` | Endpoint to use instead of the AWS one. When absent, the `AWS_ENDPOINT_URL_S3` / `AWS_ENDPOINT_URL` environment variables are consulted | | `s3.path-style-access` | `true` | Address buckets as a path (`endpoint/bucket`) instead of a virtual host (`bucket.endpoint`). Only takes effect together with a custom endpoint | +| `s3.delete.num-threads` | `8` | Number of threads `DeleteFiles` deletes with. Defaults to the number of hardware threads | The following keys are specific to iceberg-cpp; they are not part of the Java Iceberg or REST specification property set: