[fix](executor) Fix rare self-deadlock that can cause the time-sharing task executor to hang. (apache#58273)

1. Fix a self-deadlock that can occur in `_dispatch_thread` when a
destructor path attempts to re-acquire the same mutex already held by
the thread. The root cause is destructors (triggered while holding
`_lock`) performing operations that try to re-acquire the same mutex. A
safe fix should ensure destructors that may call `remove_task()` run
outside the `_lock` scope or avoid re-locking the same mutex inside
destructor paths.
2. Call `_task_executor->wait()` in
`TaskExecutorSimplifiedScanScheduler::stop()`.

---

1. `std::shared_ptr<PrioritizedSplitRunner> split =
_tokenless->_entries->take();`
3. `l.lock();` — `_dispatch_thread` acquires `_lock`.
4. After the `while` loop finishes, `split` goes out of scope and the
`shared_ptr` is destroyed.
5. `PrioritizedSplitRunner` destructor runs → destroys `_split_runner`
(`ScannerSplitRunner`).
6. `ScannerSplitRunner::_scan_func` destructor runs → destroys captured
`ctx` (`std::shared_ptr<ScannerContext>`).
7. `ScannerContext::~ScannerContext()` calls `remove_task()`.
8. `remove_task()` attempts to acquire `_lock`.
9. Result: **self-deadlock** because `_lock` is already held by
`_dispatch_thread`.

---

We must explicitly release 'split' BEFORE acquiring _lock to avoid
self-deadlock. The destructor chain (`PrioritizedSplitRunner` ->
`ScannerSplitRunner`-> `_scan_func` lambda -> captured `ScannerContext`)
may call `remove_task()` which tries to acquire `_lock`. Since `_lock`
is not a recursive mutex, this would deadlock.

Author: kaka11chen

be/src/vec/exec/executor/task_executor.h

@@ -37,6 +37,7 @@ class TaskExecutor {
     virtual Status init() = 0;
     virtual Status start() = 0;
     virtual void stop() = 0;
+    virtual void wait() = 0;
 
     virtual Result<std::shared_ptr<TaskHandle>> create_task(
             const TaskId& task_id, std::function<double()> utilization_supplier,

be/src/vec/exec/executor/time_sharing/time_sharing_task_executor.cpp

@@ -562,10 +562,6 @@ void TimeSharingTaskExecutor::_dispatch_thread() {
             std::lock_guard<std::mutex> guard(_mutex);
             _running_splits.insert(split);
         }
-        Defer defer {[&]() {
-            std::lock_guard<std::mutex> guard(_mutex);
-            _running_splits.erase(split);
-        }};
 
         Result<SharedListenableFuture<Void>> blocked_future_result = split->process();
 
@@ -577,10 +573,6 @@ void TimeSharingTaskExecutor::_dispatch_thread() {
             auto blocked_future = blocked_future_result.value();
 
             if (split->is_finished()) {
-                {
-                    std::ostringstream _oss;
-                    _oss << std::this_thread::get_id();
-                }
                 _split_finished(split, split->finished_status());
             } else {
                 if (split->is_auto_reschedule()) {
@@ -625,6 +617,17 @@ void TimeSharingTaskExecutor::_dispatch_thread() {
         // In the worst case, the destructor might even try to do something
         // with this SplitThreadPool, and produce a deadlock.
         // task.runnable.reset();
+
+        // IMPORTANT: We must explicitly release 'split' BEFORE acquiring _lock to avoid
+        // self-deadlock. The destructor chain (PrioritizedSplitRunner -> ScannerSplitRunner
+        // -> _scan_func lambda -> captured ScannerContext) may call remove_task() which
+        // tries to acquire _lock. Since _lock is not a recursive mutex, this would deadlock.
+        {
+            std::lock_guard<std::mutex> guard(_mutex);
+            _running_splits.erase(split);
+        }
+        split.reset();
+
         l.lock();
         thread_pool_task_execution_time_ns_total->increment(
                 task_execution_time_watch.elapsed_time());

be/src/vec/exec/executor/time_sharing/time_sharing_task_executor.h

@@ -97,6 +97,7 @@ class TimeSharingTaskExecutor : public TaskExecutor {
 
     Status start() override;
     void stop() override;
+    void wait() override;
 
     Result<std::shared_ptr<TaskHandle>> create_task(
             const TaskId& task_id, std::function<double()> utilization_supplier,
@@ -245,8 +246,6 @@ class TimeSharingTaskExecutor : public TaskExecutor {
             std::unique_lock<std::mutex>& lock);
     void _record_leaf_splits_size(std::unique_lock<std::mutex>& lock);
     void _split_finished(std::shared_ptr<PrioritizedSplitRunner> split, const Status& status);
-    // Waits until all the tasks are completed.
-    void wait();
 
     int64_t _get_running_tasks_for_level(int level) const;
 

be/src/vec/exec/scan/scanner_scheduler.h

@@ -268,6 +268,7 @@ class TaskExecutorSimplifiedScanScheduler final : public ScannerScheduler {
     void stop() override {
         _is_stop.store(true);
         _task_executor->stop();
+        _task_executor->wait();
     }
 
     Status start(int max_thread_num, int min_thread_num, int queue_size,