fix(core): resolve ScanContextBuilder executor lazily in Finish()

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};

src/paimon/common/executor/executor.cpp

@@ -52,22 +52,23 @@ 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 {
     return thread_count_;
 }
 
 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();
 }

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;
 }

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