diff --git a/mkdocs/docs/file-io.md b/mkdocs/docs/file-io.md
index 5b123addf7..63ce34de9a 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` | 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
diff --git a/src/iceberg/arrow/arrow_io_internal.h b/src/iceberg/arrow/arrow_io_internal.h
index a6b85b6c98..f3007b0d30 100644
--- a/src/iceberg/arrow/arrow_io_internal.h
+++ b/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
diff --git a/src/iceberg/arrow/arrow_io_util.h b/src/iceberg/arrow/arrow_io_util.h
index 925ee12a27..b486638640 100644
--- a/src/iceberg/arrow/arrow_io_util.h
+++ b/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
diff --git a/src/iceberg/arrow/s3/arrow_s3_file_io.cc b/src/iceberg/arrow/s3/arrow_s3_file_io.cc
index 4fbf330985..0676913de9 100644
--- a/src/iceberg/arrow/s3/arrow_s3_file_io.cc
+++ b/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,8 +477,7 @@ 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;
   {
@@ -347,32 +485,27 @@ Status ArrowS3FileIO::DeleteFiles(const std::vector<std::string>& file_locations
     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() {
diff --git a/src/iceberg/arrow/s3/s3_properties.h b/src/iceberg/arrow/s3/s3_properties.h
index 50dafa56ff..fe0a9aa3c9 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";
+  /// 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.
diff --git a/src/iceberg/test/arrow_s3_file_io_test.cc b/src/iceberg/test/arrow_s3_file_io_test.cc
index 24628949a9..6bb3f4178e 100644
--- a/src/iceberg/test/arrow_s3_file_io_test.cc
+++ b/src/iceberg/test/arrow_s3_file_io_test.cc
@@ -18,12 +18,17 @@
  */
 
 #include <algorithm>
+#include <array>
 #include <atomic>
+#include <barrier>
+#include <chrono>
 #include <cstdlib>
+#include <future>
 #include <iostream>
 #include <memory>
 #include <optional>
 #include <span>
+#include <stdexcept>
 #include <string>
 #include <string_view>
 #include <thread>
@@ -38,12 +43,16 @@
 #  include <arrow/filesystem/s3fs.h>
 #endif
 
+#include "iceberg/arrow/arrow_io_internal.h"
 #include "iceberg/arrow/arrow_io_util.h"
+#include "iceberg/arrow/arrow_register.h"
 #include "iceberg/arrow/s3/s3_properties.h"
 #include "iceberg/file_io.h"
+#include "iceberg/file_io_registry.h"
 #include "iceberg/logging/logger.h"
 #include "iceberg/result.h"
 #include "iceberg/storage_credential.h"
+#include "iceberg/test/executor.h"
 #include "iceberg/test/logging_test_helpers.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/util/macros.h"
@@ -103,6 +112,11 @@ Result<::arrow::fs::S3Options> ConfigureS3Options(
 
 namespace {
 
+test::ThreadExecutor& DeleteExecutor() {
+  static test::ThreadExecutor executor;
+  return executor;
+}
+
 class ArrowS3FileIOTest : public ::testing::Test {
  protected:
 #if ICEBERG_S3_ENABLED
@@ -116,6 +130,7 @@ class ArrowS3FileIOTest : public ::testing::Test {
 #  else
     ::setenv("AWS_EC2_METADATA_DISABLED", "true", /*overwrite=*/0);
 #  endif
+    ASSERT_THAT(InitializeS3FileIODeleteExecutor(&DeleteExecutor()), IsOk());
     auto io = MakeS3FileIO({});
     ASSERT_THAT(io, IsOk());
   }
@@ -156,8 +171,263 @@ Status CheckReadWrite(FileIO& io, const std::string& object_uri,
   return io.DeleteFile(object_uri);
 }
 
+class InlineDeleteExecutor : public Executor {
+ public:
+  Status Submit(ExecutorTask task) override {
+    ++submissions;
+    std::move(task)();
+    return {};
+  }
+
+  int submissions = 0;
+};
+
+enum class SubmitFailure { kStatus, kException, kAcceptedException };
+
+class FailingDeleteExecutor : public Executor {
+ public:
+  FailingDeleteExecutor(SubmitFailure failure, std::shared_future<void> release)
+      : failure_(failure), release_(std::move(release)) {}
+
+  ~FailingDeleteExecutor() override { Wait(); }
+
+  void Wait() {
+    for (auto& thread : threads_) {
+      if (thread.joinable()) {
+        thread.join();
+      }
+    }
+  }
+
+  Status Submit(ExecutorTask task) override {
+    ++submissions_;
+    if (submissions_ == 1 || failure_ == SubmitFailure::kAcceptedException) {
+      const bool first = submissions_ == 1;
+      const bool wait = !first || failure_ != SubmitFailure::kAcceptedException;
+      threads_.emplace_back([this, first, wait, task = std::move(task)]() mutable {
+        if (wait) {
+          release_.wait();
+        }
+        std::move(task)();
+        ++completed;
+        if (first) {
+          first_completed.set_value();
+        }
+      });
+    }
+    if (submissions_ == 2) {
+      submission_failed.set_value();
+      if (failure_ == SubmitFailure::kStatus) {
+        return ServiceUnavailable("executor rejected deletion");
+      }
+      throw std::runtime_error("executor rejected deletion");
+    }
+    return {};
+  }
+
+  std::promise<void> submission_failed;
+  std::promise<void> first_completed;
+  std::atomic<int> completed = 0;
+
+ private:
+  SubmitFailure failure_;
+  std::shared_future<void> release_;
+  int submissions_ = 0;
+  std::vector<std::thread> threads_;
+};
+
 }  // namespace
 
+TEST(DeleteExecutorStateTest, InitializationIsImmutable) {
+  InlineDeleteExecutor executor;
+  DeleteExecutorState state;
+  EXPECT_THAT(state.Initialize(&executor), IsOk());
+  EXPECT_THAT(state.Initialize(&executor), IsOk());
+  EXPECT_THAT(state.Initialize(nullptr), IsError(ErrorKind::kInvalidArgument));
+  EXPECT_EQ(state.Get(), &executor);
+
+  DeleteExecutorState sequential;
+  EXPECT_EQ(sequential.Get(), nullptr);
+  EXPECT_THAT(sequential.Initialize(&executor), IsError(ErrorKind::kInvalidArgument));
+}
+
+TEST(DeleteExecutorStateTest, ConcurrentInitializationChoosesOneExecutor) {
+  InlineDeleteExecutor first;
+  InlineDeleteExecutor second;
+  DeleteExecutorState state;
+  std::barrier start(2);
+  std::atomic<int> initialized = 0;
+  auto initialize = [&](Executor* executor) {
+    start.arrive_and_wait();
+    if (state.Initialize(executor).has_value()) {
+      ++initialized;
+    }
+  };
+  std::thread a(initialize, &first);
+  std::thread b(initialize, &second);
+  a.join();
+  b.join();
+  EXPECT_EQ(initialized, 1);
+  EXPECT_TRUE(state.Get() == &first || state.Get() == &second);
+}
+
+TEST(BulkDeleteFilesTest, LimitsConcurrentDeletes) {
+  const std::array<std::pair<size_t, size_t>, 4> cases = {
+      {{2, 3}, {16, 1}, {16, 3}, {128, 3}}};
+  for (const auto& [file_count, parallelism] : cases) {
+    SCOPED_TRACE(file_count);
+    SCOPED_TRACE(parallelism);
+    const auto workers = std::min(file_count, parallelism);
+    test::ThreadExecutor executor;
+    std::vector<std::string> paths;
+    for (size_t i = 0; i < file_count; ++i) {
+      paths.push_back(std::to_string(i));
+    }
+    std::vector<std::atomic<int>> attempts(file_count);
+    std::promise<void> started;
+    auto all_started = started.get_future();
+    std::promise<void> release;
+    auto released = release.get_future().share();
+    std::atomic<size_t> active = 0;
+    std::atomic<size_t> peak = 0;
+    std::atomic<size_t> attempted = 0;
+    auto run = std::async(std::launch::async, [&] {
+      return BulkDeleteFiles(
+          paths, &executor, parallelism, [&](const std::string& path) -> Status {
+            ++attempts.at(std::stoul(path));
+            const auto running = ++active;
+            auto previous = peak.load();
+            while (previous < running && !peak.compare_exchange_weak(previous, running)) {
+            }
+            if (++attempted == workers) {
+              started.set_value();
+            }
+            released.wait();
+            --active;
+            return {};
+          });
+    });
+    const auto ready = all_started.wait_for(std::chrono::seconds(5));
+    release.set_value();
+    EXPECT_THAT(run.get(), IsOk());
+    EXPECT_EQ(ready, std::future_status::ready);
+    EXPECT_EQ(peak, workers);
+    EXPECT_EQ(attempted, file_count);
+    EXPECT_EQ(executor.submit_count(), workers);
+    for (const auto& count : attempts) {
+      EXPECT_EQ(count.load(), 1);
+    }
+  }
+}
+
+TEST(BulkDeleteFilesTest, CountsThrownDeletesAndAttemptsRemainingFiles) {
+  for (bool parallel : {false, true}) {
+    SCOPED_TRACE(parallel);
+    test::ThreadExecutor executor;
+    const std::vector<std::string> paths = {"status", "exception", "unknown", "ok"};
+    const auto caller = std::this_thread::get_id();
+    std::vector<std::string> sequential_attempts;
+    std::atomic<int> attempted = 0;
+    auto logger = std::make_shared<CapturingLogger>();
+    ScopedLogger bind(logger);
+    auto status = BulkDeleteFiles(paths, parallel ? &executor : nullptr, 3,
+                                  [&](const std::string& path) -> Status {
+                                    if (!parallel) {
+                                      EXPECT_EQ(std::this_thread::get_id(), caller);
+                                      sequential_attempts.push_back(path);
+                                    }
+                                    ++attempted;
+                                    if (path == "status") {
+                                      return IOError("delete failed");
+                                    }
+                                    if (path == "exception") {
+                                      throw std::runtime_error("delete threw");
+                                    }
+                                    if (path == "unknown") {
+                                      throw 42;
+                                    }
+                                    return {};
+                                  });
+    EXPECT_THAT(status, HasErrorMessage("Failed to delete 3 of 4 files"));
+    EXPECT_EQ(attempted, 4);
+    EXPECT_EQ(logger->count(), 3);
+    if (!parallel) {
+      EXPECT_EQ(sequential_attempts, paths);
+    }
+  }
+}
+
+TEST(BulkDeleteFilesTest, DrainsAcceptedTasksWhenSubmissionFails) {
+  for (auto failure : {SubmitFailure::kStatus, SubmitFailure::kException,
+                       SubmitFailure::kAcceptedException}) {
+    SCOPED_TRACE(static_cast<int>(failure));
+    std::promise<void> release;
+    FailingDeleteExecutor executor(failure, release.get_future().share());
+    auto submission_failed = executor.submission_failed.get_future();
+    auto first_completed = executor.first_completed.get_future();
+    const std::vector<std::string> paths = {"first", "second", "third"};
+    std::atomic<int> attempted = 0;
+    auto run = std::async(std::launch::async, [&] {
+      const auto caller = std::this_thread::get_id();
+      return BulkDeleteFiles(paths, &executor, 3, [&](const std::string& path) -> Status {
+        EXPECT_NE(std::this_thread::get_id(), caller);
+        ++attempted;
+        if (path == "second") {
+          return IOError("delete failed");
+        }
+        return {};
+      });
+    });
+    const auto submitted = submission_failed.wait_for(std::chrono::seconds(5));
+    auto first = std::future_status::ready;
+    if (failure == SubmitFailure::kAcceptedException) {
+      // Leave only the last task blocked.
+      first = first_completed.wait_for(std::chrono::seconds(5));
+    }
+    const auto returned = run.wait_for(std::chrono::milliseconds(100));
+    // Unblock workers before assertions.
+    release.set_value();
+    auto status = run.get();
+    executor.Wait();
+    EXPECT_EQ(submitted, std::future_status::ready);
+    EXPECT_EQ(first, std::future_status::ready);
+    EXPECT_EQ(returned, std::future_status::timeout);
+    EXPECT_THAT(status, HasErrorMessage("executor rejected deletion"));
+    EXPECT_THAT(status, HasErrorMessage("failed to delete 1 of 3 files"));
+    EXPECT_THAT(status,
+                IsError(failure == SubmitFailure::kStatus ? ErrorKind::kServiceUnavailable
+                                                          : ErrorKind::kIOError));
+    EXPECT_EQ(attempted, 3);
+    EXPECT_EQ(executor.completed, failure == SubmitFailure::kAcceptedException ? 2 : 1);
+  }
+}
+
+TEST(BulkDeleteFilesTest, DoesNotDeleteOnTheCallerWhenSubmissionFails) {
+  test::ThreadExecutor executor(ServiceUnavailable("executor is shut down"));
+  const std::vector<std::string> paths = {"first", "second", "third"};
+  int attempted = 0;
+  EXPECT_THAT(BulkDeleteFiles(paths, &executor, 3,
+                              [&](const std::string&) -> Status {
+                                ++attempted;
+                                return {};
+                              }),
+              IsError(ErrorKind::kServiceUnavailable));
+  EXPECT_EQ(attempted, 0);
+  EXPECT_EQ(executor.submit_count(), 1);
+}
+
+TEST(BulkDeleteFilesTest, EmptyDeleteDoesNotSubmitTasks) {
+  InlineDeleteExecutor executor;
+  EXPECT_THAT(BulkDeleteFiles({}, &executor, 3,
+                              [](const std::string&) -> Status {
+                                ADD_FAILURE()
+                                    << "An empty delete should not invoke the callback";
+                                return {};
+                              }),
+              IsOk());
+  EXPECT_EQ(executor.submissions, 0);
+}
+
 TEST_F(ArrowS3FileIOTest, Create) {
   auto result = MakeS3FileIO({});
   ASSERT_THAT(result, IsOk());
@@ -299,6 +569,25 @@ TEST_F(ArrowS3FileIOTest, RejectsIncompleteStaticCredentials) {
                           "S3 client access key ID and secret access key must be set"));
 }
 
+TEST_F(ArrowS3FileIOTest, SharesDeleteExecutorWithRegistryCreatedFileIOs) {
+  RegisterAll();
+  const auto key = std::string(S3Properties::kDeleteNumThreads);
+  ICEBERG_UNWRAP_OR_FAIL(auto direct, MakeS3FileIO({{key, "1"}}));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto registered,
+      FileIORegistry::Load(FileIORegistry::kArrowS3FileIO, {{key, "2"}}));
+  auto& executor = DeleteExecutor();
+  const auto before = executor.submit_count();
+  // A bucket without a key fails before network I/O.
+  const std::vector<std::string> paths = {"s3://bucket", "s3://bucket"};
+  EXPECT_THAT(direct->DeleteFiles(paths),
+              HasErrorMessage("Failed to delete 2 of 2 files"));
+  EXPECT_EQ(executor.submit_count() - before, 1);
+  EXPECT_THAT(registered->DeleteFiles(paths),
+              HasErrorMessage("Failed to delete 2 of 2 files"));
+  EXPECT_EQ(executor.submit_count() - before, 3);
+}
+
 TEST_F(ArrowS3FileIOTest, ReadWrite) {
   if (!HasIntegrationEnv()) {
     GTEST_SKIP() << "Set ICEBERG_TEST_S3_URI to enable S3 IO test";
@@ -353,6 +642,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";
+  }
+
+  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_THAT(io->ReadFile(allowed, std::nullopt),
+              ::testing::AllOf(IsError(ErrorKind::kIOError),
+                               HasErrorMessage("Path does not exist")));
+
+  // 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
diff --git a/src/iceberg/util/executor.h b/src/iceberg/util/executor.h
index 749502b179..625c833f12 100644
--- a/src/iceberg/util/executor.h
+++ b/src/iceberg/util/executor.h
@@ -33,7 +33,7 @@ namespace iceberg {
 
 using ExecutorTask = FnOnce<void()>;
 
-/// \brief Schedules iceberg-cpp internal planning tasks.
+/// \brief Schedules iceberg-cpp tasks.
 ///
 /// Public APIs that accept an executor remain synchronous: the calling thread may block
 /// while waiting for submitted tasks to finish. Callers must ensure the executor can
@@ -41,8 +41,8 @@ using ExecutorTask = FnOnce<void()>;
 /// the same bounded executor's worker threads can deadlock unless the executor supports
 /// nested blocking work.
 ///
-/// When an executor is configured, planning callbacks may be called concurrently. Any
-/// shared mutable state captured by those callbacks must be synchronized by the caller.
+/// When an executor is configured, callbacks may be called concurrently. Any shared
+/// mutable state captured by those callbacks must be synchronized by the caller.
 class ICEBERG_EXPORT Executor {
  public:
   virtual ~Executor() = default;