[lake/hudi] Support clean and legacy Hudi lake table schemas - #4077
Open
fhan688 wants to merge 3 commits into
Open
[lake/hudi] Support clean and legacy Hudi lake table schemas#4077fhan688 wants to merge 3 commits into
fhan688 wants to merge 3 commits into
Conversation
luoyuxia
reviewed
Aug 24, 2026
| * {@code _hoodie_*} metadata columns, which are not part of the logical row type). A projected | ||
| * schema must not be passed to this method. | ||
| */ | ||
| public static boolean isLegacyTable(RowType rowType) { |
Contributor
There was a problem hiding this comment.
we don't need to consider compatiblity since hudi is never exposed to users in any public version.
Contributor
There was a problem hiding this comment.
Pull request overview
This PR implements the Hudi portion of FIP-27 in fluss-lake-hudi, introducing “clean” newly-created Hudi lake tables (user columns only) while retaining full read/write compatibility for existing legacy tables that still contain the trailing Fluss system columns (__bucket, __offset, __timestamp).
Changes:
- Stop appending legacy system columns when creating new Hudi lake tables; add explicit legacy-schema construction for compatibility checks.
- Add legacy-vs-clean layout detection and use it to drive tiering writer output and lake reader projection/trimming behavior.
- Update unit/integration tests to cover clean/legacy layouts, re-enabling tiering on legacy tables, and sentinel
-1Loffset/timestamp behavior in readers.
Reviewed changes
Copilot reviewed 13 out of 13 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeCatalog.java | Detect legacy layout for existing tables and validate schema compatibility against the correct expected layout. |
| fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRecordReader.java | Drive projection/trimming based on legacy detection and emit sentinel -1L offset/timestamp. |
| fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRowAsFlussRow.java | Make system-column stripping conditional so clean tables don’t lose trailing fields. |
| fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSortedRecordReader.java | Treat only legacy system columns as skippable when mapping user-field positions. |
| fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRow.java | Emit legacy system-column values only for legacy tables and validate business-field counts accordingly. |
| fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiConversions.java | Build clean vs legacy Flink/Hudi schemas and catalog tables; keep name-conflict checks for system column names. |
| fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiUtils.java | Introduce Hudi legacy-layout detection helper used by writer/reader code paths. |
| fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/HudiLakeCatalogTest.java | Add re-enable-tiering legacy-table test and update expected schema assertions for clean tables. |
| fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiRecordReaderTest.java | Update reader assertions to sentinel -1L and add clean-table iterator coverage. |
| fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/testutils/FlinkHudiTieringTestBase.java | Adjust integration assertions to compare business columns only for clean tables. |
| fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiTieringITCase.java | Update IT case to use revised clean-table row comparison helper. |
| fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRowTest.java | Split writer tests into clean vs legacy cases and assert correct arity/system-column behavior. |
| fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/utils/HudiUtilsTest.java | Add tests for legacy detection with clean, legacy, and partial-name-match schemas. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
+48
to
+56
| public static boolean isLegacyTable(RowType rowType) { | ||
| for (String systemColumn : LEGACY_SYSTEM_COLUMNS.keySet()) { | ||
| if (!rowType.getFieldNames().contains(systemColumn)) { | ||
| return false; | ||
| } | ||
| } | ||
| return true; | ||
| } | ||
| } |
Comment on lines
+214
to
+221
| @VisibleForTesting | ||
| boolean isLegacyTable(CatalogBaseTable existingTable) { | ||
| Set<String> columnNames = new HashSet<>(); | ||
| for (ColumnSignature column : extractColumns(existingTable)) { | ||
| columnNames.add(column.name); | ||
| } | ||
| return columnNames.containsAll(LEGACY_SYSTEM_COLUMNS.keySet()); | ||
| } |
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 12 out of 12 changed files in this pull request and generated 3 comments.
Suppressed comments (1)
fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSortedRecordReader.java:173
resolveRecordKeyInfocurrently counts every physical column from the Hudi table schema as a user field. For legacy tables that still have the trailing__bucket/__offset/__timestampcolumns, this shiftsuserFieldPositionand can produce wrong key positions (or fail to find the record-key field in projections), breaking sorted reads.
Map<String, DataType> dataTypesByName = new HashMap<>();
Map<String, Integer> userFieldPositionsByName = new HashMap<>();
List<DataType> userDataTypes = new ArrayList<>();
int userFieldPosition = 0;
for (Schema.UnresolvedColumn column :
hudiTableInfo.getHudiTable().getUnresolvedSchema().getColumns()) {
DataType dataType = getDataType(column);
dataTypesByName.put(column.getName(), dataType);
userFieldPositionsByName.put(column.getName(), userFieldPosition++);
userDataTypes.add(dataType);
}
Comment on lines
+39
to
43
| public HudiRowAsFlussRow() {} | ||
|
|
||
| public HudiRowAsFlussRow(RowData rowData) { | ||
| this(rowData, true); | ||
| } | ||
|
|
||
| HudiRowAsFlussRow(RowData rowData, boolean stripSystemColumns) { | ||
| this.rowData = rowData; | ||
| this.stripSystemColumns = stripSystemColumns; | ||
| } | ||
|
|
||
| private HudiRowAsFlussRow(boolean stripSystemColumns) { | ||
| this.stripSystemColumns = stripSystemColumns; | ||
| } |
Comment on lines
41
to
+48
| public void setFlussRecord(LogRecord logRecord) { | ||
| this.logRecord = logRecord; | ||
| this.internalRow = logRecord.getRow(); | ||
| this.originRowFieldCount = internalRow.getFieldCount(); | ||
| checkState( | ||
| originRowFieldCount == rowType.getFieldCount() - SYSTEM_COLUMNS.size(), | ||
| "Hudi table field count must equal Fluss LogRecord field count plus system fields."); | ||
| internalRow.getFieldCount() == fieldCount, | ||
| "The Fluss record's field count (%s) must equal the Hudi table field count (%s).", | ||
| internalRow.getFieldCount(), | ||
| fieldCount); |
Comment on lines
+225
to
230
| // FIP-27: the physical schema is metadata columns followed by user columns only; strip | ||
| // the Hudi metadata columns so only the business columns are emitted. | ||
| this.projectedRow = | ||
| ProjectedRow.from( | ||
| IntStream.range( | ||
| metadataFieldCount, | ||
| schema.getFields().size() - systemFieldCount) | ||
| IntStream.range(metadataFieldCount, schema.getFields().size()) | ||
| .toArray()); |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
Sub-task of the FIP-27 umbrella (#2411): Remove Mandatory System Columns From Fluss Lake Tables.
Today every Hudi 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 Hudi and other engines.This PR implements the Hudi part of FIP-27, mirroring the Paimon (#3902) and Iceberg (#3903) parts: newly created Hudi 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, and re-enabling tiering.
Unlike Iceberg, Hudi does not weave
__bucketinto the physical partition spec (bucket routing is handled by Hudi's index / file-group model), so the Hudi implementation is structurally closer to Paimon: there is no partition-spec or sort-order to reconstruct, and compaction — which operates on file groups rather than a per-bucket predicate — is unaffected and stays enabled for both layouts.Brief change log
HudiUtils.isLegacyTable(RowType), which returnstrueonly when all three system columns are present. Requiring all three guards against misdetecting a user table that merely reuses one of the system-column names (for example a table with only a__timestampcolumn being onboarded to Fluss).HudiLakeCatalog.SYSTEM_COLUMNSis renamed toLEGACY_SYSTEM_COLUMNSto make its meaning explicit.HudiConversions.convertToFlinkResolvedSchemano longer appends the system columns; new tables are clean. The user-column name-conflict check against system names is kept. NewconvertToLegacyFlinkResolvedSchema/createLegacyHudiCatalogTablereproduce the legacy layout for the compatibility check below.HudiLakeCatalogdetects the existing physical table's layout and compares it against the legacy schema (when legacy) or the clean schema (when clean), so disabling and re-enabling tiering on a legacy table is accepted and preserves its physical layout.FlussRecordAsHudiRowderivesisLegacyfrom the table'sRowType; it emits the three system values only for legacy tables, and for clean tables the business-field count equals the full row with no system fields written. The field-countcheckStateis preserved (now against the business-field count) for both layouts.HudiRecordReaderprojects the system columns only for legacy tables and emits a sentinel-1Llog offset / timestamp for both layouts, consistent with the Paimon and Iceberg readers and theLakeRecordRecordEmittercontract. The iterator trims the trailing system columns only when reading a legacy table.__bucketpredicate, so it has no equivalent of the per-bucket ownership problem that required disabling clean-table compaction for Iceberg.Tests
HudiUtilsTest(new):isLegacyTablefor clean, legacy, and partial (one/two system-named columns → clean) layouts.HudiLakeCatalogTest: create-table assertions updated to expect the clean layout; newtestReEnableTieringOnLegacyTablebuilds a legacy physical table and verifies re-enabling tiering is accepted and preserves the legacy layout.FlussRecordAsHudiRowTest: split into legacy (with system columns) and clean (business columns only) writer cases.HudiRecordReaderTest: offset/timestamp assertions updated to the-1Lsentinel; added a clean-table iterator case.FlinkHudiTieringTestBase/HudiTieringITCase: read helpers compare business columns only for the clean layout.fluss-lake-hudimodule test suite passes (118 tests). The clean writer path is exercised end-to-end byHudiTieringTest; the legacy writer, reader, and re-enable paths are covered by unit tests.API and Format
Storage format change: Hudi lake tables created after this change have no trailing
__bucket/__offset/__timestampsystem columns. Existing tables are not migrated, keep their current physical format, and remain fully readable and writable. No public Java API change.Documentation
This implements the Hudi side of FIP-27. Feature behavior (clean vs. legacy layouts and detection) is documented under the umbrella issue #2411; no standalone doc change in this PR.