[Chore](pipeline) some minor fix and refactor (#61768)

1. 调整了一些不必要的terminate调用
2. 修复set_thread_id处理的不对的问题，可能导致metric或者profile不对
3. 修复了_wake_by不是原子变量，debug string可能读的不对的问题
4. 把pipeline task的terminate方法改名为更直观的名字，避免与operator的terminate混淆
5. 一些注释更新



This pull request refactors the pipeline task termination logic to
improve thread safety and clarify responsibilities between dependency
unblocking and operator termination. The main change is replacing the
`terminate()` method of `PipelineTask` with a new method,
`unblock_all_dependencies()`, which is now responsible solely for
unblocking dependencies when a task is woken up early or canceled.
Operator termination remains the responsibility of the worker thread
within `execute()`, ensuring thread safety. This refactor also updates
related method calls and documentation for clarity.

**Refactoring of pipeline task termination and dependency management:**

* Replaced `PipelineTask::terminate()` with
`PipelineTask::unblock_all_dependencies()`, which now unblocks all
dependencies without terminating operators; updated all relevant call
sites and documentation to reflect this separation of concerns.
[[1]](diffhunk://#diff-b1379f2a0e6db93d3aaeec091b10d8a13e9461322f30af2c3f4a54df5bee702fL347-R347)
[[2]](diffhunk://#diff-b1379f2a0e6db93d3aaeec091b10d8a13e9461322f30af2c3f4a54df5bee702fL366-R367)
[[3]](diffhunk://#diff-b1379f2a0e6db93d3aaeec091b10d8a13e9461322f30af2c3f4a54df5bee702fL481-R496)
[[4]](diffhunk://#diff-b1379f2a0e6db93d3aaeec091b10d8a13e9461322f30af2c3f4a54df5bee702fL505-L507)
[[5]](diffhunk://#diff-b1379f2a0e6db93d3aaeec091b10d8a13e9461322f30af2c3f4a54df5bee702fL711-R714)
[[6]](diffhunk://#diff-b1379f2a0e6db93d3aaeec091b10d8a13e9461322f30af2c3f4a54df5bee702fL762-R768)
[[7]](diffhunk://#diff-b1379f2a0e6db93d3aaeec091b10d8a13e9461322f30af2c3f4a54df5bee702fL877-R881)
[[8]](diffhunk://#diff-3643eb9cad1657c99d794887ca26a78d9641b1d1f5dde3c86444370b7ffc9548L260-R260)
[[9]](diffhunk://#diff-2aac108aa389c146aff5acf00160da6d60f736d9a941768b62cb594d8f4f2b39L128-R128)
[[10]](diffhunk://#diff-262afd1bf43b83333335fec0b00b65ab0b0241315fd3ceb98c5b3d568971052fL438-R438)
* Updated the comment and signature for `unblock_all_dependencies()` in
`pipeline_task.h` to clarify its purpose and thread-safety
considerations, emphasizing that operator termination must occur on the
worker thread.

**Thread safety improvements:**

* Changed `_wake_by` in `PipelineTask` from a plain `int` to
`std::atomic<int>` to ensure thread-safe updates when waking up tasks
early.
* Fixed the logic in `set_thread_id()` to only update `_thread_id` and
increment the counter when the thread ID actually changes.

Author: BiteTheDDDDt

be/src/exec/pipeline/pipeline.cpp

@@ -125,11 +125,7 @@ void Pipeline::make_all_runnable(PipelineId wake_by) {
         for (auto* task : _tasks) {
             if (task) {
                 task->set_wake_up_early(wake_by);
-            }
-        }
-        for (auto* task : _tasks) {
-            if (task) {
-                task->terminate();
+                task->unblock_all_dependencies();
             }
         }
     }

be/src/exec/pipeline/pipeline_fragment_context.cpp

@@ -257,7 +257,7 @@ void PipelineFragmentContext::cancel(const Status reason) {
 
     for (auto& tasks : _tasks) {
         for (auto& task : tasks) {
-            task.first->terminate();
+            task.first->unblock_all_dependencies();
         }
     }
 }

be/src/exec/pipeline/pipeline_task.cpp

@@ -160,7 +160,7 @@ Status PipelineTask::prepare(const std::vector<TScanRangeParams>& scan_range, co
     }
     if (auto fragment = _fragment_context.lock()) {
         if (fragment->get_query_ctx()->is_cancelled()) {
-            terminate();
+            unblock_all_dependencies();
             return fragment->get_query_ctx()->exec_status();
         }
     } else {
@@ -344,7 +344,7 @@ bool PipelineTask::_is_blocked() {
            });
 }
 
-void PipelineTask::terminate() {
+void PipelineTask::unblock_all_dependencies() {
     // We use a lock to assure all dependencies are not deconstructed here.
     std::unique_lock<std::mutex> lc(_dependency_lock);
     auto fragment = _fragment_context.lock();
@@ -363,7 +363,8 @@ void PipelineTask::terminate() {
                                   [&](Dependency* dep) { dep->set_ready(); });
             _memory_sufficient_dependency->set_ready();
         } catch (const doris::Exception& e) {
-            LOG(WARNING) << "Terminate failed: " << e.code() << ", " << e.to_string();
+            LOG(WARNING) << "unblock_all_dependencies failed: " << e.code() << ", "
+                         << e.to_string();
         }
     }
 }
@@ -478,20 +479,21 @@ Status PipelineTask::execute(bool* done) {
         } else if (_eos && !_spilling &&
                    (fragment_context->is_canceled() || !_is_pending_finish())) {
             // Debug point for testing the race condition fix: inject set_wake_up_early() +
-            // terminate() here to simulate Thread B writing A then B between Thread A's two
-            // reads of _wake_up_early.
+            // unblock_all_dependencies() here to simulate Thread B writing A then B between
+            // Thread A's two reads of _wake_up_early.
             DBUG_EXECUTE_IF("PipelineTask::execute.wake_up_early_in_else_if", {
                 set_wake_up_early();
-                terminate();
+                unblock_all_dependencies();
             });
             *done = true;
         }
 
-        // NOTE: The terminate() call is intentionally placed AFTER the _is_pending_finish() check
-        // above, not before. This ordering is critical to avoid a race condition:
+        // NOTE: The operator terminate() call is intentionally placed AFTER the
+        // _is_pending_finish() check above, not before. This ordering is critical to avoid a race
+        // condition with the seq_cst memory ordering guarantee:
         //
         // Pipeline::make_all_runnable() writes in this order:
-        //   (A) set_wake_up_early()  ->  (B) terminate() [sets finish_dep._always_ready]
+        //   (A) set_wake_up_early()  ->  (B) unblock_all_dependencies() [sets finish_dep._always_ready]
         //
         // If we checked _wake_up_early (A) before _is_pending_finish() (B), there would be a
         // window where Thread A reads _wake_up_early=false, then Thread B writes both A and B,
@@ -502,9 +504,10 @@ Status PipelineTask::execute(bool* done) {
         //
         // By reading _is_pending_finish() (B) before the second read of _wake_up_early (A),
         // if Thread A observes B's effect (_always_ready=true), it is guaranteed to also observe
-        // A's effect (_wake_up_early=true) on this second read, ensuring terminate() is called.
+        // A's effect (_wake_up_early=true) on this second read, ensuring operator terminate() is
+        // called. This relies on _wake_up_early and _always_ready both being std::atomic with the
+        // default seq_cst ordering — do not weaken them to relaxed or acq/rel.
         if (_wake_up_early) {
-            terminate();
             THROW_IF_ERROR(_root->terminate(_state));
             THROW_IF_ERROR(_sink->terminate(_state));
         }
@@ -708,7 +711,7 @@ Status PipelineTask::execute(bool* done) {
                     if (required_pipeline_id == pipeline_id() && required_task_id == task_id() &&
                         fragment_context->get_fragment_id() == required_fragment_id) {
                         _wake_up_early = true;
-                        terminate();
+                        unblock_all_dependencies();
                     } else if (required_pipeline_id == pipeline_id() &&
                                fragment_context->get_fragment_id() == required_fragment_id) {
                         LOG(WARNING) << "PipelineTask::execute.terminate sleep 5s";
@@ -759,9 +762,10 @@ Status PipelineTask::do_revoke_memory(const std::shared_ptr<SpillContext>& spill
         fragment_context->get_query_ctx()->resource_ctx()->cpu_context()->update_cpu_cost_ms(
                 delta_cpu_time);
 
-        // If task is woke up early, we should terminate all operators, and this task could be closed immediately.
+        // If task is woke up early, unblock all dependencies and terminate all operators,
+        // so this task could be closed immediately.
         if (_wake_up_early) {
-            terminate();
+            unblock_all_dependencies();
             THROW_IF_ERROR(_root->terminate(_state));
             THROW_IF_ERROR(_sink->terminate(_state));
             _eos = true;
@@ -874,7 +878,7 @@ void PipelineTask::stop_if_finished() {
     if (auto sink = _sink) {
         if (sink->is_finished(_state)) {
             set_wake_up_early();
-            terminate();
+            unblock_all_dependencies();
         }
     }
 }

be/src/exec/pipeline/pipeline_task.h

@@ -71,9 +71,9 @@ class PipelineTask : public std::enable_shared_from_this<PipelineTask> {
     }
 
     virtual PipelineTask& set_thread_id(int thread_id) {
-        _thread_id = thread_id;
         if (thread_id != _thread_id) {
             COUNTER_UPDATE(_core_change_times, 1);
+            _thread_id = thread_id;
         }
         return *this;
     }
@@ -130,8 +130,12 @@ class PipelineTask : public std::enable_shared_from_this<PipelineTask> {
         _wake_by = wake_by;
     }
 
-    // Execution phase should be terminated. This is called if this task is canceled or waken up early.
-    void terminate();
+    // Unblock all dependencies so this task can never be blocked again.
+    // This is called when the task is woken up early or the fragment is canceled.
+    //
+    // NOTE: This does NOT call operator-level terminate() — operator terminate must run
+    // inside execute() on the worker thread because operator state is not thread-safe.
+    void unblock_all_dependencies();
 
     // 1 used for update priority queue
     // note(wb) an ugly implementation, need refactor later
@@ -317,7 +321,7 @@ class PipelineTask : public std::enable_shared_from_this<PipelineTask> {
     MonotonicStopWatch _state_change_watcher;
     std::atomic<bool> _spilling = false;
     const std::string _pipeline_name;
-    int _wake_by = -1;
+    std::atomic<int> _wake_by = -1;
 };
 
 using PipelineTaskSPtr = std::shared_ptr<PipelineTask>;

be/test/exec/pipeline/pipeline_task_test.cpp

@@ -435,7 +435,7 @@ TEST_F(PipelineTaskTest, TEST_TERMINATE) {
             std::this_thread::sleep_for(std::chrono::milliseconds(std::rand() % 5000));
             terminated = true;
             task->set_wake_up_early();
-            task->terminate();
+            task->unblock_all_dependencies();
         };
 
         std::thread exec_thread(exec_func);