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

[wangyong9999]

Purpose

Linked issue: none

ScanContextBuilder::Impl eagerly calls CreateDefaultExecutor() twice per builder
lifetime: once in the member initializer and once more in Reset(), which Finish()
invokes right after building the context. DefaultExecutor spawns
DEFAULT_EXECUTOR_THREAD_COUNT (4) worker threads in its constructor and joins them on
destruction, so every ScanContextBuilder costs eight thread creations even when the
caller supplies its own executor through WithExecutor().

Point-lookup style serving code builds one ScanContextBuilder per key. Measured with
strace -f -c -e trace=clone on such a path, each lookup performed about nine clone
calls and eight of them came from this builder.

This change keeps every context on a DefaultExecutor of its own (no process wide
executor is introduced) and removes the thread churn in two places:

• DefaultExecutor starts its worker threads on the first Add() instead of in the
  constructor. An executor that never receives a task never spawns a thread, and
  shutting down or destroying such an executor only cleans up state. Executors that do
  receive tasks behave as before.
• ScanContextBuilder::Impl::executor_ defaults to null and Reset() only clears it.
  Finish() resolves the executor: the one set through WithExecutor() if any,
  otherwise a fresh CreateDefaultExecutor() owned by the returned context. This
  matches what ReadContextBuilder already does.

With both changes a lookup that supplies its own executor creates no thread at all,
and one that relies on the default scan executor creates the four workers once, when
the scan actually submits work.

Tests

• DefaultExecutorTest.TestWorkersStartOnFirstTask (new): CreateDefaultExecutor(4)
  creates no thread; four appear after the first tasks run; all are gone after the
  executor is destroyed (thread count read from /proc/self/task, Linux only).
• DefaultExecutorTest.TestShutdownWithoutTasks (new): ShutdownNow() and
  destruction of an executor that never ran a task.
• ScanContextTest.TestDefaultExecutorIsCreatedPerContext (new): contexts built
  without WithExecutor() get distinct non-null executors, and an executor set on the
  builder only applies to the Finish() that follows it.
• All 12 DefaultExecutorTest.* cases (10 existing + 2 new) and the 4
  ScanContextTest.* cases pass locally in a Release build (paimon-common-test,
  paimon-core-test).

API and Format

No public API or storage format change. DefaultExecutor worker threads now start on
the first submitted task rather than at construction; this is only observable through
thread counts.

Documentation

No.

Generative AI tooling

Generated-by: Claude Code (Claude Fable 5.1)

[lxy-9602]

Please do not use GlobalDefaultExecutor. It creates a number of threads proportional to the number of CPU cores and may also introduce potential locking issues.

[wangyong9999]

One concrete failure mode is FileStoreScan::ReadFileEntries(): it submits manifest work to the context executor and then blocks in CollectAll(). If all global workers are already running scans, every worker can wait for child work queued to the same pool and deadlock. Please keep the fallback lazy but private (CreateDefaultExecutor() in Finish()), while leaving Reset() null; that removes the eager thread creation without changing executor isolation.

[wangyong9999]

Thanks, agreed: a CPU-proportional shared pool is the wrong tool for a serving path, and FileStoreScan::ReadFileEntries() blocking in CollectAll() on that same pool could deadlock once every worker is inside a scan. Pushed 3475256, which drops GetGlobalDefaultExecutor() entirely:

• DefaultExecutor now starts its workers on the first Add() instead of in the constructor. An executor that never receives a task never spawns a thread, and shutting down or destroying such an executor only cleans up state; executors that do run tasks behave exactly as before. workers_ stays guarded by the existing state_->mutex, no new lock.
• ScanContextBuilder::Finish() falls back to a fresh CreateDefaultExecutor() owned by the returned context, the same as ReadContextBuilder does. Nothing is shared across contexts.

Net effect on the per-lookup path: constructing or resetting a builder creates no thread (previously 8), a context that supplies its own executor creates none at all, and the four default workers only appear once the scan actually submits work. Covered by DefaultExecutorTest.TestWorkersStartOnFirstTask / TestShutdownWithoutTasks and ScanContextTest.TestDefaultExecutorIsCreatedPerContext; the PR description is updated to match.

[wangyong9999]

Once the fallback becomes a private CreateDefaultExecutor() (per the thread above), this test pins the opposite of the new behavior. Worth flipping it to: Finish() without WithExecutor() yields a non-null executor that is not GetGlobalDefaultExecutor() and differs between two contexts; an executor passed to WithExecutor() is used for the next Finish() only. That checks the isolation the maintainer asked for instead of the singleton.

[wangyong9999]

add_test registers the whole paimon-common-test binary, so every DefaultExecutorTest.* (and whatever ran before it) shares this process. A worker joined by the previous test can still show up in /proc/self/task when threads_before is read at line 62 and be gone by here, and the exact == threads_before + 4 / == threads_before checks then fail on an unrelated off-by-one. Capture the baseline the same way the tail does (poll until two reads agree), and relax the comparisons to >= threads_before + 4 after the tasks and <= threads_before after reset; that still proves "no thread at construction, four after the first task, none after destruction" without depending on other tests being quiet.

[wangyong9999]

Good catch, and it is not hypothetical: with the exact comparisons the first local run of this test failed on precisely that window (a worker joined by the previous executor was still listed in /proc/self/task). Addressed in c2d25dc:

• the baseline is now taken with StableProcessThreadCount(), which returns only once two consecutive reads agree;
• the checks are <= threads_before after construction, >= threads_before + 4 after the first tasks, and <= threads_before after destruction (the last one polls briefly, as before).

Validated by running the whole paimon-common-test binary, the same way add_test registers it, plus three filtered DefaultExecutorTest.* runs.