This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 7543e804 fix(io): attempt every S3 delete and report how many failed
(#966)
7543e804 is described below
commit 7543e8042180080d34b5ee44a6d4c18b86e47ac1
Author: Jiajia Li <[email protected]>
AuthorDate: Tue Oct 13 08:47:52 2026 +0800
fix(io): attempt every S3 delete and report how many failed (#966)
Follow-up to
https://github.com/apache/iceberg-cpp/pull/898#discussion_r4091927837.
`ArrowS3FileIO::DeleteFiles` returned at the first credential prefix whose
delete failed, so files under the remaining prefixes were never attempted. It
now attempts every file unless task submission fails, logs deletion failures,
and returns `Failed to delete N of M files`. Exceptions thrown by individual
deletes are counted too.
- Deletion is sequential by default.
`InitializeS3FileIODeleteExecutor(Executor*)` initializes a caller-owned
executor shared by all S3 FileIOs, including registry-created instances.
Initialize it before the first delete; otherwise, the first delete freezes the
default `nullptr`. Repeating the same pointer is allowed, but replacing it is
not. The executor must remain alive until the last delete completes.
- `s3.delete.num-threads` limits worker tasks per call, defaulting to the
hardware thread count. Workers share a file index, keeping task/future storage
bounded by the configured parallelism. The shared executor controls total
concurrency; this property does not create or resize a thread pool.
- Accepted tasks finish before the call exits, including when submission
returns an error or throws. Submission failures are returned without falling
back to the calling thread and bypassing the executor's concurrency limit. All
workers use the same delegate snapshot and the caller's logger.
Arrow has no public S3 batch-delete API, so each request still deletes one
file. `ResolvingFileIO` keeps its existing fail-fast behavior across delegates.
Tests cover deletion across allowed/denied prefixes, thrown deletes,
rejected/throwing submissions, bounded task counts and concurrency, immutable
executor initialization, and executor configuration for both direct and
registry-created FileIOs.
---
mkdocs/docs/file-io.md | 24 +++
src/iceberg/arrow/arrow_io_internal.h | 21 ++
src/iceberg/arrow/arrow_io_util.h | 9 +
src/iceberg/arrow/s3/arrow_s3_file_io.cc | 177 ++++++++++++++--
src/iceberg/arrow/s3/s3_properties.h | 2 +
src/iceberg/test/arrow_s3_file_io_test.cc | 339 ++++++++++++++++++++++++++++++
src/iceberg/util/executor.h | 6 +-
7 files changed, 553 insertions(+), 25 deletions(-)
diff --git a/mkdocs/docs/file-io.md b/mkdocs/docs/file-io.md
index 5b123add..63ce34de 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 a6b85b6c..f3007b0d 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 925ee12a..b4866386 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 4d26df5a..24b89bdd 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 50dafa56..fe0a9aa3 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 24628949..6bb3f417 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 749502b1..625c833f 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;
