Repository navigation
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()); |
luoyuxia
reviewed
Aug 25, 2026
luoyuxia
left a comment
Contributor
There was a problem hiding this comment.
@fhan688 Overall LGTM from the implementation perspective. The clean-only schema is handled consistently across table creation, writer, reader/projection, nested rows, sorted reads, and Hudi file-group-based compaction. One remaining item before merge: the current documentation and PR framing still describe behavior that was removed in the follow-up commits. website/docs/streaming-lakehouse/datalake-formats/hudi.md still says Fluss appends __bucket, __offset, and __timestamp, while the current implementation creates Hudi tables with user columns only. The PR title and description also still claim clean/legacy layout detection, legacy read/write support, and re-enable tests, none of which remain in the current head. Since Hudi has not been publicly released and this PR intentionally switches directly to clean-only, could you update hudi.md and revise the PR title/description to make that decision explicit? The guidance reserving the three system-column names can remain if that is still intended.
luoyuxia
approved these changes
Aug 25, 2026
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.
Until now, every Hudi lake table Fluss created was 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: newly created Hudi lake tables use a clean physical schema containing only user-defined columns.
Unlike Paimon (#3902) and Iceberg (#3903), the Hudi lake storage has never been exposed in any publicly released Fluss version, so there are no existing "legacy" Hudi tables that carry the three system columns. This PR therefore switches Hudi directly to the clean layout only, without any clean/legacy layout detection or legacy read/write compatibility. This keeps the Hudi implementation considerably simpler than the Paimon and Iceberg ones, which must support both layouts.
Brief change log
HudiConversionsbuilds the Hudi/Flink schema from user columns only; the three system columns are no longer appended. The name-conflict check is kept: user columns may not be named__bucket,__offset, or__timestamp(these names remain reserved for Fluss internal use).FlussRecordAsHudiRowwrites the business columns only and validates that the Fluss record's field count matches the Hudi table field count. No system values are injected.HudiRecordReaderselects the Hudi metadata columns and the projected business columns only, and emits a sentinel-1Llog offset / timestamp, consistent with the Paimon and Iceberg readers and theLakeRecordRecordEmittercontract. Nested rows (HudiArrayAsFlussArray) and sorted primary-key reads (HudiSortedRecordReader) are aligned with the clean layout.__bucketcolumn, so it works for the clean layout without changes.Tests
HudiLakeCatalogTest: create-table assertions expect tmns only).FlussRecordAsHudiRowTest: verifies the clean writer odation.HudiRecordReaderTest: verifies the reader strips Hudis the-1Loffset/timestamp sentinel.HudiRowAsFlussRowTest: uses a clean row fixture.FlinkHudiTieringTestBase/HudiTieringITCase: read olumns only.fluss-lake-hudimodule test suite passes (90API and Format
Storage format: Hudi lake tables created by Fluss containo trailing
__bucket/__offset/__timestampsystemcolumns. No public Java API change.
Documentation
`website/docs/streaming-lakehouse/datalake-formats/hudi.mt the Hudi table schema contains only user columns, andthat the three system-column names remain reserved.