[opt](query cancel) optimization for query cancel apache#28778

Author: zhiqiang-hhhh

be/src/agent/heartbeat_server.cpp

@@ -220,6 +220,11 @@ Status HeartbeatServer::_heartbeat(const TMasterInfo& master_info) {
 
     if (master_info.__isset.frontend_infos) {
         ExecEnv::GetInstance()->update_frontends(master_info.frontend_infos);
+    } else {
+        LOG_EVERY_N(WARNING, 2) << fmt::format(
+                "Heartbeat from {}:{} does not have frontend_infos, this may because we are "
+                "upgrading cluster",
+                master_info.network_address.hostname, master_info.network_address.port);
     }
 
     if (need_report) {

be/src/runtime/exec_env.cpp

@@ -18,11 +18,13 @@
 #include "runtime/exec_env.h"
 
 #include <gen_cpp/HeartbeatService_types.h>
+#include <glog/logging.h>
 
 #include <mutex>
 #include <utility>
 
 #include "common/config.h"
+#include "common/logging.h"
 #include "olap/olap_define.h"
 #include "olap/storage_engine.h"
 #include "olap/tablet_manager.h"
@@ -119,34 +121,25 @@ std::map<TNetworkAddress, FrontendInfo> ExecEnv::get_running_frontends() {
     const auto now = GetCurrentTimeMicros() / 1000;
 
     for (const auto& pair : _frontends) {
-        if (pair.second.info.process_uuid != 0) {
-            if (now - pair.second.last_reveiving_time_ms < expired_duration) {
+        auto& brpc_addr = pair.first;
+        auto& fe_info = pair.second;
+
+        if (fe_info.info.process_uuid == 0) {
+            // FE is in an unknown state, regart it as alive. conservative
+            res[brpc_addr] = fe_info;
+        } else {
+            if (now - fe_info.last_reveiving_time_ms < expired_duration) {
                 // If fe info has just been update in last expired_duration, regard it as running.
-                res[pair.first] = pair.second;
+                res[brpc_addr] = fe_info;
             } else {
                 // Fe info has not been udpate for more than expired_duration, regard it as an abnormal.
                 // Abnormal means this fe can not connect to master, and it is not dropped from cluster.
                 // or fe do not have master yet.
-                LOG(INFO) << "Frontend " << PrintFrontendInfo(pair.second.info)
-                          << " has not update its hb "
-                          << "for more than " << config::fe_expire_duration_seconds
-                          << " secs, regard it as abnormal.";
+                LOG_EVERY_N(WARNING, 50) << fmt::format(
+                        "Frontend {}:{} has not update its hb for more than {} secs, regard it as "
+                        "abnormal",
+                        brpc_addr.hostname, brpc_addr.port, config::fe_expire_duration_seconds);
             }
-
-            continue;
-        }
-
-        if (pair.second.last_reveiving_time_ms - pair.second.first_receiving_time_ms >
-            expired_duration) {
-            // A zero process-uuid that sustains more than 60 seconds(default).
-            // We will regard this fe as a abnormal frontend.
-            LOG(INFO) << "Frontend " << PrintFrontendInfo(pair.second.info)
-                      << " has not update its hb "
-                      << "for more than " << config::fe_expire_duration_seconds
-                      << " secs, regard it as abnormal.";
-            continue;
-        } else {
-            res[pair.first] = pair.second;
         }
     }
 

be/src/runtime/exec_env.h

@@ -349,6 +349,7 @@ class ExecEnv {
     std::shared_ptr<WalManager> _wal_manager;
 
     std::mutex _frontends_lock;
+    // ip:brpc_port -> frontend_indo
     std::map<TNetworkAddress, FrontendInfo> _frontends;
     GroupCommitMgr* _group_commit_mgr = nullptr;
 

be/src/runtime/fragment_mgr.cpp

@@ -1167,6 +1167,8 @@ void FragmentMgr::cancel_worker() {
             } else {
                 for (const auto& q : _query_ctx_map) {
                     if (q.second->get_fe_process_uuid() == 0) {
+                        // zero means this query is from a older version fe or
+                        // this fe is starting
                         continue;
                     }
 
@@ -1175,7 +1177,16 @@ void FragmentMgr::cancel_worker() {
                         if (q.second->get_fe_process_uuid() == itr->second.info.process_uuid ||
                             itr->second.info.process_uuid == 0) {
                             continue;
+                        } else {
+                            LOG_WARNING("Coordinator of query {} restarted, going to cancel it.",
+                                        print_id(q.second->query_id()));
                         }
+                    } else {
+                        LOG_WARNING(
+                                "Could not find target coordinator {}:{} of query {}, going to "
+                                "cancel it.",
+                                q.second->coord_addr.hostname, q.second->coord_addr.port,
+                                print_id(q.second->query_id()));
                     }
 
                     // Coorninator of this query has already dead.
@@ -1195,7 +1206,7 @@ void FragmentMgr::cancel_worker() {
 
         if (!queries_to_cancel.empty()) {
             LOG(INFO) << "There are " << queries_to_cancel.size()
-                      << " queries need to be cancelled, coordinator dead.";
+                      << " queries need to be cancelled, coordinator dead or restarted.";
         }
 
         for (const auto& qid : queries_to_cancel) {

be/src/runtime/query_context.h

@@ -184,7 +184,9 @@ class QueryContext {
         return _query_options.be_exec_version;
     }
 
-    [[nodiscard]] int64_t get_fe_process_uuid() const { return _query_options.fe_process_uuid; }
+    [[nodiscard]] int64_t get_fe_process_uuid() const {
+        return _query_options.__isset.fe_process_uuid ? _query_options.fe_process_uuid : 0;
+    }
 
     RuntimeFilterMgr* runtime_filter_mgr() { return _runtime_filter_mgr.get(); }
 

be/src/util/debug_util.cpp

@@ -125,7 +125,6 @@ std::string PrintFrontendInfos(const std::vector<TFrontendInfo>& fe_infos) {
 std::string PrintFrontendInfo(const TFrontendInfo& fe_info) {
     std::stringstream ss;
     fe_info.printTo(ss);
-
     return ss.str();
 }