[lake/iceberg] Support clean and legacy Iceberg lake table schemas - #4019
[lake/iceberg] Support clean and legacy Iceberg lake table schemas#4019fhan688 wants to merge 3 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
Warning
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Implements FIP-27 for Iceberg lake tables by introducing “clean” schemas (user columns only) while keeping legacy tables (with __bucket/__offset/__timestamp) fully readable/writable via schema-based layout detection.
Changes:
- Add schema-based legacy detection (
IcebergUtils.isLegacyTable) and thread legacy/clean behavior through writers/readers/planners. - Stop appending Iceberg system columns for newly created tables; preserve legacy layout for compatibility checks and certain evolutions.
- Update partition/sort handling and adjust tests to validate clean-table behavior (unpartitioned/unsorted where applicable).
Reviewed changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergUtils.java | Adds legacy-vs-clean layout detection helper. |
| fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergSchemaUtils.java | Introduces clean vs legacy schema builders and renames system column map. |
| fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java | Makes schema evolution and sort order conditional on legacy layout. |
| fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergPartitionSpecUtils.java | Skips legacy identity(__bucket) partitioning for clean bucket-unaware tables. |
| fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergConversions.java | Makes bucket partition key + bucket filter conditional on spec/schema shape. |
| fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/RecordWriter.java | Threads legacy-layout flag into tiering record conversion. |
| fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/FlussRecordAsIcebergRecord.java | Emits system columns only for legacy tables; adjusts positional mapping. |
| fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlanner.java | Updates bucket/partition extraction for clean vs legacy partition specs. |
| fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordReader.java | Projects/reads offset+timestamp only for legacy tables; emits sentinel otherwise. |
| fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergRecordAsFlussRow.java | Computes business-field count dynamically to handle clean vs legacy/projections. |
| fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteDataFiles.java | Skips __offset-based sorting when offset column is absent. |
| fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java | Updates expectations for clean schemas, partition specs, and unsorted tables. |
| fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkCatalogLakeTest.java | Updates schema size expectation to user columns only. |
| fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/FlinkIcebergTieringTestBase.java | Makes offset assertions conditional; handles clean tables without __offset. |
| fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergSchemaEvolutionITCase.java | Updates schema evolution assertions for clean “append-last” behavior. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
|
|
||
| public void setFlussRecord(LogRecord logRecord) { | ||
| this.logRecord = logRecord; | ||
| this.internalRow = logRecord.getRow(); |
| boolean lastIsLegacyBucket = | ||
| lastSourceField != null && lastSourceField.name().equals(BUCKET_COLUMN_NAME); | ||
| boolean lastIsBucketTransform = lastField.transform().toString().startsWith("bucket["); | ||
| if (lastIsLegacyBucket || lastIsBucketTransform) { | ||
| partitionKey.set(pos, bucket); | ||
| } |
| Types.NestedField lastSourceField = table.schema().findField(lastField.sourceId()); | ||
| boolean lastIsLegacyBucket = | ||
| lastSourceField != null && lastSourceField.name().equals(BUCKET_COLUMN_NAME); | ||
| boolean lastIsBucketTransform = lastField.transform().toString().startsWith("bucket["); |
| /** | ||
| * Returns whether the given Iceberg table is a legacy table (has the three trailing system | ||
| * columns). | ||
| * | ||
| * <p>Detection: if the {@code __timestamp} system column exists in the physical schema, this is | ||
| * a legacy table. Clean tables have no system columns. | ||
| */ | ||
| public static boolean isLegacyTable(Schema icebergSchema) { | ||
| return icebergSchema.findField(TIMESTAMP_COLUMN_NAME) != null; | ||
| } |
| // FIP-27: a clean table stores only user columns; only legacy tables carry the | ||
| // trailing __bucket/__offset/__timestamp columns (offset at idx 3). | ||
| if (actualRecord.struct().field(OFFSET_COLUMN_NAME) != null) { | ||
| assertThat(actualRecord.get(3)).isEqualTo(startingOffset++); |
| Iterable<Record> iterable = | ||
| Parquet.read(table.io().newInputFile(file.location())) | ||
| .project(table.schema()) | ||
| .createReaderFunc( | ||
| fileSchema -> | ||
| GenericParquetReaders | ||
| .buildReader( | ||
| table.schema(), | ||
| fileSchema)) | ||
| .build(); | ||
| iterable.forEach(records::add); |
| expression = Expressions.and(expression, Expressions.equal(BUCKET_COLUMN_NAME, bucket)); | ||
| // FIP-27: legacy tables carry the __bucket column and are filtered per bucket. Clean | ||
| // tables have no __bucket column, so no bucket-level filter is applied. | ||
| if (table.schema().findField(BUCKET_COLUMN_NAME) != null) { |
There was a problem hiding this comment.
Compaction for the clean schema looks difficult to preserve safely with the current per-bucket scheduling model. Once __bucket is removed, bucket-unaware clean tables no longer have a physical field or partition value that can give each IcebergLakeWriter an exclusive file set. Dropping the bucket predicate means compaction jobs for different Fluss buckets can plan the same files and attempt overlapping rewrites. Even for bucket-aware tables, we would need to scope planning through the Iceberg bucket transform rather than simply removing the predicate.
Would it be safer for this PR to explicitly disable auto compaction for clean-schema tables for now, while keeping the existing behavior for legacy tables? We can add clean-table compaction later with a separate ownership model (for example, one task per table/partition, or coordinated stable file-group sharding based on a shared snapshot).
There was a problem hiding this comment.
__bucket is not only a schema predicate for compaction. It also scopes each IcebergRewriteDataFiles instance to the bucket it was created for. IcebergLakeWriter.scheduleCompaction creates one compaction per bucket, but after dropping this predicate a clean table can cause each compaction to plan files from all buckets while the writer is still pinned to its own bucket.
This could rewrite files from bucket N into bucket M and also cause concurrent per-bucket compactions to rewrite/commit the same files. Could we derive the bucket scope from the file's partition tuple instead of the schema, or otherwise skip compaction for clean tables until we can preserve that scope?
Please also add a multi-bucket compaction IT to cover this case.
| // A legacy table has __timestamp as the last column; a clean table has no system columns. | ||
| int total = record.struct().fields().size(); | ||
| boolean isLegacy = record.struct().field(TIMESTAMP_COLUMN_NAME) != null; | ||
| return isLegacy ? total - LEGACY_SYSTEM_COLUMNS.size() : total; |
There was a problem hiding this comment.
This calculation does not seem correct for a projected legacy record. IcebergRecordReader.applyProject() appends only __offset and __timestamp; it does not project __bucket. For example, projecting one business column produces [id, __offset, __timestamp]. Since __timestamp is present, this method returns 3 - LEGACY_SYSTEM_COLUMNS.size() = 0, while the correct business field count is 1.
The current reader path wraps this adapter in ProjectedRow, whose own mapping length masks the incorrect value, but IcebergRecordAsFlussRow#getFieldCount() itself is still wrong for projected legacy records and may break direct callers or a later refactor. Could we subtract the number of system fields actually present in record.struct() instead of always subtracting all three? It would also be useful to add field-count tests for a clean full record, a legacy full record, and a legacy projected record.
| @@ -125,7 +124,8 @@ void testSchemaEvolutionLogTable() throws Exception { | |||
| .map(Types.NestedField::name) | |||
There was a problem hiding this comment.
The compatibility changes need dedicated regression tests before this can be considered complete. The current test diff only adjusts existing clean-schema assertions/helpers and adds no new @Test method, so it does not directly verify legacy Iceberg tiering, legacy projected reads, disable/re-enable tiering, or clean maintenance/compaction behavior. These paths can regress while the current tests still pass.
Please add an explicit clean/legacy test matrix covering writers, readers/projections, schema evolution, re-enabling tiering, and maintenance (or document and test that clean-schema compaction is intentionally disabled). The Paimon implementation in #3982 is a useful reference: it added independent tests for legacy schema evolution, legacy union reads, clean writer output, and legacy tiering.
| lastSourceField != null && lastSourceField.name().equals(BUCKET_COLUMN_NAME); | ||
| boolean lastIsBucketTransform = lastField.transform().toString().startsWith("bucket["); | ||
| if (lastIsLegacyBucket || lastIsBucketTransform) { | ||
| partitionKey.set(pos, bucket); |
There was a problem hiding this comment.
For an unpartitioned spec, should we return null here instead of an empty PartitionKey? A non-null empty key sends the writer through the partitioned path and can produce data//file.parquet. Returning null would use the canonical unpartitioned path data/file.parquet.
| expression = Expressions.and(expression, Expressions.equal(BUCKET_COLUMN_NAME, bucket)); | ||
| // FIP-27: legacy tables carry the __bucket column and are filtered per bucket. Clean | ||
| // tables have no __bucket column, so no bucket-level filter is applied. | ||
| if (table.schema().findField(BUCKET_COLUMN_NAME) != null) { |
There was a problem hiding this comment.
__bucket is not only a schema predicate for compaction. It also scopes each IcebergRewriteDataFiles instance to the bucket it was created for. IcebergLakeWriter.scheduleCompaction creates one compaction per bucket, but after dropping this predicate a clean table can cause each compaction to plan files from all buckets while the writer is still pinned to its own bucket.
This could rewrite files from bucket N into bucket M and also cause concurrent per-bucket compactions to rewrite/commit the same files. Could we derive the bucket scope from the file's partition tuple instead of the schema, or otherwise skip compaction for clean tables until we can preserve that scope?
Please also add a multi-bucket compaction IT to cover this case.
| : structType.fields().size(); | ||
| } | ||
|
|
||
| public void setFlussRecord(LogRecord logRecord) { |
There was a problem hiding this comment.
Could we restore the field-count validation here using businessFieldCount? The previous check prevented schema mismatches from surfacing later as a bare ArrayIndexOutOfBoundsException inside the Iceberg appender. Something like checkState(internalRow.getFieldCount() == businessFieldCount, ...) would preserve that validation for both legacy and clean layouts and fail with a much more useful error.
| // three trailing system columns. To handle re-enabling lake tiering on a legacy table, we | ||
| // build the expected schema to match what the physical table actually has. | ||
| Schema expectedSchema = | ||
| IcebergUtils.isLegacyTable(icebergSchema) |
There was a problem hiding this comment.
This is the right fix, but it looks like it only covers applySchemaChanges(). The re-enable path through createTable() still constructs the clean schema/partition/sort expectations and compares them against the legacy physical layout, causing the checks around L320/L330/L341 to fail. Could we apply the same legacy-aware handling to that existing-table path as well? An IT covering legacy table → disable tiering → re-enable tiering would help verify this.
| @@ -508,7 +528,11 @@ private void createDatabase(String databaseName) { | |||
| } | |||
|
|
|||
| private SortOrder createSortOrder(Schema icebergSchema) { | |||
There was a problem hiding this comment.
Returning SortOrder.unsorted() here makes the error message around L351 stale. It still tells users to pre-create the table with ASC(__offset), but clean tables no longer have __offset and correctly expect unsorted(). Could the error message make the ASC(__offset) guidance conditional on the legacy layout?
|
I think the legacy/clean test coverage needs to be strengthened across the affected layers.
More broadly, Paimon's #3982 added legacy coverage at each layer: writer, tiering, union read, and catalog. This PR doesn't have equivalent Iceberg coverage, and |
Purpose
Linked issue: close #3903
Sub-task of the FIP-27 umbrella (#2411): Remove Mandatory System Columns From Fluss Lake Tables.
Today every Iceberg lake table Fluss creates is forced to carry three mandatory system columns (
__bucket,__offset,__timestamp) as its last physical columns. They pollute the schema users see from Iceberg and other engines. Unlike Paimon,__bucketis additionally woven into the physical layout as anidentity(__bucket)partition field for bucket-unaware tables, and__offsetdrives the table sort order.This PR implements the Iceberg part of FIP-27, mirroring the Paimon part (#3902): newly created Iceberg lake tables use a clean physical schema containing only user-defined columns, while existing legacy tables that still carry the three system columns remain fully readable and writable without any schema migration. Both layouts are supported across create, tiering writers, readers, projections, schema evolution, compaction, and re-enabling tiering.
Brief change log
Layout detection (single source of truth): add
IcebergUtils.isLegacyTable(Schema), defined asschema.findField("__timestamp") != null. Detection is purely schema-based — no new table property or metadata, so existing tables are never migrated.Create:
IcebergSchemaUtils.createIcebergSchemano longer appends the system columns; new tables are clean.SYSTEM_COLUMNSis renamed toLEGACY_SYSTEM_COLUMNS, and a newcreateLegacyIcebergSchemareproduces the legacy layout for compatibility checks. The user-column name-conflict check against system names is kept.Partition spec (Iceberg-specific): in
IcebergPartitionSpecUtils, a bucket-unaware clean table leaves the spec unpartitioned instead of addingidentity(__bucket)(clean tables have no__bucketcolumn). Legacy tables keep theidentity(__bucket)partition; bucket-aware tables keep theirbucket(userCol)transform.Sort order (Iceberg-specific):
IcebergLakeCatalog.createSortOrderreturnsSortOrder.unsorted()for clean tables, which have no__offsetto sort by; legacy tables keepasc(__offset).Schema evolution:
applySchemaChangesinserts a new business column before the first system column only for a legacy table; for a clean table it is appended normally.AddColumnrejects names that collide withLEGACY_SYSTEM_COLUMNSviaInvalidTableException.Compatibility / re-enable tiering:
isIcebergSchemaCompatibledetects the existing layout and, for a legacy table, compares against the legacy schema (createLegacyIcebergSchema) so disabling and re-enabling tiering preserves the physical layout.IcebergPartitionSpecValidatorforks the same way when validating the target spec.Row tiering writer:
FlussRecordAsIcebergRecordemits the three system values only for legacy tables; for clean tables the business-field count equals the full row and no system fields are written. The layout flag is threaded throughRecordWriter.Split planning / bucket extraction (Iceberg-specific):
IcebergSplitPlanner.createBucketEy-spec case (clean bucket-unaware table →bucket = -1) and detects the bucket slot bytransform type (bucket[…]) rather than by source column name, so clean bucket-aware tables — which carry abucket(userCol)transform but no__bucket` column — work correctly.Partition / filter conversion (Iceberg-specific):
IcebergConversions.toPartitionsets the bucket slot only when the last partition field is a legacyidentity(__bucket)or abucket[…]transform;toFilterExpressionadds theequal(__bucket, bucket)predicate only for legacy tables.Compaction:
IcebergRewriteDataFilesnull-guards the__offsetlookup before building t, so clean tables compact without system columns.Reader / projection:
IcebergRecordReaderprojects and reads__offset/__timestamponly for legacy tables. For clean tables it emits a sentinel-1Llog offset / timestamp, consistent with the existingUNKNOWN_OFFSET = -1convention.Row adapter:
IcebergRecordAsFlussRowcomputes the business-field count dynamically fromad of hard-subtracting the three trailing system columns, which also fixes latent miscounts onprojected rows.
Tests
Adapted
IcebergLakeCatalogTest(create/alter assertions now expect user-only schemas, unparIcebergSchemaEvolutionITCase(column-order assertions changed fromisLessThan(indexOf("__bucket"))toisEqualTo(fieldNames.size() - 1)), and theFlinkIcebergoffset assertions guarded on__offset` presence) to the clean layout.Legacy coverage retained:
IcebergTieringTest,IcebergPartitionSpecValidatorTest, andIcebergSplitPlannerTestcontinue to exercise the legacyidentity(__bucket)/asc(__offset)paths;IcebergSplitPlannerTestadditionally covers the empty-spec (clean bucket-unaware) and cleanbucket(col)transform cases.Verified locally on JDK 11: unit tests
IcebergLakeCatalogTest(40),IcebergTieringTest(6lidatorTest(9),IcebergSplitPlannerTest(6) pass; ITsIcebergTieringITCase(1),IcebergSchemaEvolutionITCase(5),IcebergRewriteITCase(3), andFlinkUnionRead*IcebergITCase` union-read cases all pass in isolation.API and Format
No public API change. This changes the physical schema of newly created Iceberg lake tables (clean layout by default) and, for bucket-unaware tables, removes the
identity(__bucket)partition andasc(__offset)sortorder from newly created tables. Existing tables are not migrated and keep their current physic rolling-upgrade requirements are covered by the umbrella #2411 and documented in #3905:
Documentation
Feature behavior (clean vs. legacy layouts, detection, and the rolling-upgrade/compatibility mately under #3905. No standalone doc change in this PR.