[server] Forbid resetting table.datalake.enabled

Author: LiebingYu

fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java

@@ -37,12 +37,14 @@ public class FlussConfigUtils {
     public static final String CLIENT_SECURITY_PREFIX = "client.security.";
 
     public static final List<String> ALTERABLE_TABLE_OPTIONS;
+    public static final List<String> RESETTABLE_TABLE_OPTIONS;
 
     static {
         TABLE_OPTIONS = extractConfigOptions("table.");
         CLIENT_OPTIONS = extractConfigOptions("client.");
         ALTERABLE_TABLE_OPTIONS =
                 Collections.singletonList(ConfigOptions.TABLE_DATALAKE_ENABLED.key());
+        RESETTABLE_TABLE_OPTIONS = Collections.emptyList();
     }
 
     public static boolean isTableStorageConfig(String key) {
@@ -53,6 +55,10 @@ public static boolean isAlterableTableOption(String key) {
         return ALTERABLE_TABLE_OPTIONS.contains(key);
     }
 
+    public static boolean isResettableTableOption(String key) {
+        return RESETTABLE_TABLE_OPTIONS.contains(key);
+    }
+
     @VisibleForTesting
     static Map<String, ConfigOption<?>> extractConfigOptions(String prefix) {
         Map<String, ConfigOption<?>> options = new HashMap<>();

fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java

@@ -23,6 +23,7 @@
 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.InvalidTableException;
 import org.apache.fluss.metadata.Schema;
 import org.apache.fluss.metadata.TableChange;
@@ -429,6 +430,15 @@ void testAlterLakeEnabledLogTable() throws Exception {
                         }),
                 "log_c1,log_c2",
                 BUCKET_NUM);
+
+        // reset lake
+        TableChange.ResetOption resetLake =
+                TableChange.reset(ConfigOptions.TABLE_DATALAKE_ENABLED.key());
+        List<TableChange> resetChanges = Collections.singletonList(resetLake);
+        assertThatThrownBy(() -> admin.alterTable(logTablePath, resetChanges, false).get())
+                .cause()
+                .isInstanceOf(InvalidAlterTableException.class)
+                .hasMessage("The option 'table.datalake.enabled' is not supported to reset.");
     }
 
     @Test

fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java

@@ -324,10 +324,7 @@ public void alterTableProperties(
             TableInfo tableInfo = tableReg.toTableInfo(tablePath, schemaInfo);
 
             // validate the changes
-            validateAlterTableProperties(
-                    tableInfo,
-                    tablePropertyChanges.tableKeysToChange(),
-                    tablePropertyChanges.customKeysToChange());
+            validateAlterTableProperties(tableInfo, tablePropertyChanges);
 
             TableDescriptor tableDescriptor = tableInfo.toTableDescriptor();
             TableDescriptor newDescriptor =

fluss-server/src/main/java/org/apache/fluss/server/entity/TablePropertyChanges.java

@@ -42,6 +42,10 @@ protected TablePropertyChanges(
         this.customPropertiesToReset = customPropertiesToReset;
     }
 
+    public boolean tableKeyToReset(String key) {
+        return tablePropertiesToReset.contains(key);
+    }
+
     public Set<String> tableKeysToChange() {
         Set<String> keys = new HashSet<>(tablePropertiesToSet.keySet());
         keys.addAll(tablePropertiesToReset);

fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java

@@ -31,6 +31,7 @@
 import org.apache.fluss.metadata.MergeEngineType;
 import org.apache.fluss.metadata.TableDescriptor;
 import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.server.entity.TablePropertyChanges;
 import org.apache.fluss.types.DataType;
 import org.apache.fluss.types.DataTypeRoot;
 import org.apache.fluss.types.RowType;
@@ -48,6 +49,7 @@
 
 import static org.apache.fluss.config.FlussConfigUtils.TABLE_OPTIONS;
 import static org.apache.fluss.config.FlussConfigUtils.isAlterableTableOption;
+import static org.apache.fluss.config.FlussConfigUtils.isResettableTableOption;
 import static org.apache.fluss.config.FlussConfigUtils.isTableStorageConfig;
 import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME;
 import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME;
@@ -106,27 +108,41 @@ public static void validateTableDescriptor(TableDescriptor tableDescriptor, int
     }
 
     public static void validateAlterTableProperties(
-            TableInfo currentTable, Set<String> tableKeysToChange, Set<String> customKeysToChange) {
-        tableKeysToChange.forEach(
-                k -> {
-                    if (isTableStorageConfig(k) && !isAlterableTableOption(k)) {
-                        throw new InvalidAlterTableException(
-                                "The option '" + k + "' is not supported to alter yet.");
-                    }
-                });
+            TableInfo currentTable, TablePropertyChanges tablePropertyChanges) {
+        tablePropertyChanges
+                .tableKeysToChange()
+                .forEach(
+                        k -> {
+                            if (isTableStorageConfig(k)) {
+                                if (!isAlterableTableOption(k)) {
+                                    throw new InvalidAlterTableException(
+                                            "The option '"
+                                                    + k
+                                                    + "' is not supported to alter yet.");
+                                }
+
+                                if (tablePropertyChanges.tableKeyToReset(k)
+                                        && !isResettableTableOption(k)) {
+                                    throw new InvalidAlterTableException(
+                                            "The option '" + k + "' is not supported to reset.");
+                                }
+                            }
+                        });
 
         TableConfig currentConfig = currentTable.getTableConfig();
         if (currentConfig.isDataLakeEnabled() && currentConfig.getDataLakeFormat().isPresent()) {
             String format = currentConfig.getDataLakeFormat().get().toString();
-            customKeysToChange.forEach(
-                    k -> {
-                        if (k.startsWith(format + ".")) {
-                            throw new InvalidConfigException(
-                                    String.format(
-                                            "Property '%s' is not supported to alter which is for datalake table.",
-                                            k));
-                        }
-                    });
+            tablePropertyChanges
+                    .customKeysToChange()
+                    .forEach(
+                            k -> {
+                                if (k.startsWith(format + ".")) {
+                                    throw new InvalidConfigException(
+                                            String.format(
+                                                    "Property '%s' is not supported to alter which is for datalake table.",
+                                                    k));
+                                }
+                            });
         }
     }