(fluss) branch main updated: [lake] Forbidden schema evolution for lake enabled tables. (#2150)

[yuxia]

This is an automated email from the ASF dual-hosted git repository.

yuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git


The following commit(s) were added to refs/heads/main by this push:
     new e02837d74 [lake] Forbidden schema evolution for lake enabled tables. 
(#2150)
e02837d74 is described below

commit e02837d746b18ba19a9a921a4730847cfeaf467f
Author: Liebing <44154062+liebin...@users.noreply.github.com>
AuthorDate: Thu Dec 11 16:49:37 2025 +0800

    [lake] Forbidden schema evolution for lake enabled tables. (#2150)
---
 .../lake/paimon/LakeEnabledTableCreateITCase.java  | 56 ++++++++++++++++++++++
 .../fluss/server/coordinator/MetadataManager.java  |  7 +++
 2 files changed, 63 insertions(+)

diff --git 
a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java
 
b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java
index 1ba61dafb..699553db6 100644
--- 
a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java
+++ 
b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java
@@ -23,6 +23,7 @@ import org.apache.fluss.client.admin.Admin;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.exception.InvalidAlterTableException;
 import org.apache.fluss.exception.InvalidConfigException;
 import org.apache.fluss.exception.InvalidTableException;
 import org.apache.fluss.exception.LakeTableAlreadyExistException;
@@ -857,6 +858,61 @@ class LakeEnabledTableCreateITCase {
         admin.alterTable(tablePath, Collections.singletonList(enableLake), 
false).get();
     }
 
+    @Test
+    void testAlterLakeEnabledTableSchema() throws Exception {
+        // create table
+        TableDescriptor tableDescriptor =
+                TableDescriptor.builder()
+                        .schema(
+                                Schema.newBuilder()
+                                        .column("c1", DataTypes.INT())
+                                        .column("c2", DataTypes.STRING())
+                                        .build())
+                        .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
+                        .customProperties(new HashMap<>())
+                        .distributedBy(BUCKET_NUM, "c1", "c2")
+                        .build();
+        TablePath tablePath = TablePath.of(DATABASE, "alter_table_schema");
+        admin.createTable(tablePath, tableDescriptor, false).get();
+        Table paimonTable =
+                paimonCatalog.getTable(Identifier.create(DATABASE, 
tablePath.getTableName()));
+        verifyPaimonTable(
+                paimonTable,
+                tableDescriptor,
+                RowType.of(
+                        new DataType[] {
+                            org.apache.paimon.types.DataTypes.INT(),
+                            org.apache.paimon.types.DataTypes.STRING(),
+                            // for __bucket, __offset, __timestamp
+                            org.apache.paimon.types.DataTypes.INT(),
+                            org.apache.paimon.types.DataTypes.BIGINT(),
+                            
org.apache.paimon.types.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE()
+                        },
+                        new String[] {
+                            "c1",
+                            "c2",
+                            BUCKET_COLUMN_NAME,
+                            OFFSET_COLUMN_NAME,
+                            TIMESTAMP_COLUMN_NAME
+                        }),
+                "c1,c2",
+                BUCKET_NUM);
+
+        // test alter table schema
+        List<TableChange> tableChanges =
+                Collections.singletonList(
+                        TableChange.addColumn(
+                                "c3",
+                                DataTypes.INT(),
+                                "c3 comment",
+                                TableChange.ColumnPosition.last()));
+        assertThatThrownBy(() -> admin.alterTable(tablePath, tableChanges, 
false).get())
+                .cause()
+                .isInstanceOf(InvalidAlterTableException.class)
+                .hasMessage(
+                        "Schema evolution is currently not supported for 
tables with datalake enabled.");
+    }
+
     private void verifyPaimonTable(
             Table paimonTable,
             TableDescriptor flussTable,
diff --git 
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
 
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
index a074827d8..d62f50528 100644
--- 
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
+++ 
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
@@ -328,6 +328,13 @@ public class MetadataManager {
 
             TableInfo table = getTable(tablePath);
 
+            // TODO: remote this after lake enable table support schema 
evolution, track by
+            // https://github.com/apache/fluss/issues/2128
+            if (table.getTableConfig().isDataLakeEnabled()) {
+                throw new InvalidAlterTableException(
+                        "Schema evolution is currently not supported for 
tables with datalake enabled.");
+            }
+
             // validate the table column changes
             if (!schemaChanges.isEmpty()) {
                 Schema newSchema = SchemaUpdate.applySchemaChanges(table, 
schemaChanges);