Skip to content

Commit 404c17c

Browse files
fhan688fhan688
authored
[lake/paimon] Maintain lakestream.enabled with lake acceleration state (apache#4118)
1 parent e116bfd commit 404c17c

4 files changed

Lines changed: 290 additions & 5 deletions

File tree

‎fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@
5151
import java.util.Set;
5252
import java.util.function.Function;
5353

54+
import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_ENABLED;
5455
import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS;
5556
import static org.apache.fluss.utils.Preconditions.checkState;
5657

@@ -65,6 +66,14 @@ public class PaimonConversions {
6566
/** Option controlling whether Paimon uses legacy partition value encoding. */
6667
public static final String PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY = "partition.legacy-name";
6768

69+
/**
70+
* Native Paimon table option maintained by Fluss to mark whether the (clean-layout) Paimon
71+
* table is currently accelerated by Fluss LakeStream. Managed only for new-layout tables that
72+
* do not carry the Fluss system columns; legacy tables are left untouched. Disabling lake
73+
* acceleration removes the option instead of persisting {@code false}.
74+
*/
75+
public static final String LAKESTREAM_ENABLED_OPTION_KEY = "lakestream.enabled";
76+
6877
// for fluss config
6978
public static final String FLUSS_CONF_PREFIX = "fluss.";
7079
public static final String TABLE_DATALAKE_PAIMON_PREFIX = "table.datalake.paimon.";
@@ -186,11 +195,23 @@ public static List<SchemaChange> toPaimonSchemaChanges(
186195
String key = convertFlussPropertyKeyToPaimon(setOption.getKey());
187196
validateAlterPaimonOptions(key);
188197
schemaChanges.add(SchemaChange.setOption(key, setOption.getValue()));
198+
if (TABLE_DATALAKE_ENABLED.key().equals(setOption.getKey())) {
199+
// #4102: keep lakestream.enabled in sync with datalake acceleration state.
200+
appendLakeStreamOptionChange(
201+
Boolean.parseBoolean(setOption.getValue()),
202+
paimonIncludingSystemColumns,
203+
schemaChanges);
204+
}
189205
} else if (tableChange instanceof TableChange.ResetOption) {
190206
TableChange.ResetOption resetOption = (TableChange.ResetOption) tableChange;
191207
String key = convertFlussPropertyKeyToPaimon(resetOption.getKey());
192208
validateAlterPaimonOptions(key);
193209
schemaChanges.add(SchemaChange.removeOption(key));
210+
if (TABLE_DATALAKE_ENABLED.key().equals(resetOption.getKey())) {
211+
// #4102: resetting datalake.enabled is equivalent to disabling acceleration.
212+
appendLakeStreamOptionChange(
213+
false, paimonIncludingSystemColumns, schemaChanges);
214+
}
194215
} else if (tableChange instanceof TableChange.AddColumn) {
195216
TableChange.AddColumn addColumn = (TableChange.AddColumn) tableChange;
196217

@@ -306,6 +327,13 @@ public static Schema toPaimonSchema(TableDescriptor tableDescriptor) {
306327
tableDescriptor
307328
.getCustomProperties()
308329
.forEach((k, v) -> setFlussPropertyToPaimon(k, v, options));
330+
331+
// #4102: newly created lake tables are always clean (system columns are rejected above), so
332+
// a lake-enabled table must advertise its LakeStream state to Paimon.
333+
if (isDataLakeEnabled(tableDescriptor)) {
334+
options.set(LAKESTREAM_ENABLED_OPTION_KEY, Boolean.TRUE.toString());
335+
}
336+
309337
schemaBuilder.options(options.toMap());
310338

311339
// currently we only support string type, todo
@@ -333,6 +361,35 @@ public static Schema toPaimonSchema(TableDescriptor tableDescriptor) {
333361
return schemaBuilder.build();
334362
}
335363

364+
private static boolean isDataLakeEnabled(TableDescriptor tableDescriptor) {
365+
return Boolean.parseBoolean(
366+
tableDescriptor.getProperties().get(TABLE_DATALAKE_ENABLED.key()));
367+
}
368+
369+
/**
370+
* Maintains the {@code lakestream.enabled} Paimon option together with the {@code
371+
* table.datalake.enabled} lifecycle. Only new-layout (clean) tables are managed; legacy tables
372+
* that still carry the Fluss system columns are left untouched. Disabling removes the option
373+
* instead of persisting {@code false}.
374+
*
375+
* @param lakeStreamEnabled whether datalake acceleration is enabled after this change
376+
* @param legacyTable whether the Paimon table uses the legacy system-column layout
377+
* @param out the schema-change list to append to
378+
*/
379+
private static void appendLakeStreamOptionChange(
380+
boolean lakeStreamEnabled, boolean legacyTable, List<SchemaChange> out) {
381+
// Old-layout tables are outside the scope of this option.
382+
if (legacyTable) {
383+
return;
384+
}
385+
if (lakeStreamEnabled) {
386+
out.add(SchemaChange.setOption(LAKESTREAM_ENABLED_OPTION_KEY, Boolean.TRUE.toString()));
387+
} else {
388+
// Disabling (SetOption "false") or resetting removes the option entirely.
389+
out.add(SchemaChange.removeOption(LAKESTREAM_ENABLED_OPTION_KEY));
390+
}
391+
}
392+
336393
private static void validatePaimonOptions(Map<String, String> properties) {
337394
properties.forEach(
338395
(k, v) -> {

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

Lines changed: 149 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,7 @@
7474
import java.util.stream.Stream;
7575

7676
import static org.apache.fluss.lake.paimon.testutils.PaimonTestUtils.adjustToLegacyV1Table;
77+
import static org.apache.fluss.lake.paimon.utils.PaimonConversions.LAKESTREAM_ENABLED_OPTION_KEY;
7778
import static org.apache.fluss.lake.paimon.utils.PaimonConversions.PAIMON_UNSETTABLE_OPTIONS;
7879
import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME;
7980
import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME;
@@ -177,6 +178,7 @@ void testCreateLakeEnabledTable() throws Exception {
177178
new String[] {"log_c1", "log_c2"}),
178179
"log_c1,log_c2",
179180
BUCKET_NUM);
181+
assertThat(paimonLogTable.options()).containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true");
180182

181183
TableDescriptor logNoBucketKeyTable =
182184
TableDescriptor.builder()
@@ -234,6 +236,7 @@ void testCreateLakeEnabledTable() throws Exception {
234236
new String[] {"pk_c1", "pk_c2"}),
235237
"pk_c1",
236238
BUCKET_NUM);
239+
assertThat(paimonPkTable.options()).containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true");
237240

238241
// test partitioned table
239242
TablePath partitionedTablePath = TablePath.of(DATABASE, "partitioned_table");
@@ -708,6 +711,9 @@ void testAlterLakeEnabledLogTable() throws Exception {
708711

709712
Identifier paimonTablePath = Identifier.create(DATABASE, logTablePath.getTableName());
710713
Table enabledPaimonLogTable = paimonCatalog.getTable(paimonTablePath);
714+
// enabling lake acceleration on a clean table sets lakestream.enabled=true
715+
assertThat(enabledPaimonLogTable.options())
716+
.containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true");
711717

712718
Map<String, String> updatedProperties = new HashMap<>();
713719
updatedProperties.put(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true");
@@ -735,6 +741,9 @@ void testAlterLakeEnabledLogTable() throws Exception {
735741

736742
// verify LogTablet datalake status is disabled
737743
verifyLogTabletDataLakeEnabled(tableId, false);
744+
// disabling lake acceleration removes lakestream.enabled instead of storing false
745+
assertThat(paimonCatalog.getTable(paimonTablePath).options())
746+
.doesNotContainKey(LAKESTREAM_ENABLED_OPTION_KEY);
738747

739748
// try to enable lake table again
740749
enableLake = TableChange.set(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true");
@@ -743,6 +752,9 @@ void testAlterLakeEnabledLogTable() throws Exception {
743752

744753
// verify LogTablet datalake status is enabled again
745754
verifyLogTabletDataLakeEnabled(tableId, true);
755+
// re-enabling lake acceleration adds lakestream.enabled=true again
756+
assertThat(paimonCatalog.getTable(paimonTablePath).options())
757+
.containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true");
746758

747759
// write some data to the lake table
748760
writeData(paimonCatalog.getTable(paimonTablePath));
@@ -765,6 +777,143 @@ void testAlterLakeEnabledLogTable() throws Exception {
765777
verifyLogTabletDataLakeEnabled(tableId, true);
766778
}
767779

780+
@Test
781+
void testAlterLakeEnabledPrimaryKeyTable() throws Exception {
782+
// create pk table with lake disabled
783+
TableDescriptor pkTable =
784+
TableDescriptor.builder()
785+
.schema(
786+
Schema.newBuilder()
787+
.column("pk_c1", DataTypes.INT())
788+
.column("pk_c2", DataTypes.STRING())
789+
.primaryKey("pk_c1")
790+
.build())
791+
.property(ConfigOptions.TABLE_DATALAKE_ENABLED, false)
792+
.distributedBy(BUCKET_NUM)
793+
.build();
794+
TablePath pkTablePath = TablePath.of(DATABASE, "pk_table_alter");
795+
admin.createTable(pkTablePath, pkTable, false).get();
796+
Identifier paimonTablePath = Identifier.create(DATABASE, pkTablePath.getTableName());
797+
798+
// lake table not created yet while lake is disabled
799+
assertThatThrownBy(() -> paimonCatalog.getTable(paimonTablePath))
800+
.isInstanceOf(Catalog.TableNotExistException.class);
801+
802+
// enable lake acceleration sets lakestream.enabled=true
803+
admin.alterTable(
804+
pkTablePath,
805+
Collections.singletonList(
806+
TableChange.set(
807+
ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true")),
808+
false)
809+
.get();
810+
assertThat(paimonCatalog.getTable(paimonTablePath).options())
811+
.containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true");
812+
813+
// disable lake acceleration removes lakestream.enabled instead of storing false
814+
admin.alterTable(
815+
pkTablePath,
816+
Collections.singletonList(
817+
TableChange.set(
818+
ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "false")),
819+
false)
820+
.get();
821+
assertThat(paimonCatalog.getTable(paimonTablePath).options())
822+
.doesNotContainKey(LAKESTREAM_ENABLED_OPTION_KEY);
823+
824+
// re-enable lake acceleration adds lakestream.enabled=true again
825+
admin.alterTable(
826+
pkTablePath,
827+
Collections.singletonList(
828+
TableChange.set(
829+
ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true")),
830+
false)
831+
.get();
832+
assertThat(paimonCatalog.getTable(paimonTablePath).options())
833+
.containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true");
834+
835+
// resetting datalake.enabled is equivalent to disabling acceleration, and removes the
836+
// key from the table descriptor entirely (unlike SetOption "false")
837+
admin.alterTable(
838+
pkTablePath,
839+
Collections.singletonList(
840+
TableChange.reset(ConfigOptions.TABLE_DATALAKE_ENABLED.key())),
841+
false)
842+
.get();
843+
assertThat(paimonCatalog.getTable(paimonTablePath).options())
844+
.doesNotContainKey(LAKESTREAM_ENABLED_OPTION_KEY);
845+
846+
// re-enabling after a reset must still sync lakestream.enabled=true: since the reset
847+
// removed the key from the descriptor, MetadataManager must not rely solely on "the old
848+
// descriptor already had the key" to decide whether to sync to the lake table
849+
admin.alterTable(
850+
pkTablePath,
851+
Collections.singletonList(
852+
TableChange.set(
853+
ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true")),
854+
false)
855+
.get();
856+
assertThat(paimonCatalog.getTable(paimonTablePath).options())
857+
.containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true");
858+
}
859+
860+
@Test
861+
void testLegacyTableLakeStreamOptionUntouched() throws Exception {
862+
// create a clean, lake-enabled table, then turn it into a legacy table carrying the three
863+
// system columns. Old-layout tables are outside the scope of lakestream.enabled: altering
864+
// datalake.enabled must not add or remove the option on them.
865+
TablePath tablePath = TablePath.of(DATABASE, "legacy_lakestream_table");
866+
TableDescriptor tableDescriptor =
867+
TableDescriptor.builder()
868+
.schema(
869+
Schema.newBuilder()
870+
.column("c1", DataTypes.INT())
871+
.column("c2", DataTypes.STRING())
872+
.build())
873+
.property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
874+
.distributedBy(BUCKET_NUM, "c1")
875+
.build();
876+
admin.createTable(tablePath, tableDescriptor, false).get();
877+
Identifier paimonTablePath = Identifier.create(DATABASE, tablePath.getTableName());
878+
879+
adjustToLegacyV1Table(tablePath, paimonCatalog);
880+
String lakeStreamValueBeforeAlter =
881+
paimonCatalog
882+
.getTable(paimonTablePath)
883+
.options()
884+
.get(LAKESTREAM_ENABLED_OPTION_KEY);
885+
886+
// disable lake acceleration on a legacy table leaves the option untouched
887+
admin.alterTable(
888+
tablePath,
889+
Collections.singletonList(
890+
TableChange.set(
891+
ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "false")),
892+
false)
893+
.get();
894+
assertThat(
895+
paimonCatalog
896+
.getTable(paimonTablePath)
897+
.options()
898+
.get(LAKESTREAM_ENABLED_OPTION_KEY))
899+
.isEqualTo(lakeStreamValueBeforeAlter);
900+
901+
// re-enable lake acceleration on a legacy table also leaves the option untouched
902+
admin.alterTable(
903+
tablePath,
904+
Collections.singletonList(
905+
TableChange.set(
906+
ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true")),
907+
false)
908+
.get();
909+
assertThat(
910+
paimonCatalog
911+
.getTable(paimonTablePath)
912+
.options()
913+
.get(LAKESTREAM_ENABLED_OPTION_KEY))
914+
.isEqualTo(lakeStreamValueBeforeAlter);
915+
}
916+
768917
@Test
769918
void testThrowExceptionWhenConflictWithSystemColumn() {
770919
for (String systemColumn :

‎fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@
4545

4646
import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_ENABLED;
4747
import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_FORMAT;
48+
import static org.apache.fluss.lake.paimon.utils.PaimonConversions.LAKESTREAM_ENABLED_OPTION_KEY;
4849
import static org.apache.fluss.lake.paimon.utils.PaimonConversions.PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY;
4950
import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon;
5051
import static org.apache.fluss.lake.paimon.utils.PaimonTableValidation.isPaimonSchemaCompatible;
@@ -474,6 +475,78 @@ void testAlterTableAddColumnWhenPaimonSchemaNotMatch() throws Exception {
474475
changes));
475476
}
476477

478+
@Test
479+
void testCreateTableSetsLakeStreamEnabledForCleanTable() throws Exception {
480+
String database = "test_create_lakestream_db";
481+
String tableName = "test_create_lakestream_table";
482+
TablePath tablePath = TablePath.of(database, tableName);
483+
Identifier identifier = Identifier.create(database, tableName);
484+
485+
// getTableDescriptor sets table.datalake.enabled=true, so the clean table advertises its
486+
// LakeStream state to Paimon
487+
flussPaimonCatalog.createTable(
488+
tablePath, getTableDescriptor(FLUSS_SCHEMA), LAKE_CATALOG_CONTEXT);
489+
490+
Table table = flussPaimonCatalog.getPaimonCatalog().getTable(identifier);
491+
assertThat(table.options()).containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true");
492+
}
493+
494+
@Test
495+
void testCreateTableWithoutDataLakeEnabledHasNoLakeStreamOption() throws Exception {
496+
String database = "test_create_no_lakestream_db";
497+
String tableName = "test_create_no_lakestream_table";
498+
TablePath tablePath = TablePath.of(database, tableName);
499+
Identifier identifier = Identifier.create(database, tableName);
500+
501+
TableDescriptor tableDescriptor =
502+
TableDescriptor.builder()
503+
.schema(FLUSS_SCHEMA)
504+
.property(TABLE_DATALAKE_ENABLED.key(), "false")
505+
.property(TABLE_DATALAKE_FORMAT.key(), "paimon")
506+
.property(
507+
"table.datalake.paimon.warehouse",
508+
tempWarehouseDir.toURI().toString())
509+
.distributedBy(3)
510+
.build();
511+
flussPaimonCatalog.createTable(tablePath, tableDescriptor, LAKE_CATALOG_CONTEXT);
512+
513+
Table table = flussPaimonCatalog.getPaimonCatalog().getTable(identifier);
514+
assertThat(table.options()).doesNotContainKey(LAKESTREAM_ENABLED_OPTION_KEY);
515+
}
516+
517+
@Test
518+
void testAlterDataLakeEnabledMaintainsLakeStreamOptionForCleanTable() throws Exception {
519+
String database = "test_alter_lakestream_db";
520+
String tableName = "test_alter_lakestream_table";
521+
TablePath tablePath = TablePath.of(database, tableName);
522+
Identifier identifier = Identifier.create(database, tableName);
523+
createTable(database, tableName);
524+
525+
// disable lake acceleration removes lakestream.enabled instead of storing false
526+
flussPaimonCatalog.alterTable(
527+
tablePath,
528+
Collections.singletonList(TableChange.set(TABLE_DATALAKE_ENABLED.key(), "false")),
529+
LAKE_CATALOG_CONTEXT);
530+
Table table = flussPaimonCatalog.getPaimonCatalog().getTable(identifier);
531+
assertThat(table.options()).doesNotContainKey(LAKESTREAM_ENABLED_OPTION_KEY);
532+
533+
// re-enable lake acceleration adds lakestream.enabled=true again
534+
flussPaimonCatalog.alterTable(
535+
tablePath,
536+
Collections.singletonList(TableChange.set(TABLE_DATALAKE_ENABLED.key(), "true")),
537+
LAKE_CATALOG_CONTEXT);
538+
table = flussPaimonCatalog.getPaimonCatalog().getTable(identifier);
539+
assertThat(table.options()).containsEntry(LAKESTREAM_ENABLED_OPTION_KEY, "true");
540+
541+
// resetting datalake.enabled is equivalent to disabling acceleration
542+
flussPaimonCatalog.alterTable(
543+
tablePath,
544+
Collections.singletonList(TableChange.reset(TABLE_DATALAKE_ENABLED.key())),
545+
LAKE_CATALOG_CONTEXT);
546+
table = flussPaimonCatalog.getPaimonCatalog().getTable(identifier);
547+
assertThat(table.options()).doesNotContainKey(LAKESTREAM_ENABLED_OPTION_KEY);
548+
}
549+
477550
private org.apache.paimon.schema.Schema createPaimonSchema(
478551
List<String> primaryKeys, List<String> partitionKeys, String bucket, String bucketKey) {
479552
return createPaimonSchema(

0 commit comments

Comments
 (0)