[opt](catalog) support nested namespaces of iceberg

docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/iceberg/run20.sql

@@ -0,0 +1,2 @@
+create database if not exists nested.db1;
+

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

@@ -3375,7 +3375,7 @@ public void createDb(CreateDatabaseCommand command) throws DdlException {
         if (StringUtils.isEmpty(command.getCtlName())) {
             catalogIf = getCurrentCatalog();
         } else {
-            catalogIf = catalogMgr.getCatalog(command.getCtlName());
+            catalogIf = catalogMgr.getCatalogOrDdlException(command.getCtlName());
         }
         catalogIf.createDb(command.getDbName(), command.isIfNotExists(), command.getProperties());
     }

fe/fe-core/src/main/java/org/apache/doris/common/FeNameFormat.java

@@ -24,6 +24,7 @@
 import org.apache.doris.mysql.privilege.Role;
 import org.apache.doris.mysql.privilege.RoleManager;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.GlobalVariable;
 import org.apache.doris.qe.VariableMgr;
 
 import com.google.common.base.Strings;
@@ -53,6 +54,10 @@ public class FeNameFormat {
 
     public static final String TEMPORARY_TABLE_SIGN = "_#TEMP#_";
 
+    private static final String NESTED_DB_NAME_REGEX = "^[a-zA-Z][a-zA-Z0-9\\-_]*(\\.([a-zA-Z0-9\\-_]+))*$";
+    private static final String NESTED_UNICODE_DB_NAME_REGEX
+            = "^[a-zA-Z\\p{L}][a-zA-Z0-9\\-_\\p{L}]*(\\.([a-zA-Z0-9\\-_\\p{L}]+))*$";
+
     public static void checkCatalogName(String catalogName) throws AnalysisException {
         if (!InternalCatalog.INTERNAL_CATALOG_NAME.equals(catalogName) && (Strings.isNullOrEmpty(catalogName)
                 || !catalogName.matches(getCommonNameRegex()))) {
@@ -61,7 +66,10 @@ public static void checkCatalogName(String catalogName) throws AnalysisException
     }
 
     public static void checkDbName(String dbName) throws AnalysisException {
-        if (Strings.isNullOrEmpty(dbName) || !dbName.matches(getCommonNameRegex())) {
+        if (Strings.isNullOrEmpty(dbName)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_WRONG_DB_NAME, dbName);
+        }
+        if (!dbName.matches(getDbNameRegex())) {
             ErrorReport.reportAnalysisException(ErrorCode.ERR_WRONG_DB_NAME, dbName);
         }
     }
@@ -257,6 +265,22 @@ public static String getCommonNameRegex() {
         }
     }
 
+    public static String getDbNameRegex() {
+        if (GlobalVariable.enableNestedNamespace) {
+            if (FeNameFormat.isEnableUnicodeNameSupport()) {
+                return NESTED_UNICODE_DB_NAME_REGEX;
+            } else {
+                return NESTED_DB_NAME_REGEX;
+            }
+        } else {
+            if (FeNameFormat.isEnableUnicodeNameSupport()) {
+                return UNICODE_COMMON_NAME_REGEX;
+            } else {
+                return COMMON_NAME_REGEX;
+            }
+        }
+    }
+
     public static String getOutfileSuccessFileNameRegex() {
         if (FeNameFormat.isEnableUnicodeNameSupport()) {
             return UNICODE_UNDERSCORE_COMMON_NAME_REGEX;

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

@@ -33,6 +33,8 @@
 import org.apache.doris.datasource.ExternalDatabase;
 import org.apache.doris.datasource.ExternalTable;
 import org.apache.doris.datasource.operations.ExternalMetadataOps;
+import org.apache.doris.datasource.property.metastore.IcebergRestProperties;
+import org.apache.doris.datasource.property.metastore.MetastoreProperties;
 import org.apache.doris.nereids.trees.plans.commands.info.BranchOptions;
 import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceBranchInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceTagInfo;
@@ -54,12 +56,14 @@
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.catalog.ViewCatalog;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.expressions.Literal;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types.NestedField;
 import org.apache.iceberg.view.View;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.jetbrains.annotations.NotNull;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -68,6 +72,7 @@
 import java.util.Optional;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 public class IcebergMetadataOps implements ExternalMetadataOps {
 
@@ -129,15 +134,36 @@ public boolean databaseExist(String dbName) {
 
     public List<String> listDatabaseNames() {
         try {
-            return executionAuthenticator.execute(() -> nsCatalog.listNamespaces(getNamespace())
-                   .stream()
-                   .map(n -> n.level(n.length() - 1))
-                   .collect(Collectors.toList()));
+            return executionAuthenticator.execute(() -> listNestedNamespaces(getNamespace()));
         } catch (Exception e) {
             throw new RuntimeException("Failed to list database names, error message is:" + e.getMessage(), e);
         }
     }
 
+    @NotNull
+    private List<String> listNestedNamespaces(Namespace parentNs) {
+        // Handle nested namespaces for Iceberg REST catalog,
+        // only if "iceberg.rest.nested-namespace-enabled" is true.
+        if (dorisCatalog instanceof IcebergRestExternalCatalog) {
+            IcebergRestExternalCatalog restCatalog = (IcebergRestExternalCatalog) dorisCatalog;
+            MetastoreProperties metaProps = restCatalog.getCatalogProperty().getMetastoreProperties();
+            if (metaProps instanceof IcebergRestProperties
+                    && ((IcebergRestProperties) metaProps).isIcebergRestNestedNamespaceEnabled()) {
+                return nsCatalog.listNamespaces(parentNs)
+                        .stream()
+                        .flatMap(childNs -> Stream.concat(
+                                Stream.of(childNs.toString()),
+                                listNestedNamespaces(childNs).stream()
+                        )).collect(Collectors.toList());
+            }
+        }
+
+        return nsCatalog.listNamespaces(parentNs)
+                .stream()
+                .map(n -> n.level(n.length() - 1))
+                .collect(Collectors.toList());
+    }
+
     @Override
     public List<String> listTableNames(String dbName) {
         try {
@@ -209,7 +235,7 @@ private boolean performCreateDb(String dbName, boolean ifNotExists, Map<String,
     public void dropDbImpl(String dbName, boolean ifExists, boolean force) throws DdlException {
         try {
             executionAuthenticator.execute(() -> {
-                preformDropDb(dbName, ifExists, force);
+                performDropDb(dbName, ifExists, force);
                 return null;
             });
         } catch (Exception e) {
@@ -218,7 +244,7 @@ public void dropDbImpl(String dbName, boolean ifExists, boolean force) throws Dd
         }
     }
 
-    private void preformDropDb(String dbName, boolean ifExists, boolean force) throws DdlException {
+    private void performDropDb(String dbName, boolean ifExists, boolean force) throws DdlException {
         ExternalDatabase dorisDb = dorisCatalog.getDbNullable(dbName);
         if (dorisDb == null) {
             if (ifExists) {
@@ -229,21 +255,27 @@ private void preformDropDb(String dbName, boolean ifExists, boolean force) throw
             }
         }
         if (force) {
-            // try to drop all tables in the database
-            List<String> remoteTableNames = listTableNames(dorisDb.getRemoteName());
-            for (String remoteTableName : remoteTableNames) {
-                performDropTable(dorisDb.getRemoteName(), remoteTableName, true);
-            }
-            if (!remoteTableNames.isEmpty()) {
-                LOG.info("drop database[{}] with force, drop all tables, num: {}", dbName, remoteTableNames.size());
-            }
-            // try to drop all views in the database
-            List<String> remoteViewNames = listViewNames(dorisDb.getRemoteName());
-            for (String remoteViewName : remoteViewNames) {
-                performDropView(dorisDb.getRemoteName(), remoteViewName);
-            }
-            if (!remoteViewNames.isEmpty()) {
-                LOG.info("drop database[{}] with force, drop all views, num: {}", dbName, remoteViewNames.size());
+            try {
+                // try to drop all tables in the database
+                List<String> remoteTableNames = listTableNames(dorisDb.getRemoteName());
+                for (String remoteTableName : remoteTableNames) {
+                    performDropTable(dorisDb.getRemoteName(), remoteTableName, true);
+                }
+                if (!remoteTableNames.isEmpty()) {
+                    LOG.info("drop database[{}] with force, drop all tables, num: {}", dbName, remoteTableNames.size());
+                }
+                // try to drop all views in the database
+                List<String> remoteViewNames = listViewNames(dorisDb.getRemoteName());
+                for (String remoteViewName : remoteViewNames) {
+                    performDropView(dorisDb.getRemoteName(), remoteViewName);
+                }
+                if (!remoteViewNames.isEmpty()) {
+                    LOG.info("drop database[{}] with force, drop all views, num: {}", dbName, remoteViewNames.size());
+                }
+            } catch (NoSuchNamespaceException e) {
+                // just ignore
+                LOG.info("drop database[{}] force which does not exist", dbName);
+                return;
             }
         }
         nsCatalog.dropNamespace(getNamespace(dorisDb.getRemoteName()));

fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/IcebergRestProperties.java

@@ -110,9 +110,8 @@ public class IcebergRestProperties extends AbstractIcebergProperties {
 
     @ConnectorProperty(names = {"iceberg.rest.nested-namespace-enabled"},
             required = false,
-            supported = false,
             description = "Enable nested namespace for the iceberg rest catalog service.")
-    private String icebergRestNestedNamespaceEnabled = "true";
+    private String icebergRestNestedNamespaceEnabled = "false";
 
     @ConnectorProperty(names = {"iceberg.rest.case-insensitive-name-matching"},
             required = false,
@@ -306,6 +305,10 @@ public boolean isIcebergRestVendedCredentialsEnabled() {
         return Boolean.parseBoolean(icebergRestVendedCredentialsEnabled);
     }
 
+    public boolean isIcebergRestNestedNamespaceEnabled() {
+        return Boolean.parseBoolean(icebergRestNestedNamespaceEnabled);
+    }
+
     /**
      * Unified method to configure FileIO properties for Iceberg catalog.
      * This method handles all storage types (HDFS, S3, MinIO, etc.) and populates

fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlProto.java

@@ -18,21 +18,23 @@
 package org.apache.doris.mysql;
 
 import org.apache.doris.catalog.Env;
-import org.apache.doris.cloud.catalog.CloudEnv;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.Pair;
 import org.apache.doris.datasource.CatalogIf;
 import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.ConnectContextUtil;
 
 import com.google.common.base.Strings;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.Optional;
 
 // MySQL protocol util
 public class MysqlProto {
@@ -224,42 +226,9 @@ public static boolean negotiate(ConnectContext context) throws IOException {
         // set database
         String db = authPacket.getDb();
         if (!Strings.isNullOrEmpty(db)) {
-            String catalogName = null;
-            String dbName = null;
-            String[] dbNames = db.split("\\.");
-            if (dbNames.length == 1) {
-                dbName = db;
-            } else if (dbNames.length == 2) {
-                catalogName = dbNames[0];
-                dbName = dbNames[1];
-            } else if (dbNames.length > 2) {
-                context.getState().setError(ErrorCode.ERR_BAD_DB_ERROR, "Only one dot can be in the name: " + db);
-                sendResponsePacket(context);
-                return false;
-            }
-
-            // mysql -d
-            if (Config.isCloudMode()) {
-                try {
-                    dbName = ((CloudEnv) Env.getCurrentEnv()).analyzeCloudCluster(dbName, context);
-                } catch (DdlException e) {
-                    context.getState().setError(e.getMysqlErrorCode(), e.getMessage());
-                    sendResponsePacket(context);
-                    return false;
-                }
-
-                if (dbName == null || dbName.isEmpty()) {
-                    return true;
-                }
-            }
-
-            try {
-                if (catalogName != null) {
-                    context.getEnv().changeCatalog(context, catalogName);
-                }
-                Env.getCurrentEnv().changeDb(context, dbName);
-            } catch (DdlException e) {
-                context.getState().setError(e.getMysqlErrorCode(), e.getMessage());
+            Optional<Pair<ErrorCode, String>> res = ConnectContextUtil.initCatalogAndDb(context, db);
+            if (res.isPresent()) {
+                context.getState().setError(res.get().first, res.get().second);
                 sendResponsePacket(context);
                 return false;
             }

fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java

@@ -1042,6 +1042,7 @@
 import org.apache.doris.policy.FilterType;
 import org.apache.doris.policy.PolicyTypeEnum;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.GlobalVariable;
 import org.apache.doris.qe.SqlModeHelper;
 import org.apache.doris.resource.workloadschedpolicy.WorkloadActionMeta;
 import org.apache.doris.resource.workloadschedpolicy.WorkloadConditionMeta;
@@ -5246,15 +5247,21 @@ public RefreshDatabaseCommand visitRefreshDatabase(RefreshDatabaseContext ctx) {
         if (size == 0) {
             throw new ParseException("database name can't be empty");
         }
-        String dbName = parts.get(size - 1);
 
         // [db].
         if (size == 1) {
-            return new RefreshDatabaseCommand(dbName, properties);
+            return new RefreshDatabaseCommand(parts.get(0), properties);
         } else if (parts.size() == 2) {  // [ctl,db].
-            return new RefreshDatabaseCommand(parts.get(0), dbName, properties);
+            return new RefreshDatabaseCommand(parts.get(0), parts.get(1), properties);
+        } else {
+            if (GlobalVariable.enableNestedNamespace) {
+                // use the first part as catalog name, the rest part as db name
+                return new RefreshDatabaseCommand(parts.get(0),
+                        String.join(".", parts.subList(1, size)), properties);
+            } else {
+                throw new ParseException("Only one dot can be in the name: " + String.join(".", parts));
+            }
         }
-        throw new ParseException("Only one dot can be in the name: " + String.join(".", parts));
     }
 
     @Override

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropCatalogCommand.java

@@ -50,7 +50,7 @@ public void doRun(ConnectContext ctx, StmtExecutor executor) throws Exception {
         Util.checkCatalogAllRules(catalogName);
 
         if (catalogName.equals(InternalCatalog.INTERNAL_CATALOG_NAME)) {
-            throw new AnalysisException("Internal catalog can't be drop.");
+            throw new AnalysisException("Internal catalog can't be dropped.");
         }
 
         if (!Env.getCurrentEnv().getAccessManager().checkCtlPriv(