[Improvement](meta) support return total statistics of all databases for command show proc '/jobs

docs/en/docs/sql-manual/sql-reference/Show-Statements/SHOW-PROC.md

@@ -87,7 +87,7 @@ illustrate:
 3. frontends: Display all FE node information in the cluster, including IP address, role, status, whether it is a mater, etc., equivalent to [SHOW FRONTENDS](./SHOW-FRONTENDS.md)
 4. routine_loads: Display all routine load job information, including job name, status, etc.
 5. auth: User name and corresponding permission information
-6. jobs:
+6. jobs: show statistics of all kind of jobs. If a specific `dbId` is given, will return statistics data of the database. If `dbId` is -1, will return total statistics data of all databases
 7. bdbje: To view the bdbje database list, you need to modify the `fe.conf` file to add `enable_bdbje_debug_mode=true`, and then start `FE` through `sh start_fe.sh --daemon` to enter the `debug` mode. After entering `debug` mode, only `http server` and `MySQLServer` will be started and the `BDBJE` instance will be opened, but no metadata loading and subsequent startup processes will be entered.
 8. dbs: Mainly used to view the metadata information of each database and the tables in the Doris cluster. This information includes table structure, partitions, materialized views, data shards and replicas, and more. Through this directory and its subdirectories, you can clearly display the table metadata in the cluster, and locate some problems such as data skew, replica failure, etc.
 9. resources : View system resources, ordinary accounts can only see resources that they have USAGE_PRIV permission to use. Only the root and admin accounts can see all resources. Equivalent to [SHOW RESOURCES](./SHOW-RESOURCES.md)

docs/zh-CN/docs/sql-manual/sql-reference/Show-Statements/SHOW-PROC.md

@@ -87,7 +87,7 @@ mysql> show proc "/";
 3. frontends ：显示集群中所有的 FE 节点信息，包括IP地址、角色、状态、是否是mater等，等同于 [SHOW FRONTENDS](./SHOW-FRONTENDS.md)   
 4. routine_loads ： 显示所有的 routine load 作业信息，包括作业名称、状态等
 5. auth：用户名称及对应的权限信息
-6. jobs ：
+6. jobs ：各类任务的统计信息，可查看指定数据库的 Job 的统计信息，如果 `dbId` = -1, 则返回所有库的汇总信息
 7. bdbje：查看 bdbje 数据库列表，需要修改 `fe.conf` 文件增加 `enable_bdbje_debug_mode=true` , 然后通过 `sh start_fe.sh --daemon` 启动 `FE` 即可进入 `debug` 模式。 进入 `debug` 模式之后，仅会启动 `http server` 和  `MySQLServer` 并打开 `BDBJE` 实例，但不会进入任何元数据的加载及后续其他启动流程，
 8. dbs ： 主要用于查看 Doris 集群中各个数据库以及其中的表的元数据信息。这些信息包括表结构、分区、物化视图、数据分片和副本等等。通过这个目录和其子目录，可以清楚的展示集群中的表元数据情况，以及定位一些如数据倾斜、副本故障等问题
 9. resources : 查看系统资源，普通账户只能看到自己有 USAGE_PRIV 使用权限的资源。只有root和admin账户可以看到所有的资源。等同于 [SHOW RESOURCES](./SHOW-RESOURCES.md)

fe/fe-core/src/main/java/org/apache/doris/alter/AlterHandler.java

@@ -140,7 +140,16 @@ public Long getAlterJobV2Num(org.apache.doris.alter.AlterJobV2.JobState state, l
     }
 
     public Long getAlterJobV2Num(org.apache.doris.alter.AlterJobV2.JobState state) {
-        return alterJobsV2.values().stream().filter(e -> e.getJobState() == state).count();
+        Long counter = 0L;
+
+        for (AlterJobV2 job : alterJobsV2.values()) {
+            // no need to check priv here. This method is only called in show proc stmt,
+            // which already check the ADMIN priv.
+            if (job.getJobState() == state) {
+                counter++;
+            }
+        }
+        return counter;
     }
 
     @Override

fe/fe-core/src/main/java/org/apache/doris/alter/MaterializedViewHandler.java

@@ -65,6 +65,7 @@
 
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
@@ -1146,6 +1147,18 @@ public List<List<Comparable>> getAlterJobInfosByDb(Database db) {
         return rollupJobInfos;
     }
 
+    public List<List<Comparable>> getAllAlterJobInfos() {
+        List<List<Comparable>> rollupJobInfos = new LinkedList<List<Comparable>>();
+
+        for (AlterJobV2 alterJob : ImmutableList.copyOf(alterJobsV2.values())) {
+            // no need to check priv here. This method is only called in show proc stmt,
+            // which already check the ADMIN priv.
+            alterJob.getInfo(rollupJobInfos);
+        }
+
+        return rollupJobInfos;
+    }
+
     private void getAlterJobV2Infos(Database db, List<List<Comparable>> rollupJobInfos) {
         ConnectContext ctx = ConnectContext.get();
         for (AlterJobV2 alterJob : alterJobsV2.values()) {

fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java

@@ -1599,6 +1599,20 @@ private void runAlterJobV2() {
         });
     }
 
+    public List<List<Comparable>> getAllAlterJobInfos() {
+        List<List<Comparable>> schemaChangeJobInfos = new LinkedList<>();
+        for (AlterJobV2 alterJob : ImmutableList.copyOf(alterJobsV2.values())) {
+            // no need to check priv here. This method is only called in show proc stmt,
+            // which already check the ADMIN priv.
+            alterJob.getInfo(schemaChangeJobInfos);
+        }
+
+        // sort by "JobId", "PartitionName", "CreateTime", "FinishTime", "IndexName", "IndexState"
+        ListComparator<List<Comparable>> comparator = new ListComparator<List<Comparable>>(0, 1, 2, 3, 4, 5);
+        schemaChangeJobInfos.sort(comparator);
+        return schemaChangeJobInfos;
+    }
+
     @Override
     public List<List<Comparable>> getAlterJobInfosByDb(Database db) {
         List<List<Comparable>> schemaChangeJobInfos = new LinkedList<>();

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

@@ -51,14 +51,18 @@ public ExportProcNode(ExportMgr exportMgr, Database db) {
 
     @Override
     public ProcResult fetchResult() throws AnalysisException {
-        Preconditions.checkNotNull(db);
         Preconditions.checkNotNull(exportMgr);
 
         BaseProcResult result = new BaseProcResult();
         result.setNames(TITLE_NAMES);
 
-        List<List<String>> jobInfos = exportMgr.getExportJobInfosByIdOrState(
+        List<List<String>> jobInfos;
+        if (db == null) {
+            jobInfos = exportMgr.getExportJobInfos(LIMIT);
+        } else {
+            jobInfos = exportMgr.getExportJobInfosByIdOrState(
                 db.getId(), 0, "", false, null, null, LIMIT);
+        }
         result.setRows(jobInfos);
         return result;
     }

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

@@ -60,7 +60,8 @@ public ProcNodeInterface lookup(String dbIdStr) throws AnalysisException {
             throw new AnalysisException("Invalid db id format: " + dbIdStr);
         }
 
-        Database db = env.getInternalCatalog().getDbOrAnalysisException(dbId);
+        // dbId = -1 means need total result of all databases
+        Database db = dbId == -1 ? null : env.getInternalCatalog().getDbOrAnalysisException(dbId);
 
         return new JobsProcDir(env, db);
     }

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

@@ -24,7 +24,6 @@
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.load.ExportJob;
 import org.apache.doris.load.ExportMgr;
-import org.apache.doris.load.Load;
 import org.apache.doris.load.loadv2.LoadManager;
 
 import com.google.common.base.Preconditions;
@@ -68,9 +67,10 @@ public ProcNodeInterface lookup(String jobTypeName) throws AnalysisException {
         }
 
         if (jobTypeName.equals(LOAD)) {
-            return new LoadProcDir(env.getLoadInstance(), db);
+            return new LoadProcDir(env.getCurrentEnv().getLoadManager(), db);
         } else if (jobTypeName.equals(DELETE)) {
-            return new DeleteInfoProcDir(env.getDeleteHandler(), env.getLoadInstance(), db.getId());
+            Long dbId = db == null ? -1 : db.getId();
+            return new DeleteInfoProcDir(env.getDeleteHandler(), env.getLoadInstance(), dbId);
         } else if (jobTypeName.equals(ROLLUP)) {
             return new RollupProcDir(env.getMaterializedViewHandler(), db);
         } else if (jobTypeName.equals(SCHEMA_CHANGE)) {
@@ -90,23 +90,22 @@ public ProcResult fetchResult() throws AnalysisException {
 
         result.setNames(TITLE_NAMES);
 
+        // db is null means need total result of all databases
+        if (db == null) {
+            return fetchResultForAllDbs();
+        }
+
         long dbId = db.getId();
+
         // load
-        Load load = Env.getCurrentEnv().getLoadInstance();
         LoadManager loadManager = Env.getCurrentEnv().getLoadManager();
-        Long pendingNum = load.getLoadJobNum(org.apache.doris.load.LoadJob.JobState.PENDING, dbId)
-                + loadManager.getLoadJobNum(org.apache.doris.load.loadv2.JobState.PENDING, dbId);
-        Long runningNum = load.getLoadJobNum(org.apache.doris.load.LoadJob.JobState.ETL, dbId)
-                + load.getLoadJobNum(org.apache.doris.load.LoadJob.JobState.LOADING, dbId)
-                + loadManager.getLoadJobNum(org.apache.doris.load.loadv2.JobState.LOADING, dbId);
-        Long finishedNum = load.getLoadJobNum(org.apache.doris.load.LoadJob.JobState.QUORUM_FINISHED, dbId)
-                + load.getLoadJobNum(org.apache.doris.load.LoadJob.JobState.FINISHED, dbId)
-                + loadManager.getLoadJobNum(org.apache.doris.load.loadv2.JobState.FINISHED, dbId);
-        Long cancelledNum = load.getLoadJobNum(org.apache.doris.load.LoadJob.JobState.CANCELLED, dbId)
-                + loadManager.getLoadJobNum(org.apache.doris.load.loadv2.JobState.CANCELLED, dbId);
+        Long pendingNum = new Long(loadManager.getLoadJobNum(org.apache.doris.load.loadv2.JobState.PENDING, dbId));
+        Long runningNum = new Long(loadManager.getLoadJobNum(org.apache.doris.load.loadv2.JobState.LOADING, dbId));
+        Long finishedNum = new Long(loadManager.getLoadJobNum(org.apache.doris.load.loadv2.JobState.FINISHED, dbId));
+        Long cancelledNum = new Long(loadManager.getLoadJobNum(org.apache.doris.load.loadv2.JobState.CANCELLED, dbId));
         Long totalNum = pendingNum + runningNum + finishedNum + cancelledNum;
         result.addRow(Lists.newArrayList(LOAD, pendingNum.toString(), runningNum.toString(), finishedNum.toString(),
-                                         cancelledNum.toString(), totalNum.toString()));
+                cancelledNum.toString(), totalNum.toString()));
 
         // delete
         // TODO: find it from delete handler
@@ -155,4 +154,66 @@ public ProcResult fetchResult() throws AnalysisException {
 
         return result;
     }
+
+    public ProcResult fetchResultForAllDbs() {
+        BaseProcResult result = new BaseProcResult();
+
+        result.setNames(TITLE_NAMES);
+        // load
+        LoadManager loadManager = Env.getCurrentEnv().getLoadManager();
+        Long pendingNum = new Long(loadManager.getLoadJobNum(org.apache.doris.load.loadv2.JobState.PENDING));
+        Long runningNum = new Long(loadManager.getLoadJobNum(org.apache.doris.load.loadv2.JobState.LOADING));
+        Long finishedNum = new Long(loadManager.getLoadJobNum(org.apache.doris.load.loadv2.JobState.FINISHED));
+        Long cancelledNum = new Long(loadManager.getLoadJobNum(org.apache.doris.load.loadv2.JobState.CANCELLED));
+        Long totalNum = pendingNum + runningNum + finishedNum + cancelledNum;
+        result.addRow(Lists.newArrayList(LOAD, pendingNum.toString(), runningNum.toString(), finishedNum.toString(),
+                cancelledNum.toString(), totalNum.toString()));
+
+        // delete
+        // TODO: find it from delete handler
+        pendingNum = 0L;
+        runningNum = 0L;
+        finishedNum = 0L;
+        cancelledNum = 0L;
+        totalNum = pendingNum + runningNum + finishedNum + cancelledNum;
+        result.addRow(Lists.newArrayList(DELETE, pendingNum.toString(), runningNum.toString(), finishedNum.toString(),
+                cancelledNum.toString(), totalNum.toString()));
+
+        // rollup
+        MaterializedViewHandler materializedViewHandler = Env.getCurrentEnv().getMaterializedViewHandler();
+        pendingNum = materializedViewHandler.getAlterJobV2Num(org.apache.doris.alter.AlterJobV2.JobState.PENDING);
+        runningNum = materializedViewHandler.getAlterJobV2Num(
+            org.apache.doris.alter.AlterJobV2.JobState.WAITING_TXN)
+            + materializedViewHandler.getAlterJobV2Num(org.apache.doris.alter.AlterJobV2.JobState.RUNNING);
+        finishedNum = materializedViewHandler.getAlterJobV2Num(
+            org.apache.doris.alter.AlterJobV2.JobState.FINISHED);
+        cancelledNum = materializedViewHandler.getAlterJobV2Num(
+            org.apache.doris.alter.AlterJobV2.JobState.CANCELLED);
+        totalNum = pendingNum + runningNum + finishedNum + cancelledNum;
+        result.addRow(Lists.newArrayList(ROLLUP, pendingNum.toString(), runningNum.toString(), finishedNum.toString(),
+                cancelledNum.toString(), totalNum.toString()));
+
+        // schema change
+        SchemaChangeHandler schemaChangeHandler = Env.getCurrentEnv().getSchemaChangeHandler();
+        pendingNum = schemaChangeHandler.getAlterJobV2Num(org.apache.doris.alter.AlterJobV2.JobState.PENDING);
+        runningNum = schemaChangeHandler.getAlterJobV2Num(org.apache.doris.alter.AlterJobV2.JobState.WAITING_TXN)
+            + schemaChangeHandler.getAlterJobV2Num(org.apache.doris.alter.AlterJobV2.JobState.RUNNING);
+        finishedNum = schemaChangeHandler.getAlterJobV2Num(org.apache.doris.alter.AlterJobV2.JobState.FINISHED);
+        cancelledNum = schemaChangeHandler.getAlterJobV2Num(org.apache.doris.alter.AlterJobV2.JobState.CANCELLED);
+        totalNum = pendingNum + runningNum + finishedNum + cancelledNum;
+        result.addRow(Lists.newArrayList(SCHEMA_CHANGE, pendingNum.toString(), runningNum.toString(),
+                finishedNum.toString(), cancelledNum.toString(), totalNum.toString()));
+
+        // export
+        ExportMgr exportMgr = Env.getCurrentEnv().getExportMgr();
+        pendingNum = exportMgr.getJobNum(ExportJob.JobState.PENDING);
+        runningNum = exportMgr.getJobNum(ExportJob.JobState.EXPORTING);
+        finishedNum = exportMgr.getJobNum(ExportJob.JobState.FINISHED);
+        cancelledNum = exportMgr.getJobNum(ExportJob.JobState.CANCELLED);
+        totalNum = pendingNum + runningNum + finishedNum + cancelledNum;
+        result.addRow(Lists.newArrayList(EXPORT, pendingNum.toString(), runningNum.toString(), finishedNum.toString(),
+                cancelledNum.toString(), totalNum.toString()));
+
+        return result;
+    }
 }

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

@@ -18,25 +18,22 @@
 package org.apache.doris.common.proc;
 
 import org.apache.doris.common.AnalysisException;
-import org.apache.doris.load.Load;
+import org.apache.doris.load.loadv2.LoadManager;
 
 import com.google.common.collect.ImmutableList;
 
-import java.util.ArrayList;
-import java.util.List;
-
 public class LoadJobProcNode implements ProcNodeInterface {
 
     public static final ImmutableList<String> TITLE_NAMES = new ImmutableList.Builder<String>()
             .add("BackendId").add("TabletId").add("ReplicaId").add("Version")
             .add("PartitionId").add("LoadVersion")
             .build();
 
-    private Load load;
+    private LoadManager loadManager;
     private long jobId;
 
-    public LoadJobProcNode(Load load, long jobId) {
-        this.load = load;
+    public LoadJobProcNode(LoadManager loadManager, long jobId) {
+        this.loadManager = loadManager;
         this.jobId = jobId;
     }
 
@@ -45,17 +42,7 @@ public ProcResult fetchResult() throws AnalysisException {
         BaseProcResult result = new BaseProcResult();
         result.setNames(TITLE_NAMES);
 
-        List<List<Comparable>> infos = load.getLoadJobUnfinishedInfo(jobId);
-        // In this step, the detail of load job which is belongs to LoadManager will not be presented.
-        // The reason is that there are no detail info in load job which is streaming during loading.
-        // So it don't need to invoke the LoadManager here.
-        for (List<Comparable> info : infos) {
-            List<String> oneInfo = new ArrayList<String>(TITLE_NAMES.size());
-            for (Comparable element : info) {
-                oneInfo.add(element.toString());
-            }
-            result.addRow(oneInfo);
-        }
+        // TODO get results from LoadManager. Before do that, update implement of LoadManager：record detail info
         return result;
     }
 

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

@@ -18,16 +18,13 @@
 package org.apache.doris.common.proc;
 
 import org.apache.doris.catalog.Database;
-import org.apache.doris.catalog.Env;
 import org.apache.doris.common.AnalysisException;
-import org.apache.doris.load.Load;
+import org.apache.doris.load.loadv2.LoadManager;
 
-import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 
 import java.util.ArrayList;
 import java.util.Iterator;
-import java.util.LinkedList;
 import java.util.List;
 
 public class LoadProcDir implements ProcDirInterface {
@@ -47,30 +44,30 @@ public class LoadProcDir implements ProcDirInterface {
 
     private static final int LIMIT = 2000;
 
-    private Load load;
+    private LoadManager loadManager;
     private Database db;
 
-    public LoadProcDir(Load load, Database db) {
-        this.load = load;
+    public LoadProcDir(LoadManager loadManager, Database db) {
+        this.loadManager = loadManager;
         this.db = db;
     }
 
     @Override
     public ProcResult fetchResult() throws AnalysisException {
-        Preconditions.checkNotNull(db);
-        Preconditions.checkNotNull(load);
-
         BaseProcResult result = new BaseProcResult();
         result.setNames(TITLE_NAMES);
 
-        // merge load job from load and loadManager
-        LinkedList<List<Comparable>> loadJobInfos = load.getLoadJobInfosByDb(db.getId(), db.getFullName(),
-                null, false, null);
-        loadJobInfos.addAll(Env.getCurrentEnv().getLoadManager().getLoadJobInfosByDb(db.getId(), null,
-                false,
-                null));
+        List<List<Comparable>> loadJobInfos;
+
+        // db is null means need total result of all databases
+        if (db == null) {
+            loadJobInfos = loadManager.getAllLoadJobInfos();
+        } else {
+            loadJobInfos = loadManager.getLoadJobInfosByDb(db.getId(), null, false, null);
+        }
+
         int counter = 0;
-        Iterator<List<Comparable>> iterator = loadJobInfos.descendingIterator();
+        Iterator<List<Comparable>> iterator = loadJobInfos.iterator();
         while (iterator.hasNext()) {
             List<Comparable> infoStr = iterator.next();
             List<String> oneInfo = new ArrayList<String>(TITLE_NAMES.size());
@@ -99,7 +96,7 @@ public ProcNodeInterface lookup(String jobIdStr) throws AnalysisException {
             throw new AnalysisException("Invalid job id format: " + jobIdStr);
         }
 
-        return new LoadJobProcNode(load, jobId);
+        return new LoadJobProcNode(loadManager, jobId);
     }
 
     public static int analyzeColumn(String columnName) throws AnalysisException {

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

@@ -62,13 +62,18 @@ public RollupProcDir(MaterializedViewHandler materializedViewHandler, Database d
 
     @Override
     public ProcResult fetchResult() throws AnalysisException {
-        Preconditions.checkNotNull(db);
         Preconditions.checkNotNull(materializedViewHandler);
 
         BaseProcResult result = new BaseProcResult();
         result.setNames(TITLE_NAMES);
 
-        List<List<Comparable>> rollupJobInfos = materializedViewHandler.getAlterJobInfosByDb(db);
+        List<List<Comparable>> rollupJobInfos;
+        // db is null means need total result of all databases
+        if (db == null) {
+            rollupJobInfos = materializedViewHandler.getAllAlterJobInfos();
+        } else {
+            rollupJobInfos = materializedViewHandler.getAlterJobInfosByDb(db);
+        }
         for (List<Comparable> infoStr : rollupJobInfos) {
             List<String> oneInfo = new ArrayList<String>(TITLE_NAMES.size());
             for (Comparable element : infoStr) {