[feature](multi-catalog)support hms catalog create and drop table/db

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -2216,6 +2216,21 @@ public class Config extends ConfigBase {
         "Sample size for hive row count estimation."})
     public static int hive_stats_partition_sample_size = 3000;
 
+    @ConfField(mutable = true, masterOnly = true, description = {
+            "Hive创建外部表默认指定的input format",
+            "Default hive input format for creating table."})
+    public static String hive_default_input_format = "org.apache.hadoop.hive.ql.io.orc.OrcInputFormat";
+
+    @ConfField(mutable = true, masterOnly = true, description = {
+            "Hive创建外部表默认指定的output format",
+            "Default hive output format for creating table."})
+    public static String hive_default_output_format = "org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat";
+
+    @ConfField(mutable = true, masterOnly = true, description = {
+            "Hive创建外部表默认指定的SerDe类",
+            "Default hive serde class for creating table."})
+    public static String hive_default_serde = "org.apache.hadoop.hive.ql.io.orc.OrcSerde";
+
     @ConfField
     public static int statistics_sql_parallel_exec_instance_num = 1;
 

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

@@ -411,8 +411,8 @@ private void processModifyEngineInternal(Database db, Table externalTable,
         }
         odbcTable.writeLock();
         try {
-            db.dropTable(mysqlTable.getName());
-            db.createTable(odbcTable);
+            db.unregisterTable(mysqlTable.getName());
+            db.registerTable(odbcTable);
             if (!isReplay) {
                 ModifyTableEngineOperationLog log = new ModifyTableEngineOperationLog(db.getId(),
                         externalTable.getId(), prop);
@@ -591,17 +591,17 @@ private void replaceTableInternal(Database db, OlapTable origTable, OlapTable ne
         String newTblName = newTbl.getName();
 
         // drop origin table and new table
-        db.dropTable(oldTblName);
-        db.dropTable(newTblName);
+        db.unregisterTable(oldTblName);
+        db.unregisterTable(newTblName);
 
         // rename new table name to origin table name and add it to database
         newTbl.checkAndSetName(oldTblName, false);
-        db.createTable(newTbl);
+        db.registerTable(newTbl);
 
         if (swapTable) {
             // rename origin table name to new table name and add it to database
             origTable.checkAndSetName(newTblName, false);
-            db.createTable(origTable);
+            db.registerTable(origTable);
         } else {
             // not swap, the origin table is not used anymore, need to drop all its tablets.
             Env.getCurrentEnv().onEraseOlapTable(origTable, isReplay);
@@ -633,8 +633,8 @@ private void modifyViewDef(Database db, View view, String inlineViewDef, long sq
                 }
                 view.setNewFullSchema(newFullSchema);
                 String viewName = view.getName();
-                db.dropTable(viewName);
-                db.createTable(view);
+                db.unregisterTable(viewName);
+                db.registerTable(view);
 
                 AlterViewInfo alterViewInfo = new AlterViewInfo(db.getId(), view.getId(),
                         inlineViewDef, newFullSchema, sqlMode);
@@ -669,8 +669,8 @@ public void replayModifyViewDef(AlterViewInfo alterViewInfo) throws MetaNotFound
             }
             view.setNewFullSchema(newFullSchema);
 
-            db.dropTable(viewName);
-            db.createTable(view);
+            db.unregisterTable(viewName);
+            db.registerTable(view);
 
             LOG.info("replay modify view[{}] definition to {}", viewName, inlineViewDef);
         } finally {

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

@@ -39,7 +39,6 @@
 import org.apache.doris.common.util.ParseUtil;
 import org.apache.doris.common.util.PrintableMap;
 import org.apache.doris.common.util.PropertyAnalyzer;
-import org.apache.doris.common.util.Util;
 import org.apache.doris.datasource.es.EsUtil;
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.qe.ConnectContext;
@@ -282,16 +281,13 @@ public void analyze(Analyzer analyzer) throws UserException {
             // FIXME: MOW is not supported in cloud mode yet.
             properties.put(PropertyAnalyzer.ENABLE_UNIQUE_KEY_MERGE_ON_WRITE, "false");
         }
-
         if (Strings.isNullOrEmpty(engineName) || engineName.equalsIgnoreCase(DEFAULT_ENGINE_NAME)) {
             this.properties = maybeRewriteByAutoBucket(distributionDesc, properties);
         }
 
         super.analyze(analyzer);
         tableName.analyze(analyzer);
         FeNameFormat.checkTableName(tableName.getTbl());
-        // disallow external catalog
-        Util.prohibitExternalCatalog(tableName.getCtl(), this.getClass().getSimpleName());
         InternalDatabaseUtil.checkDatabase(tableName.getDb(), ConnectContext.get());
         if (!Env.getCurrentEnv().getAccessManager()
                 .checkTblPriv(ConnectContext.get(), tableName.getDb(), tableName.getTbl(), PrivPredicate.CREATE)) {

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

@@ -22,7 +22,6 @@
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.util.InternalDatabaseUtil;
-import org.apache.doris.common.util.Util;
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.qe.ConnectContext;
 
@@ -84,8 +83,6 @@ public void analyze(Analyzer analyzer) throws UserException {
             tableName.setDb(analyzer.getDefaultDb());
         }
         tableName.analyze(analyzer);
-        // disallow external catalog
-        Util.prohibitExternalCatalog(tableName.getCtl(), this.getClass().getSimpleName());
         InternalDatabaseUtil.checkDatabase(tableName.getDb(), ConnectContext.get());
         // check access
         if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ConnectContext.get(), tableName.getDb(),

fe/fe-core/src/main/java/org/apache/doris/backup/RestoreJob.java

@@ -905,7 +905,7 @@ private void checkAndPrepareMeta() {
                 }
                 tbl.writeLock();
                 try {
-                    if (!db.createTable(tbl)) {
+                    if (!db.registerTable(tbl)) {
                         status = new Status(ErrCode.COMMON_ERROR, "Table " + tbl.getName()
                                 + " already exist in db: " + db.getFullName());
                         return;
@@ -1286,7 +1286,7 @@ private void replayCheckAndPrepareMeta() {
             db.writeLock();
             restoreTbl.writeLock();
             try {
-                db.createTable(restoreTbl);
+                db.registerTable(restoreTbl);
             } finally {
                 restoreTbl.writeUnlock();
                 db.writeUnlock();
@@ -1936,7 +1936,7 @@ public void cancelInternal(boolean isReplay) {
                                         }
                                     }
                                 }
-                                db.dropTable(restoreTbl.getName());
+                                db.unregisterTable(restoreTbl.getName());
                             } finally {
                                 restoreTbl.writeUnlock();
                             }

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

@@ -644,7 +644,7 @@ private void recoverAllTables(RecycleDatabaseInfo dbInfo) throws DdlException {
             }
 
             Table table = tableInfo.getTable();
-            db.createTable(table);
+            db.registerTable(table);
             LOG.info("recover db[{}] with table[{}]: {}", dbId, table.getId(), table.getName());
             iterator.remove();
             idToRecycleTime.remove(table.getId());
@@ -739,7 +739,7 @@ private synchronized boolean innerRecoverTable(Database db, Table table, String
                 }
             }
 
-            db.createTable(table);
+            db.registerTable(table);
             if (isReplay) {
                 iterator.remove();
             } else {

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

@@ -416,25 +416,26 @@ public Pair<Boolean, Boolean> createTableWithLock(
         }
     }
 
-    public boolean createTable(Table table) {
+    public boolean registerTable(TableIf table) {
         boolean result = true;
-        table.setQualifiedDbName(fullQualifiedName);
-        String tableName = table.getName();
+        Table olapTable = (Table) table;
+        olapTable.setQualifiedDbName(fullQualifiedName);
+        String tableName = olapTable.getName();
         if (Env.isStoredTableNamesLowerCase()) {
             tableName = tableName.toLowerCase();
         }
         if (isTableExist(tableName)) {
             result = false;
         } else {
-            idToTable.put(table.getId(), table);
-            nameToTable.put(table.getName(), table);
+            idToTable.put(olapTable.getId(), olapTable);
+            nameToTable.put(olapTable.getName(), olapTable);
             lowerCaseToTableName.put(tableName.toLowerCase(), tableName);
         }
-        table.unmarkDropped();
+        olapTable.unmarkDropped();
         return result;
     }
 
-    public void dropTable(String tableName) {
+    public void unregisterTable(String tableName) {
         if (Env.isStoredTableNamesLowerCase()) {
             tableName = tableName.toLowerCase();
         }

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

@@ -263,7 +263,18 @@ default OlapTable getOlapTableOrAnalysisException(String tableName) throws Analy
         return (OlapTable) table;
     }
 
-    void dropTable(String tableName);
+    /**
+     * register table to memory
+     * @param table created table
+     * @return true if add to memory
+     */
+    boolean registerTable(TableIf table);
+
+    /**
+     * unregister table from memory
+     * @param tableName table name
+     */
+    void unregisterTable(String tableName);
 
     CatalogIf getCatalog();
 

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

@@ -2962,7 +2962,7 @@ public Frontend getFeByName(String name) {
 
     // The interface which DdlExecutor needs.
     public void createDb(CreateDbStmt stmt) throws DdlException {
-        getInternalCatalog().createDb(stmt);
+        getCurrentCatalog().createDb(stmt);
     }
 
     // For replay edit log, need't lock metadata
@@ -2975,7 +2975,7 @@ public void replayCreateDb(Database db) {
     }
 
     public void dropDb(DropDbStmt stmt) throws DdlException {
-        getInternalCatalog().dropDb(stmt);
+        getCurrentCatalog().dropDb(stmt);
     }
 
     public void replayDropDb(String dbName, boolean isForceDrop, Long recycleTime) throws DdlException {
@@ -3047,7 +3047,7 @@ public void replayRenameDatabase(String dbName, String newDbName) {
      * 11. add this table to ColocateGroup if necessary
      */
     public void createTable(CreateTableStmt stmt) throws UserException {
-        getInternalCatalog().createTable(stmt);
+        getCurrentCatalog().createTable(stmt);
     }
 
     public void createTableLike(CreateTableLikeStmt stmt) throws DdlException {
@@ -3677,7 +3677,7 @@ public void replayAlterExternalTableSchema(String dbName, String tableName, List
 
     // Drop table
     public void dropTable(DropTableStmt stmt) throws DdlException {
-        getInternalCatalog().dropTable(stmt);
+        getCurrentCatalog().dropTable(stmt);
     }
 
     public boolean unprotectDropTable(Database db, Table table, boolean isForceDrop, boolean isReplay,
@@ -4326,8 +4326,8 @@ public void renameTable(Database db, Table table, String newTableName) throws Dd
                     table.setName(newTableName);
                 }
 
-                db.dropTable(oldTableName);
-                db.createTable(table);
+                db.unregisterTable(oldTableName);
+                db.registerTable(table);
 
                 TableInfo tableInfo = TableInfo.createForTableRename(db.getId(), table.getId(), newTableName);
                 editLog.logTableRename(tableInfo);
@@ -4359,9 +4359,9 @@ public void replayRenameTable(TableInfo tableInfo) throws MetaNotFoundException
             table.writeLock();
             try {
                 String tableName = table.getName();
-                db.dropTable(tableName);
+                db.unregisterTable(tableName);
                 table.setName(newTableName);
-                db.createTable(table);
+                db.registerTable(table);
                 LOG.info("replay rename table[{}] to {}", tableName, newTableName);
             } finally {
                 table.writeUnlock();

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

@@ -29,12 +29,12 @@ public InfoSchemaDb() {
     @Override
     protected void initTables() {
         for (Table table : SchemaTable.TABLE_MAP.values()) {
-            super.createTable(table);
+            super.registerTable(table);
         }
     }
 
     @Override
-    public boolean createTable(Table table) {
+    public boolean registerTable(TableIf table) {
         return false;
     }
 }

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

@@ -50,12 +50,12 @@ public MysqlCompatibleDatabase(long id, String name) {
      * @note: Rename a table of mysql database in MYSQL ls allowed.
      */
     @Override
-    public boolean createTable(Table table) {
-        return super.createTable(table);
+    public boolean registerTable(TableIf table) {
+        return super.registerTable(table);
     }
 
     @Override
-    public void dropTable(String name) {
+    public void unregisterTable(String name) {
         // Do nothing
     }
 

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

@@ -52,7 +52,7 @@ public MysqlDb() {
     public void initTables() {}
 
     @Override
-    public boolean createTable(Table table) {
+    public boolean registerTable(TableIf table) {
         return false;
     }
 }

fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogIf.java

@@ -17,13 +17,18 @@
 
 package org.apache.doris.datasource;
 
+import org.apache.doris.analysis.CreateDbStmt;
+import org.apache.doris.analysis.CreateTableStmt;
+import org.apache.doris.analysis.DropDbStmt;
+import org.apache.doris.analysis.DropTableStmt;
 import org.apache.doris.catalog.DatabaseIf;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.MetaNotFoundException;
+import org.apache.doris.common.UserException;
 
 import com.google.common.base.Strings;
 import com.google.common.collect.Lists;
@@ -177,4 +182,12 @@ default CatalogLog constructEditLog() {
     boolean enableAutoAnalyze();
 
     ConcurrentHashMap<Long, DatabaseIf> getIdToDb();
+
+    void createDb(CreateDbStmt stmt) throws DdlException;
+
+    void dropDb(DropDbStmt stmt) throws DdlException;
+
+    void createTable(CreateTableStmt stmt) throws UserException;
+
+    void dropTable(DropTableStmt stmt) throws DdlException;
 }