Support Add Column at Last for DataLake Enabled Tables

[wuchong]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Support Add Column at Last for DataLake Enabled Tables

Solution

No response

Anything else?

No response

Willingness to contribute

• I'm willing to submit a PR!

[buvb]

Hi @wuchong , I'm the same person who commented on #2128
I'd like to take this task and work on the fix / implementation.
Could you please assign this issue to me?

Thanks!

[loserwang1024]

@buvb I'm currently working on this issue as well(A FIP on schema change). I'm curious about how you plan to handle schema version compatibility during tiered data processing. Maybe we can work it together.

[buvb]

Hi @wuchong,
I've submitted a PR to address this issue:
This PR implements schema evolution support for datalake-enabled tables (Paimon), specifically: #2189
Scope (MVP):

• Support ADD COLUMN ... LAST with NULLABLE columns only
• Fluss as source of truth: schema changes are first applied to Fluss, then synchronized to Paimon
• Idempotent retry handling for both Fluss and Paimon sides
• Tiering compatibility during schema transition period
  Key changes:
• [CoordinatorService]:Pass LakeCatalog to [alterTableSchema()] for Paimon sync
• [MetadataManager]: Add [syncSchemaChangesToLake()] method
• [SchemaUpdate]: Idempotent [addColumn()] - treat existing column with same type/comment as no-op
• [PaimonLakeCatalog]: Handle ColumnAlreadyExistException as success for retry scenarios
• [PaimonConversions]: Map Fluss AddColumn to Paimon SchemaChange
• [FlussRecordAsPaimonRow]: Handle tiering when Fluss record is wider than Paimon schema
  Tests:
• LakeEnabledTableCreateITCase#testAlterLakeEnabledTableSchema
• FlussRecordAsPaimonRowTest#testFlussRecordWiderThanPaimonSchema
• FlussRecordAsPaimonRowTest#testPaimonSchemaWiderThanFlussRecord
  This also closes Support schema evolution for data lake enabled table for paimon #2128.
  Please review when you have time. Thanks!

[buvb]

Hi @loserwang1024,

Thanks for reaching out! I'd be happy to collaborate.

For schema version compatibility during tiered data processing, my current implementation handles it in FlussRecordAsPaimonRow:

1. Paimon schema wider than Fluss record (Paimon already synced, old Fluss records still being consumed):

   • Padding NULL for missing business fields via isNullAt() returning true

2. Fluss record wider than Paimon schema (Fluss updated, Paimon not yet synced):

   • Using Math.min(internalRow.getFieldCount(), businessFieldCount) to ignore extra fields
   • This prevents crashes during the transition period

The key insight is that during tiering, the Fluss record and Paimon schema may temporarily be out of sync. The implementation gracefully handles both directions of mismatch.

You can check the details in PR #2189.