refactor: harden FileCleanupStrategy with retry and parallel deletes

src/iceberg/CMakeLists.txt

@@ -110,6 +110,7 @@ set(ICEBERG_SOURCES
     util/string_util.cc
     util/struct_like_set.cc
     util/temporal_util.cc
+    util/thread_pool_internal.cc
     util/timepoint.cc
     util/transform_util.cc
     util/truncate_util.cc

src/iceberg/meson.build

@@ -132,6 +132,7 @@ iceberg_sources = files(
     'util/string_util.cc',
     'util/struct_like_set.cc',
     'util/temporal_util.cc',
+    'util/thread_pool_internal.cc',
     'util/timepoint.cc',
     'util/transform_util.cc',
     'util/truncate_util.cc',

src/iceberg/test/CMakeLists.txt

@@ -132,6 +132,7 @@ add_iceberg_test(util_test
                  retry_util_test.cc
                  string_util_test.cc
                  struct_like_set_test.cc
+                 thread_pool_test.cc
                  transform_util_test.cc
                  truncate_util_test.cc
                  url_encoder_test.cc

src/iceberg/test/expire_snapshots_test.cc

@@ -19,6 +19,7 @@
 
 #include "iceberg/update/expire_snapshots.h"
 
+#include <mutex>
 #include <optional>
 #include <string>
 #include <vector>
@@ -244,10 +245,13 @@ TEST_F(ExpireSnapshotsTest, ExpireOlderThan) {
 }
 
 TEST_F(ExpireSnapshotsTest, FinalizeRequiresCommittedMetadata) {
+  std::mutex deleted_files_mu;
   std::vector<std::string> deleted_files;
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
-  update->DeleteWith(
-      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+  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);
+  });
 
   // Apply first so apply_result_ is cached
   ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
@@ -262,11 +266,14 @@ TEST_F(ExpireSnapshotsTest, FinalizeRequiresCommittedMetadata) {
 }
 
 TEST_F(ExpireSnapshotsTest, CleanupNoneSkipsDeletion) {
+  std::mutex deleted_files_mu;
   std::vector<std::string> deleted_files;
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
   update->CleanupLevel(CleanupLevel::kNone);
-  update->DeleteWith(
-      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+  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);
+  });
 
   ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
   EXPECT_EQ(result.snapshot_ids_to_remove.size(), 1);
@@ -278,10 +285,13 @@ TEST_F(ExpireSnapshotsTest, CleanupNoneSkipsDeletion) {
 }
 
 TEST_F(ExpireSnapshotsTest, FinalizeSkippedOnCommitError) {
+  std::mutex deleted_files_mu;
   std::vector<std::string> deleted_files;
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
-  update->DeleteWith(
-      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+  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);
+  });
 
   ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
   EXPECT_EQ(result.snapshot_ids_to_remove.size(), 1);
@@ -294,11 +304,14 @@ TEST_F(ExpireSnapshotsTest, FinalizeSkippedOnCommitError) {
 }
 
 TEST_F(ExpireSnapshotsTest, FinalizeSkipsWhenNothingExpired) {
+  std::mutex deleted_files_mu;
   std::vector<std::string> deleted_files;
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
   update->RetainLast(2);
-  update->DeleteWith(
-      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+  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);
+  });
 
   ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
   EXPECT_TRUE(result.snapshot_ids_to_remove.empty());
@@ -348,10 +361,13 @@ TEST_F(ExpireSnapshotsCleanupTest, IgnoresExpiredDeleteManifestReadFailures) {
                     kCurrentSequenceNumber, {});
   RewriteTableWithManifestLists(expired_manifest_list_path, current_manifest_list_path);
 
+  std::mutex deleted_files_mu;
   std::vector<std::string> deleted_files;
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
-  update->DeleteWith(
-      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+  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,
@@ -386,10 +402,13 @@ TEST_F(ExpireSnapshotsCleanupTest, DeletesExpiredFiles) {
                     kCurrentSequenceNumber, {});
   RewriteTableWithManifestLists(expired_manifest_list_path, current_manifest_list_path);
 
+  std::mutex deleted_files_mu;
   std::vector<std::string> deleted_files;
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
-  update->DeleteWith(
-      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+  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,
@@ -424,11 +443,14 @@ TEST_F(ExpireSnapshotsCleanupTest, MetadataOnlySkipsDataDeletion) {
                     kCurrentSequenceNumber, {});
   RewriteTableWithManifestLists(expired_manifest_list_path, current_manifest_list_path);
 
+  std::mutex deleted_files_mu;
   std::vector<std::string> deleted_files;
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
   update->CleanupLevel(CleanupLevel::kMetadataOnly);
-  update->DeleteWith(
-      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+  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_manifest_path,
@@ -462,10 +484,13 @@ TEST_F(ExpireSnapshotsCleanupTest, RetainedDeleteManifestSkipsDataDeletion) {
                     kCurrentSequenceNumber, {current_delete_manifest});
   RewriteTableWithManifestLists(expired_manifest_list_path, current_manifest_list_path);
 
+  std::mutex deleted_files_mu;
   std::vector<std::string> deleted_files;
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
-  update->DeleteWith(
-      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+  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_manifest_path,
@@ -487,10 +512,13 @@ TEST_F(ExpireSnapshotsCleanupTest, DeletesExpiredStats) {
       expired_manifest_list_path, current_manifest_list_path,
       {MakeStatisticsFile(kExpiredSnapshotId, expired_statistics_path)}, {});
 
+  std::mutex deleted_files_mu;
   std::vector<std::string> deleted_files;
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
-  update->DeleteWith(
-      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+  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::Contains(expired_statistics_path));
@@ -513,10 +541,13 @@ TEST_F(ExpireSnapshotsCleanupTest, KeepsReusedStats) {
        MakeStatisticsFile(kCurrentSnapshotId, reused_statistics_path)},
       {});
 
+  std::mutex deleted_files_mu;
   std::vector<std::string> deleted_files;
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
-  update->DeleteWith(
-      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+  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::Not(testing::Contains(reused_statistics_path)));
@@ -538,10 +569,13 @@ TEST_F(ExpireSnapshotsCleanupTest, DeletesExpiredPartitionStats) {
       expired_manifest_list_path, current_manifest_list_path, {},
       {MakePartitionStatisticsFile(kExpiredSnapshotId, expired_statistics_path)});
 
+  std::mutex deleted_files_mu;
   std::vector<std::string> deleted_files;
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
-  update->DeleteWith(
-      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+  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::Contains(expired_statistics_path));
@@ -564,10 +598,13 @@ TEST_F(ExpireSnapshotsCleanupTest, KeepsReusedPartitionStats) {
       {MakePartitionStatisticsFile(kExpiredSnapshotId, reused_statistics_path),
        MakePartitionStatisticsFile(kCurrentSnapshotId, reused_statistics_path)});
 
+  std::mutex deleted_files_mu;
   std::vector<std::string> deleted_files;
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
-  update->DeleteWith(
-      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+  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::Not(testing::Contains(reused_statistics_path)));

src/iceberg/test/meson.build

@@ -96,6 +96,7 @@ iceberg_tests = {
             'roaring_position_bitmap_test.cc',
             'string_util_test.cc',
             'struct_like_set_test.cc',
+            'thread_pool_test.cc',
             'transform_util_test.cc',
             'truncate_util_test.cc',
             'url_encoder_test.cc',

src/iceberg/test/thread_pool_test.cc

@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <atomic>
+#include <chrono>
+#include <cstdint>
+#include <span>
+#include <stdexcept>
+#include <thread>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include "iceberg/util/thread_pool_internal.h"
+
+namespace iceberg {
+
+TEST(ThreadPoolTest, ZeroWorkersThrows) {
+  EXPECT_THROW(ThreadPool(0), std::invalid_argument);
+}
+
+TEST(ThreadPoolTest, SubmitRunsTask) {
+  ThreadPool pool(2);
+  std::atomic<int> n{0};
+  auto fut = pool.Submit([&] { n.fetch_add(1, std::memory_order_relaxed); });
+  fut.wait();
+  EXPECT_EQ(n.load(), 1);
+}
+
+TEST(ThreadPoolTest, RunAndWaitProcessesEveryItem) {
+  ThreadPool pool(4);
+  std::vector<int> items(100);
+  for (int i = 0; i < 100; ++i) items[i] = i;
+
+  std::atomic<std::int64_t> sum{0};
+  pool.RunAndWait<int>(std::span<const int>(items),
+                       [&](int v) { sum.fetch_add(v, std::memory_order_relaxed); });
+
+  // 0 + 1 + ... + 99 = 4950
+  EXPECT_EQ(sum.load(), 4950);
+}
+
+TEST(ThreadPoolTest, RunAndWaitEmptyIsNoOp) {
+  ThreadPool pool(2);
+  std::vector<int> items;
+  std::atomic<int> seen{0};
+  pool.RunAndWait<int>(std::span<const int>(items),
+                       [&](int) { seen.fetch_add(1, std::memory_order_relaxed); });
+  EXPECT_EQ(seen.load(), 0);
+}
+
+TEST(ThreadPoolTest, RunAndWaitExecutesConcurrently) {
+  // Use a barrier-style check: each task increments an in-flight counter, sleeps
+  // briefly, decrements, and records the peak. With multiple workers the peak
+  // should exceed 1.
+  constexpr int kWorkers = 4;
+  constexpr int kItems = 8;
+  ThreadPool pool(kWorkers);
+
+  std::vector<int> items(kItems, 0);
+  std::atomic<int> in_flight{0};
+  std::atomic<int> peak{0};
+
+  pool.RunAndWait<int>(std::span<const int>(items), [&](int) {
+    int now = in_flight.fetch_add(1, std::memory_order_acq_rel) + 1;
+    int prev = peak.load(std::memory_order_relaxed);
+    while (now > prev &&
+           !peak.compare_exchange_weak(prev, now, std::memory_order_relaxed)) {
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(20));
+    in_flight.fetch_sub(1, std::memory_order_acq_rel);
+  });
+
+  EXPECT_GT(peak.load(), 1) << "expected concurrent execution across workers";
+}
+
+TEST(ThreadPoolTest, ExceptionInTaskDoesNotKillWorker) {
+  ThreadPool pool(1);
+  // packaged_task captures the exception into the future; the worker loop must
+  // continue and process the next submission.
+  auto bad = pool.Submit([] { throw std::runtime_error("boom"); });
+  EXPECT_THROW(bad.get(), std::runtime_error);
+
+  std::atomic<int> ok{0};
+  auto good = pool.Submit([&] { ok.store(1, std::memory_order_relaxed); });
+  good.wait();
+  EXPECT_EQ(ok.load(), 1);
+}
+
+TEST(ThreadPoolTest, DestructorJoinsAllPendingWork) {
+  std::atomic<int> done{0};
+  {
+    ThreadPool pool(2);
+    for (int i = 0; i < 16; ++i) {
+      // Discard futures: we rely on the destructor to drain queued work.
+      (void)pool.Submit([&] {
+        std::this_thread::sleep_for(std::chrono::milliseconds(5));
+        done.fetch_add(1, std::memory_order_relaxed);
+      });
+    }
+  }
+  EXPECT_EQ(done.load(), 16);
+}
+
+}  // namespace iceberg