[Feature](MTMV) add mv dependence system table

be/src/exec/schema_scanner.cpp

@@ -37,6 +37,7 @@
 #include "exec/schema_scanner/schema_file_cache_statistics.h"
 #include "exec/schema_scanner/schema_files_scanner.h"
 #include "exec/schema_scanner/schema_metadata_name_ids_scanner.h"
+#include "exec/schema_scanner/schema_mv_dependence_scanner.h"
 #include "exec/schema_scanner/schema_partitions_scanner.h"
 #include "exec/schema_scanner/schema_processlist_scanner.h"
 #include "exec/schema_scanner/schema_profiling_scanner.h"
@@ -231,6 +232,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_MV_DEPENDENCE:
+        return SchemaMvDependenceScanner::create_unique();
     default:
         return SchemaDummyScanner::create_unique();
         break;

be/src/exec/schema_scanner/schema_mv_dependence_scanner.cpp

@@ -0,0 +1,140 @@
+// 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 <gen_cpp/FrontendService_types.h>
+
+#include <vector>
+
+#include "exec/schema_scanner/schema_mv_dependence_scanner.h"
+#include "exec/schema_scanner/schema_helper.h"
+#include "runtime/client_cache.h"
+#include "runtime/exec_env.h"
+#include "runtime/runtime_state.h"
+#include "util/thrift_rpc_helper.h"
+#include "vec/common/string_ref.h"
+#include "vec/core/block.h"
+#include "vec/data_types/data_type_factory.hpp"
+
+namespace doris {
+
+std::vector<SchemaScanner::ColumnDesc> SchemaMvDependenceScanner::_s_mv_dependence_columns = {
+        {"SRC_CATALOG", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"SRC_DATABASE", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"SRC_TABLE", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"FROM_CATALOG", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"FROM_DATABASE", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"FROM_TABLE", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"FROM_TABLE_TYPE", TYPE_VARCHAR, sizeof(StringRef), false}
+};
+
+SchemaMvDependenceScanner::SchemaMvDependenceScanner()
+        : SchemaScanner(_s_mv_dependence_columns, TSchemaTableType::SCH_MV_DEPENDENCE) {}
+
+SchemaMvDependenceScanner::~SchemaMvDependenceScanner() = default;
+
+Status SchemaMvDependenceScanner::start(RuntimeState* state) {
+    _block_rows_limit = state->batch_size();
+    _rpc_timeout = state->execution_timeout() * 1000;
+    return Status::OK();
+}
+
+Status SchemaMvDependenceScanner::_get_mv_dependence_block_from_fe() {
+    TNetworkAddress master_addr = ExecEnv::GetInstance()->cluster_info()->master_fe_addr;
+
+    TSchemaTableRequestParams schema_table_request_params;
+    for (int i = 0; i < _s_mv_dependence_columns.size(); i++) {
+        schema_table_request_params.__isset.columns_name = true;
+        schema_table_request_params.columns_name.emplace_back(_s_mv_dependence_columns[i].name);
+    }
+    schema_table_request_params.__set_current_user_ident(*_param->common_param->current_user_ident);
+
+    TFetchSchemaTableDataRequest request;
+    request.__set_schema_table_name(TSchemaTableName::MV_DEPENDENCE);
+    request.__set_schema_table_params(schema_table_request_params);
+
+    TFetchSchemaTableDataResult result;
+
+    RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>(
+            master_addr.hostname, master_addr.port,
+            [&request, &result](FrontendServiceConnection& client) {
+                client->fetchSchemaTableData(result, request);
+            },
+            _rpc_timeout));
+
+    Status status(Status::create(result.status));
+    if (!status.ok()) {
+        LOG(WARNING) << "fetch transactions from FE failed, errmsg=" << status;
+        return status;
+    }
+    std::vector<TRow> result_data = result.data_batch;
+
+    _mv_dependence_block = vectorized::Block::create_unique();
+    for (int i = 0; i < _s_mv_dependence_columns.size(); ++i) {
+        TypeDescriptor descriptor(_s_mv_dependence_columns[i].type);
+        auto data_type = vectorized::DataTypeFactory::instance().create_data_type(descriptor, true);
+        _mv_dependence_block->insert(vectorized::ColumnWithTypeAndName(
+                data_type->create_column(), data_type, _s_mv_dependence_columns[i].name));
+    }
+
+    _mv_dependence_block->reserve(_block_rows_limit);
+
+    if (result_data.size() > 0) {
+        auto col_size = result_data[0].column_value.size();
+        if (col_size != _s_mv_dependence_columns.size()) {
+            return Status::InternalError<false>("transactions schema is not match for FE and BE");
+        }
+    }
+
+    for (int i = 0; i < result_data.size(); i++) {
+        TRow row = result_data[i];
+        for (int j = 0; j < _s_mv_dependence_columns.size(); j++) {
+            RETURN_IF_ERROR(insert_block_column(row.column_value[j], j, _mv_dependence_block.get(),
+                                                _s_mv_dependence_columns[j].type));
+        }
+    }
+    return Status::OK();
+}
+
+Status SchemaMvDependenceScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
+    if (!_is_init) {
+        return Status::InternalError("Used before initialized.");
+    }
+
+    if (nullptr == block || nullptr == eos) {
+        return Status::InternalError("input pointer is nullptr.");
+    }
+
+    if (_mv_dependence_block == nullptr) {
+        RETURN_IF_ERROR(_get_mv_dependence_block_from_fe());
+        _total_rows = (int)_mv_dependence_block->rows();
+    }
+
+    if (_row_idx == _total_rows) {
+        *eos = true;
+        return Status::OK();
+    }
+
+    int current_batch_rows = std::min(_block_rows_limit, _total_rows - _row_idx);
+    vectorized::MutableBlock mblock = vectorized::MutableBlock::build_mutable_block(block);
+    RETURN_IF_ERROR(mblock.add_rows(_mv_dependence_block.get(), _row_idx, current_batch_rows));
+    _row_idx += current_batch_rows;
+
+    *eos = _row_idx == _total_rows;
+    return Status::OK();
+}
+
+} // namespace doris

be/src/exec/schema_scanner/schema_mv_dependence_scanner.h

@@ -0,0 +1,56 @@
+// 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 <memory>
+#include <vector>
+
+#include "common/status.h"
+#include "exec/schema_scanner.h"
+
+namespace doris {
+
+class RuntimeState;
+
+namespace vectorized {
+class Block;
+}
+
+class SchemaMvDependenceScanner : public SchemaScanner {
+    ENABLE_FACTORY_CREATOR(SchemaMvDependenceScanner);
+
+public:
+    SchemaMvDependenceScanner();
+    ~SchemaMvDependenceScanner() override;
+
+    Status start(RuntimeState* state) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
+
+    static std::vector<SchemaScanner::ColumnDesc> _s_mv_dependence_columns;
+
+private:
+    Status _get_mv_dependence_block_from_fe();
+
+    int _block_rows_limit = 4096;
+    int _row_idx = 0;
+    int _total_rows = 0;
+    std::unique_ptr<vectorized::Block> _mv_dependence_block = nullptr;
+    int _rpc_timeout = 3000;
+};
+
+} // namespace doris

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

@@ -92,7 +92,9 @@ public enum SchemaTableType {
     SCH_BACKEND_KERBEROS_TICKET_CACHE("BACKEND_KERBEROS_TICKET_CACHE", "BACKEND_KERBEROS_TICKET_CACHE",
             TSchemaTableType.SCH_BACKEND_KERBEROS_TICKET_CACHE),
     SCH_ROUTINE_LOAD_JOBS("ROUTINE_LOAD_JOBS", "ROUTINE_LOAD_JOBS",
-            TSchemaTableType.SCH_ROUTINE_LOAD_JOBS);
+            TSchemaTableType.SCH_ROUTINE_LOAD_JOBS),
+    SCH_MV_DEPENDENCE("MV_DEPENDENCE", "MV_DEPENDENCE",
+                    TSchemaTableType.SCH_MV_DEPENDENCE);
 
     private static final String dbName = "INFORMATION_SCHEMA";
     private static SelectList fullSelectLists;

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

@@ -621,6 +621,16 @@ public class SchemaTable extends Table {
                                     .column("IS_ABNORMAL_PAUSE", ScalarType.createType(PrimitiveType.BOOLEAN))
                                     .build())
             )
+            .put("mv_dependence",
+                    new SchemaTable(SystemIdGenerator.getNextId(), "mv_dependence", TableType.SCHEMA,
+                        builder().column("SRC_CATALOG", ScalarType.createVarchar(NAME_CHAR_LEN))
+                            .column("SRC_DATABASE", ScalarType.createVarchar(NAME_CHAR_LEN))
+                            .column("SRC_TABLE", ScalarType.createVarchar(NAME_CHAR_LEN))
+                            .column("FROM_CATALOG", ScalarType.createVarchar(NAME_CHAR_LEN))
+                            .column("FROM_DATABASE", ScalarType.createVarchar(NAME_CHAR_LEN))
+                            .column("FROM_TABLE", ScalarType.createVarchar(NAME_CHAR_LEN))
+                            .column("FROM_TABLE_TYPE", ScalarType.createVarchar(NAME_CHAR_LEN))
+                            .build()))
             .build();
 
     private boolean fetchAllFe = false;

fe/fe-core/src/main/java/org/apache/doris/mtmv/BaseTableInfo.java

@@ -18,6 +18,7 @@
 package org.apache.doris.mtmv;
 
 import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.common.AnalysisException;
@@ -33,6 +34,7 @@
 import org.apache.logging.log4j.Logger;
 
 import java.util.List;
+import java.util.Optional;
 
 public class BaseTableInfo {
     private static final Logger LOG = LogManager.getLogger(BaseTableInfo.class);
@@ -122,6 +124,20 @@ public boolean isInternalTable() {
         }
     }
 
+    public String getType() {
+        CatalogIf catalog = Env.getCurrentEnv().getCatalogMgr().getCatalog(ctlName);
+        if (catalog != null) {
+            Optional<DatabaseIf> db = catalog.getDb(dbName);
+            if (db.isPresent()) {
+                Optional<TableIf> table = db.get().getTable(tableName);
+                if (table.isPresent()) {
+                    return table.get().getType().name();
+                }
+            }
+        }
+        return "UNKNOWN";
+    }
+
     public boolean isValid() {
         return ctlName != null && dbName != null && tableName != null;
     }

fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java

@@ -64,7 +64,9 @@
 import org.apache.doris.job.common.JobType;
 import org.apache.doris.job.extensions.mtmv.MTMVJob;
 import org.apache.doris.job.task.AbstractTask;
+import org.apache.doris.mtmv.BaseTableInfo;
 import org.apache.doris.mtmv.MTMVPartitionUtil;
+import org.apache.doris.mtmv.MTMVRelation;
 import org.apache.doris.mtmv.MTMVStatus;
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.plsql.metastore.PlsqlManager;
@@ -121,6 +123,7 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 public class MetadataGenerator {
@@ -144,6 +147,8 @@ public class MetadataGenerator {
 
     private static final ImmutableMap<String, Integer> PARTITIONS_COLUMN_TO_INDEX;
 
+    private static final ImmutableMap<String, Integer> MV_DEPENDENCE_COLUMN_TO_INDEX;
+
     static {
         ImmutableMap.Builder<String, Integer> activeQueriesbuilder = new ImmutableMap.Builder();
         List<Column> activeQueriesColList = SchemaTable.TABLE_MAP.get("active_queries").getFullSchema();
@@ -205,6 +210,13 @@ public class MetadataGenerator {
             partitionsBuilder.put(partitionsColList.get(i).getName().toLowerCase(), i);
         }
         PARTITIONS_COLUMN_TO_INDEX = partitionsBuilder.build();
+
+        ImmutableMap.Builder<String, Integer> mvDependenceBuilder = new ImmutableMap.Builder();
+        List<Column> mvDependenceBuilderColList = SchemaTable.TABLE_MAP.get("mv_dependence").getFullSchema();
+        for (int i = 0; i < mvDependenceBuilderColList.size(); i++) {
+            mvDependenceBuilder.put(mvDependenceBuilderColList.get(i).getName().toLowerCase(), i);
+        }
+        MV_DEPENDENCE_COLUMN_TO_INDEX = mvDependenceBuilder.build();
     }
 
     public static TFetchSchemaTableDataResult getMetadataTable(TFetchSchemaTableDataRequest request) throws TException {
@@ -314,6 +326,10 @@ public static TFetchSchemaTableDataResult getSchemaTableData(TFetchSchemaTableDa
                 result = partitionsMetadataResult(schemaTableParams);
                 columnIndex = PARTITIONS_COLUMN_TO_INDEX;
                 break;
+            case MV_DEPENDENCE:
+                result = mvDependenceMetadataResult(schemaTableParams);
+                columnIndex = MV_DEPENDENCE_COLUMN_TO_INDEX;
+                break;
             default:
                 return errorResult("invalid schema table name.");
         }
@@ -639,6 +655,40 @@ private static TFetchSchemaTableDataResult workloadGroupsMetadataResult(TSchemaT
         return result;
     }
 
+    private static TFetchSchemaTableDataResult mvDependenceMetadataResult(TSchemaTableRequestParams params) {
+        if (!params.isSetCurrentUserIdent()) {
+            return errorResult("current user ident is not set.");
+        }
+        Collection<DatabaseIf<? extends TableIf>> allDbs = Env.getCurrentEnv().getInternalCatalog().getAllDbs();
+        TFetchSchemaTableDataResult result = new TFetchSchemaTableDataResult();
+        List<TRow> dataBatch = Lists.newArrayList();
+        for (DatabaseIf<? extends TableIf> db : allDbs) {
+            List<? extends TableIf> tables = db.getTables();
+            String dbName = db.getFullName();
+            for (TableIf table : tables) {
+                if (table instanceof MTMV) {
+                    String tableName = table.getName();
+                    MTMVRelation relation = ((MTMV) table).getRelation();
+                    Set<BaseTableInfo> tablesOneLevel = relation.getBaseTablesOneLevel();
+                    for (BaseTableInfo info : tablesOneLevel) {
+                        TRow trow = new TRow();
+                        trow.addToColumnValue(new TCell().setStringVal("internal"));
+                        trow.addToColumnValue(new TCell().setStringVal(dbName));
+                        trow.addToColumnValue(new TCell().setStringVal(tableName));
+                        trow.addToColumnValue(new TCell().setStringVal(info.getCtlName()));
+                        trow.addToColumnValue(new TCell().setStringVal(info.getDbName()));
+                        trow.addToColumnValue(new TCell().setStringVal(info.getTableName()));
+                        trow.addToColumnValue(new TCell().setStringVal(info.getType()));
+                        dataBatch.add(trow);
+                    }
+                }
+            }
+        }
+        result.setDataBatch(dataBatch);
+        result.setStatus(new TStatus(TStatusCode.OK));
+        return result;
+    }
+
     private static TFetchSchemaTableDataResult workloadSchedPolicyMetadataResult(TSchemaTableRequestParams params) {
         if (!params.isSetCurrentUserIdent()) {
             return errorResult("current user ident is not set.");

gensrc/thrift/Descriptors.thrift

@@ -140,7 +140,8 @@ enum TSchemaTableType {
     SCH_FILE_CACHE_STATISTICS = 51,
     SCH_CATALOG_META_CACHE_STATISTICS = 52,
     SCH_BACKEND_KERBEROS_TICKET_CACHE = 53,
-    SCH_ROUTINE_LOAD_JOBS = 54;
+    SCH_ROUTINE_LOAD_JOBS = 54,
+    SCH_MV_DEPENDENCE = 55;
 }
 
 enum THdfsCompression {

gensrc/thrift/FrontendService.thrift

@@ -822,6 +822,7 @@ enum TSchemaTableName {
   TABLE_PROPERTIES = 8,
   CATALOG_META_CACHE_STATS = 9,
   PARTITIONS = 10,
+  MV_DEPENDENCE = 11,
 }
 
 struct TMetadataTableRequestParams {

regression-test/data/query_p0/schema_table/test_mv_dependence.out

@@ -0,0 +1,7 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql --
+internal	test_mv_dependence_db	mv_a	internal	test	stu	OLAP
+internal	test_mv_dependence_db	mv_b	internal	test	mv_a	MATERIALIZED_VIEW
+internal	test_mv_dependence_db	mv_c	internal	test	stu	OLAP
+internal	test_mv_dependence_db	mv_c	internal	test	grade	OLAP
+