[lake]Tolerate Iceberg lake table existent if the schema and properties matches#3601
Conversation
There was a problem hiding this comment.
Pull request overview
This PR extends the Iceberg lake catalog integration to tolerate an already-existing Iceberg table during Fluss table creation, as long as the existing table matches Fluss’s expected schema, partition spec, sort order, and required table properties. This aligns Iceberg behavior with existing “reuse existing lake table” support in other lake formats.
Changes:
- Enhanced
IcebergLakeCatalog#createTableto detect existing Iceberg tables and validate compatibility (schema/spec/sort/properties), allowing idempotent “recreate” when compatible and enforcing “empty table” whenisCreatingFlussTable=true. - Added compatibility helper methods (
isIcebergSchemaCompatibleWithSchema,isIcebergPartitionSpecCompatible,isIcebergSortOrderCompatible,isIcebergPropertiesCompatible) used for validation and unit testing. - Added comprehensive unit tests covering compatible reuse, first-time Fluss creation constraints, and various incompatibility scenarios.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java | Adds compatibility checks to tolerate existing Iceberg tables (or fail with detailed incompatibility reasons). |
| fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java | Adds tests for idempotent create behavior and compatibility validation (schema/spec/sort/properties, empty vs non-empty). |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| private void appendDummyDataFile(Table icebergTable) { | ||
| DataFile dataFile = | ||
| DataFiles.builder(PartitionSpec.unpartitioned()) | ||
| .withPath("/tmp/fluss-fake-file.parquet") | ||
| .withFileSizeInBytes(1024) | ||
| .withRecordCount(1L) | ||
| .withFormat(FileFormat.PARQUET) | ||
| .build(); | ||
| icebergTable.newAppend().appendFile(dataFile).commit(); | ||
| } |
There was a problem hiding this comment.
@Guosmilesmile Thanks. Could you please help rebase main branch
Withdrawing approval after further review.
| } | ||
| } | ||
|
|
||
| return true; |
There was a problem hiding this comment.
Blocking: This compatibility check ignores Schema.identifierFieldIds(). A pre-created Iceberg table with the same columns, partition spec, sort order, and properties but no identifier fields therefore passes this check. IcebergLakeWriter#createRecordWriter then sees an empty identifier set and selects AppendOnlyTaskWriter, so UPDATE_BEFORE and DELETE records are appended instead of emitted as equality deletes, breaking primary-key semantics. Please compare identifier fields by resolved column names (raw IDs can differ between schemas) and add coverage for missing or incorrect identifiers.
There was a problem hiding this comment.
Good Catch. I will add compare identifier fields.
23b76b9 to
c7ece59
Compare
|
CI error since lake Test meet 60min timeout. |
|
Raise a pr to split old flink module to single Test module。#3692 |
Purpose
Since we have support this feature in paimon and hudi, I think we should support it in iceberg too.
Brief change log
Tests
API and Format
Documentation