From 3f929df0bac06483c9de14bbf8883b0186258d3e Mon Sep 17 00:00:00 2001 From: fhan Date: Wed, 26 Aug 2026 23:07:39 +0800 Subject: [PATCH 1/3] [lake/paimon] Maintain lakestream.enabled with lake acceleration state --- .../lake/paimon/utils/PaimonConversions.java | 68 ++++++++++ .../paimon/LakeEnabledTableCreateITCase.java | 125 ++++++++++++++++++ .../lake/paimon/PaimonLakeCatalogTest.java | 73 ++++++++++ 3 files changed, 266 insertions(+) diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java index cf3fc03482..2a420e38ad 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java @@ -43,6 +43,8 @@ import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; +import javax.annotation.Nullable; + import java.util.ArrayList; import java.util.HashSet; import java.util.List; @@ -51,6 +53,7 @@ import java.util.Set; import java.util.function.Function; +import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_ENABLED; import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS; import static org.apache.fluss.utils.Preconditions.checkState; @@ -65,6 +68,14 @@ public class PaimonConversions { /** Option controlling whether Paimon uses legacy partition value encoding. */ public static final String PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY = "partition.legacy-name"; + /** + * Native Paimon table option maintained by Fluss to mark whether the (clean-layout) Paimon + * table is currently accelerated by Fluss LakeStream. Managed only for new-layout tables that + * do not carry the Fluss system columns; legacy tables are left untouched. Disabling lake + * acceleration removes the option instead of persisting {@code false}. + */ + public static final String LAKESTREAM_ENABLED_OPTION_KEY = "lakestream.enabled"; + // for fluss config public static final String FLUSS_CONF_PREFIX = "fluss."; public static final String TABLE_DATALAKE_PAIMON_PREFIX = "table.datalake.paimon."; @@ -186,11 +197,23 @@ public static List toPaimonSchemaChanges( String key = convertFlussPropertyKeyToPaimon(setOption.getKey()); validateAlterPaimonOptions(key); schemaChanges.add(SchemaChange.setOption(key, setOption.getValue())); + // #4102: keep lakestream.enabled in sync with datalake acceleration state. + maybeSyncLakeStreamOption( + setOption.getKey(), + setOption.getValue(), + paimonIncludingSystemColumns, + schemaChanges); } else if (tableChange instanceof TableChange.ResetOption) { TableChange.ResetOption resetOption = (TableChange.ResetOption) tableChange; String key = convertFlussPropertyKeyToPaimon(resetOption.getKey()); validateAlterPaimonOptions(key); schemaChanges.add(SchemaChange.removeOption(key)); + // #4102: resetting datalake.enabled is equivalent to disabling acceleration. + maybeSyncLakeStreamOption( + resetOption.getKey(), + null, + paimonIncludingSystemColumns, + schemaChanges); } else if (tableChange instanceof TableChange.AddColumn) { TableChange.AddColumn addColumn = (TableChange.AddColumn) tableChange; @@ -306,6 +329,13 @@ public static Schema toPaimonSchema(TableDescriptor tableDescriptor) { tableDescriptor .getCustomProperties() .forEach((k, v) -> setFlussPropertyToPaimon(k, v, options)); + + // #4102: newly created lake tables are always clean (system columns are rejected above), so + // a lake-enabled table must advertise its LakeStream state to Paimon. + if (isDataLakeEnabled(tableDescriptor)) { + options.set(LAKESTREAM_ENABLED_OPTION_KEY, Boolean.TRUE.toString()); + } + schemaBuilder.options(options.toMap()); // currently we only support string type, todo @@ -333,6 +363,44 @@ public static Schema toPaimonSchema(TableDescriptor tableDescriptor) { return schemaBuilder.build(); } + private static boolean isDataLakeEnabled(TableDescriptor tableDescriptor) { + return Boolean.parseBoolean( + tableDescriptor.getProperties().get(TABLE_DATALAKE_ENABLED.key())); + } + + /** + * Maintains the {@code lakestream.enabled} Paimon option together with the {@code + * table.datalake.enabled} lifecycle. Only new-layout (clean) tables are managed; legacy tables + * that still carry the Fluss system columns are left untouched. Disabling removes the option + * instead of persisting {@code false}. + * + * @param flussKey the original (un-prefixed) Fluss change key + * @param value the option value for a SetOption change, or {@code null} for a ResetOption + * @param legacyTable whether the Paimon table uses the legacy system-column layout + * @param out the schema-change list to append to + */ + private static void maybeSyncLakeStreamOption( + String flussKey, + @Nullable String value, + boolean legacyTable, + List out) { + if (!TABLE_DATALAKE_ENABLED.key().equals(flussKey)) { + return; + } + // Old-layout tables are outside the scope of this option. + if (legacyTable) { + return; + } + if (Boolean.parseBoolean(value)) { + out.add( + SchemaChange.setOption( + LAKESTREAM_ENABLED_OPTION_KEY, Boolean.TRUE.toString())); + } else { + // Disabling (SetOption "false") or resetting removes the option entirely. + out.add(SchemaChange.removeOption(LAKESTREAM_ENABLED_OPTION_KEY)); + } + } + private static void validatePaimonOptions(Map properties) { properties.forEach( (k, v) -> { 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 0a0c07d973..c2af2bfe74 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 @@ -74,6 +74,7 @@ import java.util.stream.Stream; import static org.apache.fluss.lake.paimon.testutils.PaimonTestUtils.adjustToLegacyV1Table; +import static org.apache.fluss.lake.paimon.utils.PaimonConversions.LAKESTREAM_ENABLED_OPTION_KEY; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.PAIMON_UNSETTABLE_OPTIONS; import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME; import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME; @@ -177,6 +178,7 @@ void testCreateLakeEnabledTable() throws Exception { new String[] {"log_c1", "log_c2"}), "log_c1,log_c2", BUCKET_NUM); + assertThat(paimonLogTable.options()).containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true"); TableDescriptor logNoBucketKeyTable = TableDescriptor.builder() @@ -234,6 +236,7 @@ void testCreateLakeEnabledTable() throws Exception { new String[] {"pk_c1", "pk_c2"}), "pk_c1", BUCKET_NUM); + assertThat(paimonPkTable.options()).containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true"); // test partitioned table TablePath partitionedTablePath = TablePath.of(DATABASE, "partitioned_table"); @@ -708,6 +711,9 @@ void testAlterLakeEnabledLogTable() throws Exception { Identifier paimonTablePath = Identifier.create(DATABASE, logTablePath.getTableName()); Table enabledPaimonLogTable = paimonCatalog.getTable(paimonTablePath); + // enabling lake acceleration on a clean table sets lakestream.enabled=true + assertThat(enabledPaimonLogTable.options()) + .containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true"); Map updatedProperties = new HashMap<>(); updatedProperties.put(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true"); @@ -735,6 +741,9 @@ void testAlterLakeEnabledLogTable() throws Exception { // verify LogTablet datalake status is disabled verifyLogTabletDataLakeEnabled(tableId, false); + // disabling lake acceleration removes lakestream.enabled instead of storing false + assertThat(paimonCatalog.getTable(paimonTablePath).options()) + .doesNotContainKey(LAKESTREAM_ENABLED_OPTION_KEY); // try to enable lake table again enableLake = TableChange.set(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true"); @@ -743,6 +752,9 @@ void testAlterLakeEnabledLogTable() throws Exception { // verify LogTablet datalake status is enabled again verifyLogTabletDataLakeEnabled(tableId, true); + // re-enabling lake acceleration adds lakestream.enabled=true again + assertThat(paimonCatalog.getTable(paimonTablePath).options()) + .containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true"); // write some data to the lake table writeData(paimonCatalog.getTable(paimonTablePath)); @@ -765,6 +777,119 @@ void testAlterLakeEnabledLogTable() throws Exception { verifyLogTabletDataLakeEnabled(tableId, true); } + @Test + void testAlterLakeEnabledPrimaryKeyTable() throws Exception { + // create pk table with lake disabled + TableDescriptor pkTable = + TableDescriptor.builder() + .schema( + Schema.newBuilder() + .column("pk_c1", DataTypes.INT()) + .column("pk_c2", DataTypes.STRING()) + .primaryKey("pk_c1") + .build()) + .property(ConfigOptions.TABLE_DATALAKE_ENABLED, false) + .distributedBy(BUCKET_NUM) + .build(); + TablePath pkTablePath = TablePath.of(DATABASE, "pk_table_alter"); + admin.createTable(pkTablePath, pkTable, false).get(); + Identifier paimonTablePath = Identifier.create(DATABASE, pkTablePath.getTableName()); + + // lake table not created yet while lake is disabled + assertThatThrownBy(() -> paimonCatalog.getTable(paimonTablePath)) + .isInstanceOf(Catalog.TableNotExistException.class); + + // enable lake acceleration sets lakestream.enabled=true + admin.alterTable( + pkTablePath, + Collections.singletonList( + TableChange.set( + ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true")), + false) + .get(); + assertThat(paimonCatalog.getTable(paimonTablePath).options()) + .containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true"); + + // disable lake acceleration removes lakestream.enabled instead of storing false + admin.alterTable( + pkTablePath, + Collections.singletonList( + TableChange.set( + ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "false")), + false) + .get(); + assertThat(paimonCatalog.getTable(paimonTablePath).options()) + .doesNotContainKey(LAKESTREAM_ENABLED_OPTION_KEY); + + // re-enable lake acceleration adds lakestream.enabled=true again + admin.alterTable( + pkTablePath, + Collections.singletonList( + TableChange.set( + ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true")), + false) + .get(); + assertThat(paimonCatalog.getTable(paimonTablePath).options()) + .containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true"); + } + + @Test + void testLegacyTableLakeStreamOptionUntouched() throws Exception { + // create a clean, lake-enabled table, then turn it into a legacy table carrying the three + // system columns. Old-layout tables are outside the scope of lakestream.enabled: altering + // datalake.enabled must not add or remove the option on them. + TablePath tablePath = TablePath.of(DATABASE, "legacy_lakestream_table"); + TableDescriptor tableDescriptor = + TableDescriptor.builder() + .schema( + Schema.newBuilder() + .column("c1", DataTypes.INT()) + .column("c2", DataTypes.STRING()) + .build()) + .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true) + .distributedBy(BUCKET_NUM, "c1") + .build(); + admin.createTable(tablePath, tableDescriptor, false).get(); + Identifier paimonTablePath = Identifier.create(DATABASE, tablePath.getTableName()); + + adjustToLegacyV1Table(tablePath, paimonCatalog); + String lakeStreamValueBeforeAlter = + paimonCatalog + .getTable(paimonTablePath) + .options() + .get(LAKESTREAM_ENABLED_OPTION_KEY); + + // disable lake acceleration on a legacy table leaves the option untouched + admin.alterTable( + tablePath, + Collections.singletonList( + TableChange.set( + ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "false")), + false) + .get(); + assertThat( + paimonCatalog + .getTable(paimonTablePath) + .options() + .get(LAKESTREAM_ENABLED_OPTION_KEY)) + .isEqualTo(lakeStreamValueBeforeAlter); + + // re-enable lake acceleration on a legacy table also leaves the option untouched + admin.alterTable( + tablePath, + Collections.singletonList( + TableChange.set( + ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true")), + false) + .get(); + assertThat( + paimonCatalog + .getTable(paimonTablePath) + .options() + .get(LAKESTREAM_ENABLED_OPTION_KEY)) + .isEqualTo(lakeStreamValueBeforeAlter); + } + @Test void testThrowExceptionWhenConflictWithSystemColumn() { for (String systemColumn : diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java index 1e67a92a28..c704bef954 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java @@ -45,6 +45,7 @@ import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_ENABLED; import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_FORMAT; +import static org.apache.fluss.lake.paimon.utils.PaimonConversions.LAKESTREAM_ENABLED_OPTION_KEY; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon; import static org.apache.fluss.lake.paimon.utils.PaimonTableValidation.isPaimonSchemaCompatible; @@ -474,6 +475,78 @@ void testAlterTableAddColumnWhenPaimonSchemaNotMatch() throws Exception { changes)); } + @Test + void testCreateTableSetsLakeStreamEnabledForCleanTable() throws Exception { + String database = "test_create_lakestream_db"; + String tableName = "test_create_lakestream_table"; + TablePath tablePath = TablePath.of(database, tableName); + Identifier identifier = Identifier.create(database, tableName); + + // getTableDescriptor sets table.datalake.enabled=true, so the clean table advertises its + // LakeStream state to Paimon + flussPaimonCatalog.createTable( + tablePath, getTableDescriptor(FLUSS_SCHEMA), LAKE_CATALOG_CONTEXT); + + Table table = flussPaimonCatalog.getPaimonCatalog().getTable(identifier); + assertThat(table.options()).containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true"); + } + + @Test + void testCreateTableWithoutDataLakeEnabledHasNoLakeStreamOption() throws Exception { + String database = "test_create_no_lakestream_db"; + String tableName = "test_create_no_lakestream_table"; + TablePath tablePath = TablePath.of(database, tableName); + Identifier identifier = Identifier.create(database, tableName); + + TableDescriptor tableDescriptor = + TableDescriptor.builder() + .schema(FLUSS_SCHEMA) + .property(TABLE_DATALAKE_ENABLED.key(), "false") + .property(TABLE_DATALAKE_FORMAT.key(), "paimon") + .property( + "table.datalake.paimon.warehouse", + tempWarehouseDir.toURI().toString()) + .distributedBy(3) + .build(); + flussPaimonCatalog.createTable(tablePath, tableDescriptor, LAKE_CATALOG_CONTEXT); + + Table table = flussPaimonCatalog.getPaimonCatalog().getTable(identifier); + assertThat(table.options()).doesNotContainKey(LAKESTREAM_ENABLED_OPTION_KEY); + } + + @Test + void testAlterDataLakeEnabledMaintainsLakeStreamOptionForCleanTable() throws Exception { + String database = "test_alter_lakestream_db"; + String tableName = "test_alter_lakestream_table"; + TablePath tablePath = TablePath.of(database, tableName); + Identifier identifier = Identifier.create(database, tableName); + createTable(database, tableName); + + // disable lake acceleration removes lakestream.enabled instead of storing false + flussPaimonCatalog.alterTable( + tablePath, + Collections.singletonList(TableChange.set(TABLE_DATALAKE_ENABLED.key(), "false")), + LAKE_CATALOG_CONTEXT); + Table table = flussPaimonCatalog.getPaimonCatalog().getTable(identifier); + assertThat(table.options()).doesNotContainKey(LAKESTREAM_ENABLED_OPTION_KEY); + + // re-enable lake acceleration adds lakestream.enabled=true again + flussPaimonCatalog.alterTable( + tablePath, + Collections.singletonList(TableChange.set(TABLE_DATALAKE_ENABLED.key(), "true")), + LAKE_CATALOG_CONTEXT); + table = flussPaimonCatalog.getPaimonCatalog().getTable(identifier); + assertThat(table.options()).containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true"); + + // resetting datalake.enabled is equivalent to disabling acceleration + flussPaimonCatalog.alterTable( + tablePath, + Collections.singletonList(TableChange.reset(TABLE_DATALAKE_ENABLED.key())), + LAKE_CATALOG_CONTEXT); + table = flussPaimonCatalog.getPaimonCatalog().getTable(identifier); + assertThat(table.options()).doesNotContainKey(LAKESTREAM_ENABLED_OPTION_KEY); + } + private org.apache.paimon.schema.Schema createPaimonSchema( List primaryKeys, List partitionKeys, String bucket, String bucketKey) { return createPaimonSchema( From 247669c2ead9a48c866ea56af05e672b54f9a3be Mon Sep 17 00:00:00 2001 From: fhan Date: Thu, 27 Aug 2026 10:37:14 +0800 Subject: [PATCH 2/3] [lake/paimon] fix format violations --- .../fluss/lake/paimon/utils/PaimonConversions.java | 14 +++----------- 1 file changed, 3 insertions(+), 11 deletions(-) diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java index 2a420e38ad..d6f5c85767 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java @@ -210,10 +210,7 @@ public static List toPaimonSchemaChanges( schemaChanges.add(SchemaChange.removeOption(key)); // #4102: resetting datalake.enabled is equivalent to disabling acceleration. maybeSyncLakeStreamOption( - resetOption.getKey(), - null, - paimonIncludingSystemColumns, - schemaChanges); + resetOption.getKey(), null, paimonIncludingSystemColumns, schemaChanges); } else if (tableChange instanceof TableChange.AddColumn) { TableChange.AddColumn addColumn = (TableChange.AddColumn) tableChange; @@ -380,10 +377,7 @@ private static boolean isDataLakeEnabled(TableDescriptor tableDescriptor) { * @param out the schema-change list to append to */ private static void maybeSyncLakeStreamOption( - String flussKey, - @Nullable String value, - boolean legacyTable, - List out) { + String flussKey, @Nullable String value, boolean legacyTable, List out) { if (!TABLE_DATALAKE_ENABLED.key().equals(flussKey)) { return; } @@ -392,9 +386,7 @@ private static void maybeSyncLakeStreamOption( return; } if (Boolean.parseBoolean(value)) { - out.add( - SchemaChange.setOption( - LAKESTREAM_ENABLED_OPTION_KEY, Boolean.TRUE.toString())); + out.add(SchemaChange.setOption(LAKESTREAM_ENABLED_OPTION_KEY, Boolean.TRUE.toString())); } else { // Disabling (SetOption "false") or resetting removes the option entirely. out.add(SchemaChange.removeOption(LAKESTREAM_ENABLED_OPTION_KEY)); From 8be347d4270d5392f60b17bcf6a18675ed2b514b Mon Sep 17 00:00:00 2001 From: fhan Date: Sat, 29 Aug 2026 23:53:11 +0800 Subject: [PATCH 3/3] [lake/paimon] Fix lakestream.enabled loss after reset-then-enable --- .../lake/paimon/utils/PaimonConversions.java | 21 ++++++++-------- .../paimon/LakeEnabledTableCreateITCase.java | 24 +++++++++++++++++++ .../server/coordinator/MetadataManager.java | 16 +++++++++---- 3 files changed, 46 insertions(+), 15 deletions(-) diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java index d6f5c85767..af6198c21d 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java @@ -43,8 +43,6 @@ import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; -import javax.annotation.Nullable; - import java.util.ArrayList; import java.util.HashSet; import java.util.List; @@ -198,9 +196,9 @@ public static List toPaimonSchemaChanges( validateAlterPaimonOptions(key); schemaChanges.add(SchemaChange.setOption(key, setOption.getValue())); // #4102: keep lakestream.enabled in sync with datalake acceleration state. - maybeSyncLakeStreamOption( + appendLakeStreamOptionChange( setOption.getKey(), - setOption.getValue(), + Boolean.parseBoolean(setOption.getValue()), paimonIncludingSystemColumns, schemaChanges); } else if (tableChange instanceof TableChange.ResetOption) { @@ -209,8 +207,8 @@ public static List toPaimonSchemaChanges( validateAlterPaimonOptions(key); schemaChanges.add(SchemaChange.removeOption(key)); // #4102: resetting datalake.enabled is equivalent to disabling acceleration. - maybeSyncLakeStreamOption( - resetOption.getKey(), null, paimonIncludingSystemColumns, schemaChanges); + appendLakeStreamOptionChange( + resetOption.getKey(), false, paimonIncludingSystemColumns, schemaChanges); } else if (tableChange instanceof TableChange.AddColumn) { TableChange.AddColumn addColumn = (TableChange.AddColumn) tableChange; @@ -372,12 +370,15 @@ private static boolean isDataLakeEnabled(TableDescriptor tableDescriptor) { * instead of persisting {@code false}. * * @param flussKey the original (un-prefixed) Fluss change key - * @param value the option value for a SetOption change, or {@code null} for a ResetOption + * @param lakeStreamEnabled whether datalake acceleration is enabled after this change * @param legacyTable whether the Paimon table uses the legacy system-column layout * @param out the schema-change list to append to */ - private static void maybeSyncLakeStreamOption( - String flussKey, @Nullable String value, boolean legacyTable, List out) { + private static void appendLakeStreamOptionChange( + String flussKey, + boolean lakeStreamEnabled, + boolean legacyTable, + List out) { if (!TABLE_DATALAKE_ENABLED.key().equals(flussKey)) { return; } @@ -385,7 +386,7 @@ private static void maybeSyncLakeStreamOption( if (legacyTable) { return; } - if (Boolean.parseBoolean(value)) { + if (lakeStreamEnabled) { out.add(SchemaChange.setOption(LAKESTREAM_ENABLED_OPTION_KEY, Boolean.TRUE.toString())); } else { // Disabling (SetOption "false") or resetting removes the option entirely. 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 c2af2bfe74..16479f51ba 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 @@ -831,6 +831,30 @@ void testAlterLakeEnabledPrimaryKeyTable() throws Exception { .get(); assertThat(paimonCatalog.getTable(paimonTablePath).options()) .containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true"); + + // resetting datalake.enabled is equivalent to disabling acceleration, and removes the + // key from the table descriptor entirely (unlike SetOption "false") + admin.alterTable( + pkTablePath, + Collections.singletonList( + TableChange.reset(ConfigOptions.TABLE_DATALAKE_ENABLED.key())), + false) + .get(); + assertThat(paimonCatalog.getTable(paimonTablePath).options()) + .doesNotContainKey(LAKESTREAM_ENABLED_OPTION_KEY); + + // re-enabling after a reset must still sync lakestream.enabled=true: since the reset + // removed the key from the descriptor, MetadataManager must not rely solely on "the old + // descriptor already had the key" to decide whether to sync to the lake table + admin.alterTable( + pkTablePath, + Collections.singletonList( + TableChange.set( + ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true")), + false) + .get(); + assertThat(paimonCatalog.getTable(paimonTablePath).options()) + .containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true"); } @Test 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 43d9843425..a729adee11 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 @@ -618,12 +618,18 @@ private void preAlterTableProperties( // We should always alter lake table even though datalake is disabled. // Otherwise, if user alter the fluss table when datalake is disabled, then enable datalake // again, the lake table will mismatch. - // Only sync to lake if this table has ever opted into datalake (key present regardless of - // value). + // Sync to lake if this table has ever opted into datalake (key present regardless of + // value), or if this change is enabling datalake now. The latter is needed because + // resetting table.datalake.enabled removes the key entirely (see + // #getUpdatedTableDescriptor), so a later re-enable would otherwise see an old descriptor + // without the key and skip the alterTable call that re-applies lakestream.enabled. + boolean enablingDataLake = + isDataLakeEnabled(newDescriptor) && !isDataLakeEnabled(tableDescriptor); if (lakeCatalog != null - && tableDescriptor - .getProperties() - .containsKey(ConfigOptions.TABLE_DATALAKE_ENABLED.key())) { + && (enablingDataLake + || tableDescriptor + .getProperties() + .containsKey(ConfigOptions.TABLE_DATALAKE_ENABLED.key()))) { try { lakeCatalog.alterTable(tablePath, tableChanges, lakeCatalogContext); } catch (TableNotExistException e) {