[refactor](query ctx) Remove useless state in query ctx (#39948)

Author: Gabriel39

be/src/pipeline/exec/scan_operator.cpp

@@ -32,6 +32,7 @@
 #include "pipeline/exec/operator.h"
 #include "runtime/types.h"
 #include "util/runtime_profile.h"
+#include "vec/exec/scan/scanner_context.h"
 #include "vec/exprs/vcast_expr.h"
 #include "vec/exprs/vcompound_pred.h"
 #include "vec/exprs/vectorized_fn_call.h"

be/src/runtime/query_context.cpp

@@ -87,7 +87,6 @@ QueryContext::QueryContext(TUniqueId query_id, ExecEnv* exec_env,
     SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(query_mem_tracker);
     _query_watcher.start();
     _shared_hash_table_controller.reset(new vectorized::SharedHashTableController());
-    _shared_scanner_controller.reset(new vectorized::SharedScannerController());
     _execution_dependency = pipeline::Dependency::create_unique(-1, -1, "ExecutionDependency");
     _runtime_filter_mgr = std::make_unique<RuntimeFilterMgr>(
             TUniqueId(), RuntimeFilterParamsContext::create(this), query_mem_tracker);
@@ -193,7 +192,6 @@ QueryContext::~QueryContext() {
     _runtime_filter_mgr.reset();
     _execution_dependency.reset();
     _shared_hash_table_controller.reset();
-    _shared_scanner_controller.reset();
     _runtime_predicates.clear();
     file_scan_range_params_map.clear();
     obj_pool.clear();
@@ -206,24 +204,14 @@ QueryContext::~QueryContext() {
 
 void QueryContext::set_ready_to_execute(Status reason) {
     set_execution_dependency_ready();
-    {
-        std::lock_guard<std::mutex> l(_start_lock);
-        _exec_status.update(reason);
-        _ready_to_execute = true;
-    }
+    _exec_status.update(reason);
     if (query_mem_tracker && !reason.ok()) {
         query_mem_tracker->set_is_query_cancelled(!reason.ok());
     }
-    _start_cond.notify_all();
 }
 
 void QueryContext::set_ready_to_execute_only() {
     set_execution_dependency_ready();
-    {
-        std::lock_guard<std::mutex> l(_start_lock);
-        _ready_to_execute = true;
-    }
-    _start_cond.notify_all();
 }
 
 void QueryContext::set_execution_dependency_ready() {
@@ -284,21 +272,6 @@ std::string QueryContext::print_all_pipeline_context() {
     return fmt::to_string(debug_string_buffer);
 }
 
-Status QueryContext::cancel_pipeline_context(const int fragment_id, const Status& reason) {
-    std::weak_ptr<pipeline::PipelineFragmentContext> ctx_to_cancel;
-    {
-        std::lock_guard<std::mutex> lock(_pipeline_map_write_lock);
-        if (!_fragment_id_to_pipeline_ctx.contains(fragment_id)) {
-            return Status::InternalError("fragment_id_to_pipeline_ctx is empty!");
-        }
-        ctx_to_cancel = _fragment_id_to_pipeline_ctx[fragment_id];
-    }
-    if (auto pipeline_ctx = ctx_to_cancel.lock()) {
-        pipeline_ctx->cancel(reason);
-    }
-    return Status::OK();
-}
-
 void QueryContext::set_pipeline_context(
         const int fragment_id, std::shared_ptr<pipeline::PipelineFragmentContext> pip_ctx) {
     std::lock_guard<std::mutex> lock(_pipeline_map_write_lock);

be/src/runtime/query_context.h

@@ -39,7 +39,6 @@
 #include "util/threadpool.h"
 #include "vec/exec/scan/scanner_scheduler.h"
 #include "vec/runtime/shared_hash_table_controller.h"
-#include "vec/runtime/shared_scanner_controller.h"
 #include "workload_group/workload_group.h"
 
 namespace doris {
@@ -111,41 +110,20 @@ class QueryContext {
 
     void cancel_all_pipeline_context(const Status& reason, int fragment_id = -1);
     std::string print_all_pipeline_context();
-    Status cancel_pipeline_context(const int fragment_id, const Status& reason);
     void set_pipeline_context(const int fragment_id,
                               std::shared_ptr<pipeline::PipelineFragmentContext> pip_ctx);
     void cancel(Status new_status, int fragment_id = -1);
 
-    void set_exec_status(Status new_status) { _exec_status.update(new_status); }
-
     [[nodiscard]] Status exec_status() { return _exec_status.status(); }
 
     void set_execution_dependency_ready();
 
     void set_ready_to_execute_only();
 
-    bool is_ready_to_execute() {
-        std::lock_guard<std::mutex> l(_start_lock);
-        return _ready_to_execute;
-    }
-
-    bool wait_for_start() {
-        int wait_time = config::max_fragment_start_wait_time_seconds;
-        std::unique_lock<std::mutex> l(_start_lock);
-        while (!_ready_to_execute.load() && _exec_status.ok() && --wait_time > 0) {
-            _start_cond.wait_for(l, std::chrono::seconds(1));
-        }
-        return _ready_to_execute.load() && _exec_status.ok();
-    }
-
     std::shared_ptr<vectorized::SharedHashTableController> get_shared_hash_table_controller() {
         return _shared_hash_table_controller;
     }
 
-    std::shared_ptr<vectorized::SharedScannerController> get_shared_scanner_controller() {
-        return _shared_scanner_controller;
-    }
-
     bool has_runtime_predicate(int source_node_id) {
         return _runtime_predicates.contains(source_node_id);
     }
@@ -283,16 +261,9 @@ class QueryContext {
     // If this token is not set, the scanner will be executed in "_scan_thread_pool" in exec env.
     std::unique_ptr<ThreadPoolToken> _thread_token;
 
-    std::mutex _start_lock;
-    std::condition_variable _start_cond;
-    // Only valid when _need_wait_execution_trigger is set to true in PlanFragmentExecutor.
-    // And all fragments of this query will start execution when this is set to true.
-    std::atomic<bool> _ready_to_execute {false};
-
     void _init_query_mem_tracker();
 
     std::shared_ptr<vectorized::SharedHashTableController> _shared_hash_table_controller;
-    std::shared_ptr<vectorized::SharedScannerController> _shared_scanner_controller;
     std::unordered_map<int, vectorized::RuntimePredicate> _runtime_predicates;
 
     WorkloadGroupPtr _workload_group = nullptr;