[feature] Support alter database comment and custom properties

LiebingYu · LiebingYu · commit 791bffc392e8 · 2025-06-21T16:02:02.000+08:00
diff --git a/fluss-client/src/main/java/com/alibaba/fluss/client/admin/Admin.java b/fluss-client/src/main/java/com/alibaba/fluss/client/admin/Admin.java
@@ -161,6 +161,24 @@ CompletableFuture<Void> dropDatabase(
     /** List all databases in fluss cluster asynchronously. */
     CompletableFuture<List<String>> listDatabases();
 
+    /**
+     * Alter a database asynchronously.
+     *
+     * <p>The following exceptions can be anticipated when calling {@code get()} on returned future.
+     *
+     * <ul>
+     *   <li>{@link DatabaseNotExistException} if the database not exists and {@code
+     *       ignoreIfNotExists} is false.
+     * </ul>
+     *
+     * @param databaseName The name of the database to alter.
+     * @param databaseDescriptor The descriptor of the database to alter.
+     * @param ignoreIfNotExists Flag to specify behavior when a database with the given name not
+     *     exists: if set to false, throw a DatabaseNotExistException, if set to true, do nothing.
+     */
+    CompletableFuture<Void> alterDatabase(
+            String databaseName, DatabaseDescriptor databaseDescriptor, boolean ignoreIfNotExists);
+
     /**
      * Create a new table asynchronously.
      *
diff --git a/fluss-client/src/main/java/com/alibaba/fluss/client/admin/FlussAdmin.java b/fluss-client/src/main/java/com/alibaba/fluss/client/admin/FlussAdmin.java
@@ -39,6 +39,7 @@
 import com.alibaba.fluss.rpc.RpcClient;
 import com.alibaba.fluss.rpc.gateway.AdminGateway;
 import com.alibaba.fluss.rpc.gateway.TabletServerGateway;
+import com.alibaba.fluss.rpc.messages.AlterDatabaseRequest;
 import com.alibaba.fluss.rpc.messages.CreateAclsRequest;
 import com.alibaba.fluss.rpc.messages.CreateDatabaseRequest;
 import com.alibaba.fluss.rpc.messages.CreateTableRequest;
@@ -214,6 +215,16 @@ public CompletableFuture<List<String>> listDatabases() {
                 .thenApply(ListDatabasesResponse::getDatabaseNamesList);
     }
 
+    @Override
+    public CompletableFuture<Void> alterDatabase(
+            String databaseName, DatabaseDescriptor databaseDescriptor, boolean ignoreIfNotExists) {
+        AlterDatabaseRequest request = new AlterDatabaseRequest();
+        request.setDatabaseJson(databaseDescriptor.toJsonBytes())
+                .setDatabaseName(databaseName)
+                .setIgnoreIdNotExists(ignoreIfNotExists);
+        return gateway.alterDatabase(request).thenApply(r -> null);
+    }
+
     @Override
     public CompletableFuture<Void> createTable(
             TablePath tablePath, TableDescriptor tableDescriptor, boolean ignoreIfExists) {
diff --git a/fluss-client/src/test/java/com/alibaba/fluss/client/admin/FlussAdminITCase.java b/fluss-client/src/test/java/com/alibaba/fluss/client/admin/FlussAdminITCase.java
@@ -134,6 +134,7 @@ void testGetDatabaseInfo() throws Exception {
                 DatabaseDescriptor.builder()
                         .comment("test comment")
                         .customProperty("key1", "value1")
+                        .customProperty("key2", "value2")
                         .build(),
                 false);
         DatabaseInfo databaseInfo = admin.getDatabaseInfo("test_db_2").get();
@@ -144,9 +145,30 @@ void testGetDatabaseInfo() throws Exception {
                 .isEqualTo("test comment");
         assertThat(databaseInfo.getDatabaseDescriptor().getCustomProperties())
                 .containsEntry("key1", "value1");
-        assertThat(databaseInfo.getDatabaseDescriptor().getCustomProperties()).hasSize(1);
+        assertThat(databaseInfo.getDatabaseDescriptor().getCustomProperties()).hasSize(2);
         assertThat(databaseInfo.getCreatedTime())
                 .isBetween(timestampBeforeCreate, timestampAfterCreate);
+
+        long timesBeforeAlter = System.currentTimeMillis();
+        admin.alterDatabase(
+                "test_db_2",
+                DatabaseDescriptor.builder()
+                        .comment("alter comment")
+                        .customProperty("key1", "new_value1")
+                        .customProperty("key3", "val3")
+                        .build(),
+                false);
+        databaseInfo = admin.getDatabaseInfo("test_db_2").get();
+        long timestampAfterAlter = System.currentTimeMillis();
+        assertThat(databaseInfo.getCreatedTime()).isLessThan(databaseInfo.getModifiedTime());
+        assertThat(databaseInfo.getDatabaseName()).isEqualTo("test_db_2");
+        assertThat(databaseInfo.getDatabaseDescriptor().getComment().get())
+                .isEqualTo("alter comment");
+        assertThat(databaseInfo.getDatabaseDescriptor().getCustomProperties())
+                .containsEntry("key1", "new_value1")
+                .containsEntry("key3", "val3");
+        assertThat(databaseInfo.getDatabaseDescriptor().getCustomProperties()).hasSize(2);
+        assertThat(databaseInfo.getModifiedTime()).isBetween(timesBeforeAlter, timestampAfterAlter);
     }
 
     @Test
diff --git a/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/catalog/FlinkCatalog.java b/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/catalog/FlinkCatalog.java
@@ -230,9 +230,21 @@ public void dropDatabase(String databaseName, boolean ignoreIfNotExists, boolean
     }
 
     @Override
-    public void alterDatabase(String databaseName, CatalogDatabase catalogDatabase, boolean b)
+    public void alterDatabase(
+            String databaseName, CatalogDatabase catalogDatabase, boolean ignoreIfNotExists)
             throws DatabaseNotExistException, CatalogException {
-        throw new UnsupportedOperationException();
+        try {
+            admin.alterDatabase(databaseName, toFlussDatabase(catalogDatabase), ignoreIfNotExists);
+        } catch (Exception e) {
+            Throwable t = ExceptionUtils.stripExecutionException(e);
+            if (CatalogExceptionUtils.isDatabaseNotExist(t)) {
+                throw new DatabaseNotExistException(getName(), databaseName);
+            } else {
+                throw new CatalogException(
+                        String.format("Failed to alter database %s in %s", databaseName, getName()),
+                        t);
+            }
+        }
     }
 
     @Override
diff --git a/fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/catalog/FlinkCatalogITCase.java b/fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/catalog/FlinkCatalogITCase.java
@@ -511,6 +511,7 @@ void testCreateWithUnSupportDataType() {
 
     @Test
     void testCreateDatabase() {
+        // create database without anything
         tEnv.executeSql("create database test_db");
         List<Row> databases =
                 CollectionUtil.iteratorToList(tEnv.executeSql("show databases").collect());
@@ -521,6 +522,16 @@ void testCreateDatabase() {
         tEnv.executeSql("drop database test_db");
         databases = CollectionUtil.iteratorToList(tEnv.executeSql("show databases").collect());
         assertThat(databases.toString()).isEqualTo(String.format("[+I[%s]]", DEFAULT_DB));
+
+        // create database with comment and custom properties
+        tEnv.executeSql("create database test_db comment 'test_db' with('key1' = 'val1')");
+        databases = CollectionUtil.iteratorToList(tEnv.executeSql("show databases").collect());
+
+        assertThat(databases.stream().map(Row::toString).collect(Collectors.toList()))
+                .containsExactlyInAnyOrderElementsOf(
+                        Arrays.asList(String.format("+I[%s]", DEFAULT_DB), "+I[test_db]"));
+        // alter database
+        tEnv.executeSql("alter database test_db set('key1' = 'new_val1', 'key2' = 'val2')");
     }
 
     @Test
diff --git a/fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/catalog/FlinkCatalogTest.java b/fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/catalog/FlinkCatalogTest.java
@@ -394,6 +394,17 @@ void testDatabase() throws Exception {
         assertThat(db2.getComment()).isEqualTo("test comment");
         assertThat(db2.getProperties())
                 .isEqualTo(Collections.singletonMap(SCAN_STARTUP_MODE.key(), "earliest"));
+        // test alter db2
+        catalog.alterDatabase(
+                "db2",
+                new CatalogDatabaseImpl(
+                        Collections.singletonMap(SCAN_STARTUP_MODE.key(), "latest"),
+                        "alter comment"),
+                false);
+        db2 = catalog.getDatabase("db2");
+        assertThat(db2.getComment()).isEqualTo("alter comment");
+        assertThat(db2.getProperties())
+                .isEqualTo(Collections.singletonMap(SCAN_STARTUP_MODE.key(), "latest"));
         // test create table in db1
         ObjectPath path1 = new ObjectPath("db1", "t1");
         CatalogTable table = this.newCatalogTable(new HashMap<>());
@@ -432,8 +443,6 @@ void testDatabase() throws Exception {
         assertThatThrownBy(() -> catalog.listTables("unknown"))
                 .isInstanceOf(DatabaseNotExistException.class)
                 .hasMessage("Database %s does not exist in Catalog %s.", "unknown", CATALOG_NAME);
-        assertThatThrownBy(() -> catalog.alterDatabase("db2", null, false))
-                .isInstanceOf(UnsupportedOperationException.class);
         assertThat(catalog.getDefaultDatabase()).isEqualTo(DEFAULT_DB);
 
         // Test catalog with null default database
diff --git a/fluss-rpc/src/main/java/com/alibaba/fluss/rpc/gateway/AdminGateway.java b/fluss-rpc/src/main/java/com/alibaba/fluss/rpc/gateway/AdminGateway.java
@@ -17,6 +17,8 @@
 
 package com.alibaba.fluss.rpc.gateway;
 
+import com.alibaba.fluss.rpc.messages.AlterDatabaseRequest;
+import com.alibaba.fluss.rpc.messages.AlterDatabaseResponse;
 import com.alibaba.fluss.rpc.messages.CreateAclsRequest;
 import com.alibaba.fluss.rpc.messages.CreateAclsResponse;
 import com.alibaba.fluss.rpc.messages.CreateDatabaseRequest;
@@ -56,6 +58,14 @@ public interface AdminGateway extends AdminReadOnlyGateway {
     @RPC(api = ApiKeys.DROP_DATABASE)
     CompletableFuture<DropDatabaseResponse> dropDatabase(DropDatabaseRequest request);
 
+    /**
+     * Alter a database.
+     *
+     * @param request Alter database request.
+     */
+    @RPC(api = ApiKeys.ALTER_DATABASE)
+    CompletableFuture<AlterDatabaseResponse> alterDatabase(AlterDatabaseRequest request);
+
     /**
      * Creates a new table.
      *
diff --git a/fluss-rpc/src/main/java/com/alibaba/fluss/rpc/protocol/ApiKeys.java b/fluss-rpc/src/main/java/com/alibaba/fluss/rpc/protocol/ApiKeys.java
@@ -70,7 +70,8 @@ public enum ApiKeys {
     CREATE_ACLS(1039, 0, 0, PUBLIC),
     LIST_ACLS(1040, 0, 0, PUBLIC),
     DROP_ACLS(1041, 0, 0, PUBLIC),
-    LAKE_TIERING_HEARTBEAT(1042, 0, 0, PRIVATE);
+    LAKE_TIERING_HEARTBEAT(1042, 0, 0, PRIVATE),
+    ALTER_DATABASE(1043, 0, 0, PUBLIC);
 
     private static final Map<Integer, ApiKeys> ID_TO_TYPE =
             Arrays.stream(ApiKeys.values())
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -63,6 +63,16 @@ message CreateDatabaseRequest {
 message CreateDatabaseResponse {
 }
 
+// alter database request and response
+message AlterDatabaseRequest {
+  required string database_name = 1;
+  required bool ignore_id_not_exists = 2;
+  required bytes database_json = 3;
+}
+
+message AlterDatabaseResponse {
+}
+
 // get table request and response
 message GetDatabaseInfoRequest {
   required string database_name = 1;
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/CoordinatorService.java b/fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/CoordinatorService.java
@@ -42,6 +42,8 @@
 import com.alibaba.fluss.rpc.gateway.CoordinatorGateway;
 import com.alibaba.fluss.rpc.messages.AdjustIsrRequest;
 import com.alibaba.fluss.rpc.messages.AdjustIsrResponse;
+import com.alibaba.fluss.rpc.messages.AlterDatabaseRequest;
+import com.alibaba.fluss.rpc.messages.AlterDatabaseResponse;
 import com.alibaba.fluss.rpc.messages.CommitKvSnapshotRequest;
 import com.alibaba.fluss.rpc.messages.CommitKvSnapshotResponse;
 import com.alibaba.fluss.rpc.messages.CommitLakeTableSnapshotRequest;
@@ -220,6 +222,23 @@ public CompletableFuture<DropDatabaseResponse> dropDatabase(DropDatabaseRequest
         return CompletableFuture.completedFuture(response);
     }
 
+    @Override
+    public CompletableFuture<AlterDatabaseResponse> alterDatabase(AlterDatabaseRequest request) {
+        if (authorizer != null) {
+            authorizer.authorize(
+                    currentSession(),
+                    OperationType.ALTER,
+                    Resource.database(request.getDatabaseName()));
+        }
+
+        AlterDatabaseResponse response = new AlterDatabaseResponse();
+        DatabaseDescriptor databaseDescriptor =
+                DatabaseDescriptor.fromJsonBytes(request.getDatabaseJson());
+        metadataManager.alterDatabase(
+                request.getDatabaseName(), databaseDescriptor, request.isIgnoreIdNotExists());
+        return CompletableFuture.completedFuture(response);
+    }
+
     @Override
     public CompletableFuture<CreateTableResponse> createTable(CreateTableRequest request) {
         TablePath tablePath = toTablePath(request.getTablePath());
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/MetadataManager.java b/fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/MetadataManager.java
@@ -209,6 +209,32 @@ public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade
         uncheck(() -> zookeeperClient.deleteDatabase(name), "Fail to drop database: " + name);
     }
 
+    public void alterDatabase(
+            String databaseName, DatabaseDescriptor databaseDescriptor, boolean ignoreIfNotExists) {
+        if (!databaseExists(databaseName)) {
+            if (ignoreIfNotExists) {
+                return;
+            }
+            throw new DatabaseNotExistException("Database " + databaseName + " does not exist.");
+        }
+
+        try {
+            DatabaseInfo databaseInfo = getDatabase(databaseName);
+            DatabaseRegistration databaseRegistration =
+                    DatabaseRegistration.of(databaseDescriptor, databaseInfo.getCreatedTime());
+            zookeeperClient.alterDatabase(databaseName, databaseRegistration);
+        } catch (Exception e) {
+            if (e instanceof KeeperException.NoNodeException) {
+                if (ignoreIfNotExists) {
+                    return;
+                }
+                throw new DatabaseNotExistException("Database " + databaseName + " not exists.");
+            } else {
+                throw new FlussRuntimeException("Failed to alter database: " + databaseName, e);
+            }
+        }
+    }
+
     public void dropTable(TablePath tablePath, boolean ignoreIfNotExists)
             throws TableNotExistException {
         if (!tableExists(tablePath)) {
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/zk/ZooKeeperClient.java b/fluss-server/src/main/java/com/alibaba/fluss/server/zk/ZooKeeperClient.java
@@ -342,6 +342,12 @@ public void deleteDatabase(String database) throws Exception {
         zkClient.delete().deletingChildrenIfNeeded().forPath(path);
     }
 
+    public void alterDatabase(String database, DatabaseRegistration databaseRegistration)
+            throws Exception {
+        String path = DatabaseZNode.path(database);
+        zkClient.setData().forPath(path, DatabaseZNode.encode(databaseRegistration));
+    }
+
     public boolean databaseExists(String database) throws Exception {
         String path = DatabaseZNode.path(database);
         return zkClient.checkExists().forPath(path) != null;
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/zk/data/DatabaseRegistration.java b/fluss-server/src/main/java/com/alibaba/fluss/server/zk/data/DatabaseRegistration.java
@@ -63,6 +63,15 @@ public static DatabaseRegistration of(DatabaseDescriptor databaseDescriptor) {
                 currentMillis);
     }
 
+    public static DatabaseRegistration of(DatabaseDescriptor databaseDescriptor, long createdTime) {
+        final long currentMillis = System.currentTimeMillis();
+        return new DatabaseRegistration(
+                databaseDescriptor.getComment().orElse(null),
+                databaseDescriptor.getCustomProperties(),
+                createdTime,
+                currentMillis);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (o == null || getClass() != o.getClass()) {
diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/coordinator/TestCoordinatorGateway.java b/fluss-server/src/test/java/com/alibaba/fluss/server/coordinator/TestCoordinatorGateway.java
@@ -22,6 +22,8 @@
 import com.alibaba.fluss.rpc.gateway.CoordinatorGateway;
 import com.alibaba.fluss.rpc.messages.AdjustIsrRequest;
 import com.alibaba.fluss.rpc.messages.AdjustIsrResponse;
+import com.alibaba.fluss.rpc.messages.AlterDatabaseRequest;
+import com.alibaba.fluss.rpc.messages.AlterDatabaseResponse;
 import com.alibaba.fluss.rpc.messages.ApiVersionsRequest;
 import com.alibaba.fluss.rpc.messages.ApiVersionsResponse;
 import com.alibaba.fluss.rpc.messages.CommitKvSnapshotRequest;
@@ -127,6 +129,11 @@ public CompletableFuture<DropDatabaseResponse> dropDatabase(DropDatabaseRequest
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    public CompletableFuture<AlterDatabaseResponse> alterDatabase(AlterDatabaseRequest request) {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public CompletableFuture<CreateTableResponse> createTable(CreateTableRequest request) {
         throw new UnsupportedOperationException();
diff --git a/website/docs/engine-flink/ddl.md b/website/docs/engine-flink/ddl.md
@@ -75,6 +75,15 @@ USE fluss;
 DROP DATABASE my_db;
 ```
 
+## Alter Database
+
+The following SQL sets one or more properties in the specified database. If a particular property is already set in the database, override the old value with the new one.
+
+```sql title="Flink SQL"
+-- alter the database
+ALTER DATABASE my_db SET ('key1' = 'val1', 'key2' = 'val2');
+```
+
 ## Create Table
 
 ### PrimaryKey Table
