[feature](info) Append RunningTasks into BE info (apache#56277)

To help us determine whether any query/loading tasks are running,
`RunningTasks ` metrics is appended in this PR and users could retrieve
it by `show backends` command.

<img width="3800" height="668" alt="image"
src="https://github.com/user-attachments/assets/50124384-3427-4621-91b3-b165bfda8d93"
/>

Author: Gabriel39

be/src/agent/task_worker_pool.cpp

@@ -1061,6 +1061,7 @@ void report_task_callback(const ClusterInfo* cluster_info) {
         }
     }
     request.__set_backend(BackendOptions::get_local_backend());
+    request.__set_running_tasks(ExecEnv::GetInstance()->fragment_mgr()->running_query_num());
     bool succ = handle_report(request, cluster_info, "task");
     report_task_total << 1;
     if (!succ) [[unlikely]] {

be/src/runtime/fragment_mgr.cpp

@@ -653,6 +653,12 @@ void FragmentMgr::remove_pipeline_context(
     _pipeline_map.erase({query_id, f_context->get_fragment_id()});
 }
 
+void FragmentMgr::remove_query_context(const TUniqueId& key) {
+#ifndef BE_TEST
+    _query_ctx_map.erase(key);
+#endif
+}
+
 std::shared_ptr<QueryContext> FragmentMgr::get_query_ctx(const TUniqueId& query_id) {
     auto val = _query_ctx_map.find(query_id);
     if (auto q_ctx = val.lock()) {
@@ -891,7 +897,7 @@ void FragmentMgr::cancel_query(const TUniqueId query_id, const Status reason) {
         }
     }
     query_ctx->cancel(reason);
-    _query_ctx_map.erase(query_id);
+    remove_query_context(query_id);
     LOG(INFO) << "Query " << print_id(query_id)
               << " is cancelled and removed. Reason: " << reason.to_string();
 }

be/src/runtime/fragment_mgr.h

@@ -134,6 +134,7 @@ class FragmentMgr : public RestMonitorIface {
     Status exec_plan_fragment(const TPipelineFragmentParams& params, const QuerySource query_type,
                               const FinishCallback& cb);
 
+    void remove_query_context(const TUniqueId& key);
     Status start_query_execution(const PExecPlanFragmentStartRequest* request);
 
     Status trigger_pipeline_context_report(const ReportStatusRequest,

be/src/runtime/query_context.cpp

@@ -196,6 +196,8 @@ QueryContext::~QueryContext() {
 
     _exec_env->spill_stream_mgr()->async_cleanup_query(_query_id);
     DorisMetrics::instance()->query_ctx_cnt->increment(-1);
+    // TODO(gabriel): we need to clear outdated query contexts on time
+    // ExecEnv::GetInstance()->fragment_mgr()->remove_query_context(this->_query_id);
     // the only one msg shows query's end. any other msg should append to it if need.
     LOG_INFO("Query {} deconstructed, mem_tracker: {}", print_id(this->_query_id), mem_tracker_msg);
 }

fe/fe-core/src/main/java/org/apache/doris/common/proc/BackendsProcDir.java

@@ -51,7 +51,7 @@ public class BackendsProcDir implements ProcDirInterface {
             .add("DataUsedCapacity").add("TrashUsedCapacity").add("AvailCapacity").add("TotalCapacity").add("UsedPct")
             .add("MaxDiskUsedPct").add("RemoteUsedCapacity").add("Tag").add("ErrMsg").add("Version").add("Status")
             .add("HeartbeatFailureCounter").add("NodeRole").add("CpuCores").add("Memory")
-            .build();
+            .add("RunningTasks").build();
 
     public static final ImmutableList<String> DISK_TITLE_NAMES = new ImmutableList.Builder<String>()
             .add("BackendId").add("Host").add("RootPath").add("DirType").add("DiskState")
@@ -173,6 +173,9 @@ public static List<List<String>> getBackendInfos() {
 
             // memory
             backendInfo.add(RuntimeProfile.printCounter(backend.getBeMemory(), TUnit.BYTES));
+
+            // runningFragments
+            backendInfo.add(String.valueOf(backend.getRunningTasks()));
             comparableBackendInfos.add(backendInfo);
         }
 

fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java

@@ -224,7 +224,8 @@ public TMasterResult handleReport(TReportRequest request) throws TException {
 
         ReportTask reportTask = new ReportTask(beId, reportType, tasks, disks, tablets, partitionsVersion,
                 reportVersion, request.getStoragePolicy(), request.getResource(), request.getNumCores(),
-                request.getPipelineExecutorSize(), numTablets, request.getIndexPolicy());
+                request.getPipelineExecutorSize(), numTablets, request.getIndexPolicy(),
+                request.isSetRunningTasks() ? request.getRunningTasks() : -1);
         try {
             putToQueue(reportTask);
         } catch (Exception e) {
@@ -314,14 +315,15 @@ private class ReportTask extends MasterTask {
         private List<TStorageResource> storageResources;
         private int cpuCores;
         private int pipelineExecutorSize;
+        private long runningTasks;
         private long numTablets;
         private List<TIndexPolicy> indexPolicys;
 
         public ReportTask(long beId, ReportType reportType, Map<TTaskType, Set<Long>> tasks,
                 Map<String, TDisk> disks, Map<Long, TTablet> tablets,
                 Map<Long, Long> partitionsVersion, long reportVersion,
                 List<TStoragePolicy> storagePolicies, List<TStorageResource> storageResources, int cpuCores,
-                int pipelineExecutorSize, long numTablets, List<TIndexPolicy> indexPolicys) {
+                int pipelineExecutorSize, long numTablets, List<TIndexPolicy> indexPolicys, long runningTasks) {
             this.beId = beId;
             this.reportType = reportType;
             this.tasks = tasks;
@@ -335,12 +337,13 @@ public ReportTask(long beId, ReportType reportType, Map<TTaskType, Set<Long>> ta
             this.pipelineExecutorSize = pipelineExecutorSize;
             this.numTablets = numTablets;
             this.indexPolicys = indexPolicys;
+            this.runningTasks = runningTasks;
         }
 
         @Override
         protected void exec() {
             if (tasks != null) {
-                ReportHandler.taskReport(beId, tasks);
+                ReportHandler.taskReport(beId, tasks, runningTasks);
             }
             if (disks != null) {
                 ReportHandler.diskReport(beId, disks);
@@ -704,7 +707,8 @@ private static void debugBlock() {
         }
     }
 
-    private static void taskReport(long backendId, Map<TTaskType, Set<Long>> runningTasks) {
+    private static void taskReport(long backendId, Map<TTaskType, Set<Long>> runningTasks,
+            long numRunningTasks) {
         debugBlock();
         if (LOG.isDebugEnabled()) {
             LOG.debug("begin to handle task report from backend {}", backendId);
@@ -728,6 +732,7 @@ private static void taskReport(long backendId, Map<TTaskType, Set<Long>> running
                 ? runningTasks.get(TTaskType.PUBLISH_VERSION).size() : 0;
         if (be != null) {
             be.setPublishTaskLastTimeAccumulated((long) publishTaskSize);
+            be.setRunningTasks(numRunningTasks);
         }
 
         List<AgentTask> diffTasks = AgentTaskQueue.getDiffTasks(backendId, runningTasks);

fe/fe-core/src/main/java/org/apache/doris/system/Backend.java

@@ -106,6 +106,8 @@ public class Backend implements Writable {
 
     private Long lastPublishTaskAccumulatedNum = 0L;
 
+    private Long runningTasks = 0L;
+
     private String heartbeatErrMsg = "";
 
     // This is used for the first time we init pathHashToDishInfo in SystemInfoService.
@@ -959,6 +961,14 @@ private int getDiskNum() {
         return disksRef.size();
     }
 
+    public Long getRunningTasks() {
+        return runningTasks;
+    }
+
+    public void setRunningTasks(Long runningTasks) {
+        this.runningTasks = runningTasks;
+    }
+
     /**
      * Note: This class must be a POJO in order to display in JSON format
      * Add additional information in the class to show in `show backends`

fe/fe-core/src/test/java/org/apache/doris/utframe/DemoMultiBackendsTest.java

@@ -200,13 +200,13 @@ public void testCreateDbAndTable() throws Exception {
         ProcResult result = dir.fetchResult();
         Assert.assertEquals(BackendsProcDir.TITLE_NAMES.size(), result.getColumnNames().size());
         Assert.assertEquals("{\"location\" : \"default\"}",
-                result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 8));
+                result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 9));
         Assert.assertEquals(
                 "{\"lastSuccessReportTabletsTime\":\"N/A\",\"lastStreamLoadTime\":-1,\"isQueryDisabled\":false,"
                         + "\"isLoadDisabled\":false,\"isActive\":true,\"currentFragmentNum\":0,\"lastFragmentUpdateTime\":0}",
-                result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 5));
-        Assert.assertEquals("0", result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 4));
-        Assert.assertEquals(Tag.VALUE_MIX, result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 3));
+                result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 6));
+        Assert.assertEquals("0", result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 5));
+        Assert.assertEquals(Tag.VALUE_MIX, result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 4));
     }
 
     private static void updateReplicaPathHash() {

gensrc/thrift/MasterService.thrift

@@ -121,6 +121,8 @@ struct TReportRequest {
     // tablet num in be, in cloud num_tablets may not eq tablet_list.size()
     14: optional i64 num_tablets
     15: optional list<AgentService.TIndexPolicy> index_policy
+    // Running query/loading tasks
+    16: optional i64 running_tasks
 }
 
 struct TMasterResult {