[improve](information schema) introduce routine load job system table (#48963)

### What problem does this PR solve?

Part IV of #48511

doc apache/doris-website#2196

**Introduce routine load job statistic system table:**
```
mysql> show create table information_schema.routine_load_job\G
*************************** 1. row ***************************
       Table: routine_load_job
Create Table: CREATE TABLE `routine_load_job` (
  `JOB_ID` text NULL,
  `JOB_NAME` text NULL,
  `CREATE_TIME` text NULL,
  `PAUSE_TIME` text NULL,
  `END_TIME` text NULL,
  `DB_NAME` text NULL,
  `TABLE_NAME` text NULL,
  `STATE` text NULL,
  `CURRENT_TASK_NUM` text NULL,
  `JOB_PROPERTIES` text NULL,
  `DATA_SOURCE_PROPERTIES` text NULL,
  `CUSTOM_PROPERTIES` text NULL,
  `STATISTIC` text NULL,
  `PROGRESS` text NULL,
  `LAG` text NULL,
  `REASON_OF_STATE_CHANGED` text NULL,
  `ERROR_LOG_URLS` text NULL,
  `USER_NAME` text NULL,
  `CURRENT_ABORT_TASK_NUM` int NULL,
  `IS_ABNORMAL_PAUSE` boolean NULL
) ENGINE=SCHEMA;
1 row in set (0.00 sec)
```

**There are some benefits to empower job with SQL query capability for
statistical information:**

- It can be used in conjunction with metrics add through
#48209 to roughly locate abnormal
jobs when Grafana alarms, and the following SQL can be used:

```
SELECT JOB_NAME
FROM information_schema.routine_load_job_statistics
WHERE CURRENT_ABORT_TASK_NUM > 0
   OR IS_ABNORMAL_PAUSE = TRUE;
```

- User can use the `select * from information_schema.routine_load_job`
instead of the `show routine load`. The advantage is that the `show
routine load` can only be searched by name, but SQL can be very flexible
in locating jobs

Author: sollhui

be/src/exec/schema_scanner.cpp

@@ -40,6 +40,7 @@
 #include "exec/schema_scanner/schema_partitions_scanner.h"
 #include "exec/schema_scanner/schema_processlist_scanner.h"
 #include "exec/schema_scanner/schema_profiling_scanner.h"
+#include "exec/schema_scanner/schema_routine_load_job_scanner.h"
 #include "exec/schema_scanner/schema_routine_scanner.h"
 #include "exec/schema_scanner/schema_rowsets_scanner.h"
 #include "exec/schema_scanner/schema_schema_privileges_scanner.h"
@@ -228,6 +229,8 @@ std::unique_ptr<SchemaScanner> SchemaScanner::create(TSchemaTableType::type type
         return SchemaCatalogMetaCacheStatsScanner::create_unique();
     case TSchemaTableType::SCH_BACKEND_KERBEROS_TICKET_CACHE:
         return SchemaBackendKerberosTicketCacheScanner::create_unique();
+    case TSchemaTableType::SCH_ROUTINE_LOAD_JOB:
+        return SchemaRoutineLoadJobScanner::create_unique();
     default:
         return SchemaDummyScanner::create_unique();
         break;

be/src/exec/schema_scanner/schema_helper.cpp

@@ -142,4 +142,13 @@ Status SchemaHelper::show_user(const std::string& ip, const int32_t port,
             });
 }
 
+Status SchemaHelper::fetch_routine_load_job(const std::string& ip, const int32_t port,
+                                            const TFetchRoutineLoadJobRequest& request,
+                                            TFetchRoutineLoadJobResult* result) {
+    return ThriftRpcHelper::rpc<FrontendServiceClient>(
+            ip, port, [&request, &result](FrontendServiceConnection& client) {
+                client->fetchRoutineLoadJob(*result, request);
+            });
+}
+
 } // namespace doris

be/src/exec/schema_scanner/schema_helper.h

@@ -26,6 +26,8 @@
 namespace doris {
 class TDescribeTablesParams;
 class TDescribeTablesResult;
+class TFetchRoutineLoadJobRequest;
+class TFetchRoutineLoadJobResult;
 class TGetDbsParams;
 class TGetDbsResult;
 class TGetTablesParams;
@@ -84,6 +86,10 @@ class SchemaHelper {
                                     TShowProcessListResult* result);
     static Status show_user(const std::string& ip, const int32_t port,
                             const TShowUserRequest& request, TShowUserResult* result);
+
+    static Status fetch_routine_load_job(const std::string& ip, const int32_t port,
+                                         const TFetchRoutineLoadJobRequest& request,
+                                         TFetchRoutineLoadJobResult* result);
 };
 
 } // namespace doris

be/src/exec/schema_scanner/schema_routine_load_job_scanner.cpp

@@ -0,0 +1,199 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "exec/schema_scanner/schema_routine_load_job_scanner.h"
+
+#include <gen_cpp/Descriptors_types.h>
+#include <gen_cpp/FrontendService_types.h>
+
+#include <string>
+
+#include "exec/schema_scanner/schema_helper.h"
+#include "runtime/runtime_state.h"
+#include "vec/common/string_ref.h"
+#include "vec/core/block.h"
+#include "vec/data_types/data_type_factory.hpp"
+
+namespace doris {
+class RuntimeState;
+namespace vectorized {
+class Block;
+} // namespace vectorized
+
+std::vector<SchemaScanner::ColumnDesc> SchemaRoutineLoadJobScanner::_s_tbls_columns = {
+        //   name,       type,          size,     is_null
+        {"JOB_ID", TYPE_STRING, sizeof(StringRef), true},
+        {"JOB_NAME", TYPE_STRING, sizeof(StringRef), true},
+        {"CREATE_TIME", TYPE_STRING, sizeof(StringRef), true},
+        {"PAUSE_TIME", TYPE_STRING, sizeof(StringRef), true},
+        {"END_TIME", TYPE_STRING, sizeof(StringRef), true},
+        {"DB_NAME", TYPE_STRING, sizeof(StringRef), true},
+        {"TABLE_NAME", TYPE_STRING, sizeof(StringRef), true},
+        {"STATE", TYPE_STRING, sizeof(StringRef), true},
+        {"CURRENT_TASK_NUM", TYPE_STRING, sizeof(StringRef), true},
+        {"JOB_PROPERTIES", TYPE_STRING, sizeof(StringRef), true},
+        {"DATA_SOURCE_PROPERTIES", TYPE_STRING, sizeof(StringRef), true},
+        {"CUSTOM_PROPERTIES", TYPE_STRING, sizeof(StringRef), true},
+        {"STATISTIC", TYPE_STRING, sizeof(StringRef), true},
+        {"PROGRESS", TYPE_STRING, sizeof(StringRef), true},
+        {"LAG", TYPE_STRING, sizeof(StringRef), true},
+        {"REASON_OF_STATE_CHANGED", TYPE_STRING, sizeof(StringRef), true},
+        {"ERROR_LOG_URLS", TYPE_STRING, sizeof(StringRef), true},
+        {"USER_NAME", TYPE_STRING, sizeof(StringRef), true},
+        {"CURRENT_ABORT_TASK_NUM", TYPE_INT, sizeof(int32_t), true},
+        {"IS_ABNORMAL_PAUSE", TYPE_BOOLEAN, sizeof(int8_t), true},
+};
+
+SchemaRoutineLoadJobScanner::SchemaRoutineLoadJobScanner()
+        : SchemaScanner(_s_tbls_columns, TSchemaTableType::SCH_ROUTINE_LOAD_JOB) {}
+
+SchemaRoutineLoadJobScanner::~SchemaRoutineLoadJobScanner() {}
+
+Status SchemaRoutineLoadJobScanner::start(RuntimeState* state) {
+    if (!_is_init) {
+        return Status::InternalError("used before initialized.");
+    }
+    TFetchRoutineLoadJobRequest request;
+    RETURN_IF_ERROR(SchemaHelper::fetch_routine_load_job(
+            *(_param->common_param->ip), _param->common_param->port, request, &_result));
+    return Status::OK();
+}
+
+Status SchemaRoutineLoadJobScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
+    if (!_is_init) {
+        return Status::InternalError("call this before initial.");
+    }
+    if (block == nullptr || eos == nullptr) {
+        return Status::InternalError("invalid parameter.");
+    }
+
+    *eos = true;
+    if (_result.routineLoadJobs.empty()) {
+        return Status::OK();
+    }
+
+    return _fill_block_impl(block);
+}
+
+Status SchemaRoutineLoadJobScanner::_fill_block_impl(vectorized::Block* block) {
+    SCOPED_TIMER(_fill_block_timer);
+
+    const auto& jobs_info = _result.routineLoadJobs;
+    size_t row_num = jobs_info.size();
+    if (row_num == 0) {
+        return Status::OK();
+    }
+
+    for (size_t col_idx = 0; col_idx < _s_tbls_columns.size(); ++col_idx) {
+        const auto& col_desc = _s_tbls_columns[col_idx];
+
+        std::vector<StringRef> str_refs(row_num);
+        std::vector<int32_t> int_vals(row_num);
+        std::vector<int8_t> bool_vals(row_num);
+        std::vector<void*> datas(row_num);
+        std::vector<std::string> column_values(row_num);
+
+        for (size_t row_idx = 0; row_idx < row_num; ++row_idx) {
+            const auto& job_info = jobs_info[row_idx];
+            std::string& column_value = column_values[row_idx];
+
+            if (col_desc.type == TYPE_STRING) {
+                switch (col_idx) {
+                case 0: // JOB_ID
+                    column_value = job_info.__isset.job_id ? job_info.job_id : "";
+                    break;
+                case 1: // JOB_NAME
+                    column_value = job_info.__isset.job_name ? job_info.job_name : "";
+                    break;
+                case 2: // CREATE_TIME
+                    column_value = job_info.__isset.create_time ? job_info.create_time : "";
+                    break;
+                case 3: // PAUSE_TIME
+                    column_value = job_info.__isset.pause_time ? job_info.pause_time : "";
+                    break;
+                case 4: // END_TIME
+                    column_value = job_info.__isset.end_time ? job_info.end_time : "";
+                    break;
+                case 5: // DB_NAME
+                    column_value = job_info.__isset.db_name ? job_info.db_name : "";
+                    break;
+                case 6: // TABLE_NAME
+                    column_value = job_info.__isset.table_name ? job_info.table_name : "";
+                    break;
+                case 7: // STATE
+                    column_value = job_info.__isset.state ? job_info.state : "";
+                    break;
+                case 8: // CURRENT_TASK_NUM
+                    column_value =
+                            job_info.__isset.current_task_num ? job_info.current_task_num : "";
+                    break;
+                case 9: // JOB_PROPERTIES
+                    column_value = job_info.__isset.job_properties ? job_info.job_properties : "";
+                    break;
+                case 10: // DATA_SOURCE_PROPERTIES
+                    column_value = job_info.__isset.data_source_properties
+                                           ? job_info.data_source_properties
+                                           : "";
+                    break;
+                case 11: // CUSTOM_PROPERTIES
+                    column_value =
+                            job_info.__isset.custom_properties ? job_info.custom_properties : "";
+                    break;
+                case 12: // STATISTIC
+                    column_value = job_info.__isset.statistic ? job_info.statistic : "";
+                    break;
+                case 13: // PROGRESS
+                    column_value = job_info.__isset.progress ? job_info.progress : "";
+                    break;
+                case 14: // LAG
+                    column_value = job_info.__isset.lag ? job_info.lag : "";
+                    break;
+                case 15: // REASON_OF_STATE_CHANGED
+                    column_value = job_info.__isset.reason_of_state_changed
+                                           ? job_info.reason_of_state_changed
+                                           : "";
+                    break;
+                case 16: // ERROR_LOG_URLS
+                    column_value = job_info.__isset.error_log_urls ? job_info.error_log_urls : "";
+                    break;
+                case 17: // USER_NAME
+                    column_value = job_info.__isset.user_name ? job_info.user_name : "";
+                    break;
+                }
+
+                str_refs[row_idx] =
+                        StringRef(column_values[row_idx].data(), column_values[row_idx].size());
+                datas[row_idx] = &str_refs[row_idx];
+            } else if (col_desc.type == TYPE_INT) {
+                int_vals[row_idx] = job_info.__isset.current_abort_task_num
+                                            ? job_info.current_abort_task_num
+                                            : 0;
+                datas[row_idx] = &int_vals[row_idx];
+            } else if (col_desc.type == TYPE_BOOLEAN) {
+                bool_vals[row_idx] =
+                        job_info.__isset.is_abnormal_pause ? job_info.is_abnormal_pause : false;
+                datas[row_idx] = &bool_vals[row_idx];
+            }
+        }
+
+        RETURN_IF_ERROR(fill_dest_column_for_range(block, col_idx, datas));
+    }
+
+    return Status::OK();
+}
+
+} // namespace doris

be/src/exec/schema_scanner/schema_routine_load_job_scanner.h

@@ -0,0 +1,50 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <gen_cpp/FrontendService_types.h>
+
+#include <vector>
+
+#include "common/status.h"
+#include "exec/schema_scanner.h"
+
+namespace doris {
+class RuntimeState;
+namespace vectorized {
+class Block;
+} // namespace vectorized
+
+class SchemaRoutineLoadJobScanner : public SchemaScanner {
+    ENABLE_FACTORY_CREATOR(SchemaRoutineLoadJobScanner);
+
+public:
+    SchemaRoutineLoadJobScanner();
+    ~SchemaRoutineLoadJobScanner() override;
+
+    Status start(RuntimeState* state) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
+
+private:
+    Status _fill_block_impl(vectorized::Block* block);
+
+    TFetchRoutineLoadJobResult _result;
+    static std::vector<SchemaScanner::ColumnDesc> _s_tbls_columns;
+};
+
+} // namespace doris

fe/fe-core/src/main/java/org/apache/doris/analysis/SchemaTableType.java

@@ -90,7 +90,9 @@ public enum SchemaTableType {
     SCH_CATALOG_META_CACHE_STATISTICS("CATALOG_META_CACHE_STATISTICS", "CATALOG_META_CACHE_STATISTICS",
             TSchemaTableType.SCH_CATALOG_META_CACHE_STATISTICS),
     SCH_BACKEND_KERBEROS_TICKET_CACHE("BACKEND_KERBEROS_TICKET_CACHE", "BACKEND_KERBEROS_TICKET_CACHE",
-            TSchemaTableType.SCH_BACKEND_KERBEROS_TICKET_CACHE);
+            TSchemaTableType.SCH_BACKEND_KERBEROS_TICKET_CACHE),
+    SCH_ROUTINE_LOAD_JOB("ROUTINE_LOAD_JOB", "ROUTINE_LOAD_JOB",
+            TSchemaTableType.SCH_ROUTINE_LOAD_JOB);
 
     private static final String dbName = "INFORMATION_SCHEMA";
     private static SelectList fullSelectLists;

fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java

@@ -597,6 +597,30 @@ public class SchemaTable extends Table {
                                     .column("REFRESH_INTERVAL_SECOND", ScalarType.createType(PrimitiveType.BIGINT))
                                     .build())
             )
+            .put("routine_load_job",
+                    new SchemaTable(SystemIdGenerator.getNextId(), "routine_load_job", TableType.SCHEMA,
+                            builder().column("JOB_ID", ScalarType.createStringType())
+                                    .column("JOB_NAME", ScalarType.createStringType())
+                                    .column("CREATE_TIME", ScalarType.createStringType())
+                                    .column("PAUSE_TIME", ScalarType.createStringType())
+                                    .column("END_TIME", ScalarType.createStringType())
+                                    .column("DB_NAME", ScalarType.createStringType())
+                                    .column("TABLE_NAME", ScalarType.createStringType())
+                                    .column("STATE", ScalarType.createStringType())
+                                    .column("CURRENT_TASK_NUM", ScalarType.createStringType())
+                                    .column("JOB_PROPERTIES", ScalarType.createStringType())
+                                    .column("DATA_SOURCE_PROPERTIES", ScalarType.createStringType())
+                                    .column("CUSTOM_PROPERTIES", ScalarType.createStringType())
+                                    .column("STATISTIC", ScalarType.createStringType())
+                                    .column("PROGRESS", ScalarType.createStringType())
+                                    .column("LAG", ScalarType.createStringType())
+                                    .column("REASON_OF_STATE_CHANGED", ScalarType.createStringType())
+                                    .column("ERROR_LOG_URLS", ScalarType.createStringType())
+                                    .column("USER_NAME", ScalarType.createStringType())
+                                    .column("CURRENT_ABORT_TASK_NUM", ScalarType.createType(PrimitiveType.INT))
+                                    .column("IS_ABNORMAL_PAUSE", ScalarType.createType(PrimitiveType.BOOLEAN))
+                                    .build())
+            )
             .build();
 
     private boolean fetchAllFe = false;

fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java

@@ -474,7 +474,7 @@ protected boolean needAutoResume() {
     }
 
     @Override
-    protected String getStatistic() {
+    public String getStatistic() {
         Map<String, Object> summary = this.jobStatistic.summary();
         Gson gson = new GsonBuilder().disableHtmlEscaping().create();
         return gson.toJson(summary);
@@ -635,7 +635,7 @@ private void setCustomKafkaProperties(Map<String, String> kafkaProperties) {
     }
 
     @Override
-    protected String dataSourcePropertiesJsonToString() {
+    public String dataSourcePropertiesJsonToString() {
         Map<String, String> dataSourceProperties = Maps.newHashMap();
         dataSourceProperties.put("brokerList", brokerList);
         dataSourceProperties.put("topic", topic);
@@ -647,21 +647,21 @@ protected String dataSourcePropertiesJsonToString() {
     }
 
     @Override
-    protected String customPropertiesJsonToString() {
+    public String customPropertiesJsonToString() {
         Gson gson = new GsonBuilder().disableHtmlEscaping().create();
         return gson.toJson(customProperties);
     }
 
     @Override
-    protected Map<String, String> getDataSourceProperties() {
+    public Map<String, String> getDataSourceProperties() {
         Map<String, String> dataSourceProperties = Maps.newHashMap();
         dataSourceProperties.put("kafka_broker_list", brokerList);
         dataSourceProperties.put("kafka_topic", topic);
         return dataSourceProperties;
     }
 
     @Override
-    protected Map<String, String> getCustomProperties() {
+    public Map<String, String> getCustomProperties() {
         Map<String, String> ret = new HashMap<>();
         customProperties.forEach((k, v) -> ret.put("property." + k, v));
         return ret;
@@ -910,7 +910,7 @@ public boolean hasMoreDataToConsume(UUID taskId, Map<Integer, Long> partitionIdT
     }
 
     @Override
-    protected String getLag() {
+    public String getLag() {
         Map<Integer, Long> partitionIdToOffsetLag = ((KafkaProgress) progress).getLag(cachedPartitionWithLatestOffsets);
         Gson gson = new Gson();
         return gson.toJson(partitionIdToOffsetLag);