@Override
    public void alterDatabase(String name, CatalogDatabase newDatabase, boolean ignoreIfNotExists)
            throws DatabaseNotExistException, CatalogException {
        if (asNamespaceCatalog != null) {
            Namespace namespace = appendLevel(baseNamespace, name);
            Map<String, String> updates = Maps.newHashMap();
            Set<String> removals = Sets.newHashSet();

            try {
                Map<String, String> oldProperties = asNamespaceCatalog.loadNamespaceMetadata(namespace);
                Map<String, String> newProperties =
                        mergeComment(newDatabase.getProperties(), newDatabase.getComment());

                for (String key : oldProperties.keySet()) {
                    if (!newProperties.containsKey(key)) {
                        removals.add(key);
                    }
                }

                for (Map.Entry<String, String> entry : newProperties.entrySet()) {
                    if (!entry.getValue().equals(oldProperties.get(entry.getKey()))) {
                        updates.put(entry.getKey(), entry.getValue());
                    }
                }

                if (!updates.isEmpty()) {
                    asNamespaceCatalog.setProperties(namespace, updates);
                }

                if (!removals.isEmpty()) {
                    asNamespaceCatalog.removeProperties(namespace, removals);
                }

            } catch (NoSuchNamespaceException e) {
                if (!ignoreIfNotExists) {
                    throw new DatabaseNotExistException(getName(), name, e);
                }
            }
        } else {
            if (getDefaultDatabase().equals(name)) {
                throw new CatalogException(
                        "Can not alter the default database when the iceberg catalog doesn't support namespaces.");
            }
            if (!ignoreIfNotExists) {
                throw new DatabaseNotExistException(getName(), name);
            }
        }
    }
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