@Override
    public CatalogDatabase getDatabase(String databaseName)
            throws DatabaseNotExistException, CatalogException {
        if (asNamespaceCatalog == null) {
            if (!getDefaultDatabase().equals(databaseName)) {
                throw new DatabaseNotExistException(getName(), databaseName);
            } else {
                return new CatalogDatabaseImpl(Maps.newHashMap(), "");
            }
        } else {
            try {
                Map<String, String> metadata =
                        Maps.newHashMap(
                                asNamespaceCatalog.loadNamespaceMetadata(appendLevel(baseNamespace, databaseName)));
                String comment = metadata.remove("comment");
                return new CatalogDatabaseImpl(metadata, comment);
            } catch (NoSuchNamespaceException e) {
                throw new DatabaseNotExistException(getName(), databaseName, e);
            }
        }
    }
Append a new level to the base namespace
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/FlinkCatalog.java