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` | Maximum worker tasks per `DeleteFiles` call; defaults to the hardware thread count. Requires a delete executor |
 
 The following keys are specific to iceberg-cpp; they are not part of the Java
 Iceberg or REST specification property set:
@@ -80,6 +81,29 @@ Without credentials, the AWS default credential chain is used, which covers
 environment variables, the shared configuration file, and the various role and
 identity providers.
 
+### Bulk deletion
+
+S3 `DeleteFiles` attempts each file and returns the failure count, including
+exceptions thrown by a delete. Deletion is sequential by default. To enable
+concurrency, call `iceberg::arrow::InitializeS3FileIODeleteExecutor(executor)`
+from `iceberg/arrow/arrow_io_util.h` before the first delete. The application
+owns the executor; the setting applies to direct and registry-created FileIOs.
+
+Initialization is permanent. Repeating the same pointer is allowed, but changing
+it returns an error. The first delete freezes the default `nullptr` if no executor
+was initialized. Keep the executor alive until the last delete finishes, and do
+not start more deletes after destroying it.
+
+`s3.delete.num-threads` limits workers within one call, not across concurrent
+calls. The shared executor controls total concurrency; this property does not
+create or resize a thread pool. Workers share the file list, so task and future
+storage is bounded by the configured parallelism.
+
+`DeleteFiles` waits for accepted work before returning, even when submission
+fails or throws. A submission failure is returned without deleting on the caller.
+The executor must make progress while callers wait; do not call `DeleteFiles`
+from its workers unless it supports nested blocking work.
+
 ### S3-compatible storage
 
 Stores that speak the S3 API are served by the same implementation. The scheme

src/iceberg/arrow/arrow_io_internal.h

@@ -20,8 +20,11 @@
 #pragma once
 
 #include <cstddef>
+#include <functional>
 #include <memory>
+#include <mutex>
 #include <optional>
+#include <span>
 #include <string>
 #include <vector>
 
@@ -30,9 +33,27 @@
 
 #include "iceberg/file_io.h"
 #include "iceberg/iceberg_bundle_export.h"
+#include "iceberg/util/executor.h"
 
 namespace iceberg::arrow {
 
+// The first initialization or lookup fixes the executor for this state.
+class ICEBERG_BUNDLE_EXPORT DeleteExecutorState {
+ public:
+  Status Initialize(Executor* executor);
+  Executor* Get();
+
+ private:
+  std::mutex mutex_;
+  bool initialized_ = false;
+  Executor* executor_ = nullptr;
+};
+
+// Attempt each file unless submission fails; always wait for accepted work.
+ICEBERG_BUNDLE_EXPORT Status BulkDeleteFiles(
+    std::span<const std::string> file_locations, Executor* executor, size_t parallelism,
+    const std::function<Status(const std::string&)>& delete_file);
+
 /// \brief Open a FileIO input as an Arrow input stream.
 ///
 /// Uses ArrowFileSystemFileIO's native Arrow stream directly when possible and falls

src/iceberg/arrow/arrow_io_util.h

@@ -29,6 +29,7 @@
 #include "iceberg/file_io.h"
 #include "iceberg/iceberg_bundle_export.h"
 #include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
 
 namespace iceberg::arrow {
 
@@ -47,6 +48,14 @@ ICEBERG_BUNDLE_EXPORT std::unique_ptr<FileIO> MakeLocalFileIO();
 ICEBERG_BUNDLE_EXPORT Result<std::unique_ptr<FileIO>> MakeS3FileIO(
     const std::unordered_map<std::string, std::string>& properties = {});
 
+/// \brief Initialize the shared executor for S3 bulk deletes.
+///
+/// nullptr runs sequentially. The first delete freezes this default if not initialized.
+/// Repeating the same executor is allowed; replacing it is not. The caller owns it
+/// and must keep it alive until the last delete completes. Includes registry-created
+/// FileIOs. See Executor for blocking requirements.
+ICEBERG_BUNDLE_EXPORT Status InitializeS3FileIODeleteExecutor(Executor* executor);
+
 /// \brief Finalize (clean up) the Arrow S3 subsystem.
 ///
 /// Must be called before process exit if S3 was initialized, otherwise Arrow's

src/iceberg/arrow/s3/arrow_s3_file_io.cc

@@ -18,13 +18,21 @@
  */
 
 #include <algorithm>
+#include <atomic>
+#include <cstdint>
 #include <cstdlib>
+#include <exception>
+#include <format>
+#include <functional>
+#include <future>
 #include <memory>
 #include <mutex>
 #include <optional>
 #include <shared_mutex>
+#include <span>
 #include <string>
 #include <string_view>
+#include <thread>
 #include <unordered_map>
 #include <utility>
 #include <vector>
@@ -39,12 +47,140 @@
 #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/executor.h"
 #include "iceberg/util/macros.h"
 #include "iceberg/util/property_util.h"
 #include "iceberg/util/string_util.h"
 
 namespace iceberg::arrow {
 
+namespace {
+
+struct DeleteTasks {
+  std::vector<std::future<void>> futures;
+
+  ~DeleteTasks() {
+    for (auto& future : futures) {
+      if (future.valid()) {
+        future.wait();
+      }
+    }
+  }
+};
+
+// Submit may throw instead of returning an error; both reject the worker.
+Status SubmitWorker(Executor& executor, ExecutorTask task) {
+  try {
+    return executor.Submit(std::move(task));
+  } catch (const std::exception& e) {
+    return IOError("Submit threw an exception: {}", e.what());
+  } catch (...) {
+    return IOError("Submit threw an unknown exception");
+  }
+}
+
+DeleteExecutorState delete_executor;
+
+}  // namespace
+
+Status DeleteExecutorState::Initialize(Executor* executor) {
+  std::lock_guard lock(mutex_);
+  if (initialized_) {
+    if (executor_ != executor) {
+      return InvalidArgument("Delete executor is already initialized");
+    }
+    return {};
+  }
+  executor_ = executor;
+  initialized_ = true;
+  return {};
+}
+
+Executor* DeleteExecutorState::Get() {
+  std::lock_guard lock(mutex_);
+  initialized_ = true;
+  return executor_;
+}
+
+Status BulkDeleteFiles(std::span<const std::string> file_locations, Executor* executor,
+                       size_t parallelism,
+                       const std::function<Status(const std::string&)>& delete_file) {
+  ICEBERG_PRECHECK(parallelism > 0, "Delete parallelism must be positive");
+  auto logger = GetCurrentLogger();
+  std::atomic<size_t> next = 0;
+  std::atomic<size_t> failed = 0;
+  auto work = [&] {
+    ScopedLogger bind(logger);
+    while (true) {
+      const auto index = next.fetch_add(1, std::memory_order_relaxed);
+      if (index >= file_locations.size()) {
+        return;
+      }
+      const auto& location = file_locations[index];
+      Status status;
+      try {
+        status = delete_file(location);
+      } catch (const std::exception& e) {
+        status = IOError("Delete threw an exception: {}", e.what());
+      } catch (...) {
+        status = IOError("Delete threw an unknown exception");
+      }
+      if (!status.has_value()) {
+        failed.fetch_add(1, std::memory_order_relaxed);
+        ICEBERG_LOG_WARN("Failed to delete {}: {}", location, status.error().message);
+      }
+    }
+  };
+
+  std::optional<Error> submission_error;
+  try {
+    if (executor == nullptr) {
+      work();
+    } else {
+      const auto workers = std::min(parallelism, file_locations.size());
+      DeleteTasks tasks;
+      tasks.futures.reserve(workers);
+      size_t accepted = 0;
+      for (; accepted < workers; ++accepted) {
+        std::packaged_task<void()> task(work);
+        // Submit may accept the task before failing.
+        tasks.futures.push_back(task.get_future());
+        ExecutorTask executor_task([task = std::move(task)]() mutable { task(); });
+        if (auto status = SubmitWorker(*executor, std::move(executor_task));
+            !status.has_value()) {
+          submission_error = std::move(status.error());
+          break;
+        }
+      }
+      // A rejected worker's future is only waited for, by `tasks`.
+      for (size_t i = 0; i < accepted; ++i) {
+        tasks.futures[i].get();
+      }
+    }
+  } catch (const std::exception& e) {
+    return IOError("Bulk delete execution failed: {}", e.what());
+  } catch (...) {
+    return IOError("Bulk delete execution failed with an unknown exception");
+  }
+  if (submission_error.has_value()) {
+    if (failed > 0) {
+      submission_error->message += std::format("; failed to delete {} of {} files",
+                                               failed.load(), file_locations.size());
+    }
+    return Status::unexpected_type(std::move(*submission_error));
+  }
+  if (failed > 0) {
+    return IOError("Failed to delete {} of {} files", failed.load(),
+                   file_locations.size());
+  }
+  return {};
+}
+
+Status InitializeS3FileIODeleteExecutor(Executor* executor) {
+  return delete_executor.Initialize(executor);
+}
+
 #if ICEBERG_S3_ENABLED
 
 namespace {
@@ -181,9 +317,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_parallelism)
       : default_file_io_(std::make_shared<ArrowFileSystemFileIO>(std::move(arrow_fs))),
-        default_properties_(std::move(default_properties)) {}
+        default_properties_(std::move(default_properties)),
+        delete_parallelism_(delete_parallelism) {}
 
   Result<std::unique_ptr<InputFile>> NewInputFile(std::string file_location) override;
 
@@ -237,6 +375,7 @@ class ArrowS3FileIO final : public FileIO, public SupportsStorageCredentials {
 
   std::shared_ptr<ArrowFileSystemFileIO> default_file_io_;
   std::unordered_map<std::string, std::string> default_properties_;
+  const size_t delete_parallelism_;
   // Guards everything below; shared because reads happen per file operation.
   mutable std::shared_mutex mutex_;
   std::vector<StorageCredential> storage_credentials_;
@@ -338,41 +477,35 @@ 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.
+  // Keep one delegate snapshot for the whole batch.
   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);
-    }
-  }
-  for (auto& [file_io, locations] : locations_by_io) {
-    ICEBERG_RETURN_UNEXPECTED(file_io->DeleteFiles(locations));
-  }
-  return {};
+  return BulkDeleteFiles(
+      file_locations, delete_executor.Get(), delete_parallelism_,
+      [&](const std::string& location) {
+        return MatchDelegate(fallback, by_prefix, location)->DeleteFile(location);
+      });
 }
 
 }  // namespace
 
 Result<std::unique_ptr<FileIO>> MakeS3FileIO(
     const std::unordered_map<std::string, std::string>& properties) {
+  size_t delete_parallelism = std::max(1u, std::thread::hardware_concurrency());
+  if (const auto* value = FindProperty(properties, S3Properties::kDeleteNumThreads);
+      value != nullptr) {
+    ICEBERG_ASSIGN_OR_RAISE(auto count, StringUtils::ParseNumber<int32_t>(*value));
+    ICEBERG_PRECHECK(count > 0, "{} must be positive", S3Properties::kDeleteNumThreads);
+    delete_parallelism = static_cast<size_t>(count);
+  }
   // Uses default credentials if properties are empty.
   ICEBERG_ASSIGN_OR_RAISE(auto fs, BuildArrowS3FileSystem(properties));
-  return std::make_unique<ArrowS3FileIO>(std::move(fs), properties);
+  return std::make_unique<ArrowS3FileIO>(std::move(fs), properties, delete_parallelism);
 }
 
 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";
+  /// Maximum workers per DeleteFiles call; defaults to the hardware thread count.
+  static constexpr std::string_view kDeleteNumThreads = "s3.delete.num-threads";
 };
 
 /// \brief URI schemes served by the Arrow S3 FileIO, lower-case.