This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 5cc51a7c fix(core): resolve ScanContextBuilder executor lazily in
Finish() (#282)
5cc51a7c is described below
commit 5cc51a7c1466b259b8c37d4f250d321458500f7d
Author: wangyong9999 <[email protected]>
AuthorDate: Mon Sep 7 19:11:00 2026 +0800
fix(core): resolve ScanContextBuilder executor lazily in Finish() (#282)
---
.../common/executor/default_executor_test.cpp | 87 ++++++++++++++++++++++
src/paimon/common/executor/executor.cpp | 16 +++-
src/paimon/core/operation/scan_context.cpp | 12 +--
src/paimon/core/operation/scan_context_test.cpp | 25 +++++++
4 files changed, 131 insertions(+), 9 deletions(-)
diff --git a/src/paimon/common/executor/default_executor_test.cpp
b/src/paimon/common/executor/default_executor_test.cpp
index 07e78681..80775f2f 100644
--- a/src/paimon/common/executor/default_executor_test.cpp
+++ b/src/paimon/common/executor/default_executor_test.cpp
@@ -26,6 +26,10 @@
#include <thread>
#include <vector>
+#ifdef __linux__
+#include <dirent.h>
+#endif
+
#include "gtest/gtest.h"
#include "paimon/common/executor/future.h"
#include "paimon/executor.h"
@@ -35,6 +39,89 @@
namespace paimon::test {
+#ifdef __linux__
+// Number of threads of the current process according to /proc.
+int32_t CountProcessThreads() {
+ DIR* dir = opendir("/proc/self/task");
+ if (dir == nullptr) {
+ return -1;
+ }
+ int32_t count = 0;
+ while (struct dirent* entry = readdir(dir)) {
+ if (entry->d_name[0] != '.') {
+ ++count;
+ }
+ }
+ closedir(dir);
+ return count;
+}
+
+// A worker joined by an earlier test can trail in /proc for a moment, so take
+// the count only once two consecutive reads agree.
+int32_t StableProcessThreadCount() {
+ int32_t last = CountProcessThreads();
+ for (int32_t i = 0; i < 100; ++i) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ int32_t current = CountProcessThreads();
+ if (current == last) {
+ return current;
+ }
+ last = current;
+ }
+ return last;
+}
+#endif
+
+TEST(DefaultExecutorTest, TestWorkersStartOnFirstTask) {
+#ifdef __linux__
+ const int32_t threads_before = StableProcessThreadCount();
+ ASSERT_GT(threads_before, 0);
+#endif
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<Executor> executor,
CreateDefaultExecutor(4));
+ ASSERT_EQ(4u, executor->GetThreadNum());
+#ifdef __linux__
+ // Constructing the executor does not spawn any worker thread.
+ ASSERT_LE(CountProcessThreads(), threads_before);
+#endif
+
+ std::atomic<int64_t> sum = {0};
+ std::vector<std::future<void>> futures;
+ for (int32_t index = 0; index < 8; ++index) {
+ futures.push_back(Via(executor.get(), [&sum]() { sum++; }));
+ }
+ Wait(futures);
+ ASSERT_EQ(8, sum.load());
+#ifdef __linux__
+ // The first task started all four workers.
+ ASSERT_GE(CountProcessThreads(), threads_before + 4);
+#endif
+ executor.reset();
+#ifdef __linux__
+ // Destroying the executor joined them; the joined threads may trail in
+ // /proc for a moment, so poll briefly.
+ int32_t threads_after = CountProcessThreads();
+ for (int32_t i = 0; i < 100 && threads_after > threads_before; ++i) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ threads_after = CountProcessThreads();
+ }
+ ASSERT_LE(threads_after, threads_before);
+#endif
+}
+
+TEST(DefaultExecutorTest, TestShutdownWithoutTasks) {
+ // Shutting down or destroying an executor that never ran a task must not
+ // block or touch workers that were never started.
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<Executor> executor,
CreateDefaultExecutor(2));
+ executor->ShutdownNow();
+ std::atomic<bool> ran = {false};
+ executor->Add([&ran]() { ran = true; });
+ std::this_thread::sleep_for(std::chrono::milliseconds(50));
+ ASSERT_FALSE(ran.load());
+ executor.reset();
+ std::unique_ptr<Executor> idle_executor = CreateDefaultExecutor();
+ idle_executor.reset();
+}
+
TEST(DefaultExecutorTest, TestViaVoidFunc) {
auto executor = GetGlobalDefaultExecutor();
std::atomic<int64_t> sum = {0};
diff --git a/src/paimon/common/executor/executor.cpp
b/src/paimon/common/executor/executor.cpp
index 45253fe1..92aef4a0 100644
--- a/src/paimon/common/executor/executor.cpp
+++ b/src/paimon/common/executor/executor.cpp
@@ -52,15 +52,15 @@ class DefaultExecutor : public Executor {
private:
uint32_t thread_count_;
+ // Guarded by state_->mutex; populated on the first Add().
std::vector<std::thread> workers_;
std::shared_ptr<State> state_ = std::make_shared<State>();
};
DefaultExecutor::DefaultExecutor(uint32_t thread_count) :
thread_count_(thread_count) {
assert(thread_count > 0);
- for (uint32_t i = 0; i < thread_count_; ++i) {
- workers_.emplace_back(&DefaultExecutor::WorkerThread, state_);
- }
+ // Worker threads are started lazily by the first Add(): an executor that
+ // never receives a task never spawns a thread.
}
uint32_t DefaultExecutor::GetThreadNum() const {
@@ -68,6 +68,7 @@ uint32_t DefaultExecutor::GetThreadNum() const {
}
void DefaultExecutor::ShutdownInternal(bool wait_for_pending_tasks) {
+ std::vector<std::thread> workers;
{
std::unique_lock<std::mutex> lock(state_->mutex);
if (state_->stop) {
@@ -80,8 +81,9 @@ void DefaultExecutor::ShutdownInternal(bool
wait_for_pending_tasks) {
state_->tasks.swap(empty);
}
state_->condition.notify_all();
+ workers.swap(workers_);
}
- for (std::thread& worker : workers_) {
+ for (std::thread& worker : workers) {
if (worker.joinable()) {
if (worker.get_id() == std::this_thread::get_id()) {
worker.detach();
@@ -112,6 +114,12 @@ void DefaultExecutor::Add(std::function<void()> func) {
return;
}
state_->tasks.emplace(std::move(func));
+ if (workers_.empty()) {
+ workers_.reserve(thread_count_);
+ for (uint32_t i = 0; i < thread_count_; ++i) {
+ workers_.emplace_back(&DefaultExecutor::WorkerThread, state_);
+ }
+ }
}
state_->condition.notify_one();
}
diff --git a/src/paimon/core/operation/scan_context.cpp
b/src/paimon/core/operation/scan_context.cpp
index 16a80731..0b2ab855 100644
--- a/src/paimon/core/operation/scan_context.cpp
+++ b/src/paimon/core/operation/scan_context.cpp
@@ -67,7 +67,7 @@ class ScanContextBuilder::Impl {
global_index_result_.reset();
realtime_context_.reset();
memory_pool_ = GetDefaultPool();
- executor_ = CreateDefaultExecutor();
+ executor_.reset();
specific_file_system_.reset();
table_schema_ = std::nullopt;
options_.clear();
@@ -84,7 +84,8 @@ class ScanContextBuilder::Impl {
std::shared_ptr<GlobalIndexResult> global_index_result_;
std::shared_ptr<RealtimeContext> realtime_context_;
std::shared_ptr<MemoryPool> memory_pool_ = GetDefaultPool();
- std::shared_ptr<Executor> executor_ = CreateDefaultExecutor();
+ // Resolved in Finish(); a builder never owns an executor of its own.
+ std::shared_ptr<Executor> executor_;
std::shared_ptr<FileSystem> specific_file_system_;
std::optional<std::string> table_schema_;
std::map<std::string, std::string> options_;
@@ -178,13 +179,14 @@ Result<std::unique_ptr<ScanContext>>
ScanContextBuilder::Finish() {
if (impl_->path_.empty()) {
return Status::Invalid("cannot scan with empty table path");
}
+ std::shared_ptr<Executor> executor =
+ impl_->executor_ ? impl_->executor_ : CreateDefaultExecutor();
auto ctx = std::make_unique<ScanContext>(
impl_->path_, impl_->is_streaming_mode_, impl_->limit_,
std::make_shared<ScanFilter>(impl_->predicates_,
impl_->partition_filters_,
impl_->bucket_filter_),
- impl_->global_index_result_, impl_->realtime_context_,
impl_->memory_pool_,
- impl_->executor_, impl_->specific_file_system_, impl_->table_schema_,
impl_->options_,
- impl_->cache_);
+ impl_->global_index_result_, impl_->realtime_context_,
impl_->memory_pool_, executor,
+ impl_->specific_file_system_, impl_->table_schema_, impl_->options_,
impl_->cache_);
impl_->Reset();
return ctx;
}
diff --git a/src/paimon/core/operation/scan_context_test.cpp
b/src/paimon/core/operation/scan_context_test.cpp
index 6c86c049..3b965889 100644
--- a/src/paimon/core/operation/scan_context_test.cpp
+++ b/src/paimon/core/operation/scan_context_test.cpp
@@ -101,4 +101,29 @@ TEST(ScanContextTest, TestSetOptionsOverridesAddedOptions)
{
ASSERT_EQ(expected_options, ctx->GetOptions());
}
+TEST(ScanContextTest, TestDefaultExecutorIsCreatedPerContext) {
+ // A builder without WithExecutor() gives every context a default executor
+ // of its own; nothing is shared across contexts.
+ ScanContextBuilder first_builder("table_root_path");
+ ASSERT_OK_AND_ASSIGN(auto first_ctx, first_builder.Finish());
+ ScanContextBuilder second_builder("table_root_path");
+ ASSERT_OK_AND_ASSIGN(auto second_ctx, second_builder.Finish());
+ ASSERT_TRUE(first_ctx->GetExecutor());
+ ASSERT_TRUE(second_ctx->GetExecutor());
+ ASSERT_NE(first_ctx->GetExecutor(), second_ctx->GetExecutor());
+ // Neither falls back to the process wide singleton.
+ ASSERT_NE(GetGlobalDefaultExecutor(), first_ctx->GetExecutor());
+ ASSERT_NE(GetGlobalDefaultExecutor(), second_ctx->GetExecutor());
+
+ // Finish() resets the builder; an explicit executor set before does not
+ // leak into the next context built from the same builder.
+ std::shared_ptr<Executor> executor = CreateDefaultExecutor();
+ first_builder.WithExecutor(executor);
+ ASSERT_OK_AND_ASSIGN(auto explicit_ctx, first_builder.Finish());
+ ASSERT_EQ(executor, explicit_ctx->GetExecutor());
+ ASSERT_OK_AND_ASSIGN(auto reset_ctx, first_builder.Finish());
+ ASSERT_TRUE(reset_ctx->GetExecutor());
+ ASSERT_NE(executor, reset_ctx->GetExecutor());
+}
+
} // namespace paimon::test
