[feat](load) introduce load job system table

be/src/exec/schema_scanner.cpp

@@ -41,6 +41,7 @@
 #include "exec/schema_scanner/schema_encryption_keys_scanner.h"
 #include "exec/schema_scanner/schema_file_cache_statistics.h"
 #include "exec/schema_scanner/schema_files_scanner.h"
+#include "exec/schema_scanner/schema_load_job_scanner.h"
 #include "exec/schema_scanner/schema_metadata_name_ids_scanner.h"
 #include "exec/schema_scanner/schema_partitions_scanner.h"
 #include "exec/schema_scanner/schema_processlist_scanner.h"
@@ -240,6 +241,8 @@ std::unique_ptr<SchemaScanner> SchemaScanner::create(TSchemaTableType::type type
         return SchemaBackendKerberosTicketCacheScanner::create_unique();
     case TSchemaTableType::SCH_ROUTINE_LOAD_JOBS:
         return SchemaRoutineLoadJobScanner::create_unique();
+    case TSchemaTableType::SCH_LOAD_JOBS:
+        return SchemaLoadJobScanner::create_unique();
     case TSchemaTableType::SCH_BACKEND_TABLETS:
         return SchemaTabletsScanner::create_unique();
     case TSchemaTableType::SCH_VIEW_DEPENDENCY:

be/src/exec/schema_scanner/schema_helper.cpp

@@ -151,6 +151,15 @@ Status SchemaHelper::fetch_routine_load_job(const std::string& ip, const int32_t
             });
 }
 
+Status SchemaHelper::fetch_load_job(const std::string& ip, const int32_t port,
+                                    const TFetchLoadJobRequest& request,
+                                    TFetchLoadJobResult* result) {
+    return ThriftRpcHelper::rpc<FrontendServiceClient>(
+            ip, port, [&request, &result](FrontendServiceConnection& client) {
+                client->fetchLoadJob(*result, request);
+            });
+}
+
 Status SchemaHelper::fetch_schema_table_data(const std::string& ip, const int32_t port,
                                              const TFetchSchemaTableDataRequest& request,
                                              TFetchSchemaTableDataResult* result) {

be/src/exec/schema_scanner/schema_helper.h

@@ -29,6 +29,8 @@ class TDescribeTablesParams;
 class TDescribeTablesResult;
 class TFetchRoutineLoadJobRequest;
 class TFetchRoutineLoadJobResult;
+class TFetchLoadJobRequest;
+class TFetchLoadJobResult;
 class TFetchSchemaTableDataRequest;
 class TFetchSchemaTableDataResult;
 class TGetDbsParams;
@@ -94,6 +96,9 @@ class SchemaHelper {
                                          const TFetchRoutineLoadJobRequest& request,
                                          TFetchRoutineLoadJobResult* result);
 
+    static Status fetch_load_job(const std::string& ip, const int32_t port,
+                                 const TFetchLoadJobRequest& request, TFetchLoadJobResult* result);
+
     static Status fetch_schema_table_data(const std::string& ip, const int32_t port,
                                           const TFetchSchemaTableDataRequest& request,
                                           TFetchSchemaTableDataResult* result);

be/src/exec/schema_scanner/schema_load_job_scanner.cpp

@@ -0,0 +1,189 @@
+// 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_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> SchemaLoadJobScanner::_s_tbls_columns = {
+        //   name,       type,          size,     is_null
+        {"JOB_ID", TYPE_STRING, sizeof(StringRef), true},
+        {"LABEL", TYPE_STRING, sizeof(StringRef), true},
+        {"STATE", TYPE_STRING, sizeof(StringRef), true},
+        {"PROGRESS", TYPE_STRING, sizeof(StringRef), true},
+        {"TYPE", TYPE_STRING, sizeof(StringRef), true},
+        {"ETL_INFO", TYPE_STRING, sizeof(StringRef), true},
+        {"TASK_INFO", TYPE_STRING, sizeof(StringRef), true},
+        {"ERROR_MSG", TYPE_STRING, sizeof(StringRef), true},
+        {"CREATE_TIME", TYPE_STRING, sizeof(StringRef), true},
+        {"ETL_START_TIME", TYPE_STRING, sizeof(StringRef), true},
+        {"ETL_FINISH_TIME", TYPE_STRING, sizeof(StringRef), true},
+        {"LOAD_START_TIME", TYPE_STRING, sizeof(StringRef), true},
+        {"LOAD_FINISH_TIME", TYPE_STRING, sizeof(StringRef), true},
+        {"URL", TYPE_STRING, sizeof(StringRef), true},
+        {"JOB_DETAILS", TYPE_STRING, sizeof(StringRef), true},
+        {"TRANSACTION_ID", TYPE_STRING, sizeof(StringRef), true},
+        {"ERROR_TABLETS", TYPE_STRING, sizeof(StringRef), true},
+        {"USER", TYPE_STRING, sizeof(StringRef), true},
+        {"COMMENT", TYPE_STRING, sizeof(StringRef), true},
+        {"FIRST_ERROR_MSG", TYPE_STRING, sizeof(StringRef), true},
+};
+
+SchemaLoadJobScanner::SchemaLoadJobScanner()
+        : SchemaScanner(_s_tbls_columns, TSchemaTableType::SCH_LOAD_JOBS) {}
+
+SchemaLoadJobScanner::~SchemaLoadJobScanner() {}
+
+Status SchemaLoadJobScanner::start(RuntimeState* state) {
+    if (!_is_init) {
+        return Status::InternalError("used before initialized.");
+    }
+    TFetchLoadJobRequest request;
+    RETURN_IF_ERROR(SchemaHelper::fetch_load_job(*(_param->common_param->ip),
+                                                 _param->common_param->port, request, &_result));
+    return Status::OK();
+}
+
+Status SchemaLoadJobScanner::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.loadJobs.empty()) {
+        return Status::OK();
+    }
+
+    return _fill_block_impl(block);
+}
+
+Status SchemaLoadJobScanner::_fill_block_impl(vectorized::Block* block) {
+    SCOPED_TIMER(_fill_block_timer);
+
+    const auto& jobs_info = _result.loadJobs;
+    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<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: // LABEL
+                    column_value = job_info.__isset.label ? job_info.label : "";
+                    break;
+                case 2: // STATE
+                    column_value = job_info.__isset.state ? job_info.state : "";
+                    break;
+                case 3: // PROGRESS
+                    column_value = job_info.__isset.progress ? job_info.progress : "";
+                    break;
+                case 4: // TYPE
+                    column_value = job_info.__isset.type ? job_info.type : "";
+                    break;
+                case 5: // ETL_INFO
+                    column_value = job_info.__isset.etl_info ? job_info.etl_info : "";
+                    break;
+                case 6: // TASK_INFO
+                    column_value = job_info.__isset.task_info ? job_info.task_info : "";
+                    break;
+                case 7: // ERROR_MSG
+                    column_value = job_info.__isset.error_msg ? job_info.error_msg : "";
+                    break;
+                case 8: // CREATE_TIME
+                    column_value = job_info.__isset.create_time ? job_info.create_time : "";
+                    break;
+                case 9: // ETL_START_TIME
+                    column_value = job_info.__isset.etl_start_time ? job_info.etl_start_time : "";
+                    break;
+                case 10: // ETL_FINISH_TIME
+                    column_value = job_info.__isset.etl_finish_time ? job_info.etl_finish_time : "";
+                    break;
+                case 11: // LOAD_START_TIME
+                    column_value = job_info.__isset.load_start_time ? job_info.load_start_time : "";
+                    break;
+                case 12: // LOAD_FINISH_TIME
+                    column_value =
+                            job_info.__isset.load_finish_time ? job_info.load_finish_time : "";
+                    break;
+                case 13: // URL
+                    column_value = job_info.__isset.url ? job_info.url : "";
+                    break;
+                case 14: // JOB_DETAILS
+                    column_value = job_info.__isset.job_details ? job_info.job_details : "";
+                    break;
+                case 15: // TRANSACTION_ID
+                    column_value = job_info.__isset.transaction_id ? job_info.transaction_id : "";
+                    break;
+                case 16: // ERROR_TABLETS
+                    column_value = job_info.__isset.error_tablets ? job_info.error_tablets : "";
+                    break;
+                case 17: // USER
+                    column_value = job_info.__isset.user ? job_info.user : "";
+                    break;
+                case 18: // COMMENT
+                    column_value = job_info.__isset.comment ? job_info.comment : "";
+                    break;
+                case 19: // FIRST_ERROR_MSG
+                    column_value = job_info.__isset.first_error_msg ? job_info.first_error_msg : "";
+                    break;
+                }
+
+                str_refs[row_idx] =
+                        StringRef(column_values[row_idx].data(), column_values[row_idx].size());
+                datas[row_idx] = &str_refs[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_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 SchemaLoadJobScanner : public SchemaScanner {
+    ENABLE_FACTORY_CREATOR(SchemaLoadJobScanner);
+
+public:
+    SchemaLoadJobScanner();
+    ~SchemaLoadJobScanner() 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);
+
+    TFetchLoadJobResult _result;
+    static std::vector<SchemaScanner::ColumnDesc> _s_tbls_columns;
+};
+
+} // namespace doris

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

@@ -96,6 +96,8 @@ public enum SchemaTableType {
             TSchemaTableType.SCH_BACKEND_KERBEROS_TICKET_CACHE),
     SCH_ROUTINE_LOAD_JOBS("ROUTINE_LOAD_JOBS", "ROUTINE_LOAD_JOBS",
             TSchemaTableType.SCH_ROUTINE_LOAD_JOBS),
+    SCH_LOAD_JOBS("LOAD_JOBS", "LOAD_JOBS",
+            TSchemaTableType.SCH_LOAD_JOBS),
     SCH_VIEW_DEPENDENCY("VIEW_DEPENDENCY", "VIEW_DEPENDENCY",
                     TSchemaTableType.SCH_VIEW_DEPENDENCY),
     SQL_BLOCK_RULE_STATUS("SQL_BLOCK_RULE_STATUS", "SQL_BLOCK_RULE_STATUS",

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

@@ -674,6 +674,30 @@ public class SchemaTable extends Table {
                                     .column("IS_ABNORMAL_PAUSE", ScalarType.createType(PrimitiveType.BOOLEAN))
                                     .build())
             )
+            .put("load_jobs",
+                    new SchemaTable(SystemIdGenerator.getNextId(), "load_jobs", TableType.SCHEMA,
+                            builder().column("JOB_ID", ScalarType.createStringType())
+                                    .column("LABEL", ScalarType.createStringType())
+                                    .column("STATE", ScalarType.createStringType())
+                                    .column("PROGRESS", ScalarType.createStringType())
+                                    .column("TYPE", ScalarType.createStringType())
+                                    .column("ETL_INFO", ScalarType.createStringType())
+                                    .column("TASK_INFO", ScalarType.createStringType())
+                                    .column("ERROR_MSG", ScalarType.createStringType())
+                                    .column("CREATE_TIME", ScalarType.createStringType())
+                                    .column("ETL_START_TIME", ScalarType.createStringType())
+                                    .column("ETL_FINISH_TIME", ScalarType.createStringType())
+                                    .column("LOAD_START_TIME", ScalarType.createStringType())
+                                    .column("LOAD_FINISH_TIME", ScalarType.createStringType())
+                                    .column("URL", ScalarType.createStringType())
+                                    .column("JOB_DETAILS", ScalarType.createStringType())
+                                    .column("TRANSACTION_ID", ScalarType.createStringType())
+                                    .column("ERROR_TABLETS", ScalarType.createStringType())
+                                    .column("USER", ScalarType.createStringType())
+                                    .column("COMMENT", ScalarType.createStringType())
+                                    .column("FIRST_ERROR_MSG", ScalarType.createStringType())
+                                    .build())
+            )
             .put("backend_tablets", new SchemaTable(SystemIdGenerator.getNextId(), "backend_tablets", TableType.SCHEMA,
                     builder().column("BE_ID", ScalarType.createType(PrimitiveType.BIGINT))
                             .column("TABLET_ID", ScalarType.createType(PrimitiveType.BIGINT))