Skip to content

[lake/paimon] Maintain lakestream.enabled with lake acceleration state #4118

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fhan688 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

[lake/paimon] Maintain lakestream.enabled with lake acceleration state #4118

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3f929df
[lake/paimon] Maintain lakestream.enabled with lake acceleration state
fhan688 Aug 26, 2026
247669c
[lake/paimon] fix format violations
fhan688 Aug 27, 2026
8be347d
[lake/paimon] Fix lakestream.enabled loss after reset-then-enable
fhan688 Aug 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 61 additions & 0 deletions ...fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,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;

Expand All @@ -65,6 +66,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.";
Expand Down Expand Up @@ -186,11 +195,20 @@ public static List<SchemaChange> 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.
appendLakeStreamOptionChange(
setOption.getKey(),
Boolean.parseBoolean(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.
appendLakeStreamOptionChange(
resetOption.getKey(), false, paimonIncludingSystemColumns, schemaChanges);
} else if (tableChange instanceof TableChange.AddColumn) {
TableChange.AddColumn addColumn = (TableChange.AddColumn) tableChange;

Expand Down Expand Up @@ -306,6 +324,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)) {

@luoyuxia luoyuxia Aug 29, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

RESET removes table.datalake.enabled from the current descriptor. When it is enabled again, createTable only validates the existing Paimon table and alterTable is skipped because the old key is absent, so lakestream.enabled is not restored.

Suggested change in MetadataManager#preAlterTableProperties:

boolean enablingDataLake =
        isDataLakeEnabled(newDescriptor)
                && !isDataLakeEnabled(tableDescriptor);

if (lakeCatalog != null
        && (enablingDataLake
                || tableDescriptor
                        .getProperties()
                        .containsKey(ConfigOptions.TABLE_DATALAKE_ENABLED.key()))) {
    lakeCatalog.alterTable(tablePath, tableChanges, lakeCatalogContext);
}

Please also add an Admin-level true -> RESET -> true regression test.

options.set(LAKESTREAM_ENABLED_OPTION_KEY, Boolean.TRUE.toString());
}

schemaBuilder.options(options.toMap());

// currently we only support string type, todo
Expand Down Expand Up @@ -333,6 +358,42 @@ 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 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 appendLakeStreamOptionChange(
String flussKey,
boolean lakeStreamEnabled,
boolean legacyTable,
List<SchemaChange> out) {
if (!TABLE_DATALAKE_ENABLED.key().equals(flussKey)) {
return;
}
// Old-layout tables are outside the scope of this option.
if (legacyTable) {
return;
}
if (lakeStreamEnabled) {
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<String, String> properties) {
properties.forEach(
(k, v) -> {
Expand Down
149 changes: 149 additions & 0 deletions ...-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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");
Expand Down Expand Up @@ -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<String, String> updatedProperties = new HashMap<>();
updatedProperties.put(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true");
Expand Down Expand Up @@ -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");
Expand All @@ -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));
Expand All @@ -765,6 +777,143 @@ 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");

// 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
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 :
Expand Down
73 changes: 73 additions & 0 deletions ...e/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<String> primaryKeys, List<String> partitionKeys, String bucket, String bucketKey) {
return createPaimonSchema(
Expand Down
Loading
Loading