From 62e85f739af572b1a613167581bfdb8cc834ffdc Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Sun, 6 Sep 2026 23:05:54 -0400 Subject: [PATCH 1/4] fix(core): resolve ScanContextBuilder executor lazily in Finish() ScanContextBuilder::Impl created a DefaultExecutor in its member initializer and again in Reset(), spawning and joining eight worker threads per builder even when the caller supplied an executor. Keep the executor unset until Finish(), which falls back to GetGlobalDefaultExecutor() when nothing was set, so ScanContext still carries a non-null executor. --- src/paimon/core/operation/scan_context.cpp | 12 ++++++----- .../core/operation/scan_context_test.cpp | 21 +++++++++++++++++++ 2 files changed, 28 insertions(+), 5 deletions(-) diff --git a/src/paimon/core/operation/scan_context.cpp b/src/paimon/core/operation/scan_context.cpp index 16a80731d..01cea9fc8 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 global_index_result_; std::shared_ptr realtime_context_; std::shared_ptr memory_pool_ = GetDefaultPool(); - std::shared_ptr executor_ = CreateDefaultExecutor(); + // Lazily resolved in Finish(): a builder must not spawn executor threads. + std::shared_ptr executor_; std::shared_ptr specific_file_system_; std::optional table_schema_; std::map options_; @@ -178,13 +179,14 @@ Result> ScanContextBuilder::Finish() { if (impl_->path_.empty()) { return Status::Invalid("cannot scan with empty table path"); } + std::shared_ptr executor = + impl_->executor_ ? impl_->executor_ : GetGlobalDefaultExecutor(); auto ctx = std::make_unique( impl_->path_, impl_->is_streaming_mode_, impl_->limit_, std::make_shared(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 6c86c0498..c6c4f783c 100644 --- a/src/paimon/core/operation/scan_context_test.cpp +++ b/src/paimon/core/operation/scan_context_test.cpp @@ -101,4 +101,25 @@ TEST(ScanContextTest, TestSetOptionsOverridesAddedOptions) { ASSERT_EQ(expected_options, ctx->GetOptions()); } +TEST(ScanContextTest, TestDefaultExecutorIsGlobalSingleton) { + // A builder without WithExecutor() must not own executor threads of its + // own: every context falls back to the process wide default executor. + 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_EQ(GetGlobalDefaultExecutor(), first_ctx->GetExecutor()); + ASSERT_EQ(first_ctx->GetExecutor(), 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 = 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_EQ(GetGlobalDefaultExecutor(), reset_ctx->GetExecutor()); +} + } // namespace paimon::test From 90a152310bcd0c5926775f7958ac67a830466511 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Sun, 6 Sep 2026 23:05:54 -0400 Subject: [PATCH 2/4] fix(common): start DefaultExecutor workers on the first task Review feedback on #282: do not fall back to the process wide GetGlobalDefaultExecutor() in ScanContextBuilder::Finish(). Each context keeps a DefaultExecutor of its own again; the thread churn is removed by having DefaultExecutor spawn its workers on the first Add() instead of in the constructor, so an executor that never receives a task never creates a thread. --- .../common/executor/default_executor_test.cpp | 70 +++++++++++++++++++ src/paimon/common/executor/executor.cpp | 16 +++-- src/paimon/core/operation/scan_context.cpp | 4 +- .../core/operation/scan_context_test.cpp | 13 ++-- 4 files changed, 91 insertions(+), 12 deletions(-) diff --git a/src/paimon/common/executor/default_executor_test.cpp b/src/paimon/common/executor/default_executor_test.cpp index 07e78681b..a9bbae559 100644 --- a/src/paimon/common/executor/default_executor_test.cpp +++ b/src/paimon/common/executor/default_executor_test.cpp @@ -26,6 +26,10 @@ #include #include +#ifdef __linux__ +#include +#endif + #include "gtest/gtest.h" #include "paimon/common/executor/future.h" #include "paimon/executor.h" @@ -35,6 +39,72 @@ 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; +} +#endif + +TEST(DefaultExecutorTest, TestWorkersStartOnFirstTask) { +#ifdef __linux__ + int32_t threads_before = CountProcessThreads(); + ASSERT_GT(threads_before, 0); +#endif + ASSERT_OK_AND_ASSIGN(std::unique_ptr executor, CreateDefaultExecutor(4)); + ASSERT_EQ(4u, executor->GetThreadNum()); +#ifdef __linux__ + // Constructing the executor does not spawn any worker thread. + ASSERT_EQ(threads_before, CountProcessThreads()); +#endif + + std::atomic sum = {0}; + std::vector> 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__ + ASSERT_EQ(threads_before + 4, CountProcessThreads()); +#endif + executor.reset(); +#ifdef __linux__ + // A joined worker 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_EQ(threads_before, threads_after); +#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, CreateDefaultExecutor(2)); + executor->ShutdownNow(); + std::atomic 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 idle_executor = CreateDefaultExecutor(); + idle_executor.reset(); +} + TEST(DefaultExecutorTest, TestViaVoidFunc) { auto executor = GetGlobalDefaultExecutor(); std::atomic sum = {0}; diff --git a/src/paimon/common/executor/executor.cpp b/src/paimon/common/executor/executor.cpp index 45253fe1f..92aef4a09 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 workers_; std::shared_ptr state_ = std::make_shared(); }; 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 workers; { std::unique_lock 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 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 01cea9fc8..0b2ab8558 100644 --- a/src/paimon/core/operation/scan_context.cpp +++ b/src/paimon/core/operation/scan_context.cpp @@ -84,7 +84,7 @@ class ScanContextBuilder::Impl { std::shared_ptr global_index_result_; std::shared_ptr realtime_context_; std::shared_ptr memory_pool_ = GetDefaultPool(); - // Lazily resolved in Finish(): a builder must not spawn executor threads. + // Resolved in Finish(); a builder never owns an executor of its own. std::shared_ptr executor_; std::shared_ptr specific_file_system_; std::optional table_schema_; @@ -180,7 +180,7 @@ Result> ScanContextBuilder::Finish() { return Status::Invalid("cannot scan with empty table path"); } std::shared_ptr executor = - impl_->executor_ ? impl_->executor_ : GetGlobalDefaultExecutor(); + impl_->executor_ ? impl_->executor_ : CreateDefaultExecutor(); auto ctx = std::make_unique( impl_->path_, impl_->is_streaming_mode_, impl_->limit_, std::make_shared(impl_->predicates_, impl_->partition_filters_, diff --git a/src/paimon/core/operation/scan_context_test.cpp b/src/paimon/core/operation/scan_context_test.cpp index c6c4f783c..b977b32d8 100644 --- a/src/paimon/core/operation/scan_context_test.cpp +++ b/src/paimon/core/operation/scan_context_test.cpp @@ -101,16 +101,16 @@ TEST(ScanContextTest, TestSetOptionsOverridesAddedOptions) { ASSERT_EQ(expected_options, ctx->GetOptions()); } -TEST(ScanContextTest, TestDefaultExecutorIsGlobalSingleton) { - // A builder without WithExecutor() must not own executor threads of its - // own: every context falls back to the process wide default executor. +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_EQ(GetGlobalDefaultExecutor(), first_ctx->GetExecutor()); - ASSERT_EQ(first_ctx->GetExecutor(), second_ctx->GetExecutor()); + ASSERT_TRUE(second_ctx->GetExecutor()); + ASSERT_NE(first_ctx->GetExecutor(), second_ctx->GetExecutor()); // Finish() resets the builder; an explicit executor set before does not // leak into the next context built from the same builder. @@ -119,7 +119,8 @@ TEST(ScanContextTest, TestDefaultExecutorIsGlobalSingleton) { 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_EQ(GetGlobalDefaultExecutor(), reset_ctx->GetExecutor()); + ASSERT_TRUE(reset_ctx->GetExecutor()); + ASSERT_NE(executor, reset_ctx->GetExecutor()); } } // namespace paimon::test From 1c63cdd1ad88056c95a7fc1374ddef6678ac417f Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Sun, 6 Sep 2026 23:05:54 -0400 Subject: [PATCH 3/4] test(common): make the lazy worker test robust to trailing threads paimon-common-test runs every suite in one process, so a worker joined by an earlier test can still be listed in /proc/self/task when the baseline is read. Take the baseline once two consecutive reads agree and compare with <= / >= instead of exact counts. --- .../common/executor/default_executor_test.cpp | 29 +++++++++++++++---- 1 file changed, 23 insertions(+), 6 deletions(-) diff --git a/src/paimon/common/executor/default_executor_test.cpp b/src/paimon/common/executor/default_executor_test.cpp index a9bbae559..80775f2f6 100644 --- a/src/paimon/common/executor/default_executor_test.cpp +++ b/src/paimon/common/executor/default_executor_test.cpp @@ -55,18 +55,33 @@ int32_t CountProcessThreads() { 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__ - int32_t threads_before = CountProcessThreads(); + const int32_t threads_before = StableProcessThreadCount(); ASSERT_GT(threads_before, 0); #endif ASSERT_OK_AND_ASSIGN(std::unique_ptr executor, CreateDefaultExecutor(4)); ASSERT_EQ(4u, executor->GetThreadNum()); #ifdef __linux__ // Constructing the executor does not spawn any worker thread. - ASSERT_EQ(threads_before, CountProcessThreads()); + ASSERT_LE(CountProcessThreads(), threads_before); #endif std::atomic sum = {0}; @@ -77,17 +92,19 @@ TEST(DefaultExecutorTest, TestWorkersStartOnFirstTask) { Wait(futures); ASSERT_EQ(8, sum.load()); #ifdef __linux__ - ASSERT_EQ(threads_before + 4, CountProcessThreads()); + // The first task started all four workers. + ASSERT_GE(CountProcessThreads(), threads_before + 4); #endif executor.reset(); #ifdef __linux__ - // A joined worker may trail in /proc for a moment, so poll briefly. + // 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) { + 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_EQ(threads_before, threads_after); + ASSERT_LE(threads_after, threads_before); #endif } From e1412ef2e26ac1296bef6605cc158d44c99c648d Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Sun, 6 Sep 2026 23:05:54 -0400 Subject: [PATCH 4/4] test(core): assert the scan default executor is not the global one Pin the isolation explicitly: a context built without WithExecutor() must not fall back to GetGlobalDefaultExecutor(). --- src/paimon/core/operation/scan_context_test.cpp | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/paimon/core/operation/scan_context_test.cpp b/src/paimon/core/operation/scan_context_test.cpp index b977b32d8..3b9658892 100644 --- a/src/paimon/core/operation/scan_context_test.cpp +++ b/src/paimon/core/operation/scan_context_test.cpp @@ -111,6 +111,9 @@ TEST(ScanContextTest, TestDefaultExecutorIsCreatedPerContext) { 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.