[lake/paimon] Remove system columns from Paimon

[luoyuxia]

Purpose

Linked issue: close #xxx

Brief change log

Tests

API and Format

Documentation

[Copilot]

Pull request overview

This pull request removes system columns (__bucket, __offset, __timestamp) from Paimon datalake tables and introduces a storage version mechanism (version 2) to track the schema format. The change maintains backward compatibility with legacy v1 tables that include system columns.

Changes:

• Introduces TABLE_DATALAKE_STORAGE_VERSION configuration option to track datalake schema version
• Removes system columns from Paimon schemas for new v2 tables
• Implements conditional logic throughout the codebase to handle both v1 (with system columns) and v2 (without system columns) tables
• Updates tests to reflect the new schema format and adds utilities for testing legacy table compatibility

Reviewed changes

Copilot reviewed 20 out of 20 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java	Adds new TABLE_DATALAKE_STORAGE_VERSION config option
fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java	Adds method to retrieve datalake storage version
fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeCatalog.java	Defines CURRENT_LAKE_STORAGE_VERSION constant (version 2)
fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java	Adds validation to prevent users from manually setting storage version
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java	Automatically sets storage version when creating tables with datalake format
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java	Conditionally adds system columns based on storage version presence
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java	Updates schema change handling to get paimon table for checking system columns
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRow.java	Adds conditional logic to handle system columns based on flag
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/RecordWriter.java	Passes system column flag to FlussRecordAsPaimonRow
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/AppendOnlyWriter.java	Propagates system column flag parameter
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java	Propagates system column flag parameter
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeWriter.java	Determines whether Paimon table includes system columns based on storage version
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonRecordReader.java	Adds logic to conditionally handle system columns when reading from Paimon
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java	Handles legacy v1 tables and v2 tables differently for timestamp-based reads
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/PaimonTestUtils.java	Adds utility to adjust tables to legacy v1 format for testing
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/FlussRecordAsPaimonRowTest.java	Updates tests to remove system column expectations
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java	Updates assertions to remove system column checks
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java	Updates schema expectations and adds legacy table compatibility tests
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java	Adds tests for reading from legacy v1 tables
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadFromTimestampITCase.java	Updates test to use legacy v1 table for timestamp-based reads

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.