Re: [PR] feat: add executor pool support [iceberg-cpp]
wgtmac merged PR #687: URL: https://github.com/apache/iceberg-cpp/pull/687 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] feat: add executor pool support [iceberg-cpp]
wgtmac commented on code in PR #687: URL: https://github.com/apache/iceberg-cpp/pull/687#discussion_r3361454453 ## src/iceberg/util/functional.h: ## @@ -0,0 +1,82 @@ +/* + * 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. + */ + +// Borrowed the file from Apache Arrow: +// https://github.com/apache/arrow/blob/main/cpp/src/arrow/util/functional.h Review Comment: We need to update LICENSE file -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] feat: add executor pool support [iceberg-cpp]
HuaHuaY commented on code in PR #687:
URL: https://github.com/apache/iceberg-cpp/pull/687#discussion_r3322982125
##
src/iceberg/manifest/manifest_merge_manager.h:
##
@@ -56,6 +57,9 @@ class ICEBERG_EXPORT ManifestMergeManager {
ManifestMergeManager(const ManifestMergeManager&) = delete;
ManifestMergeManager& operator=(const ManifestMergeManager&) = delete;
+ /// \brief Configure an optional executor for manifest merging.
+ ManifestMergeManager& PlanWith(OptionalExecutor executor);
Review Comment:
The documentation is on the Executor interface. There are too many
`PlanWith` and putting comments there will result in too much redundancy.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] feat: add executor pool support [iceberg-cpp]
HuaHuaY commented on code in PR #687:
URL: https://github.com/apache/iceberg-cpp/pull/687#discussion_r3322984290
##
src/iceberg/delete_file_index.cc:
##
@@ -528,107 +531,159 @@ DeleteFileIndex::Builder&
DeleteFileIndex::Builder::IgnoreResiduals() {
return *this;
}
+DeleteFileIndex::Builder& DeleteFileIndex::Builder::PlanWith(OptionalExecutor
executor) {
+ executor_ = executor;
+ return *this;
+}
+
Result> DeleteFileIndex::Builder::LoadDeleteFiles()
{
- // Build expression caches per spec ID
- std::unordered_map> part_expr_cache;
+ // TODO(zehua): Replace with a thread-safe LRU cache.
+ std::shared_mutex projected_expr_cache_mutex;
+ std::unordered_map>
projected_expr_cache;
+ std::shared_mutex eval_cache_mutex;
std::unordered_map> eval_cache;
auto data_filter = ignore_residuals_ ? True::Instance() : data_filter_;
- // Filter and read manifests into manifest entries
- std::vector files;
- for (const auto& manifest : delete_manifests_) {
-if (manifest.content != ManifestContent::kDeletes) {
- continue;
+ auto and_filters =
+ [](std::shared_ptr left,
+ std::shared_ptr right) ->
Result> {
+if (left && right) {
+ return And::MakeFolded(std::move(left), std::move(right));
}
-if (!manifest.has_added_files() && !manifest.has_existing_files()) {
- continue;
+if (right) {
+ return right;
+}
+return left;
+ };
+
+ auto get_projected_expr = [&](int32_t spec_id,
+const std::shared_ptr& spec)
+ -> Result> {
+if (!data_filter_) {
+ return std::shared_ptr();
}
-const int32_t spec_id = manifest.partition_spec_id;
-auto spec_iter = specs_by_id_.find(spec_id);
-ICEBERG_CHECK(spec_iter != specs_by_id_.cend(),
- "Partition spec ID {} not found when loading delete files",
spec_id);
+{
+ std::shared_lock lock(projected_expr_cache_mutex);
+ auto iter = projected_expr_cache.find(spec_id);
+ if (iter != projected_expr_cache.end()) {
+return iter->second;
+ }
+}
-const auto& spec = spec_iter->second;
+std::lock_guard lock(projected_expr_cache_mutex);
+auto iter = projected_expr_cache.find(spec_id);
+if (iter != projected_expr_cache.end()) {
+ return iter->second;
+}
-// Get or compute projected partition expression
-if (!part_expr_cache.contains(spec_id) && data_filter_) {
- auto projector = Projections::Inclusive(*spec, *schema_,
case_sensitive_);
- ICEBERG_ASSIGN_OR_RAISE(auto projected,
projector->Project(data_filter_));
- part_expr_cache[spec_id] = std::move(projected);
+auto projector = Projections::Inclusive(*spec, *schema_, case_sensitive_);
+ICEBERG_ASSIGN_OR_RAISE(auto projected, projector->Project(data_filter_));
+auto [inserted_iter, _] = projected_expr_cache.emplace(spec_id,
std::move(projected));
+return inserted_iter->second;
+ };
+
+ auto get_manifest_evaluator =
+ [&](int32_t spec_id, const std::shared_ptr& spec,
+ const std::shared_ptr& filter) ->
Result {
+if (!filter) {
+ return nullptr;
}
-// Get or create manifest evaluator
-if (!eval_cache.contains(spec_id)) {
- auto filter = partition_filter_;
- if (auto it = part_expr_cache.find(spec_id); it !=
part_expr_cache.cend()) {
-if (filter) {
- ICEBERG_ASSIGN_OR_RAISE(filter, And::Make(filter, it->second));
-} else {
- filter = it->second;
-}
- }
- if (filter) {
-ICEBERG_ASSIGN_OR_RAISE(auto evaluator,
-ManifestEvaluator::MakePartitionFilter(
-std::move(filter), spec, *schema_,
case_sensitive_));
-eval_cache[spec_id] = std::move(evaluator);
+{
+ std::shared_lock lock(eval_cache_mutex);
+ auto iter = eval_cache.find(spec_id);
+ if (iter != eval_cache.end()) {
+return iter->second.get();
}
}
-// Evaluate manifest against filter
-if (auto it = eval_cache.find(spec_id); it != eval_cache.end()) {
- ICEBERG_ASSIGN_OR_RAISE(auto should_match,
it->second->Evaluate(manifest));
- if (!should_match) {
-continue; // Manifest doesn't match filter
- }
+std::lock_guard lock(eval_cache_mutex);
+auto iter = eval_cache.find(spec_id);
+if (iter != eval_cache.end()) {
+ return iter->second.get();
}
-// Read manifest entries
-ICEBERG_ASSIGN_OR_RAISE(auto reader,
-ManifestReader::Make(manifest, io_, schema_,
spec));
-
-auto partition_filter = partition_filter_;
-if (auto it = part_expr_cache.find(spec_id); it != part_expr_cache.cend())
{
- if (partition_filter) {
-ICEBERG_ASSIGN_OR_RAISE(partition_filter,
-And::Make(partition_filter, it->second));
- } else {
-partition_filter = it->second;
+ICEBERG_
Re: [PR] feat: add executor pool support [iceberg-cpp]
HuaHuaY commented on code in PR #687:
URL: https://github.com/apache/iceberg-cpp/pull/687#discussion_r3322974610
##
src/iceberg/update/snapshot_update.h:
##
@@ -77,6 +79,16 @@ class ICEBERG_EXPORT SnapshotUpdate : public PendingUpdate {
return self;
}
+ /// \brief Configure an executor for manifest planning work.
+ ///
+ /// 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) {
Review Comment:
This is the behavior of `apache/iceberg` repo.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] feat: add executor pool support [iceberg-cpp]
HuaHuaY commented on code in PR #687:
URL: https://github.com/apache/iceberg-cpp/pull/687#discussion_r3322969077
##
src/iceberg/util/executor.h:
##
@@ -0,0 +1,43 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include
+#include
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/util/functional.h"
+
+namespace iceberg {
+
+using ExecutorTask = FnOnce;
+
+class ICEBERG_EXPORT Executor {
+ public:
+ virtual ~Executor() = default;
+
+ /// \brief Schedule a task for execution.
+ virtual Status Submit(ExecutorTask task) = 0;
+};
+
+using OptionalExecutor = std::optional>;
Review Comment:
Holding a pointer requires a check to determine whether it needs to be
destructed, which is detrimental to future code maintainability.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] feat: add executor pool support [iceberg-cpp]
wgtmac commented on code in PR #687:
URL: https://github.com/apache/iceberg-cpp/pull/687#discussion_r3322782516
##
src/iceberg/manifest/manifest_merge_manager.h:
##
@@ -56,6 +57,9 @@ class ICEBERG_EXPORT ManifestMergeManager {
ManifestMergeManager(const ManifestMergeManager&) = delete;
ManifestMergeManager& operator=(const ManifestMergeManager&) = delete;
+ /// \brief Configure an optional executor for manifest merging.
+ ManifestMergeManager& PlanWith(OptionalExecutor executor);
Review Comment:
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.
##
src/iceberg/delete_file_index.cc:
##
@@ -528,107 +531,159 @@ DeleteFileIndex::Builder&
DeleteFileIndex::Builder::IgnoreResiduals() {
return *this;
}
+DeleteFileIndex::Builder& DeleteFileIndex::Builder::PlanWith(OptionalExecutor
executor) {
+ executor_ = executor;
+ return *this;
+}
+
Result> DeleteFileIndex::Builder::LoadDeleteFiles()
{
- // Build expression caches per spec ID
- std::unordered_map> part_expr_cache;
+ // TODO(zehua): Replace with a thread-safe LRU cache.
+ std::shared_mutex projected_expr_cache_mutex;
+ std::unordered_map>
projected_expr_cache;
+ std::shared_mutex eval_cache_mutex;
std::unordered_map> eval_cache;
auto data_filter = ignore_residuals_ ? True::Instance() : data_filter_;
- // Filter and read manifests into manifest entries
- std::vector files;
- for (const auto& manifest : delete_manifests_) {
-if (manifest.content != ManifestContent::kDeletes) {
- continue;
+ auto and_filters =
+ [](std::shared_ptr left,
+ std::shared_ptr right) ->
Result> {
+if (left && right) {
+ return And::MakeFolded(std::move(left), std::move(right));
}
-if (!manifest.has_added_files() && !manifest.has_existing_files()) {
- continue;
+if (right) {
+ return right;
+}
+return left;
+ };
+
+ auto get_projected_expr = [&](int32_t spec_id,
+const std::shared_ptr& spec)
+ -> Result> {
+if (!data_filter_) {
+ return std::shared_ptr();
}
-const int32_t spec_id = manifest.partition_spec_id;
-auto spec_iter = specs_by_id_.find(spec_id);
-ICEBERG_CHECK(spec_iter != specs_by_id_.cend(),
- "Partition spec ID {} not found when loading delete files",
spec_id);
+{
+ std::shared_lock lock(projected_expr_cache_mutex);
+ auto iter = projected_expr_cache.find(spec_id);
+ if (iter != projected_expr_cache.end()) {
+return iter->second;
+ }
+}
-const auto& spec = spec_iter->second;
+std::lock_guard lock(projected_expr_cache_mutex);
+auto iter = projected_expr_cache.find(spec_id);
+if (iter != projected_expr_cache.end()) {
+ return iter->second;
+}
-// Get or compute projected partition expression
-if (!part_expr_cache.contains(spec_id) && data_filter_) {
- auto projector = Projections::Inclusive(*spec, *schema_,
case_sensitive_);
- ICEBERG_ASSIGN_OR_RAISE(auto projected,
projector->Project(data_filter_));
- part_expr_cache[spec_id] = std::move(projected);
+auto projector = Projections::Inclusive(*spec, *schema_, case_sensitive_);
+ICEBERG_ASSIGN_OR_RAISE(auto projected, projector->Project(data_filter_));
+auto [inserted_iter, _] = projected_expr_cache.emplace(spec_id,
std::move(projected));
+return inserted_iter->second;
+ };
+
+ auto get_manifest_evaluator =
+ [&](int32_t spec_id, const std::shared_ptr& spec,
+ const std::shared_ptr& filter) ->
Result {
+if (!filter) {
+ return nullptr;
}
-// Get or create manifest evaluator
-if (!eval_cache.contains(spec_id)) {
- auto filter = partition_filter_;
- if (auto it = part_expr_cache.find(spec_id); it !=
part_expr_cache.cend()) {
-if (filter) {
- ICEBERG_ASSIGN_OR_RAISE(filter, And::Make(filter, it->second));
-} else {
- filter = it->second;
-}
- }
- if (filter) {
-ICEBERG_ASSIGN_OR_RAISE(auto evaluator,
-ManifestEvaluator::MakePartitionFilter(
-std::move(filter), spec, *schema_,
case_sensitive_));
-eval_cache[spec_id] = std::move(evaluator);
+{
+ std::shared_lock lock(eval_cache_mutex);
+ auto iter = eval_cache.find(spec_id);
+ if (iter != eval_cache.end()) {
+return iter->second.get(
Re: [PR] feat: add executor pool support [iceberg-cpp]
wgtmac commented on code in PR #687:
URL: https://github.com/apache/iceberg-cpp/pull/687#discussion_r3322704869
##
src/iceberg/util/executor.h:
##
@@ -0,0 +1,43 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include
+#include
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/util/functional.h"
+
+namespace iceberg {
+
+using ExecutorTask = FnOnce;
+
+class ICEBERG_EXPORT Executor {
+ public:
+ virtual ~Executor() = default;
+
+ /// \brief Schedule a task for execution.
+ virtual Status Submit(ExecutorTask task) = 0;
Review Comment:
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.
##
src/iceberg/util/task_group.cc:
##
@@ -0,0 +1,100 @@
+/*
+ * 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 "iceberg/util/task_group.h"
+
+#include
+#include
+#include
+#include
+
+#include "iceberg/util/macros.h"
+
+namespace iceberg::internal {
+
+namespace {
+
+Status AggregateTaskErrors(std::vector errors) {
+ if (errors.empty()) {
+return {};
+ }
+ if (errors.size() == 1) {
+return std::unexpected(std::move(errors.front()));
+ }
+
+ ErrorKind kind = errors.front().kind;
+ std::string message = std::format("Task group failed with {} errors:",
errors.size());
+ for (const auto& error : errors) {
+message += std::format("\n - {}", error.message);
+ }
+ return std::unexpected(Error{.kind = kind, .message = std::move(message)});
+}
+
+Result> SubmitTask(Executor& executor, FnOnce
task) {
+ std::promise promise;
+ auto future = promise.get_future();
+
+ ExecutorTask executor_task(
+ [promise = std::move(promise), task = std::move(task)]() mutable {
+promise.set_value(std::move(task)());
+ });
+
+ ICEBERG_RETURN_UNEXPECTED(executor.Submit(std::move(executor_task)));
+
+ return future;
+}
+
+} // namespace
+
+Status RunTasksSingleThreaded(std::vector> tasks) {
+ std::vector errors;
+ for (auto& task : tasks) {
+auto status = std::move(task)();
+if (!status.has_value()) {
+ errors.push_back(std::move(status.error()));
+}
+ }
+ return AggregateTaskErrors(std::move(errors));
+}
+
+Status RunTasksParallel(Executor& executor, std::vector>
tasks) {
+ std::vector> futures;
+ futures.reserve(tasks.size());
+
+ std::vector errors;
+ for (auto& task : tasks) {
Review Comment:
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
Re: [PR] feat: add executor pool support [iceberg-cpp]
HuaHuaY commented on code in PR #687:
URL: https://github.com/apache/iceberg-cpp/pull/687#discussion_r3322038965
##
src/iceberg/manifest/manifest_filter_manager.cc:
##
@@ -401,19 +451,35 @@ Result>
ManifestFilterManager::FilterManifests(
}
bool trust_manifest_references = CanTrustManifestReferences(manifests);
- manifest_evaluator_cache_.clear();
- residual_evaluator_cache_.clear();
+ {
+std::lock_guard lock(manifest_evaluator_cache_mutex_);
+manifest_evaluator_cache_.clear();
+ }
+ {
+std::lock_guard lock(residual_evaluator_cache_mutex_);
+residual_evaluator_cache_.clear();
+ }
+
+ std::vector filter_results(manifests.size());
+ auto filter_tasks = TaskGroup().SetExecutor(executor_);
+ for (auto&& [manifest, result] : std::views::zip(manifests, filter_results))
{
+filter_tasks.Submit([&]() -> Status {
Review Comment:
There are too many variables captured by this lambda. It is inappropriate to
write the names clearly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] feat: add executor pool support [iceberg-cpp]
HuaHuaY commented on code in PR #687:
URL: https://github.com/apache/iceberg-cpp/pull/687#discussion_r3322023058
##
src/iceberg/util/task_group.h:
##
@@ -0,0 +1,109 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include
+#include
+#include
+#include
+#include
+#include
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/util/executor.h"
+#include "iceberg/util/functional.h"
+#include "iceberg/util/retry_util.h"
+
+namespace iceberg {
+
+namespace internal {
+
+template
+concept OnceStatusTask = RvalueInvocable;
+
+template
+concept RepeatableStatusTask =
+std::is_invocable_r_v ||
+(std::copy_constructible && std::is_invocable_r_v);
+
+template
+concept RetryableStatusTask = std::constructible_from,
F> &&
+ RepeatableStatusTask>;
+
+ICEBERG_EXPORT Status RunTasksSingleThreaded(std::vector>
tasks);
+
+ICEBERG_EXPORT Status RunTasksParallel(Executor& executor,
+ std::vector> tasks);
+
+} // namespace internal
+
+template
+class ICEBERG_TEMPLATE_CLASS_EXPORT TaskGroup {
+ private:
+ static constexpr bool kRetryEnabled = !std::same_as;
+
+ struct Empty {};
+
+ using RetryConfigStorage = std::conditional_t;
+
+ public:
+ TaskGroup() = default;
+
+ explicit TaskGroup(RetryConfig retry_config)
+requires(kRetryEnabled)
+ : retry_config_(std::move(retry_config)) {}
+
+ auto&& SetExecutor(this auto&& self, OptionalExecutor executor) {
+self.executor_ = std::move(executor);
+return std::forward(self);
+ }
+
+ template
+requires((!kRetryEnabled && internal::OnceStatusTask) ||
+ (kRetryEnabled && internal::RetryableStatusTask))
+ auto&& Submit(this auto&& self, F&& task) {
+self.tasks_.emplace_back([&] {
+ if constexpr (!kRetryEnabled) {
+return std::forward(task);
+ } else {
+return [retry_config = self.retry_config_,
+task = std::forward(task)]() mutable -> Status {
+ return RetryRunner(retry_config).Run(task);
+};
+ }
+}());
+return std::forward(self);
+ }
+
+ Status Run() && {
+if (!executor_.has_value()) {
+ return internal::RunTasksSingleThreaded(std::move(tasks_));
+}
+return internal::RunTasksParallel(executor_->get(), std::move(tasks_));
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] feat: add executor pool support [iceberg-cpp]
HuaHuaY commented on code in PR #687:
URL: https://github.com/apache/iceberg-cpp/pull/687#discussion_r3322011268
##
src/iceberg/manifest/manifest_group.cc:
##
@@ -376,57 +399,73 @@ ManifestGroup::ReadEntries() {
Evaluator::Make(*DataFileFilterSchema(), file_filter_,
case_sensitive_));
}
- std::unordered_map> result;
+ std::vector>>
manifest_results(
+ data_manifests_.size());
- // TODO(gangwu): Parallelize reading manifests
- for (const auto& manifest : data_manifests_) {
-const int32_t spec_id = manifest.partition_spec_id;
+ auto read_tasks = TaskGroup().SetExecutor(executor_);
+ for (auto&& [manifest, manifest_result] :
+ std::views::zip(data_manifests_, manifest_results)) {
+read_tasks.Submit([&]() -> Status {
+ const int32_t spec_id = manifest.partition_spec_id;
-ICEBERG_ASSIGN_OR_RAISE(auto manifest_evaluator,
get_manifest_evaluator(spec_id));
-ICEBERG_ASSIGN_OR_RAISE(bool should_match,
manifest_evaluator->Evaluate(manifest));
-if (!should_match) {
- // Skip this manifest because it doesn't match partition filter
- continue;
-}
+ ICEBERG_ASSIGN_OR_RAISE(auto manifest_evaluator,
get_manifest_evaluator(spec_id));
+ ICEBERG_ASSIGN_OR_RAISE(bool should_match,
manifest_evaluator->Evaluate(manifest));
+ if (!should_match) {
+// Skip this manifest because it doesn't match partition filter
+return {};
+ }
-if (ignore_deleted_) {
- // only scan manifests that have entries other than deletes
- if (!manifest.has_added_files() && !manifest.has_existing_files()) {
-continue;
+ if (ignore_deleted_) {
+// only scan manifests that have entries other than deletes
+if (!manifest.has_added_files() && !manifest.has_existing_files()) {
+ return {};
+}
}
-}
-if (ignore_existing_) {
- // only scan manifests that have entries other than existing
- if (!manifest.has_added_files() && !manifest.has_deleted_files()) {
-continue;
+ if (ignore_existing_) {
+// only scan manifests that have entries other than existing
+if (!manifest.has_added_files() && !manifest.has_deleted_files()) {
+ return {};
+}
}
-}
-// Read manifest entries
-ICEBERG_ASSIGN_OR_RAISE(auto reader, MakeReader(manifest));
-ICEBERG_ASSIGN_OR_RAISE(auto entries,
-ignore_deleted_ ? reader->LiveEntries() :
reader->Entries());
+ // Read manifest entries
+ ICEBERG_ASSIGN_OR_RAISE(auto reader, MakeReader(manifest));
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto entries, ignore_deleted_ ? reader->LiveEntries() :
reader->Entries());
-for (auto& entry : entries) {
- if (ignore_existing_ && entry.status == ManifestStatus::kExisting) {
-continue;
- }
+ for (auto& entry : entries) {
+if (ignore_existing_ && entry.status == ManifestStatus::kExisting) {
+ continue;
+}
- if (data_file_evaluator != nullptr) {
-DataFileStructLike data_file(*entry.data_file);
-ICEBERG_ASSIGN_OR_RAISE(bool should_match,
-data_file_evaluator->Evaluate(data_file));
-if (!should_match) {
+if (data_file_evaluator != nullptr) {
+ DataFileStructLike data_file(*entry.data_file);
+ ICEBERG_ASSIGN_OR_RAISE(bool should_match,
+ data_file_evaluator->Evaluate(data_file));
+ if (!should_match) {
+continue;
+ }
+}
+
+if (!manifest_entry_predicate_(entry)) {
continue;
}
- }
- if (!manifest_entry_predicate_(entry)) {
-continue;
+manifest_result[spec_id].push_back(std::move(entry));
}
+ return {};
+});
+ }
+ ICEBERG_RETURN_UNEXPECTED(std::move(read_tasks).Run());
- result[spec_id].push_back(std::move(entry));
+ std::unordered_map> result;
+ for (auto& manifest_result : manifest_results) {
+result.merge(manifest_result);
Review Comment:
`merge()` does not involve memory allocation and is more efficient.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] feat: add executor pool support [iceberg-cpp]
wgtmac commented on code in PR #687:
URL: https://github.com/apache/iceberg-cpp/pull/687#discussion_r3321997084
##
src/iceberg/update/snapshot_update.cc:
##
@@ -250,13 +251,17 @@ Result
SnapshotUpdate::Apply() {
}
ICEBERG_ASSIGN_OR_RAISE(auto manifests, Apply(base(), parent_snapshot));
+ auto metadata_tasks = TaskGroup().SetExecutor(plan_executor_);
for (auto& manifest : manifests) {
if (manifest.added_snapshot_id != kInvalidSnapshotId) {
continue;
}
-// TODO(xxx): read in parallel and cache enriched manifests for retries
-ICEBERG_ASSIGN_OR_RAISE(manifest, AddMetadata(manifest, ctx_->table->io(),
base()));
+metadata_tasks.Submit([&manifest, this]() -> Status {
Review Comment:
This one is nicely done: each task captures a distinct element of
`manifests` (not an iterator or index), so there's no aliasing, and the
`manifest_count_`/`attempt_` switch to atomics is backed by the
`ConcurrentManifestPaths` test.
##
src/iceberg/manifest/manifest_filter_manager.cc:
##
@@ -401,19 +451,35 @@ Result>
ManifestFilterManager::FilterManifests(
}
bool trust_manifest_references = CanTrustManifestReferences(manifests);
- manifest_evaluator_cache_.clear();
- residual_evaluator_cache_.clear();
+ {
+std::lock_guard lock(manifest_evaluator_cache_mutex_);
+manifest_evaluator_cache_.clear();
+ }
+ {
+std::lock_guard lock(residual_evaluator_cache_mutex_);
+residual_evaluator_cache_.clear();
+ }
+
+ std::vector filter_results(manifests.size());
+ auto filter_tasks = TaskGroup().SetExecutor(executor_);
+ for (auto&& [manifest, result] : std::views::zip(manifests, filter_results))
{
+filter_tasks.Submit([&]() -> Status {
Review Comment:
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.
##
src/iceberg/util/retry_util.h:
##
@@ -69,76 +53,104 @@ struct ICEBERG_EXPORT RetryConfig {
double scale_factor = 2.0;
};
-/// \brief Utility class for running tasks with retry logic
-///
-/// When retries are enabled (`num_retries > 0`), callers must explicitly
configure
-/// retry policy with `OnlyRetryOn(...)` or `StopRetryOn(...)`.
-class ICEBERG_EXPORT RetryRunner {
- public:
- /// \brief Construct a RetryRunner with the given configuration
- explicit RetryRunner(RetryConfig config = {}) : config_(std::move(config)) {}
+namespace detail {
- /// \brief Specify error types that should trigger a retry.
- ///
- /// When set, only errors matching one of these kinds will be retried.
- /// All other errors will stop retries immediately.
- ///
- /// \note OnlyRetryOn takes priority over StopRetryOn. If OnlyRetryOn is set,
- /// StopRetryOn is ignored.
- RetryRunner& OnlyRetryOn(std::initializer_list error_kinds) {
-retry_policy_mode_ = RetryPolicyMode::kOnlyRetryOn;
-retry_error_kinds_ = std::vector(error_kinds);
-return *this;
- }
+class ICEBERG_EXPORT RetryRunnerBase {
+ protected:
+ explicit RetryRunnerBase(RetryConfig config) : config_(std::move(config)) {}
- /// \brief Specify a single error type that should trigger a retry.
- ///
- /// \note OnlyRetryOn takes priority over StopRetryOn. If OnlyRetryOn is set,
- /// StopRetryOn is ignored.
- RetryRunner& OnlyRetryOn(ErrorKind error_kind) { return
OnlyRetryOn({error_kind}); }
+ using Clock = std::chrono::steady_clock;
+ using Duration = std::chrono::milliseconds;
+ using TimePoint = Clock::time_point;
- /// \brief Specify error types that should stop retries immediately.
- ///
- /// When set, errors matching one of these kinds will not be retried.
- /// All other errors will be retried.
- ///
- /// \note OnlyRetryOn takes priority over StopRetryOn. If OnlyRetryOn is set,
- /// StopRetryOn is ignored.
- RetryRunner& StopRetryOn(std::initializer_list error_kinds) {
-if (retry_policy_mode_ == RetryPolicyMode::kOnlyRetryOn) {
- return *this;
-}
+ /// \brief Validate retry counts and timing bounds.
+ Status ValidateConfig() const;
+ std::optional ComputeDeadline() const;
+ bool HasTimedOut(const std::optional& deadline) const;
+ std::optional RetryDelayWithinBudget(
+ int32_t attempt, const std::optional& deadline) const;
+ bool WaitForNextAttempt(int32_t attempt,
+ const std::optional& deadline) const;
+ /// \brief Calculate delay with exponential backoff and jitter
+ int32_t CalculateDelay(int32_t attempt) const;
+
+ RetryConfig config_;
+};
+
+} // namespace detail
+
+namespace retry {
+
+enum class RetryPolicyMode {
+ kNoRetry,
+ kOnlyRetryOn,
+ kStopRetryOn,
+};
-retry_policy_mode_ = RetryPolicyMode::kStopRetryOn;
-retry_error_kinds_ = std::vector(error_kinds);
-
Re: [PR] feat: add executor pool support [iceberg-cpp]
HuaHuaY commented on code in PR #687:
URL: https://github.com/apache/iceberg-cpp/pull/687#discussion_r3321997571
##
src/iceberg/util/task_group.cc:
##
@@ -0,0 +1,100 @@
+/*
+ * 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 "iceberg/util/task_group.h"
+
+#include
+#include
+#include
+#include
+
+#include "iceberg/util/macros.h"
+
+namespace iceberg::internal {
+
+namespace {
+
+Status AggregateTaskErrors(std::vector errors) {
+ if (errors.empty()) {
+return {};
+ }
+ if (errors.size() == 1) {
+return std::unexpected(std::move(errors.front()));
+ }
+
+ ErrorKind kind = errors.front().kind;
+ std::string message = std::format("Task group failed with {} errors:",
errors.size());
+ for (const auto& error : errors) {
+message += std::format("\n - {}", error.message);
+ }
+ return std::unexpected(Error{.kind = kind, .message = std::move(message)});
+}
+
+Result> SubmitTask(Executor& executor, FnOnce
task) {
+ std::promise promise;
+ auto future = promise.get_future();
+
+ ExecutorTask executor_task(
+ [promise = std::move(promise), task = std::move(task)]() mutable {
+promise.set_value(std::move(task)());
Review Comment:
Errors resulting from violating calling conventions are the user's
responsibility, not the library's.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] feat: add executor pool support [iceberg-cpp]
wgtmac commented on code in PR #687:
URL: https://github.com/apache/iceberg-cpp/pull/687#discussion_r3321987627
##
src/iceberg/manifest/manifest_group.cc:
##
@@ -376,57 +399,73 @@ ManifestGroup::ReadEntries() {
Evaluator::Make(*DataFileFilterSchema(), file_filter_,
case_sensitive_));
}
- std::unordered_map> result;
+ std::vector>>
manifest_results(
+ data_manifests_.size());
- // TODO(gangwu): Parallelize reading manifests
- for (const auto& manifest : data_manifests_) {
-const int32_t spec_id = manifest.partition_spec_id;
+ auto read_tasks = TaskGroup().SetExecutor(executor_);
+ for (auto&& [manifest, manifest_result] :
+ std::views::zip(data_manifests_, manifest_results)) {
+read_tasks.Submit([&]() -> Status {
+ const int32_t spec_id = manifest.partition_spec_id;
-ICEBERG_ASSIGN_OR_RAISE(auto manifest_evaluator,
get_manifest_evaluator(spec_id));
-ICEBERG_ASSIGN_OR_RAISE(bool should_match,
manifest_evaluator->Evaluate(manifest));
-if (!should_match) {
- // Skip this manifest because it doesn't match partition filter
- continue;
-}
+ ICEBERG_ASSIGN_OR_RAISE(auto manifest_evaluator,
get_manifest_evaluator(spec_id));
+ ICEBERG_ASSIGN_OR_RAISE(bool should_match,
manifest_evaluator->Evaluate(manifest));
+ if (!should_match) {
+// Skip this manifest because it doesn't match partition filter
+return {};
+ }
-if (ignore_deleted_) {
- // only scan manifests that have entries other than deletes
- if (!manifest.has_added_files() && !manifest.has_existing_files()) {
-continue;
+ if (ignore_deleted_) {
+// only scan manifests that have entries other than deletes
+if (!manifest.has_added_files() && !manifest.has_existing_files()) {
+ return {};
+}
}
-}
-if (ignore_existing_) {
- // only scan manifests that have entries other than existing
- if (!manifest.has_added_files() && !manifest.has_deleted_files()) {
-continue;
+ if (ignore_existing_) {
+// only scan manifests that have entries other than existing
+if (!manifest.has_added_files() && !manifest.has_deleted_files()) {
+ return {};
+}
}
-}
-// Read manifest entries
-ICEBERG_ASSIGN_OR_RAISE(auto reader, MakeReader(manifest));
-ICEBERG_ASSIGN_OR_RAISE(auto entries,
-ignore_deleted_ ? reader->LiveEntries() :
reader->Entries());
+ // Read manifest entries
+ ICEBERG_ASSIGN_OR_RAISE(auto reader, MakeReader(manifest));
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto entries, ignore_deleted_ ? reader->LiveEntries() :
reader->Entries());
-for (auto& entry : entries) {
- if (ignore_existing_ && entry.status == ManifestStatus::kExisting) {
-continue;
- }
+ for (auto& entry : entries) {
+if (ignore_existing_ && entry.status == ManifestStatus::kExisting) {
+ continue;
+}
- if (data_file_evaluator != nullptr) {
-DataFileStructLike data_file(*entry.data_file);
-ICEBERG_ASSIGN_OR_RAISE(bool should_match,
-data_file_evaluator->Evaluate(data_file));
-if (!should_match) {
+if (data_file_evaluator != nullptr) {
+ DataFileStructLike data_file(*entry.data_file);
+ ICEBERG_ASSIGN_OR_RAISE(bool should_match,
+ data_file_evaluator->Evaluate(data_file));
+ if (!should_match) {
+continue;
+ }
+}
+
+if (!manifest_entry_predicate_(entry)) {
continue;
}
- }
- if (!manifest_entry_predicate_(entry)) {
-continue;
+manifest_result[spec_id].push_back(std::move(entry));
}
+ return {};
+});
+ }
+ ICEBERG_RETURN_UNEXPECTED(std::move(read_tasks).Run());
- result[spec_id].push_back(std::move(entry));
+ std::unordered_map> result;
+ for (auto& manifest_result : manifest_results) {
+result.merge(manifest_result);
Review Comment:
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.
##
src/iceberg/util/task_group.h:
##
@@ -0,0 +1,109 @@
+/*
+ * 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
