[Core] Support alter database

docs/content/program-api/catalog-api.md

@@ -82,7 +82,7 @@ public class ListDatabases {
 
 ## Drop Database
 
-You can use the catalog to drop databases.
+You can use the catalog to drop database.
 
 ```java
 import org.apache.paimon.catalog.Catalog;
@@ -102,6 +102,30 @@ public class DropDatabase {
 }
 ```
 
+## Alter Database
+
+You can use the catalog to alter database's properties.(ps: only support hive and jdbc catalog)
+
+```java
+import java.util.ArrayList;
+import org.apache.paimon.catalog.Catalog;
+
+public class AlterDatabase {
+
+    public static void main(String[] args) {
+        try {
+            Catalog catalog = CreateCatalog.createHiveCatalog();
+            List<DatabaseChange> changes = new ArrayList<>();
+            changes.add(DatabaseChange.setProperty("k1", "v1"));
+            changes.add(DatabaseChange.removeProperty("k2"));
+            catalog.alterDatabase("my_db", changes, true);
+        } catch (Catalog.DatabaseNotExistException e) {
+            // do something
+        }
+    }
+}
+```
+
 ## Determine Whether Table Exists
 
 You can use the catalog to determine whether the table exists

paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java

@@ -229,6 +229,26 @@ public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade
 
     protected abstract void dropDatabaseImpl(String name);
 
+    @Override
+    public void alterDatabase(String name, List<PropertyChange> changes, boolean ignoreIfNotExists)
+            throws DatabaseNotExistException {
+        checkNotSystemDatabase(name);
+        try {
+            if (changes == null || changes.isEmpty()) {
+                return;
+            }
+            alterDatabaseImpl(name, changes);
+        } catch (DatabaseNotExistException e) {
+            if (ignoreIfNotExists) {
+                return;
+            }
+            throw new DatabaseNotExistException(name);
+        }
+    }
+
+    protected abstract void alterDatabaseImpl(String name, List<PropertyChange> changes)
+            throws DatabaseNotExistException;
+
     @Override
     public List<String> listTables(String databaseName) throws DatabaseNotExistException {
         if (isSystemDatabase(databaseName)) {

paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java

@@ -187,6 +187,13 @@ public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade
         }
     }
 
+    @Override
+    public void alterDatabase(String name, List<PropertyChange> changes, boolean ignoreIfNotExists)
+            throws DatabaseNotExistException {
+        super.alterDatabase(name, changes, ignoreIfNotExists);
+        databaseCache.invalidate(name);
+    }
+
     @Override
     public void dropTable(Identifier identifier, boolean ignoreIfNotExists)
             throws TableNotExistException {

paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java

@@ -125,6 +125,19 @@ void createDatabase(String name, boolean ignoreIfExists, Map<String, String> pro
     void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade)
             throws DatabaseNotExistException, DatabaseNotEmptyException;
 
+    /**
+     * Alter a database.
+     *
+     * @param name Name of the database to alter.
+     * @param changes the property changes
+     * @param ignoreIfNotExists Flag to specify behavior when the database does not exist: if set to
+     *     false, throw an exception, if set to true, do nothing.
+     * @throws DatabaseNotExistException if the given database is not exist and ignoreIfNotExists is
+     *     false
+     */
+    void alterDatabase(String name, List<PropertyChange> changes, boolean ignoreIfNotExists)
+            throws DatabaseNotExistException;
+
     /**
      * Return a {@link Table} identified by the given {@link Identifier}.
      *

paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java

@@ -84,6 +84,12 @@ public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade
         wrapped.dropDatabase(name, ignoreIfNotExists, cascade);
     }
 
+    @Override
+    public void alterDatabase(String name, List<PropertyChange> changes, boolean ignoreIfNotExists)
+            throws DatabaseNotExistException {
+        wrapped.alterDatabase(name, changes, ignoreIfNotExists);
+    }
+
     @Override
     public List<String> listTables(String databaseName) throws DatabaseNotExistException {
         return wrapped.listTables(databaseName);

paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java

@@ -92,6 +92,11 @@ protected void dropDatabaseImpl(String name) {
         uncheck(() -> fileIO.delete(newDatabasePath(name), true));
     }
 
+    @Override
+    protected void alterDatabaseImpl(String name, List<PropertyChange> changes) {
+        throw new UnsupportedOperationException("Alter database is not supported.");
+    }
+
     @Override
     protected List<String> listTablesImpl(String databaseName) {
         return uncheck(() -> listTablesInFileSystem(newDatabasePath(databaseName)));

paimon-core/src/main/java/org/apache/paimon/catalog/PropertyChange.java

@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.catalog;
+
+import org.apache.paimon.utils.Pair;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.Maps;
+import org.apache.paimon.shade.guava30.com.google.common.collect.Sets;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** define change to the database property. */
+public interface PropertyChange {
+
+    static PropertyChange setProperty(String property, String value) {
+        return new SetProperty(property, value);
+    }
+
+    static PropertyChange removeProperty(String property) {
+        return new RemoveProperty(property);
+    }
+
+    static Pair<Map<String, String>, Set<String>> getSetPropertiesToRemoveKeys(
+            List<PropertyChange> changes) {
+        Map<String, String> setProperties = Maps.newHashMap();
+        Set<String> removeKeys = Sets.newHashSet();
+        changes.forEach(
+                change -> {
+                    if (change instanceof PropertyChange.SetProperty) {
+                        PropertyChange.SetProperty setProperty =
+                                (PropertyChange.SetProperty) change;
+                        setProperties.put(setProperty.property(), setProperty.value());
+                    } else {
+                        removeKeys.add(((PropertyChange.RemoveProperty) change).property());
+                    }
+                });
+        return Pair.of(setProperties, removeKeys);
+    }
+
+    /** Set property for database change. */
+    final class SetProperty implements PropertyChange {
+
+        private final String property;
+        private final String value;
+
+        private SetProperty(String property, String value) {
+            this.property = property;
+            this.value = value;
+        }
+
+        public String property() {
+            return this.property;
+        }
+
+        public String value() {
+            return this.value;
+        }
+    }
+
+    /** Remove property for database change. */
+    final class RemoveProperty implements PropertyChange {
+
+        private final String property;
+
+        private RemoveProperty(String property) {
+            this.property = property;
+        }
+
+        public String property() {
+            return this.property;
+        }
+    }
+}

paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java

@@ -24,6 +24,7 @@
 import org.apache.paimon.catalog.CatalogLockFactory;
 import org.apache.paimon.catalog.Database;
 import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.catalog.PropertyChange;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.operation.Lock;
@@ -33,11 +34,13 @@
 import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.utils.Pair;
 import org.apache.paimon.utils.Preconditions;
 
 import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap;
 import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
 import org.apache.paimon.shade.guava30.com.google.common.collect.Maps;
+import org.apache.paimon.shade.guava30.com.google.common.collect.Sets;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -52,12 +55,15 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 import static org.apache.paimon.jdbc.JdbcCatalogLock.acquireTimeout;
 import static org.apache.paimon.jdbc.JdbcCatalogLock.checkMaxSleep;
+import static org.apache.paimon.jdbc.JdbcUtils.deleteProperties;
 import static org.apache.paimon.jdbc.JdbcUtils.execute;
 import static org.apache.paimon.jdbc.JdbcUtils.insertProperties;
+import static org.apache.paimon.jdbc.JdbcUtils.updateProperties;
 import static org.apache.paimon.jdbc.JdbcUtils.updateTable;
 
 /* This file is based on source code from the Iceberg Project (http://iceberg.apache.org/), licensed by the Apache
@@ -197,6 +203,45 @@ protected void dropDatabaseImpl(String name) {
         execute(connections, JdbcUtils.DELETE_ALL_DATABASE_PROPERTIES_SQL, catalogKey, name);
     }
 
+    @Override
+    protected void alterDatabaseImpl(String name, List<PropertyChange> changes) {
+        Pair<Map<String, String>, Set<String>> setPropertiesToRemoveKeys =
+                PropertyChange.getSetPropertiesToRemoveKeys(changes);
+        Map<String, String> setProperties = setPropertiesToRemoveKeys.getLeft();
+        Set<String> removeKeys = setPropertiesToRemoveKeys.getRight();
+        Map<String, String> startingProperties = fetchProperties(name);
+        Map<String, String> inserts = Maps.newHashMap();
+        Map<String, String> updates = Maps.newHashMap();
+        Set<String> removes = Sets.newHashSet();
+        if (!setProperties.isEmpty()) {
+            setProperties.forEach(
+                    (k, v) -> {
+                        if (!startingProperties.containsKey(k)) {
+                            inserts.put(k, v);
+                        } else {
+                            updates.put(k, v);
+                        }
+                    });
+        }
+        if (!removeKeys.isEmpty()) {
+            removeKeys.forEach(
+                    k -> {
+                        if (startingProperties.containsKey(k)) {
+                            removes.add(k);
+                        }
+                    });
+        }
+        if (!inserts.isEmpty()) {
+            insertProperties(connections, catalogKey, name, inserts);
+        }
+        if (!updates.isEmpty()) {
+            updateProperties(connections, catalogKey, name, updates);
+        }
+        if (!removes.isEmpty()) {
+            deleteProperties(connections, catalogKey, name, removes);
+        }
+    }
+
     @Override
     protected List<String> listTablesImpl(String databaseName) {
         return fetch(