refactor: harden FileCleanupStrategy with retry and parallel deletes (#649)

### What

* `FileCleanupStrategy` now takes an `OptionalExecutor` in its
constructor. When `DeleteWith()` is configured, per-path deletes fan out
through `TaskGroup<>` -- uses the supplied executor when set, otherwise
runs the callbacks serially on the calling thread via
`RunTasksSingleThreaded`. Preserves prior single-threaded behavior by
default.
* The `FileIO::DeleteFiles` bulk path is wrapped in
`RetryRunner<retry::StopRetryOn<ErrorKind::kNotFound>>` with a tight
budget (2 retries, 100ms-1s backoff, 5s total). Mirrors Java's
`Tasks.foreach(...).stopRetryOn(NotFoundException.class).retry(N)`. The
retry primarily helps atomic-bulk FileIO impls (e.g. an S3
`DeleteObjects`-backed FileIO) ride out transient throttles --
best-effort, see code comment for the fail-fast-iterative caveat.
* New public builder
`ExpireSnapshots::ExecuteDeleteWith(OptionalExecutor)` (named after
Java's `executeDeleteWith(ExecutorService)`) so callers opt in to
parallel deletion; threaded through both `IncrementalFileCleanup` and
`ReachableFileCleanup`.
* `DeleteWith()` doc note clarifies that the user-supplied callback may
be invoked concurrently from worker threads and must be thread-safe.
* Drops the `std::async` / `std::thread` / `std::span` machinery and the
ad-hoc retry loop -- replaced by `util/task_group.h` and
`util/retry_util.h` from #687.

### Test coverage

Existing 25 `ExpireSnapshots*` tests continue to pass unchanged (they
hit the default no-executor path).

Adds `ExpireSnapshotsCleanupTest.ExecutorDispatchesDeletesConcurrently`
-- wires a `test::ThreadExecutor` through `ExecuteDeleteWith()`, runs an
`ExpireSnapshotId` cleanup with a custom `DeleteWith` callback, and
asserts the executor received one submission per file (data + manifest +
manifest-list = 3) and that all paths show up in the mutex-guarded sink.

Author: shangxinli

src/iceberg/test/expire_snapshots_test.cc

@@ -19,8 +19,12 @@
 
 #include "iceberg/update/expire_snapshots.h"
 
+#include <functional>
+#include <mutex>
 #include <optional>
+#include <stdexcept>
 #include <string>
+#include <unordered_map>
 #include <vector>
 
 #include <gmock/gmock.h>
@@ -33,6 +37,7 @@
 #include "iceberg/snapshot.h"
 #include "iceberg/statistics_file.h"
 #include "iceberg/table_metadata.h"
+#include "iceberg/test/executor.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/update_test_base.h"
 
@@ -444,6 +449,115 @@ TEST_F(ExpireSnapshotsCleanupTest, DeletesExpiredFiles) {
                                                            expired_manifest_list_path));
 }
 
+TEST_F(ExpireSnapshotsCleanupTest, ExecutorDispatchesDeletesConcurrently) {
+  const auto expired_data_file_path = table_location_ + "/data/expired-data.parquet";
+  const auto expired_data_manifest_path = table_location_ + "/metadata/expired-data.avro";
+  const auto expired_manifest_list_path =
+      table_location_ + "/metadata/expired-manifest-list.avro";
+  const auto current_manifest_list_path =
+      table_location_ + "/metadata/current-manifest-list.avro";
+
+  auto expired_data_manifest = WriteDataManifest(
+      expired_data_manifest_path, kExpiredSnapshotId,
+      {MakeEntry(ManifestStatus::kAdded, kExpiredSnapshotId, kExpiredSequenceNumber,
+                 MakeDataFile(expired_data_file_path))});
+  WriteManifestList(expired_manifest_list_path, kExpiredSnapshotId,
+                    /*parent_snapshot_id=*/0, kExpiredSequenceNumber,
+                    {expired_data_manifest});
+  WriteManifestList(current_manifest_list_path, kCurrentSnapshotId, kExpiredSnapshotId,
+                    kCurrentSequenceNumber, {});
+  RewriteTableWithManifestLists(expired_manifest_list_path, current_manifest_list_path);
+
+  test::ThreadExecutor executor;
+  std::mutex deleted_files_mu;
+  std::vector<std::string> deleted_files;
+
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
+  update->ExpireSnapshotId(kExpiredSnapshotId);
+  update->ExecuteDeleteWith(std::ref(executor));
+  update->DeleteWith([&deleted_files, &deleted_files_mu](const std::string& path) {
+    std::lock_guard<std::mutex> lock(deleted_files_mu);
+    deleted_files.push_back(path);
+  });
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_THAT(deleted_files, testing::UnorderedElementsAre(expired_data_file_path,
+                                                           expired_data_manifest_path,
+                                                           expired_manifest_list_path));
+  EXPECT_EQ(executor.submit_count(), 3);
+}
+
+TEST_F(ExpireSnapshotsCleanupTest, ExecuteDeleteWithWithoutDeleteWithDoesNotUseExecutor) {
+  const auto expired_data_file_path = table_location_ + "/data/expired-data.parquet";
+  const auto expired_data_manifest_path = table_location_ + "/metadata/expired-data.avro";
+  const auto expired_manifest_list_path =
+      table_location_ + "/metadata/expired-manifest-list.avro";
+  const auto current_manifest_list_path =
+      table_location_ + "/metadata/current-manifest-list.avro";
+
+  auto expired_data_manifest = WriteDataManifest(
+      expired_data_manifest_path, kExpiredSnapshotId,
+      {MakeEntry(ManifestStatus::kAdded, kExpiredSnapshotId, kExpiredSequenceNumber,
+                 MakeDataFile(expired_data_file_path))});
+  WriteManifestList(expired_manifest_list_path, kExpiredSnapshotId,
+                    /*parent_snapshot_id=*/0, kExpiredSequenceNumber,
+                    {expired_data_manifest});
+  WriteManifestList(current_manifest_list_path, kCurrentSnapshotId, kExpiredSnapshotId,
+                    kCurrentSequenceNumber, {});
+  RewriteTableWithManifestLists(expired_manifest_list_path, current_manifest_list_path);
+
+  test::ThreadExecutor executor(ServiceUnavailable("executor should be unused"));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
+  update->ExpireSnapshotId(kExpiredSnapshotId);
+  update->ExecuteDeleteWith(std::ref(executor));
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_EQ(executor.submit_count(), 0);
+}
+
+TEST_F(ExpireSnapshotsCleanupTest, DeleteWithRetriesTransientFailures) {
+  const auto expired_data_file_path = table_location_ + "/data/expired-data.parquet";
+  const auto expired_data_manifest_path = table_location_ + "/metadata/expired-data.avro";
+  const auto expired_manifest_list_path =
+      table_location_ + "/metadata/expired-manifest-list.avro";
+  const auto current_manifest_list_path =
+      table_location_ + "/metadata/current-manifest-list.avro";
+
+  auto expired_data_manifest = WriteDataManifest(
+      expired_data_manifest_path, kExpiredSnapshotId,
+      {MakeEntry(ManifestStatus::kAdded, kExpiredSnapshotId, kExpiredSequenceNumber,
+                 MakeDataFile(expired_data_file_path))});
+  WriteManifestList(expired_manifest_list_path, kExpiredSnapshotId,
+                    /*parent_snapshot_id=*/0, kExpiredSequenceNumber,
+                    {expired_data_manifest});
+  WriteManifestList(current_manifest_list_path, kCurrentSnapshotId, kExpiredSnapshotId,
+                    kCurrentSequenceNumber, {});
+  RewriteTableWithManifestLists(expired_manifest_list_path, current_manifest_list_path);
+
+  std::unordered_map<std::string, int> attempts_by_path;
+  std::vector<std::string> deleted_files;
+
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
+  update->ExpireSnapshotId(kExpiredSnapshotId);
+  update->DeleteWith([&attempts_by_path, &deleted_files](const std::string& path) {
+    auto& attempts = attempts_by_path[path];
+    ++attempts;
+    if (attempts == 1) {
+      throw std::runtime_error("transient delete failure");
+    }
+    deleted_files.push_back(path);
+  });
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_THAT(deleted_files, testing::UnorderedElementsAre(expired_data_file_path,
+                                                           expired_data_manifest_path,
+                                                           expired_manifest_list_path));
+  EXPECT_EQ(attempts_by_path[expired_data_file_path], 2);
+  EXPECT_EQ(attempts_by_path[expired_data_manifest_path], 2);
+  EXPECT_EQ(attempts_by_path[expired_manifest_list_path], 2);
+}
+
 TEST_F(ExpireSnapshotsCleanupTest, MetadataOnlySkipsDataDeletion) {
   const auto expired_data_file_path = table_location_ + "/data/expired-data.parquet";
   const auto expired_delete_manifest_path =

src/iceberg/update/expire_snapshots.cc

@@ -21,6 +21,7 @@
 
 #include <algorithm>
 #include <cstdint>
+#include <exception>
 #include <iterator>
 #include <memory>
 #include <string>
@@ -41,8 +42,10 @@
 #include "iceberg/transaction.h"
 #include "iceberg/util/error_collector.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/retry_util.h"
 #include "iceberg/util/snapshot_util_internal.h"
 #include "iceberg/util/string_util.h"
+#include "iceberg/util/task_group.h"
 
 namespace iceberg {
 
@@ -61,8 +64,11 @@ Result<std::unique_ptr<ManifestReader>> MakeManifestReader(
 class FileCleanupStrategy {
  public:
   FileCleanupStrategy(std::shared_ptr<FileIO> file_io,
-                      std::function<void(const std::string&)> delete_func)
-      : file_io_(std::move(file_io)), delete_func_(std::move(delete_func)) {}
+                      std::function<void(const std::string&)> delete_func,
+                      OptionalExecutor executor)
+      : file_io_(std::move(file_io)),
+        delete_func_(std::move(delete_func)),
+        executor_(std::move(executor)) {}
 
   virtual ~FileCleanupStrategy() = default;
 
@@ -94,20 +100,36 @@ class FileCleanupStrategy {
     return expired;
   }
 
-  /// \brief Delete files at the given locations.
+  /// \brief Best-effort delete with bounded retry.
   void DeleteFiles(const std::unordered_set<std::string>& paths) {
-    try {
-      if (delete_func_) {
-        for (const auto& path : paths) {
+    if (paths.empty()) return;
+
+    if (!delete_func_) {
+      std::vector<std::string> path_list(paths.begin(), paths.end());
+
+      TaskGroup<retry::StopRetryOn<ErrorKind::kNotFound>> group(kDeleteRetryConfig);
+      group.Submit([this, paths = std::move(path_list)]() -> Status {
+        return file_io_->DeleteFiles(paths);
+      });
+      std::ignore = std::move(group).Run();
+      return;
+    }
+
+    TaskGroup<retry::StopRetryOn<ErrorKind::kNotFound>> group(kDeleteRetryConfig);
+    group.SetExecutor(executor_);
+    for (const auto& path : paths) {
+      group.Submit([this, path]() -> Status {
+        try {
           delete_func_(path);
+          return {};
+        } catch (const std::exception& e) {
+          return IOError("Delete callback failed for {}: {}", path, e.what());
+        } catch (...) {
+          return IOError("Delete callback failed for {}", path);
         }
-      } else {
-        std::vector<std::string> path_list(paths.begin(), paths.end());
-        std::ignore = file_io_->DeleteFiles(path_list);
-      }
-    } catch (...) {
-      // TODO(shangxinli): add retry
+      });
     }
+    std::ignore = std::move(group).Run();
   }
 
   bool HasAnyStatisticsFiles(const TableMetadata& metadata) const {
@@ -149,6 +171,18 @@ class FileCleanupStrategy {
 
   std::shared_ptr<FileIO> file_io_;
   std::function<void(const std::string&)> delete_func_;
+  OptionalExecutor executor_;
+
+ private:
+  /// Retry budget for the FileIO bulk `DeleteFiles` path. Tight on purpose: file
+  /// cleanup is best-effort and runs after a successful commit, so we'd rather give
+  /// up than block the caller for minutes on a flaky storage layer.
+  static constexpr RetryConfig kDeleteRetryConfig{
+      .num_retries = 2,
+      .min_wait_ms = 100,
+      .max_wait_ms = 1000,
+      .total_timeout_ms = 5000,
+  };
 };
 
 /// \brief File cleanup strategy that determines safe deletions via full reachability.
@@ -157,7 +191,7 @@ class FileCleanupStrategy {
 /// still referenced by retained snapshots, then deletes orphaned manifests, data
 /// files, and manifest lists.
 ///
-/// TODO(shangxinli): Add multi-threaded manifest reading and file deletion support.
+/// TODO(shangxinli): Add multi-threaded manifest reading support.
 class ReachableFileCleanup : public FileCleanupStrategy {
  public:
   using FileCleanupStrategy::FileCleanupStrategy;
@@ -362,7 +396,7 @@ class ReachableFileCleanup : public FileCleanupStrategy {
 /// logically introduced by a snapshot whose changes are still present in the
 /// current state under a different id.
 ///
-/// TODO(shangxinli): Add multi-threaded manifest reading and file deletion support.
+/// TODO(shangxinli): Add multi-threaded manifest reading support.
 class IncrementalFileCleanup : public FileCleanupStrategy {
  public:
   using FileCleanupStrategy::FileCleanupStrategy;
@@ -699,6 +733,11 @@ ExpireSnapshots& ExpireSnapshots::CleanExpiredMetadata(bool clean) {
   return *this;
 }
 
+ExpireSnapshots& ExpireSnapshots::ExecuteDeleteWith(OptionalExecutor executor) {
+  executor_ = std::move(executor);
+  return *this;
+}
+
 Result<std::unordered_set<int64_t>> ExpireSnapshots::ComputeBranchSnapshotsToRetain(
     int64_t snapshot_id, TimePointMs expire_snapshot_older_than,
     int32_t min_snapshots_to_keep) const {
@@ -930,11 +969,11 @@ Status ExpireSnapshots::Finalize(Result<const TableMetadata*> commit_result) {
       !HasNonMainSnapshots(metadata_after_expiration);
 
   if (can_use_incremental) {
-    return IncrementalFileCleanup(ctx_->table->io(), delete_func_)
+    return IncrementalFileCleanup(ctx_->table->io(), delete_func_, executor_)
         .CleanFiles(metadata_before_expiration, metadata_after_expiration,
                     cleanup_level_);
   }
-  return ReachableFileCleanup(ctx_->table->io(), delete_func_)
+  return ReachableFileCleanup(ctx_->table->io(), delete_func_, executor_)
       .CleanFiles(metadata_before_expiration, metadata_after_expiration, cleanup_level_);
 }
 

src/iceberg/update/expire_snapshots.h

@@ -32,6 +32,7 @@
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
 #include "iceberg/update/pending_update.h"
+#include "iceberg/util/executor.h"
 #include "iceberg/util/timepoint.h"
 
 /// \file iceberg/update/expire_snapshots.h
@@ -115,6 +116,8 @@ class ICEBERG_EXPORT ExpireSnapshots : public PendingUpdate {
   /// If this method is not called, unnecessary manifests and data files will still be
   /// deleted.
   ///
+  /// \note With ExecuteDeleteWith(), callbacks may run concurrently.
+  ///
   /// \param delete_func A function that will be called to delete manifests and data files
   /// \return Reference to this for method chaining.
   ExpireSnapshots& DeleteWith(std::function<void(const std::string&)> delete_func);
@@ -140,6 +143,15 @@ class ICEBERG_EXPORT ExpireSnapshots : public PendingUpdate {
   /// \return Reference to this for method chaining.
   ExpireSnapshots& CleanExpiredMetadata(bool clean);
 
+  /// \brief Configure an executor for DeleteWith() callbacks.
+  ///
+  /// Only used with DeleteWith(). The caller must keep the executor alive until
+  /// Finalize() returns.
+  ///
+  /// \param executor An executor reference, or std::nullopt for serial deletion.
+  /// \return Reference to this for method chaining.
+  ExpireSnapshots& ExecuteDeleteWith(OptionalExecutor executor);
+
   Kind kind() const final { return Kind::kExpireSnapshots; }
   bool IsRetryable() const override { return true; }
 
@@ -184,6 +196,7 @@ class ICEBERG_EXPORT ExpireSnapshots : public PendingUpdate {
   enum CleanupLevel cleanup_level_ { CleanupLevel::kAll };
   bool clean_expired_metadata_{false};
   bool specified_snapshot_id_{false};
+  OptionalExecutor executor_;
 
   /// Cached result from Apply(), consumed by Finalize() and cleared after use.
   std::optional<ApplyResult> apply_result_;