[enhancement](mysql compatible) add user and procs_priv tables to mysql db in all catalogs

be/src/exec/schema_scanner.cpp

@@ -44,6 +44,7 @@
 #include "exec/schema_scanner/schema_table_privileges_scanner.h"
 #include "exec/schema_scanner/schema_tables_scanner.h"
 #include "exec/schema_scanner/schema_user_privileges_scanner.h"
+#include "exec/schema_scanner/schema_user_scanner.h"
 #include "exec/schema_scanner/schema_variables_scanner.h"
 #include "exec/schema_scanner/schema_views_scanner.h"
 #include "exec/schema_scanner/schema_workload_groups_scanner.h"
@@ -164,6 +165,8 @@ std::unique_ptr<SchemaScanner> SchemaScanner::create(TSchemaTableType::type type
         return SchemaProcessListScanner::create_unique();
     case TSchemaTableType::SCH_PROCEDURES:
         return SchemaRoutinesScanner::create_unique();
+    case TSchemaTableType::SCH_USER:
+        return SchemaUserScanner::create_unique();
     default:
         return SchemaDummyScanner::create_unique();
         break;

be/src/exec/schema_scanner/schema_helper.cpp

@@ -37,6 +37,8 @@ class TShowVariableRequest;
 class TShowVariableResult;
 class TShowProcessListRequest;
 class TShowProcessListResult;
+class TShowUserRequest;
+class TShowUserResult;
 
 Status SchemaHelper::get_db_names(const std::string& ip, const int32_t port,
                                   const TGetDbsParams& request, TGetDbsResult* result) {
@@ -134,4 +136,12 @@ Status SchemaHelper::show_process_list(const std::string& ip, const int32_t port
             });
 }
 
+Status SchemaHelper::show_user(const std::string& ip, const int32_t port,
+                               const TShowUserRequest& request, TShowUserResult* result) {
+    return ThriftRpcHelper::rpc<FrontendServiceClient>(
+            ip, port, [&request, &result](FrontendServiceConnection& client) {
+                client->showUser(*result, request);
+            });
+}
+
 } // namespace doris

be/src/exec/schema_scanner/schema_helper.h

@@ -39,6 +39,8 @@ class TShowVariableRequest;
 class TShowVariableResult;
 class TShowProcessListRequest;
 class TShowProcessListResult;
+class TShowUserRequest;
+class TShowUserResult;
 
 // this class is a helper for getting schema info from FE
 class SchemaHelper {
@@ -82,6 +84,8 @@ class SchemaHelper {
     static Status show_process_list(const std::string& ip, const int32_t port,
                                     const TShowProcessListRequest& request,
                                     TShowProcessListResult* result);
+    static Status show_user(const std::string& ip, const int32_t port,
+                            const TShowUserRequest& request, TShowUserResult* result);
 };
 
 } // namespace doris

be/src/exec/schema_scanner/schema_user_scanner.cpp

@@ -0,0 +1,134 @@
+// 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_user_scanner.h"
+
+#include <gen_cpp/FrontendService_types.h>
+
+#include <vector>
+
+#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 {
+
+std::vector<SchemaScanner::ColumnDesc> SchemaUserScanner::_s_user_columns = {
+        {"Host", TYPE_CHAR, sizeof(StringRef), false},
+        {"User", TYPE_CHAR, sizeof(StringRef), false},
+        {"Node_priv", TYPE_CHAR, sizeof(StringRef), false},
+        {"Admin_priv", TYPE_CHAR, sizeof(StringRef), false},
+        {"Grant_priv", TYPE_CHAR, sizeof(StringRef), false},
+        {"Select_priv", TYPE_CHAR, sizeof(StringRef), false},
+        {"Load_priv", TYPE_CHAR, sizeof(StringRef), false},
+        {"Alter_priv", TYPE_CHAR, sizeof(StringRef), false},
+        {"Create_priv", TYPE_CHAR, sizeof(StringRef), false},
+        {"Drop_priv", TYPE_CHAR, sizeof(StringRef), false},
+        {"Usage_priv", TYPE_CHAR, sizeof(StringRef), false},
+        {"Show_view_priv", TYPE_CHAR, sizeof(StringRef), false},
+        {"Cluster_usage_priv", TYPE_CHAR, sizeof(StringRef), false},
+        {"Stage_usage_priv", TYPE_CHAR, sizeof(StringRef), false},
+        {"ssl_type", TYPE_CHAR, sizeof(StringRef), false},
+        {"ssl_cipher", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"x509_issuer", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"x509_subject", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"max_questions", TYPE_BIGINT, sizeof(int64_t), false},
+        {"max_updates", TYPE_BIGINT, sizeof(int64_t), false},
+        {"max_connections", TYPE_BIGINT, sizeof(int64_t), false},
+        {"max_user_connections", TYPE_BIGINT, sizeof(int64_t), false},
+        {"plugin", TYPE_CHAR, sizeof(StringRef), false},
+        {"authentication_string", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"password_policy.expiration_seconds", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"password_policy.password_creation_time", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"password_policy.history_num", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"password_policy.history_passwords", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"password_policy.num_failed_login", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"password_policy.password_lock_seconds", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"password_policy.failed_login_counter", TYPE_VARCHAR, sizeof(StringRef), false},
+        {"password_policy.lock_time", TYPE_VARCHAR, sizeof(StringRef), false}};
+
+SchemaUserScanner::SchemaUserScanner()
+        : SchemaScanner(_s_user_columns, TSchemaTableType::SCH_USER) {}
+
+SchemaUserScanner::~SchemaUserScanner() = default;
+
+Status SchemaUserScanner::start(RuntimeState* state) {
+    TShowUserRequest request;
+    RETURN_IF_ERROR(SchemaHelper::show_user(*(_param->common_param->ip), _param->common_param->port,
+                                            request, &_user_result));
+
+    return Status::OK();
+}
+
+Status SchemaUserScanner::get_next_block(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 (_user_result.userinfo_list.empty()) {
+        return Status::OK();
+    }
+
+    return _fill_block_impl(block);
+}
+
+Status SchemaUserScanner::_fill_block_impl(vectorized::Block* block) {
+    SCOPED_TIMER(_fill_block_timer);
+
+    const auto& userinfo_list = _user_result.userinfo_list;
+    size_t row_num = userinfo_list.size();
+    if (row_num == 0) {
+        return Status::OK();
+    }
+
+    for (size_t col_idx = 0; col_idx < _s_user_columns.size(); ++col_idx) {
+        std::vector<StringRef> str_refs(row_num);
+        std::vector<int64_t> int_vals(row_num);
+        std::vector<void*> datas(row_num);
+        std::vector<std::string> column_values(
+                row_num); // Store the strings to ensure their lifetime
+
+        for (size_t row_idx = 0; row_idx < row_num; ++row_idx) {
+            const auto& row = userinfo_list[row_idx];
+            std::string column_value = row.size() > col_idx ? row[col_idx] : "";
+
+            // Ensure column value assignment and type casting
+            column_values[row_idx] = column_value;
+            if (_s_user_columns[col_idx].type == TYPE_BIGINT) {
+                int64_t val = !column_value.empty() ? std::stoll(column_value) : 0;
+                int_vals[row_idx] = val;
+                datas[row_idx] = &int_vals[row_idx];
+            } else {
+                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_user_scanner.h

@@ -0,0 +1,53 @@
+// 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"
+#include "gen_cpp/FrontendService_types.h"
+
+namespace doris {
+
+class RuntimeState;
+
+namespace vectorized {
+class Block;
+}
+
+class SchemaUserScanner : public SchemaScanner {
+    ENABLE_FACTORY_CREATOR(SchemaUserScanner);
+
+public:
+    SchemaUserScanner();
+    ~SchemaUserScanner() override;
+
+    Status start(RuntimeState* state) override;
+    Status get_next_block(vectorized::Block* block, bool* eos) override;
+
+    static std::vector<SchemaScanner::ColumnDesc> _s_user_columns;
+
+private:
+    Status _fill_block_impl(vectorized::Block* block);
+
+    TShowUserResult _user_result;
+};
+
+} // namespace doris

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

@@ -19,6 +19,7 @@
 
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.InfoSchemaDb;
+import org.apache.doris.catalog.MysqlDb;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
@@ -79,6 +80,11 @@ public void analyze(Analyzer analyzer) throws AnalysisException, UserException {
             ErrorReport.reportAnalysisException(
                     ErrorCode.ERR_DBACCESS_DENIED_ERROR, analyzer.getQualifiedUser(), dbName);
         }
+        // Don't allow dropping 'mysql' database
+        if (dbName.equalsIgnoreCase(MysqlDb.DATABASE_NAME)) {
+            ErrorReport.reportAnalysisException(
+                    ErrorCode.ERR_DBACCESS_DENIED_ERROR, analyzer.getQualifiedUser(), dbName);
+        }
         // check access
         if (!Env.getCurrentEnv().getAccessManager().checkDbPriv(ConnectContext.get(), catalogName,
                 dbName, PrivPredicate.DROP)) {

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

@@ -72,7 +72,9 @@ public enum SchemaTableType {
     SCH_PROFILING("PROFILING", "PROFILING", TSchemaTableType.SCH_PROFILING),
     SCH_BACKEND_ACTIVE_TASKS("BACKEND_ACTIVE_TASKS", "BACKEND_ACTIVE_TASKS", TSchemaTableType.SCH_BACKEND_ACTIVE_TASKS),
     SCH_ACTIVE_QUERIES("ACTIVE_QUERIES", "ACTIVE_QUERIES", TSchemaTableType.SCH_ACTIVE_QUERIES),
-    SCH_WORKLOAD_GROUPS("WORKLOAD_GROUPS", "WORKLOAD_GROUPS", TSchemaTableType.SCH_WORKLOAD_GROUPS);
+    SCH_WORKLOAD_GROUPS("WORKLOAD_GROUPS", "WORKLOAD_GROUPS", TSchemaTableType.SCH_WORKLOAD_GROUPS),
+    SCHE_USER("user", "user", TSchemaTableType.SCH_USER),
+    SCH_PROCS_PRIV("procs_priv", "procs_priv", TSchemaTableType.SCH_PROCS_PRIV);
     private static final String dbName = "INFORMATION_SCHEMA";
     private static SelectList fullSelectLists;
 

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

@@ -0,0 +1,98 @@
+// 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.
+
+package org.apache.doris.catalog;
+
+import org.apache.doris.analysis.SchemaTableType;
+import org.apache.doris.common.SystemIdGenerator;
+import org.apache.doris.thrift.TSchemaTable;
+import org.apache.doris.thrift.TTableDescriptor;
+import org.apache.doris.thrift.TTableType;
+
+import com.google.common.collect.ImmutableMap;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Doris's representation of MySQL mysql table metadata.
+ */
+public class MysqlDBTable extends SchemaTable {
+
+    public static Map<String, Table> TABLE_MAP = ImmutableMap.<String, Table>builder().put("user",
+                    new MysqlDBTable(SystemIdGenerator.getNextId(), "user", TableType.SCHEMA,
+                            builder().column("Host", ScalarType.createCharType(255))
+                                    .column("User", ScalarType.createCharType(32))
+                                    .column("Node_priv", ScalarType.createCharType(1))
+                                    .column("Admin_priv", ScalarType.createCharType(1))
+                                    .column("Grant_priv", ScalarType.createCharType(1))
+                                    .column("Select_priv", ScalarType.createCharType(1))
+                                    .column("Load_priv", ScalarType.createCharType(1))
+                                    .column("Alter_priv", ScalarType.createCharType(1))
+                                    .column("Create_priv", ScalarType.createCharType(1))
+                                    .column("Drop_priv", ScalarType.createCharType(1))
+                                    .column("Usage_priv", ScalarType.createCharType(1))
+                                    .column("Show_view_priv", ScalarType.createCharType(1))
+                                    .column("Cluster_usage_priv", ScalarType.createCharType(1))
+                                    .column("Stage_usage_priv", ScalarType.createCharType(1))
+                                    .column("ssl_type", ScalarType.createCharType(9))
+                                    .column("ssl_cipher", ScalarType.createVarcharType(ScalarType.MAX_VARCHAR_LENGTH))
+                                    .column("x509_issuer", ScalarType.createVarcharType(ScalarType.MAX_VARCHAR_LENGTH))
+                                    .column("x509_subject", ScalarType.createVarcharType(ScalarType.MAX_VARCHAR_LENGTH))
+                                    .column("max_questions", Type.BIGINT)
+                                    .column("max_updates", Type.BIGINT)
+                                    .column("max_connections", Type.BIGINT)
+                                    .column("max_user_connections", Type.BIGINT)
+                                    .column("plugin", ScalarType.createCharType(64))
+                                    .column("authentication_string",
+                                            ScalarType.createVarcharType(ScalarType.MAX_VARCHAR_LENGTH))
+                                    .column("password_policy.expiration_seconds", ScalarType.createVarcharType(32))
+                                    .column("password_policy.password_creation_time", ScalarType.createVarcharType(32))
+                                    .column("password_policy.history_num", ScalarType.createVarcharType(32))
+                                    .column("password_policy.history_passwords",
+                                            ScalarType.createVarcharType(ScalarType.MAX_VARCHAR_LENGTH))
+                                    .column("password_policy.num_failed_login", ScalarType.createVarcharType(32))
+                                    .column("password_policy.password_lock_seconds", ScalarType.createVarcharType(32))
+                                    .column("password_policy.failed_login_counter", ScalarType.createVarcharType(32))
+                                    .column("password_policy.lock_time", ScalarType.createVarcharType(32))
+                                    .build()))
+            .put("procs_priv",
+                    new MysqlDBTable(SystemIdGenerator.getNextId(), "procs_priv", TableType.SCHEMA,
+                            builder().column("Host", ScalarType.createCharType(60))
+                                    .column("Db", ScalarType.createCharType(64))
+                                    .column("User", ScalarType.createCharType(32))
+                                    .column("Routine_name", ScalarType.createCharType(64))
+                                    .column("Routine_type", ScalarType.createCharType(9))
+                                    .column("Grantor", ScalarType.createCharType(93))
+                                    .column("Proc_priv", ScalarType.createCharType(16))
+                                    .column("Timestamp", ScalarType.createCharType(1))
+                                    .build()))
+            .build();
+
+    public MysqlDBTable(long id, String tableName, TableType type, List<Column> fullSchema) {
+        super(id, tableName, type, fullSchema);
+    }
+
+    @Override
+    public TTableDescriptor toThrift() {
+        TSchemaTable tSchemaTable = new TSchemaTable(SchemaTableType.getThriftType(this.name));
+        TTableDescriptor tTableDescriptor = new TTableDescriptor(getId(), TTableType.SCHEMA_TABLE,
+                TABLE_MAP.get(this.name).getBaseSchema().size(), 0, this.name, "");
+        tTableDescriptor.setSchemaTable(tSchemaTable);
+        return tTableDescriptor;
+    }
+}

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

@@ -49,7 +49,11 @@ public MysqlDb() {
      * If we need tables of mysql database in the future, create a MysqlTable class like {@link SchemaTable}
      */
     @Override
-    public void initTables() {}
+    public void initTables() {
+        for (Table table : MysqlDBTable.TABLE_MAP.values()) {
+            super.registerTable(table);
+        }
+    }
 
     @Override
     public boolean registerTable(TableIf table) {