fix(io): attempt every S3 delete and report how many failed

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:

src/iceberg/arrow/s3/arrow_s3_file_io.cc

@@ -18,13 +18,16 @@
  */
 
 #include <algorithm>
+#include <atomic>
 #include <cstdlib>
+#include <future>
 #include <memory>
 #include <mutex>
 #include <optional>
 #include <shared_mutex>
 #include <string>
 #include <string_view>
+#include <thread>
 #include <unordered_map>
 #include <utility>
 #include <vector>
@@ -39,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"
@@ -181,9 +185,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<std::string, std::string> default_properties)
+                std::unordered_map<std::string, std::string> default_properties,
+                size_t delete_threads)
       : default_file_io_(std::make_shared<ArrowFileSystemFileIO>(std::move(arrow_fs))),
-        default_properties_(std::move(default_properties)) {}
+        default_properties_(std::move(default_properties)),
+        delete_threads_(delete_threads) {}
 
   Result<std::unique_ptr<InputFile>> NewInputFile(std::string file_location) override;
 
@@ -237,6 +243,7 @@ class ArrowS3FileIO final : public FileIO, public SupportsStorageCredentials {
 
   std::shared_ptr<ArrowFileSystemFileIO> default_file_io_;
   std::unordered_map<std::string, std::string> default_properties_;
+  size_t delete_threads_;
   // Guards everything below; shared because reads happen per file operation.
   mutable std::shared_mutex mutex_;
   std::vector<StorageCredential> storage_credentials_;
@@ -338,30 +345,46 @@ Status ArrowS3FileIO::DeleteFile(const std::string& file_location) {
 }
 
 Status ArrowS3FileIO::DeleteFiles(const std::vector<std::string>& 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: 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<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) {
-    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);
+  std::atomic<size_t> next = 0;
+  std::atomic<size_t> 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. Helpers log where the caller does.
+  auto logger = GetCurrentLogger();
+  std::vector<std::future<void>> 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.get();  // Rethrows what a helper threw.
   }
-  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.load(),
+                   file_locations.size());
   }
   return {};
 }
@@ -372,7 +395,16 @@ Result<std::unique_ptr<FileIO>> MakeS3FileIO(
     const std::unordered_map<std::string, std::string>& properties) {
   // Uses default credentials if properties are empty.
   ICEBERG_ASSIGN_OR_RAISE(auto fs, BuildArrowS3FileSystem(properties));
-  return std::make_unique<ArrowS3FileIO>(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<size_t>(*value));
+    if (delete_threads == 0) {
+      return InvalidArgument(R"("{}" must be positive)", S3Properties::kDeleteNumThreads);
+    }
+  }
+  return std::make_unique<ArrowS3FileIO>(std::move(fs), properties, delete_threads);
 }
 
 Status FinalizeS3() {

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.

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";
@@ -353,6 +362,56 @@ 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";
+  }
+  // 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());
+  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<std::string> 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());
+  // Each failure reaches the caller's logger, whichever thread hit it.
+  auto logger = std::make_shared<CapturingLogger>();
+  {
+    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.
+  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