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

[fhan688]

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.enabled in 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:

Operation	Paimon option change
Create a new lake-enabled Paimon table	set lakestream.enabled=true
Enable lake acceleration for a clean table	set lakestream.enabled=true
Disable lake acceleration for a clean table	remove lakestream.enabled
Re-enable lake acceleration	set lakestream.enabled=true again

Disabling removes the option rather than storing lakestream.enabled=false.

Brief change log

All changes are in PaimonConversions (the Paimon conversion layer); the upstream coordination layer
is untouched.

• New option key: add LAKESTREAM_ENABLED_OPTION_KEY = "lakestream.enabled", written as a native Paimon option (it does not go through the fluss. property-prefix path).
• Create path (toPaimonSchema): when the table descriptor is lake-enabled, set lakestream.enabled=true on 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.
• Alter path (toPaimonSchemaChanges): the method already resolves the target layout via PaimonUtils.isLegacyTable(rowType). A new helper maybeSyncLakeStreamOption(...) is invoked from both the SetOption and ResetOption branches:
  • only new-layout (clean) tables are managed; legacy tables are short-circuited and left untouched;
  • table.datalake.enabled=true -> append SchemaChange.setOption("lakestream.enabled","true");
  • table.datalake.enabled=false or a reset of the key -> append SchemaChange.removeOption("lakestream.enabled").
• The existing table.datalake.enabled property handling is unchanged; lakestream.enabled is 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.isPaimonSchemaCompatible does not compare lakestream.enabled, so injecting it in toPaimonSchema (which is reused during the alter compatibility check) does not
affect re-enable schema compatibility.

Tests

• PaimonLakeCatalogTest (unit, no cluster):
  • a lake-enabled clean table gets lakestream.enabled=true on create;
  • a non-lake table has no lakestream.enabled;
  • enable/disable/reset of table.datalake.enabled on a clean table sets / removes / removes the option respectively.
• LakeEnabledTableCreateITCase (integration):
  • testCreateLakeEnabledTable -- assert lakestream.enabled=true on created log and primary-key tables;
  • testAlterLakeEnabledLogTable -- assert set on enable, removed on disable, set again on re-enable;
  • new testAlterLakeEnabledPrimaryKeyTable -- same enable -> disable -> re-enable lifecycle for a primary-key table;
  • new testLegacyTableLakeStreamOptionUntouched -- a legacy table (built via adjustToLegacyV1Table) keeps its lakestream.enabled unchanged across disable/enable.
• Covers both primary-key and log tables, as required by the acceptance criteria.
• Local run (Maven 3.8.6 + JBR 21, source level 8):
  • PaimonLakeCatalogTest -- 12 tests passed;
  • LakeEnabledTableCreateITCase -- 16 tests passed.
  • spotless:check could not run locally (google-java-format 1.15.0 is incompatible with JDK 21); formatting was verified manually (git diff --check clean, import order, line width). Full mvn clean verify should run in CI on JDK 11.

API and Format

No public API change. This adds a native Paimon table option lakestream.enabled on 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.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[luoyuxia]

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.

[luoyuxia]

@fhan688 Thanks for the pr. Left one comment. PTAL

[luoyuxia]

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.

[luoyuxia]

@fhan688 Thanks for the pr. I left minor comments. PTAL