fix: review

Author: gyang94

fluss-client/src/main/java/com/alibaba/fluss/client/admin/Admin.java

@@ -240,7 +240,9 @@ CompletableFuture<Void> createTable(
      * <p>The following exceptions can be anticipated when calling {@code get()} on returned future.
      *
      * <ul>
-     *   <li>{@link DatabaseNotExistException} if the database does not exist.
+     *   <li>{@link DatabaseNotExistException} when the database does not exist.
+     *   <li>{@link TableNotExistException} when the table does not exist, if ignoreIfNotExists is
+     *       false.
      * </ul>
      *
      * @param tablePath The table path of the table.

fluss-client/src/test/java/com/alibaba/fluss/client/admin/FlussAdminITCase.java

@@ -197,6 +197,69 @@ void testGetTableInfoAndSchema() throws Exception {
                 .isBetween(timestampBeforeCreate, timestampAfterCreate);
     }
 
+    @Test
+    void testAlterTable() throws Exception {
+        // create table
+        TablePath tablePath = TablePath.of("test_db", "alter_table_1");
+        admin.createTable(tablePath, DEFAULT_TABLE_DESCRIPTOR, false).get();
+
+        TableInfo tableInfo = admin.getTableInfo(tablePath).get();
+
+        TableDescriptor existingTableDescriptor = tableInfo.toTableDescriptor();
+        Map<String, String> updateProperties =
+                new HashMap<>(existingTableDescriptor.getProperties());
+        Map<String, String> updateCustomProperties =
+                new HashMap<>(existingTableDescriptor.getCustomProperties());
+        updateProperties.put("table.datalake.enabled", "true");
+        updateCustomProperties.put("table.datalake.enabled", "true");
+
+        TableDescriptor newTableDescriptor =
+                TableDescriptor.builder()
+                        .schema(existingTableDescriptor.getSchema())
+                        .comment(existingTableDescriptor.getComment().orElse("test table"))
+                        .partitionedBy(existingTableDescriptor.getPartitionKeys())
+                        .distributedBy(
+                                existingTableDescriptor
+                                        .getTableDistribution()
+                                        .get()
+                                        .getBucketCount()
+                                        .orElse(3),
+                                existingTableDescriptor.getBucketKeys())
+                        .properties(updateProperties)
+                        .customProperties(updateCustomProperties)
+                        .build();
+        // alter table
+        admin.alterTable(tablePath, newTableDescriptor, false).get();
+
+        TableInfo alteredTableInfo = admin.getTableInfo(tablePath).get();
+        TableDescriptor alteredTableDescriptor = alteredTableInfo.toTableDescriptor();
+        assertThat(alteredTableDescriptor).isEqualTo(newTableDescriptor);
+
+        // throw exception if table not exist
+        assertThatThrownBy(
+                        () ->
+                                admin.alterTable(
+                                                TablePath.of("test_db", "alter_table_not_exist"),
+                                                newTableDescriptor,
+                                                false)
+                                        .get())
+                .cause()
+                .isInstanceOf(TableNotExistException.class);
+
+        // throw exception if database not exist
+        assertThatThrownBy(
+                        () ->
+                                admin.alterTable(
+                                                TablePath.of(
+                                                        "test_db_not_exist",
+                                                        "alter_table_not_exist"),
+                                                newTableDescriptor,
+                                                false)
+                                        .get())
+                .cause()
+                .isInstanceOf(DatabaseNotExistException.class);
+    }
+
     @Test
     void testCreateInvalidDatabaseAndTable() {
         assertThatThrownBy(

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/catalog/FlinkCatalog.java

@@ -399,7 +399,7 @@ public void alterTable(
             admin.alterTable(tablePath, tableDescriptor, ignoreIfNotExist).get();
         } catch (Exception e) {
             Throwable t = ExceptionUtils.stripExecutionException(e);
-            if (CatalogExceptionUtils.isTableAlreadyExist(t)) {
+            if (CatalogExceptionUtils.isTableNotExist(t)) {
                 throw new TableNotExistException(getName(), objectPath);
             } else if (isTableInvalid(t)) {
                 throw new InvalidTableException(t.getMessage());

fluss-rpc/src/main/proto/FlussApi.proto

@@ -116,7 +116,6 @@ message AlterTableRequest {
 }
 
 message AlterTableResponse {
-
 }
 
 // get table request and response

fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/MetadataManager.java

@@ -305,8 +305,8 @@ public long createTable(
 
     public void alterTable(
             TablePath tablePath, TableDescriptor tableDescriptor, boolean ignoreIfNotExists) {
-        // validate table properties before creating table
-        validateTableDescriptor(tableDescriptor, maxBucketNum);
+        // validate table properties before altering table
+        validateAlterTableProperties(tableDescriptor);
 
         if (!databaseExists(tablePath.getDatabaseName())) {
             throw new DatabaseNotExistException(