[feature](multi-catalog) add specified_database_list for hms,iceberg

docs/en/docs/lakehouse/multi-catalog/hive.md

@@ -55,6 +55,11 @@ CREATE CATALOG hive PROPERTIES (
 ```
 
  In addition to `type` and  `hive.metastore.uris` , which are required, you can specify other parameters regarding the connection.
+
+> `specified_database_list`:
+> 
+> only synchronize the specified databases, split with ','. Default values is '' will synchronize all databases. db name is case sensitive.
+> 
 
 For example, to specify HDFS HA:
 

docs/en/docs/lakehouse/multi-catalog/iceberg.md

@@ -53,6 +53,11 @@ CREATE CATALOG iceberg PROPERTIES (
 );
 ```
 
+> `specified_database_list`:
+> 
+> only synchronize the specified databases, split with ','. Default values is '' will synchronize all databases. db name is case sensitive.
+> 
+
 ### Iceberg Native Catalog
 
 <version since="dev">

docs/zh-CN/docs/lakehouse/multi-catalog/hive.md

@@ -53,6 +53,11 @@ CREATE CATALOG hive PROPERTIES (
 ```
 
 除了 `type` 和 `hive.metastore.uris` 两个必须参数外，还可以通过更多参数来传递连接所需要的信息。
+
+> `specified_database_list`:
+> 
+> 支持只同步指定的同步多个database，以','分隔。默认为''，同步所有database。db名称是大小写敏感的。
+> 
 
 如提供 HDFS HA 信息，示例如下：
 

docs/zh-CN/docs/lakehouse/multi-catalog/iceberg.md

@@ -51,6 +51,11 @@ CREATE CATALOG iceberg PROPERTIES (
 );
 ```
 
+> `specified_database_list`:
+> 
+> 支持只同步指定的同步多个database，以','分隔。默认为''，同步所有database。db名称是大小写敏感的。
+> 
+
 ### 基于Iceberg API创建Catalog
 
 <version since="dev">

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

@@ -46,6 +46,7 @@
 public abstract class Resource implements Writable, GsonPostProcessable {
     private static final Logger LOG = LogManager.getLogger(OdbcCatalogResource.class);
     public static final String REFERENCE_SPLIT = "@";
+    public static final String SPECIFIED_DATABASE_LIST = "specified_database_list";
 
     public enum ResourceType {
         UNKNOWN,

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

@@ -19,6 +19,7 @@
 
 import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.Resource;
 import org.apache.doris.catalog.external.EsExternalDatabase;
 import org.apache.doris.catalog.external.ExternalDatabase;
 import org.apache.doris.catalog.external.ExternalTable;
@@ -417,4 +418,21 @@ public void dropDatabase(String dbName) {
     public void createDatabase(long dbId, String dbName) {
         throw new NotImplementedException();
     }
+
+    public Map getSpecifiedDatabaseMap() {
+        String specifiedDatabaseList = catalogProperty.getOrDefault(Resource.SPECIFIED_DATABASE_LIST, "");
+        Map<String, Boolean> specifiedDatabaseMap = Maps.newHashMap();
+        specifiedDatabaseList = specifiedDatabaseList.trim();
+        if (specifiedDatabaseList.isEmpty()) {
+            return specifiedDatabaseMap;
+        }
+        String[] databaseList = specifiedDatabaseList.split(",");
+        for (int i = 0; i < databaseList.length; i++) {
+            String dbname = databaseList[i].trim();
+            if (!dbname.isEmpty()) {
+                specifiedDatabaseMap.put(dbname, true);
+            }
+        }
+        return specifiedDatabaseMap;
+    }
 }

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

@@ -112,8 +112,12 @@ protected void init() {
         initCatalogLog.setCatalogId(id);
         initCatalogLog.setType(InitCatalogLog.Type.HMS);
         List<String> allDatabases = client.getAllDatabases();
+        Map<String, Boolean> specifiedDatabaseMap = getSpecifiedDatabaseMap();
         // Update the db name to id map.
         for (String dbName : allDatabases) {
+            if (!specifiedDatabaseMap.isEmpty() && specifiedDatabaseMap.get(dbName) == null) {
+                continue;
+            }
             long dbId;
             if (dbNameToId != null && dbNameToId.containsKey(dbName)) {
                 dbId = dbNameToId.get(dbName);

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

@@ -67,7 +67,11 @@ protected void init() {
         initCatalogLog.setCatalogId(id);
         initCatalogLog.setType(InitCatalogLog.Type.ICEBERG);
         List<String> allDatabaseNames = listDatabaseNames();
+        Map<String, Boolean> specifiedDatabaseMap = getSpecifiedDatabaseMap();
         for (String dbName : allDatabaseNames) {
+            if (!specifiedDatabaseMap.isEmpty() && specifiedDatabaseMap.get(dbName) == null) {
+                continue;
+            }
             long dbId;
             if (dbNameToId != null && dbNameToId.containsKey(dbName)) {
                 dbId = dbNameToId.get(dbName);