refine doc

Author: luoyuxia

fluss-client/src/test/java/org/apache/fluss/client/table/LakeEnableTableITCase.java

@@ -230,7 +230,7 @@ void testCannotEnableTableWhenTableFormatDiffersFromClusterFormat() throws Excep
     @Test
     void testEnableTableAfterClusterEnablesDataLake() throws Exception {
         String databaseName = "test_db";
-        String tableName = "test_table_before_cluster_enable";
+        String tableName = "test_enable_datalake_table";
         TablePath tablePath = TablePath.of(databaseName, tableName);
 
         admin.createDatabase(databaseName, DatabaseDescriptor.EMPTY, true).get();

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

@@ -75,8 +75,10 @@ public void validate(Configuration newConfig) throws ConfigException {
         // set, rather than only the merged result. We accept this for now because
         // table-level enablement is still validated, and enabling datalake for a table
         // will fail if datalake.format is not configured.
-        boolean explicitDataLakeEnabled = newConfig.getOptional(DATALAKE_ENABLED).orElse(false);
-        if (explicitDataLakeEnabled && newDatalakeFormat == null) {
+        Optional<Boolean> optDataLakeEnabled = newConfig.getOptional(DATALAKE_ENABLED);
+        if (optDataLakeEnabled.isPresent()
+                && optDataLakeEnabled.get()
+                && newDatalakeFormat == null) {
             throw new ConfigException(
                     String.format(
                             "'%s' must be configured when '%s' is explicitly set.",

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

@@ -129,11 +129,17 @@ public static void validateTableDescriptor(
 
     private static void checkTableLakeFormatMatchesCluster(
             Configuration tableConf, @Nullable DataLakeFormat clusterDataLakeFormat) {
+        if (clusterDataLakeFormat == null) {
+            return;
+        }
+
+        if (!tableConf.get(ConfigOptions.TABLE_DATALAKE_ENABLED)) {
+            return;
+        }
+
         Optional<DataLakeFormat> tableDataLakeFormat =
                 tableConf.getOptional(ConfigOptions.TABLE_DATALAKE_FORMAT);
-        if (clusterDataLakeFormat != null
-                && tableDataLakeFormat.isPresent()
-                && tableDataLakeFormat.get() != clusterDataLakeFormat) {
+        if (tableDataLakeFormat.isPresent() && tableDataLakeFormat.get() != clusterDataLakeFormat) {
             throw new InvalidConfigException(
                     String.format(
                             "'%s' ('%s') must match cluster '%s' ('%s') when '%s' is enabled.",

fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java

@@ -531,7 +531,7 @@ public void reconfigure(Configuration newConfig) {
     }
 
     @Test
-    void testExplicitDatalakeEnabledRequiresFormat() throws Exception {
+    void testExplicitDataLakeEnabledRequiresDataLakeFormat() throws Exception {
         try (LakeCatalogDynamicLoader lakeCatalogDynamicLoader =
                 new LakeCatalogDynamicLoader(new Configuration(), null, true)) {
             DynamicConfigManager dynamicConfigManager =
@@ -552,41 +552,4 @@ void testExplicitDatalakeEnabledRequiresFormat() throws Exception {
                             "'datalake.format' must be configured when 'datalake.enabled' is explicitly set.");
         }
     }
-
-    @Test
-    void testPreBindOnlyModeDoesNotCreateLakeCatalog() throws Exception {
-        try (LakeCatalogDynamicLoader lakeCatalogDynamicLoader =
-                new LakeCatalogDynamicLoader(new Configuration(), null, true)) {
-            DynamicConfigManager dynamicConfigManager =
-                    new DynamicConfigManager(zookeeperClient, new Configuration(), true);
-            dynamicConfigManager.register(lakeCatalogDynamicLoader);
-            dynamicConfigManager.startup();
-
-            dynamicConfigManager.alterConfigs(
-                    Arrays.asList(
-                            new AlterConfig(DATALAKE_FORMAT.key(), "paimon", AlterConfigOpType.SET),
-                            new AlterConfig(DATALAKE_ENABLED.key(), "false", AlterConfigOpType.SET),
-                            new AlterConfig(
-                                    "datalake.paimon.metastore",
-                                    "filesystem",
-                                    AlterConfigOpType.SET)));
-
-            assertThat(lakeCatalogDynamicLoader.getLakeCatalogContainer().getDataLakeFormat())
-                    .isEqualTo(PAIMON);
-            assertThat(
-                            lakeCatalogDynamicLoader
-                                    .getLakeCatalogContainer()
-                                    .isClusterDataLakeTableEnabled())
-                    .isFalse();
-            assertThat(lakeCatalogDynamicLoader.getLakeCatalogContainer().getLakeCatalog())
-                    .isNull();
-            assertThat(
-                            lakeCatalogDynamicLoader
-                                    .getLakeCatalogContainer()
-                                    .getDefaultTableLakeOptions())
-                    .isEqualTo(
-                            Collections.singletonMap(
-                                    "table.datalake.paimon.metastore", "filesystem"));
-        }
-    }
 }

website/docs/maintenance/operations/updating-configs.md

@@ -32,16 +32,10 @@ admin.alterClusterConfigs(
                 new AlterConfig(DATALAKE_ENABLED.key(), "true", AlterConfigOpType.SET),
                 new AlterConfig(DATALAKE_FORMAT.key(), "paimon", AlterConfigOpType.SET)));
 
-// Pre-bind the lakehouse format without enabling lakehouse tables
-admin.alterClusterConfigs(
-        Arrays.asList(
-                new AlterConfig(DATALAKE_ENABLED.key(), "false", AlterConfigOpType.SET),
-                new AlterConfig(DATALAKE_FORMAT.key(), "paimon", AlterConfigOpType.SET)));
-
-// Return to legacy behavior where configuring datalake.format alone also enables lakehouse tables
+// Disable lakehouse storage
 admin.alterClusterConfigs(
         Collections.singletonList(
-                new AlterConfig(DATALAKE_ENABLED.key(), null, AlterConfigOpType.DELETE)));
+                new AlterConfig(DATALAKE_ENABLED.key(), "false", AlterConfigOpType.SET)));
 
 // Set RocksDB shared rate limiter to 200MB/sec
 admin.alterClusterConfigs(
@@ -54,8 +48,6 @@ The `AlterConfig` class contains three properties:
 * `value`: The configuration value to be set (e.g., `paimon`)
 * `opType`: The operation type, either `AlterConfigOpType.SET` or `AlterConfigOpType.DELETE`
 
-If `datalake.enabled` is explicitly configured, `datalake.format` must also be configured in the cluster.
-
 ### Using Flink Stored Procedures
 
 For certain configurations, Fluss provides convenient Flink stored procedures that can be called directly from Flink SQL. See [Procedures](engine-flink/procedures.md#cluster-configuration-procedures) for detailed documentation on using `get_cluster_config` and `set_cluster_config` procedures.

website/docs/maintenance/operations/upgrade-notes-0.10.md

@@ -5,35 +5,40 @@ sidebar_position: 4
 
 # Upgrade Notes from v0.9 to v0.10
 
-## New `datalake.enabled` Cluster Configuration
+## Cluster Configuration Changes
 
-Starting from v0.10, Fluss introduces the cluster-level configuration `datalake.enabled` to control whether the cluster is ready to create and manage lakehouse tables.
+### New `datalake.enabled` Cluster Configuration
 
-### Behavior Changes
+Starting in v0.10, Fluss introduces the cluster-level configuration `datalake.enabled` to control whether the cluster is ready to create and manage lakehouse tables.
 
-- If `datalake.enabled` is unset, Fluss keeps the legacy behavior: configuring `datalake.format` alone also enables lakehouse tables.
-- If `datalake.enabled = false`, Fluss only pre-binds the lake format for newly created tables and does not allow lakehouse tables yet.
+#### Behavior Changes
+
+- If `datalake.enabled` is unset, Fluss preserves the legacy behavior: configuring `datalake.format` alone also enables lakehouse tables.
+- If `datalake.enabled = false`, Fluss pre-binds the lake format for newly created tables but does not allow lakehouse tables yet.
 - If `datalake.enabled = true`, Fluss fully enables lakehouse tables.
 - If `datalake.enabled` is explicitly configured, `datalake.format` must also be configured.
 
-### Recommended Configuration
+#### Recommended Configuration
 
-If you want to enable lakehouse tables on the cluster, configure both options together:
+To enable lakehouse tables for the cluster, configure both options together:
 
 ```yaml
 datalake.enabled: true
 datalake.format: paimon
 ```
 
-If you only want to pre-bind the lake format without enabling lakehouse tables yet, configure:
+To pre-bind the lake format without enabling lakehouse tables yet, configure:
 
 ```yaml
 datalake.enabled: false
 datalake.format: paimon
 ```
 
-### Documentation Updates for Existing Deployments
+This mode is useful when you want newly created tables to carry the lake format in advance, while postponing lakehouse enablement at the cluster level.
+After `datalake.enabled` is later set to `true`, tables created under this configuration can still turn on `table.datalake.enabled` without being recreated.
+
+#### Notes for Existing Deployments
 
-If your existing deployment or internal scripts only set `datalake.format`, they will continue to work with the legacy behavior as long as `datalake.enabled` is left unset.
+If your existing deployment or internal scripts only set `datalake.format`, they will continue to work with the legacy behavior as long as `datalake.enabled` remains unset.
 
-However, for new configuration examples and operational guidance, prefer configuring `datalake.enabled` explicitly together with `datalake.format`.
+For new configuration examples and operational guidance, we recommend explicitly configuring `datalake.enabled` together with `datalake.format`.