branch-3.1: [opt](catalog) support nested namespaces of iceberg #56415 #56874

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/analysis/DropCatalogStmt.java

@@ -53,7 +53,7 @@ public void analyze(Analyzer analyzer) throws UserException {
         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(

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

@@ -3350,12 +3350,12 @@ public void createDb(CreateDbStmt stmt) throws DdlException {
         if (StringUtils.isEmpty(stmt.getCtlName())) {
             catalogIf = getCurrentCatalog();
         } else {
-            catalogIf = catalogMgr.getCatalog(stmt.getCtlName());
+            catalogIf = catalogMgr.getCatalogOrDdlException(stmt.getCtlName());
         }
         catalogIf.createDb(stmt.getFullDbName(), stmt.isSetIfNotExists(), stmt.getProperties());
     }
 
-    // For replay edit log, need't lock metadata
+    // For replay edit log, no need to lock metadata
     public void unprotectCreateDb(Database db) {
         getInternalCatalog().unprotectCreateDb(db);
     }

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

@@ -25,6 +25,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);
         }
     }
@@ -253,6 +261,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/CatalogMgr.java

@@ -160,6 +160,12 @@ public CatalogIf getCatalogOrAnalysisException(long id) throws AnalysisException
                         ErrorCode.ERR_UNKNOWN_CATALOG));
     }
 
+    public CatalogIf getCatalogOrDdlException(long id) throws DdlException {
+        return getCatalogOrException(id,
+                catalog -> new DdlException(ErrorCode.ERR_UNKNOWN_CATALOG.formatErrorMsg(catalog),
+                        ErrorCode.ERR_UNKNOWN_CATALOG));
+    }
+
     public <E extends Exception> CatalogIf<? extends DatabaseIf<? extends TableIf>>
             getCatalogOrException(long id, Function<Long, E> e) throws E {
         CatalogIf catalog = idToCatalog.get(id);

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

@@ -34,13 +34,17 @@
 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;
 import org.apache.doris.nereids.trees.plans.commands.info.DropBranchInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.DropTagInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.TagOptions;
 
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Splitter;
 import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.apache.iceberg.ManageSnapshots;
 import org.apache.iceberg.PartitionSpec;
@@ -54,20 +58,24 @@
 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.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 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 +137,35 @@ 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) {
@@ -210,7 +238,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) {
@@ -219,7 +247,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) {
@@ -230,21 +258,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()));
@@ -761,7 +795,8 @@ public View loadView(String dbName, String tblName) {
         }
         try {
             ViewCatalog viewCatalog = (ViewCatalog) catalog;
-            return executionAuthenticator.execute(() -> viewCatalog.loadView(TableIdentifier.of(dbName, tblName)));
+            return executionAuthenticator.execute(
+                    () -> viewCatalog.loadView(TableIdentifier.of(getNamespace(dbName), tblName)));
         } catch (Exception e) {
             throw new RuntimeException("Failed to load view, error message is:" + e.getMessage(), e);
         }
@@ -774,23 +809,30 @@ public List<String> listViewNames(String db) {
         }
         try {
             return executionAuthenticator.execute(() ->
-                ((ViewCatalog) catalog).listViews(Namespace.of(db))
+                    ((ViewCatalog) catalog).listViews(getNamespace(db))
                     .stream().map(TableIdentifier::name).collect(Collectors.toList()));
         } catch (Exception e) {
             throw new RuntimeException("Failed to list view names, error message is:" + e.getMessage(), e);
         }
     }
 
     private TableIdentifier getTableIdentifier(String dbName, String tblName) {
-        return externalCatalogName
-            .map(s -> TableIdentifier.of(s, dbName, tblName))
-            .orElseGet(() -> TableIdentifier.of(dbName, tblName));
+        Namespace ns = getNamespace(dbName);
+        return TableIdentifier.of(ns, tblName);
     }
 
     private Namespace getNamespace(String dbName) {
-        return externalCatalogName
-            .map(s -> Namespace.of(s, dbName))
-            .orElseGet(() -> Namespace.of(dbName));
+        return getNamespace(externalCatalogName, dbName);
+    }
+
+    @VisibleForTesting
+    public static Namespace getNamespace(Optional<String> catalogName, String dbName) {
+        String[] splits = Splitter.on(".").omitEmptyStrings().trimResults().splitToList(dbName).toArray(new String[0]);
+        if (catalogName.isPresent()) {
+            splits = Arrays.copyOf(splits, splits.length + 1);
+            splits[splits.length - 1] = catalogName.get();
+        }
+        return Namespace.of(splits);
     }
 
     private Namespace getNamespace() {

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,
@@ -201,11 +200,7 @@ private ParamRules buildRules() {
         ParamRules rules = new ParamRules()
                 // OAuth2 requires either credential or token, but not both
                 .mutuallyExclusive(icebergRestOauth2Credential, icebergRestOauth2Token,
-                        "OAuth2 cannot have both credential and token configured")
-                // If using credential flow, server URI is required
-                .requireAllIfPresent(icebergRestOauth2Credential,
-                        new String[] {icebergRestOauth2ServerUri},
-                        "OAuth2 credential flow requires server-uri");
+                        "OAuth2 cannot have both credential and token configured");
 
         // Custom validation: OAuth2 scope should not be used with token
         if (Strings.isNotBlank(icebergRestOauth2Token) && Strings.isNotBlank(icebergRestOauth2Scope)) {
@@ -275,7 +270,9 @@ private void addOAuth2Properties() {
         if (Strings.isNotBlank(icebergRestOauth2Credential)) {
             // Client Credentials Flow
             icebergRestCatalogProperties.put(OAuth2Properties.CREDENTIAL, icebergRestOauth2Credential);
-            icebergRestCatalogProperties.put(OAuth2Properties.OAUTH2_SERVER_URI, icebergRestOauth2ServerUri);
+            if (Strings.isNotBlank(icebergRestOauth2ServerUri)) {
+                icebergRestCatalogProperties.put(OAuth2Properties.OAUTH2_SERVER_URI, icebergRestOauth2ServerUri);
+            }
             if (Strings.isNotBlank(icebergRestOauth2Scope)) {
                 icebergRestCatalogProperties.put(OAuth2Properties.SCOPE, icebergRestOauth2Scope);
             }
@@ -306,6 +303,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