feat: add executor pool support by HuaHuaY · Pull Request #687 · apache/iceberg-cpp

HuaHuaY · 2026-05-28T12:46:59Z

No description provided.

wgtmac

Nice work. The abstraction is clean and the test coverage is solid. Since the default executor is nullopt, existing paths stay single-threaded, so the risk is well contained.

A couple of things before merge:

The PR has no description. Worth adding the motivation and a short design note, ideally with the parallel/serial numbers that justify it.
The RetryRunner change from a runtime fluent API (OnlyRetryOn/StopRetryOn) to the compile-time RetryRunner<Policy> is a breaking API change that's independent of executor support. Consider splitting it into its own PR so it gets reviewed on its own merits.

Left a few inline notes.

wgtmac · 2026-05-29T03:40:37Z

+
+  ExecutorTask executor_task(
+      [promise = std::move(promise), task = std::move(task)]() mutable {
+        promise.set_value(std::move(task)());


If a task throws instead of returning a Status, the promise is never set and the exception escapes on the worker thread, while future.get() sees a broken promise. The repo uses Result/Status rather than exceptions, so this is unlikely in our own code, but Executor is a public extension point and users may plug in pools where it matters. Wrapping the call in try/catch and turning the exception into a Status would be safer.

Errors resulting from violating calling conventions are the user's responsibility, not the library's.

wgtmac · 2026-05-29T03:40:37Z

+    if (!executor_.has_value()) {
+      return internal::RunTasksSingleThreaded(std::move(tasks_));
+    }
+    return internal::RunTasksParallel(executor_->get(), std::move(tasks_));


Run() blocks on future.get() from the calling thread. That's fine for the current sequential call sites, but PlanWith is now public API. If someone drives a TaskGroup from a worker thread of the same bounded executor, it can deadlock (pool saturated while waiting on a task queued behind it). Worth documenting that the driving thread must not be one of the executor's own workers.

The issue of a thread pool's tasks spawning more tasks and then blocking themselves is a problem that I believe any thread pool with a thread limit will encounter, not just a problem with this particular PR.

wgtmac · 2026-05-29T03:40:37Z

-      result[spec_id].push_back(std::move(entry));
+  std::unordered_map<int32_t, std::vector<ManifestEntry>> result;
+  for (auto& manifest_result : manifest_results) {
+    result.merge(manifest_result);


This depends on merge() moving out the non-conflicting nodes and leaving only the conflicting keys, which are then appended. It's correct, but takes a moment to parse. A plain loop doing result[spec_id].insert(...) for every entry reads more directly, at negligible cost.

merge() does not involve memory allocation and is more efficient.

wgtmac

A few more minor notes, non-blocking.

wgtmac · 2026-05-29T03:43:50Z

+  std::vector<FilterManifestResult> filter_results(manifests.size());
+  auto filter_tasks = TaskGroup().SetExecutor(executor_);
+  for (auto&& [manifest, result] : std::views::zip(manifests, filter_results)) {
+    filter_tasks.Submit([&]() -> Status {


This is safe today: the bindings refer to elements of manifests and filter_results, both of which outlive Run(). But capturing a loop's structured bindings by [&] into a deferred closure is an easy footgun. Capturing explicitly (e.g. [&manifest, &result, ...]) makes the intent clearer and harder to break later.

There are too many variables captured by this lambda. It is inappropriate to write the names clearly.

wgtmac

Another pass, this time on the interface design and extensibility.

The overall shape is good: one virtual Executor that engines implement, threaded through the builders via PlanWith. The Arrow adapter test is a nice proof that an external pool drops in with basically one line, so "bring your own threadpool" is well covered.

On future async directions (C++23 coroutines, P2300 std::execution): the model here is synchronous and blocking, TaskGroup::Run() fans out and blocks on std::future::get(), and the planning APIs return Result<...> directly. So this is a parallel-for primitive, not a step toward a sender/receiver pipeline. That's a reasonable scope for now, I'd just flag it explicitly so nobody expects this interface to extend into async later, it'll be a separate one. Details inline.

wgtmac · 2026-05-29T07:08:17Z

+  virtual ~Executor() = default;
+
+  /// \brief Schedule a task for execution.
+  virtual Status Submit(ExecutorTask task) = 0;


This is a fire-and-forget execute-style primitive (closer to the abandoned P0443 executor.execute than to P2300's scheduler/sender). Completion is tracked outside, via the std::promise/future plumbing in RunTasksParallel. Fine for a blocking parallel-for, but it doesn't lay groundwork for coroutines or std::execution: those need the executor to hand back something awaitable/composable, and the planning APIs (PlanFiles() -> Result<...>) are synchronous anyway. Going async later would be a separate interface, not an extension of this one. Worth stating in the header that Executor is a parallel-dispatch primitive, not an async scheduler.

Separately: ExecutorTask being move-only is the right call and matches Arrow, but pools whose submit takes a copyable std::function will need std::move_only_function or a small shim to adapt.

wgtmac · 2026-05-29T07:08:17Z

+  virtual Status Submit(ExecutorTask task) = 0;
+};
+
+using OptionalExecutor = std::optional<std::reference_wrapper<Executor>>;


std::optional<std::reference_wrapper<Executor>> is a little awkward to use (executor_->get() at the call sites). A plain Executor* expresses "nullable, non-owning borrow" just as well and reads cleaner everywhere it's passed. Minor.

Holding a pointer requires a check to determine whether it needs to be destructed, which is detrimental to future code maintainability.

wgtmac · 2026-05-29T07:08:17Z

+  futures.reserve(tasks.size());
+
+  std::vector<Error> errors;
+  for (auto& task : tasks) {


All tasks are submitted up front, each with its own promise/future. For a handful of manifests that's fine, but as a general primitive there's no concurrency bound and no fail-fast: N tasks always queue N and allocate N futures, and if one fails early the rest still run to completion. Probably out of scope for this PR, but worth a note if engines may push large fan-outs through here.

wgtmac · 2026-05-29T07:08:17Z

+  /// The executor is borrowed and must outlive this update. Planning callbacks may be
+  /// called concurrently; callers must synchronize shared mutable state captured by
+  /// those callbacks.
+  auto& ScanManifestsWith(this auto& self, Executor& executor) {


Naming: every other entry point uses PlanWith (TableScanBuilder, ExpireSnapshots, ManifestGroup, ManifestFilterManager, ManifestMergeManager), but this one is ScanManifestsWith. Worth settling on one name for a consistent public surface.

This is the behavior of apache/iceberg repo.

wgtmac

One more pass. The main thing I found is a documentation gap around a new concurrency contract that now leaks onto user-supplied callbacks.

Once an executor is set, the user's ManifestWriterFactory and the shared FileIO get called from multiple worker threads. The tests already account for this (the factories use an atomic counter plus a barrier), so the requirement is understood, it's just not written down anywhere a downstream engine would see it. Worth documenting on the public surface.

wgtmac · 2026-05-29T07:24:36Z

  ManifestMergeManager& operator=(const ManifestMergeManager&) = delete;

+  /// \brief Configure an optional executor for manifest merging.
+  ManifestMergeManager& PlanWith(OptionalExecutor executor);


Once an executor is set here, the user-supplied ManifestWriterFactory gets invoked concurrently from worker threads (parallel FlushBin). The tests already guard against this with an atomic path counter + barrier, which confirms the intent, but the requirement isn't documented. A downstream engine implementing a factory won't know it has to be thread-safe. Worth a \note on this PlanWith (and the other PlanWith entry points / the ManifestWriterFactory typedef) stating the factory must be safe to call concurrently when an executor is configured.

The documentation is on the Executor interface. There are too many PlanWith and putting comments there will result in too much redundancy.

HuaHuaY force-pushed the infra branch 2 times, most recently from 08fa54a to 8ac58e2 Compare May 28, 2026 12:52

feat: add executor pool support

03e8d03

HuaHuaY force-pushed the infra branch from 8ac58e2 to 03e8d03 Compare May 28, 2026 14:39

fix factory thread safe and adjust concept

6694105

HuaHuaY force-pushed the infra branch from 27790b7 to 6694105 Compare May 28, 2026 17:33

HuaHuaY mentioned this pull request May 29, 2026

refactor: harden FileCleanupStrategy with retry and parallel deletes #649

Open

wgtmac reviewed May 29, 2026

View reviewed changes

update comments

14e231c

wgtmac reviewed May 29, 2026

View reviewed changes

Conversation

HuaHuaY commented May 28, 2026

Uh oh!

wgtmac left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wgtmac left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

wgtmac left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wgtmac left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants