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

src/iceberg/arrow/s3/arrow_s3_file_io.cc

@@ -17,11 +17,15 @@
  * under the License.
  */
 
+#include <algorithm>
+#include <atomic>
 #include <cstdlib>
+#include <future>
 #include <memory>
 #include <optional>
 #include <string>
 #include <string_view>
+#include <thread>
 #include <unordered_map>
 #include <utility>
 #include <vector>
@@ -36,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"
@@ -178,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<std::string, std::string> default_properties)
+                std::unordered_map<std::string, std::string> 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<std::unique_ptr<InputFile>> NewInputFile(std::string file_location) override;
 
@@ -207,6 +214,7 @@ class ArrowS3FileIO final : public FileIO, public SupportsStorageCredentials {
 
   ArrowFileSystemFileIO default_file_io_;
   std::unordered_map<std::string, std::string> default_properties_;
+  size_t delete_threads_;
   std::vector<StorageCredential> storage_credentials_;
   std::vector<std::pair<std::string, std::unique_ptr<ArrowFileSystemFileIO>>>
       file_io_by_prefix_;
@@ -284,12 +292,37 @@ Status ArrowS3FileIO::DeleteFile(const std::string& file_location) {
 }
 
 Status ArrowS3FileIO::DeleteFiles(const std::vector<std::string>& file_locations) {
-  std::unordered_map<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);
+  // Like Java's S3FileIO: delete concurrently, keep going after a failure and
+  // report the count. Arrow has no batch delete for S3.
+  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 = 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. 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 {};
 }
@@ -300,7 +333,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

@@ -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";
@@ -299,6 +308,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