[fix](workload policy)query time based be policy not work due to wrong finish_time_ (#60456)

### What problem does this PR solve?

Issue Number: close
[#60454](#60454)

Problem Summary:

### Release note

None

### Check List (For Author)

- Test <!-- At least one of them must be included. -->
    - [x] Regression test
    - [x] Unit Test
    - [x] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
        - [ ] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason <!-- Add your reason?  -->

- Behavior changed:
    - [x] No.
    - [ ] Yes. <!-- Explain the behavior change -->

- Does this need documentation?
    - [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
apache/doris-website#1214 -->

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->

Author: wenzhenghu

be/src/runtime/workload_management/task_controller.h

@@ -64,7 +64,16 @@ class TaskController {
     virtual void finish_impl() {}
     int64_t start_time() const { return start_time_; }
     int64_t finish_time() const { return finish_time_; }
-    int64_t running_time() const { return finish_time() - start_time(); }
+    int64_t running_time() const {
+        if (start_time() == 0) {
+            return 0;
+        }
+        if (is_finished_) {
+            return finish_time() - start_time();
+        } else {
+            return MonotonicMillis() - start_time();
+        }
+    }
 
     /* cancel action
     */

be/test/runtime/workload_sched_policy_test.cpp

@@ -101,6 +101,10 @@ TEST_F(WorkloadSchedPolicyTest, one_policy_one_condition) {
                      std::move(action_ptr_list));
 
         WorkloadAction::RuntimeContext action_runtime_ctx = create_runtime_context();
+        TUniqueId task_id;
+        task_id.hi = 1;
+        task_id.lo = 1;
+        action_runtime_ctx.resource_ctx->task_controller()->set_task_id(task_id);
         std::this_thread::sleep_for(std::chrono::milliseconds(50));
         action_runtime_ctx.resource_ctx->task_controller()->finish();
         EXPECT_TRUE(policy->is_match(&action_runtime_ctx))
@@ -208,6 +212,10 @@ TEST_F(WorkloadSchedPolicyTest, one_policy_mutl_condition) {
 
     // 2. is match
     WorkloadAction::RuntimeContext action_runtime_ctx = create_runtime_context();
+    TUniqueId task_id;
+    task_id.hi = 1;
+    task_id.lo = 1;
+    action_runtime_ctx.resource_ctx->task_controller()->set_task_id(task_id);
     std::shared_ptr<MemTrackerLimiter> mem_tracker =
             MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::QUERY, "Test");
     action_runtime_ctx.resource_ctx->memory_context()->set_mem_tracker(mem_tracker);
@@ -283,4 +291,37 @@ TEST_F(WorkloadSchedPolicyTest, test_wg_id_set) {
     action_runtime_ctx.resource_ctx->set_workload_group(workload_group);
     EXPECT_TRUE(policy->is_match(&action_runtime_ctx));
 }
+
+TEST_F(WorkloadSchedPolicyTest, test_task_controller_running_time) {
+    std::shared_ptr<TaskController> task_controller = std::make_shared<TaskController>();
+    TUniqueId task_id;
+    task_id.hi = 1;
+    task_id.lo = 1;
+
+    // 1. Before set_task_id, start_time is 0
+    EXPECT_EQ(task_controller->start_time(), 0);
+
+    // 2. set_task_id will init start_time
+    task_controller->set_task_id(task_id);
+    EXPECT_GT(task_controller->start_time(), 0);
+
+    // 3. running_time should be positive when running
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    int64_t running_time = task_controller->running_time();
+    EXPECT_GT(running_time, 0);
+    EXPECT_GE(running_time, 10);
+
+    // 4. finish task
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    task_controller->finish();
+    EXPECT_TRUE(task_controller->is_finished());
+
+    // 5. running_time should be fixed after finish
+    int64_t finished_running_time = task_controller->running_time();
+    EXPECT_GT(finished_running_time, running_time);
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    EXPECT_EQ(task_controller->running_time(), finished_running_time);
+}
+
 } // namespace doris