[lake/paimon] Maintain lakestream.enabled with lake acceleration state - #4118
[lake/paimon] Maintain lakestream.enabled with lake acceleration state#4118fhan688 wants to merge 3 commits into
Conversation
|
|
||
| // #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)) { |
There was a problem hiding this comment.
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.
| * @param out the schema-change list to append to | ||
| */ | ||
| private static void maybeSyncLakeStreamOption( | ||
| String flussKey, @Nullable String value, boolean legacyTable, List<SchemaChange> out) { |
There was a problem hiding this comment.
Could we make the SET/RESET semantics explicit here? The nullable value uses null as a hidden RESET signal, and maybeSync does not show that this method appends a schema change.
Suggested complete refactor:
// SetOption
appendLakeStreamOptionChange(
setOption.getKey(),
Boolean.parseBoolean(setOption.getValue()),
paimonIncludingSystemColumns,
schemaChanges);
// ResetOption
appendLakeStreamOptionChange(
resetOption.getKey(), false, paimonIncludingSystemColumns, schemaChanges);
private static void appendLakeStreamOptionChange(
String changedOptionKey,
boolean lakeStreamEnabled,
boolean legacyTable,
List<SchemaChange> schemaChanges) {
if (legacyTable
|| !TABLE_DATALAKE_ENABLED.key().equals(changedOptionKey)) {
return;
}
if (lakeStreamEnabled) {
schemaChanges.add(
SchemaChange.setOption(
LAKESTREAM_ENABLED_OPTION_KEY,
Boolean.TRUE.toString()));
} else {
schemaChanges.add(
SchemaChange.removeOption(LAKESTREAM_ENABLED_OPTION_KEY));
}
}
This removes the nullable overload, makes RESET = false explicit, and names the list mutation. The javax.annotation.Nullable import can then be removed.
Purpose
Linked issue: close #4102
Follow-up to FIP-27 (clean Paimon lake table schema, umbrella #2411). Under FIP-27 newly created Paimon lake tables use a clean physical layout without the Fluss system columns (
__bucket,__offset,__timestamp), while pre-FIP-27 legacy tables still carry them.Paimon needs an explicit signal to know whether a table is currently accelerated by Fluss LakeStream. Today Fluss drives the lake-acceleration lifecycle but leaves Paimon unaware of the table's LakeStream state. This PR makes Fluss maintain a native Paimon table option
lakestream.enabledin lock-step with that lifecycle.Scope is limited to new-layout (clean) tables. Legacy tables that still carry the system columns are out of scope and keep their current behavior untouched.
Option lifecycle:
lakestream.enabled=truelakestream.enabled=truelakestream.enabledlakestream.enabled=trueagainDisabling removes the option rather than storing
lakestream.enabled=false.Brief change log
All changes are in
PaimonConversions(the Paimon conversion layer); the upstream coordination layeris untouched.
LAKESTREAM_ENABLED_OPTION_KEY = "lakestream.enabled", written as a native Paimon option (it does not go through thefluss.property-prefix path).toPaimonSchema): when the table descriptor is lake-enabled, setlakestream.enabled=trueon the generated Paimon options. Newly created tables are always clean (system columns are rejected earlier in this method), so no layout check is needed here.toPaimonSchemaChanges): the method already resolves the target layout viaPaimonUtils.isLegacyTable(rowType). A new helpermaybeSyncLakeStreamOption(...)is invoked from both theSetOptionandResetOptionbranches:table.datalake.enabled=true-> appendSchemaChange.setOption("lakestream.enabled","true");table.datalake.enabled=falseor a reset of the key -> appendSchemaChange.removeOption("lakestream.enabled").table.datalake.enabledproperty handling is unchanged;lakestream.enabledis maintained in addition, not as a replacement.Why the conversion layer (not the coordinator): only this layer holds the Paimon physical schema, which is required to distinguish clean vs. legacy tables. Verified that
PaimonTableValidation.isPaimonSchemaCompatibledoes not comparelakestream.enabled, so injecting it intoPaimonSchema(which is reused during the alter compatibility check) does notaffect re-enable schema compatibility.
Tests
PaimonLakeCatalogTest(unit, no cluster):lakestream.enabled=trueon create;lakestream.enabled;table.datalake.enabledon a clean table sets / removes / removes the option respectively.LakeEnabledTableCreateITCase(integration):testCreateLakeEnabledTable-- assertlakestream.enabled=trueon created log and primary-key tables;testAlterLakeEnabledLogTable-- assert set on enable, removed on disable, set again on re-enable;testAlterLakeEnabledPrimaryKeyTable-- same enable -> disable -> re-enable lifecycle for a primary-key table;testLegacyTableLakeStreamOptionUntouched-- a legacy table (built viaadjustToLegacyV1Table) keeps itslakestream.enabledunchanged across disable/enable.PaimonLakeCatalogTest-- 12 tests passed;LakeEnabledTableCreateITCase-- 16 tests passed.spotless:checkcould not run locally (google-java-format 1.15.0 is incompatible with JDK 21); formatting was verified manually (git diff --checkclean, import order, line width). Fullmvn clean verifyshould run in CI on JDK 11.API and Format
No public API change. This adds a native Paimon table option
lakestream.enabledon new-layout (clean) Paimon lake tables, maintained together with the lake-acceleration lifecycle.Legacy tables carrying the Fluss system columns are not affected. Fluss's own storage format is unchanged.
Documentation
No documentation change. This is an internal Paimon table option maintained by Fluss; it introduces no user-facing configuration or feature surface.