diff --git a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiArrayAsFlussArray.java b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiArrayAsFlussArray.java index 9e256995708..28eb10f40e7 100644 --- a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiArrayAsFlussArray.java +++ b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiArrayAsFlussArray.java @@ -149,7 +149,7 @@ public InternalMap getMap(int pos) { @Override public InternalRow getRow(int pos, int numFields) { RowData nestedRow = hudiArray.getRow(pos, numFields); - return nestedRow == null ? null : new HudiRowAsFlussRow(nestedRow, false); + return nestedRow == null ? null : new HudiRowAsFlussRow(nestedRow); } @Override diff --git a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRecordReader.java b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRecordReader.java index ea88aac94bc..ea05cca99c6 100644 --- a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRecordReader.java +++ b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRecordReader.java @@ -55,10 +55,7 @@ import java.util.stream.Collectors; import java.util.stream.IntStream; -import static org.apache.fluss.lake.hudi.HudiLakeCatalog.SYSTEM_COLUMNS; import static org.apache.fluss.lake.hudi.utils.HudiConversions.toChangeType; -import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME; -import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME; /** Record reader for Hudi tables. */ public class HudiRecordReader implements RecordReader { @@ -126,10 +123,7 @@ public HudiRecordReader( unifiedHudiTableReader.readFileSlice(fileSlice); this.iterator = new HudiRecordAsFlussRecordIterator( - hudiRecordIterator, - requiredSchema, - metadataFieldCount, - SYSTEM_COLUMNS.size()); + hudiRecordIterator, requiredSchema, metadataFieldCount); } } @@ -164,22 +158,16 @@ private static int[] selectedFields( return IntStream.range(0, schema.getFields().size()).toArray(); } + // FIP-27: Hudi lake tables contain only user columns, so the selected fields are the Hudi + // metadata columns followed by the projected business columns. int[] hudiMetadataFields = IntStream.range(0, metadataFieldCount).toArray(); int[] projectedDataFields = Arrays.stream(project) .filter(projectPath -> projectPath.length > 0) .mapToInt(projectPath -> projectPath[0] + metadataFieldCount) .toArray(); - int[] systemFields = - SYSTEM_COLUMNS.keySet().stream() - .mapToInt(systemColumn -> requiredFieldPosition(schema, systemColumn)) - .toArray(); - return IntStream.concat( - IntStream.concat( - IntStream.of(hudiMetadataFields), - IntStream.of(projectedDataFields)), - IntStream.of(systemFields)) + return IntStream.concat(IntStream.of(hudiMetadataFields), IntStream.of(projectedDataFields)) .toArray(); } @@ -216,42 +204,29 @@ private static DataType producedDataType(RowType rowType, int[] selectedFields) .bridgedTo(RowData.class); } - private static int requiredFieldPosition(Schema schema, String fieldName) { - Schema.Field field = schema.getField(fieldName); - if (field == null) { - throw new IllegalArgumentException( - String.format( - "Required Hudi system column '%s' does not exist in Hudi schema.", - fieldName)); - } - return field.pos(); - } - /** Iterator for Hudi {@link RowData} as Fluss {@link LogRecord}. */ public static class HudiRecordAsFlussRecordIterator implements CloseableIterator { + /** Sentinel offset / timestamp emitted for rows read from a lake table. */ + private static final long NO_SYSTEM_COLUMN_VALUE = -1L; + private final ClosableIterator hudiRecordIterator; private final ProjectedRow projectedRow; private final HudiRowAsFlussRow hudiRowAsFlussRow; - private final int logOffsetColIndex; - private final int timestampColIndex; private boolean closed; public HudiRecordAsFlussRecordIterator( ClosableIterator hudiRecordIterator, Schema schema, - int metadataFieldCount, - int systemFieldCount) { + int metadataFieldCount) { this.hudiRecordIterator = hudiRecordIterator; - this.logOffsetColIndex = requiredFieldPosition(schema, OFFSET_COLUMN_NAME); - this.timestampColIndex = requiredFieldPosition(schema, TIMESTAMP_COLUMN_NAME); this.hudiRowAsFlussRow = new HudiRowAsFlussRow(); + // 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()); } @@ -278,12 +253,12 @@ public boolean hasNext() { public LogRecord next() { RowData rowData = hudiRecordIterator.next(); ChangeType changeType = toChangeType(rowData.getRowKind()); - long offset = rowData.getLong(logOffsetColIndex); - long timestamp = rowData.getTimestamp(timestampColIndex, 6).getMillisecond(); - + // The lake table does not carry a per-record log offset / timestamp meaningful to + // downstream consumers, so a sentinel -1 is emitted, consistent with the Paimon and + // Iceberg readers and the LakeRecordRecordEmitter contract. return new GenericRecord( - offset, - timestamp, + NO_SYSTEM_COLUMN_VALUE, + NO_SYSTEM_COLUMN_VALUE, changeType, projectedRow.replaceRow(hudiRowAsFlussRow.replaceRow(rowData))); } diff --git a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRowAsFlussRow.java b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRowAsFlussRow.java index 059dd317f8e..9a0e04a2a44 100644 --- a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRowAsFlussRow.java +++ b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRowAsFlussRow.java @@ -31,29 +31,15 @@ import org.apache.flink.table.data.RowData; import org.apache.flink.table.data.TimestampData; -import static org.apache.fluss.lake.hudi.HudiLakeCatalog.SYSTEM_COLUMNS; - /** Wraps a Hudi/Flink {@link RowData} as a Fluss {@link InternalRow}. */ public class HudiRowAsFlussRow implements InternalRow { private RowData rowData; - private final boolean stripSystemColumns; - public HudiRowAsFlussRow() { - this(true); - } + 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; } public HudiRowAsFlussRow replaceRow(RowData rowData) { @@ -63,7 +49,9 @@ public HudiRowAsFlussRow replaceRow(RowData rowData) { @Override public int getFieldCount() { - return stripSystemColumns ? rowData.getArity() - SYSTEM_COLUMNS.size() : rowData.getArity(); + // FIP-27: Hudi lake tables contain only user columns; the arity is the business field + // count. + return rowData.getArity(); } @Override @@ -170,6 +158,6 @@ public InternalMap getMap(int pos) { @Override public InternalRow getRow(int pos, int numFields) { RowData value = rowData.getRow(pos, numFields); - return value == null ? null : new HudiRowAsFlussRow(value, false); + return value == null ? null : new HudiRowAsFlussRow(value); } } diff --git a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSortedRecordReader.java b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSortedRecordReader.java index efa3f1bc60f..a6e6781768e 100644 --- a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSortedRecordReader.java +++ b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSortedRecordReader.java @@ -54,7 +54,6 @@ import java.util.Map; import java.util.stream.Collectors; -import static org.apache.fluss.lake.hudi.HudiLakeCatalog.SYSTEM_COLUMNS; import static org.apache.fluss.utils.Preconditions.checkState; /** Sorted Hudi record reader for primary key table union read. */ @@ -167,9 +166,6 @@ private static RecordKeyInfo resolveRecordKeyInfo(HudiTableInfo hudiTableInfo) { int userFieldPosition = 0; for (Schema.UnresolvedColumn column : hudiTableInfo.getHudiTable().getUnresolvedSchema().getColumns()) { - if (SYSTEM_COLUMNS.containsKey(column.getName())) { - continue; - } DataType dataType = getDataType(column); dataTypesByName.put(column.getName(), dataType); userFieldPositionsByName.put(column.getName(), userFieldPosition++); diff --git a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/RecordWriter.java b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/RecordWriter.java index c730ae9376c..4be86027a39 100644 --- a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/RecordWriter.java +++ b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/RecordWriter.java @@ -61,8 +61,7 @@ public RecordWriter( HudiWriteTableInfo hudiTableInfo, CkpMetadata ckpMetadata) { this.bucketNum = writerInitContext.tableBucket().getBucket(); - this.flussRecordAsHudiRecord = - new FlussRecordAsHudiRow(bucketNum, hudiTableInfo.getRowType()); + this.flussRecordAsHudiRecord = new FlussRecordAsHudiRow(hudiTableInfo.getRowType()); this.hudiTableInfo = hudiTableInfo; this.ckpMetadata = ckpMetadata; this.recordWriteBuffer = diff --git a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRow.java b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRow.java index f62adc76adf..97fd5dc2d64 100644 --- a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRow.java +++ b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRow.java @@ -19,72 +19,37 @@ import org.apache.fluss.record.LogRecord; -import org.apache.flink.table.data.TimestampData; import org.apache.flink.table.types.logical.RowType; import org.apache.flink.types.RowKind; -import static org.apache.fluss.lake.hudi.HudiLakeCatalog.SYSTEM_COLUMNS; import static org.apache.fluss.lake.hudi.utils.HudiConversions.toRowKind; import static org.apache.fluss.utils.Preconditions.checkState; -/** Wraps a Fluss {@link LogRecord} as a Hudi/Flink row with Fluss system columns. */ +/** Wraps a Fluss {@link LogRecord} as a Hudi/Flink row. */ public class FlussRecordAsHudiRow extends FlussRowAsHudiRow { - private final int bucket; + // FIP-27: Hudi lake tables contain only user columns; no Fluss system columns are written. + private final int fieldCount; private LogRecord logRecord; - private int originRowFieldCount; - public FlussRecordAsHudiRow(int bucket, RowType rowType) { + public FlussRecordAsHudiRow(RowType rowType) { super(rowType); - this.bucket = bucket; + this.fieldCount = rowType.getFieldCount(); } 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); } @Override public RowKind getRowKind() { return toRowKind(logRecord.getChangeType()); } - - @Override - public boolean isNullAt(int pos) { - if (pos < originRowFieldCount) { - return super.isNullAt(pos); - } - return false; - } - - @Override - public int getInt(int pos) { - if (pos == originRowFieldCount) { - return bucket; - } - return super.getInt(pos); - } - - @Override - public long getLong(int pos) { - if (pos == originRowFieldCount + 1) { - return logRecord.logOffset(); - } else if (pos == originRowFieldCount + 2) { - return logRecord.timestamp(); - } - return super.getLong(pos); - } - - @Override - public TimestampData getTimestamp(int pos, int precision) { - if (pos == originRowFieldCount + 2) { - return TimestampData.fromEpochMillis(logRecord.timestamp()); - } - return super.getTimestamp(pos, precision); - } } diff --git a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiConversions.java b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiConversions.java index 84d5416dc60..72f19a3b4de 100644 --- a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiConversions.java +++ b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiConversions.java @@ -33,7 +33,6 @@ import org.apache.flink.table.catalog.ResolvedSchema; import org.apache.flink.table.catalog.UniqueConstraint; import org.apache.flink.table.factories.FactoryUtil; -import org.apache.flink.table.types.DataType; import org.apache.flink.types.RowKind; import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.configuration.FlinkOptions; @@ -112,7 +111,8 @@ public static ResolvedSchema convertToFlinkResolvedSchema( List columns = new ArrayList<>(); - // Add regular columns + // FIP-27: Hudi lake tables contain only user columns; the Fluss system columns + // (__bucket/__offset/__timestamp) are not written to the physical schema. for (org.apache.fluss.metadata.Schema.Column column : tableDescriptor.getSchema().getColumns()) { String columnName = column.getName(); @@ -133,11 +133,6 @@ public static ResolvedSchema convertToFlinkResolvedSchema( columns.add(Column.physical(columnName, column.getDataType().accept(converter))); } - // add system metadata columns to schema - for (Map.Entry systemColumn : SYSTEM_COLUMNS.entrySet()) { - columns.add(Column.physical(systemColumn.getKey(), systemColumn.getValue())); - } - UniqueConstraint constraint = null; // Set primary key if this is a PK table if (isPkTable && tableDescriptor.hasPrimaryKey()) { diff --git a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/HudiLakeCatalogTest.java b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/HudiLakeCatalogTest.java index 8acb742b034..05dec7b896b 100644 --- a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/HudiLakeCatalogTest.java +++ b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/HudiLakeCatalogTest.java @@ -562,12 +562,10 @@ void testHudiMetadataColumnPrefixConflictThrowsException() { private org.apache.flink.table.api.Schema buildExpectedHudiSchema( DataType idType, String primaryKeyName) { + // FIP-27: newly created tables are clean and carry only user columns. return org.apache.flink.table.api.Schema.newBuilder() .column("id", idType) .column("name", org.apache.flink.table.api.DataTypes.STRING()) - .column("__bucket", org.apache.flink.table.api.DataTypes.INT()) - .column("__offset", org.apache.flink.table.api.DataTypes.BIGINT()) - .column("__timestamp", org.apache.flink.table.api.DataTypes.TIMESTAMP(6)) .primaryKeyNamed(primaryKeyName, "id") .build(); } diff --git a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiRecordReaderTest.java b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiRecordReaderTest.java index 64f802a9afb..6a2f699916e 100644 --- a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiRecordReaderTest.java +++ b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiRecordReaderTest.java @@ -24,7 +24,6 @@ import org.apache.flink.table.data.GenericRowData; import org.apache.flink.table.data.RowData; import org.apache.flink.table.data.StringData; -import org.apache.flink.table.data.TimestampData; import org.apache.flink.types.RowKind; import org.apache.hudi.common.util.collection.ClosableIterator; import org.apache.hudi.org.apache.avro.Schema; @@ -35,26 +34,25 @@ import java.util.NoSuchElementException; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test for {@link HudiRecordReader}. */ class HudiRecordReaderTest { @Test void testIteratorConvertsHudiRowDataToFlussLogRecord() { + // FIP-27: a Hudi lake table has only Hudi metadata columns followed by user columns; the + // reader strips the metadata columns and emits a sentinel -1 offset / timestamp. TestingClosableIterator hudiIterator = - new TestingClosableIterator( - rowData(RowKind.UPDATE_AFTER, 11, "value", 3, 42L, 1234L)); + new TestingClosableIterator(rowData(RowKind.UPDATE_AFTER, 11, "value")); HudiRecordReader.HudiRecordAsFlussRecordIterator iterator = - new HudiRecordReader.HudiRecordAsFlussRecordIterator( - hudiIterator, fullSchema(), 2, 3); + new HudiRecordReader.HudiRecordAsFlussRecordIterator(hudiIterator, fullSchema(), 2); assertThat(iterator.hasNext()).isTrue(); LogRecord logRecord = iterator.next(); assertThat(logRecord.getChangeType()).isEqualTo(ChangeType.UPDATE_AFTER); - assertThat(logRecord.logOffset()).isEqualTo(42L); - assertThat(logRecord.timestamp()).isEqualTo(1234L); + assertThat(logRecord.logOffset()).isEqualTo(-1L); + assertThat(logRecord.timestamp()).isEqualTo(-1L); assertThat(logRecord.getRow().getFieldCount()).isEqualTo(2); assertThat(logRecord.getRow().getInt(0)).isEqualTo(11); assertThat(logRecord.getRow().getString(1)).isEqualTo(BinaryString.fromString("value")); @@ -66,19 +64,18 @@ void testIteratorConvertsHudiRowDataToFlussLogRecord() { } @Test - void testIteratorSkipsMetadataAndAllSystemColumnsForProjectedSchema() { + void testIteratorSkipsMetadataForProjectedSchema() { TestingClosableIterator hudiIterator = - new TestingClosableIterator( - projectedRowData(RowKind.DELETE, "projected", 5, 7L, 999L)); + new TestingClosableIterator(projectedRowData(RowKind.DELETE, "projected")); HudiRecordReader.HudiRecordAsFlussRecordIterator iterator = new HudiRecordReader.HudiRecordAsFlussRecordIterator( - hudiIterator, projectedSchema(), 2, 3); + hudiIterator, projectedSchema(), 2); LogRecord logRecord = iterator.next(); assertThat(logRecord.getChangeType()).isEqualTo(ChangeType.DELETE); - assertThat(logRecord.logOffset()).isEqualTo(7L); - assertThat(logRecord.timestamp()).isEqualTo(999L); + assertThat(logRecord.logOffset()).isEqualTo(-1L); + assertThat(logRecord.timestamp()).isEqualTo(-1L); assertThat(logRecord.getRow().getFieldCount()).isEqualTo(1); assertThat(logRecord.getRow().getString(0)).isEqualTo(BinaryString.fromString("projected")); @@ -86,45 +83,22 @@ void testIteratorSkipsMetadataAndAllSystemColumnsForProjectedSchema() { assertThat(hudiIterator.getCloseCount()).isEqualTo(1); } - @Test - void testIteratorFailsClearlyWhenRequiredSystemColumnIsMissing() { - TestingClosableIterator hudiIterator = - new TestingClosableIterator( - projectedRowData(RowKind.DELETE, "projected", 5, 7L, 999L)); - - assertThatThrownBy( - () -> - new HudiRecordReader.HudiRecordAsFlussRecordIterator( - hudiIterator, schemaWithoutOffsetColumn(), 2, 2)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("__offset") - .hasMessageContaining("does not exist"); - } - - private static RowData rowData( - RowKind rowKind, int id, String value, int bucket, long offset, long timestamp) { - GenericRowData rowData = new GenericRowData(7); + private static RowData rowData(RowKind rowKind, int id, String value) { + GenericRowData rowData = new GenericRowData(4); rowData.setRowKind(rowKind); rowData.setField(0, StringData.fromString("commit")); rowData.setField(1, StringData.fromString("record-key")); rowData.setField(2, id); rowData.setField(3, StringData.fromString(value)); - rowData.setField(4, bucket); - rowData.setField(5, offset); - rowData.setField(6, TimestampData.fromEpochMillis(timestamp)); return rowData; } - private static RowData projectedRowData( - RowKind rowKind, String value, int bucket, long offset, long timestamp) { - GenericRowData rowData = new GenericRowData(6); + private static RowData projectedRowData(RowKind rowKind, String value) { + GenericRowData rowData = new GenericRowData(3); rowData.setRowKind(rowKind); rowData.setField(0, StringData.fromString("commit")); rowData.setField(1, StringData.fromString("record-key")); rowData.setField(2, StringData.fromString(value)); - rowData.setField(3, bucket); - rowData.setField(4, offset); - rowData.setField(5, TimestampData.fromEpochMillis(timestamp)); return rowData; } @@ -134,10 +108,7 @@ private static Schema fullSchema() { + "{\"name\":\"_hoodie_commit_time\",\"type\":\"string\"}," + "{\"name\":\"_hoodie_record_key\",\"type\":\"string\"}," + "{\"name\":\"id\",\"type\":\"int\"}," - + "{\"name\":\"value\",\"type\":\"string\"}," - + "{\"name\":\"__bucket\",\"type\":\"int\"}," - + "{\"name\":\"__offset\",\"type\":\"long\"}," - + "{\"name\":\"__timestamp\",\"type\":\"long\"}" + + "{\"name\":\"value\",\"type\":\"string\"}" + "]"); } @@ -146,21 +117,7 @@ private static Schema projectedSchema() { "[" + "{\"name\":\"_hoodie_commit_time\",\"type\":\"string\"}," + "{\"name\":\"_hoodie_record_key\",\"type\":\"string\"}," - + "{\"name\":\"value\",\"type\":\"string\"}," - + "{\"name\":\"__bucket\",\"type\":\"int\"}," - + "{\"name\":\"__offset\",\"type\":\"long\"}," - + "{\"name\":\"__timestamp\",\"type\":\"long\"}" - + "]"); - } - - private static Schema schemaWithoutOffsetColumn() { - return parseSchema( - "[" - + "{\"name\":\"_hoodie_commit_time\",\"type\":\"string\"}," - + "{\"name\":\"_hoodie_record_key\",\"type\":\"string\"}," - + "{\"name\":\"value\",\"type\":\"string\"}," - + "{\"name\":\"__bucket\",\"type\":\"int\"}," - + "{\"name\":\"__timestamp\",\"type\":\"long\"}" + + "{\"name\":\"value\",\"type\":\"string\"}" + "]"); } diff --git a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiRowAsFlussRowTest.java b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiRowAsFlussRowTest.java index 814c896515e..5b556bc80f7 100644 --- a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiRowAsFlussRowTest.java +++ b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiRowAsFlussRowTest.java @@ -46,7 +46,8 @@ void testAccessPrimitiveAndNestedValues() { GenericRowData nestedRow = new GenericRowData(1); nestedRow.setField(0, StringData.fromString("nested")); - GenericRowData rowData = new GenericRowData(19); + // FIP-27: a clean Hudi row has only user columns (no trailing system columns). + GenericRowData rowData = new GenericRowData(16); rowData.setField(0, true); rowData.setField(1, (byte) 1); rowData.setField(2, (short) 2); @@ -67,9 +68,6 @@ void testAccessPrimitiveAndNestedValues() { Collections.singletonMap( StringData.fromString("key"), StringData.fromString("value")))); rowData.setField(15, nestedRow); - rowData.setField(16, 0); - rowData.setField(17, 1L); - rowData.setField(18, TimestampData.fromEpochMillis(3000L)); HudiRowAsFlussRow row = new HudiRowAsFlussRow(rowData); diff --git a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/testutils/FlinkHudiTieringTestBase.java b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/testutils/FlinkHudiTieringTestBase.java index 40de1ed424c..edb6dc3aef4 100644 --- a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/testutils/FlinkHudiTieringTestBase.java +++ b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/testutils/FlinkHudiTieringTestBase.java @@ -73,7 +73,6 @@ import java.util.Arrays; import java.util.HashMap; import java.util.HashSet; -import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Optional; @@ -443,22 +442,13 @@ record -> } protected void checkDataInHudiCOWTable( - TablePath tablePath, - String partition, - List expectedRows, - long startingOffset, - int bucket) + TablePath tablePath, String partition, List expectedRows, int bucket) throws Exception { + // FIP-27: a clean table carries only user columns, so compare business columns only. The + // physical __offset system column is no longer present on clean tables. List expectedRecords = new ArrayList<>(); - Iterator flussRowIterator = expectedRows.iterator(); - while (flussRowIterator.hasNext()) { - InternalRow flussRow = flussRowIterator.next(); - expectedRecords.add( - flussRow.getInt(0) - + "," - + flussRow.getString(1).toString() - + "," - + startingOffset++); + for (InternalRow flussRow : expectedRows) { + expectedRecords.add(flussRow.getInt(0) + "," + flussRow.getString(1).toString()); } List actualRecords = @@ -466,15 +456,9 @@ protected void checkDataInHudiCOWTable( tablePath, partition, bucket, - record -> - record.getInt(5) - + "," - + record.getString(6).toString() - + "," - + record.getLong(8)); + record -> record.getInt(5) + "," + record.getString(6).toString()); assertThat(actualRecords).containsExactlyInAnyOrderElementsOf(expectedRecords); - assertThat(flussRowIterator.hasNext()).isFalse(); } protected void checkFlussOffsetsInSnapshot( diff --git a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiTieringITCase.java b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiTieringITCase.java index 707fb6d3932..895a81a4b8b 100644 --- a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiTieringITCase.java +++ b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiTieringITCase.java @@ -184,7 +184,7 @@ private void testLogTableTiering() throws Exception { assertReplicaStatus(logTableBucket, 30); assertThat(getLeaderReplica(logTableBucket).getLogTablet().getLakeMaxTimestamp()) .isGreaterThan(-1); - checkDataInHudiCOWTable(logTablePath, "", flussRows, 0, 0); + checkDataInHudiCOWTable(logTablePath, "", flussRows, 0); checkFlussOffsetsInSnapshot(logTablePath, Collections.singletonMap(logTableBucket, 30L)); } diff --git a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRowTest.java b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRowTest.java index 1e291fc1305..ce98baf0110 100644 --- a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRowTest.java +++ b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRowTest.java @@ -49,7 +49,9 @@ class FlussRecordAsHudiRowTest { @Test - void testLogRecordFieldsAndSystemColumns() { + void testLogRecordFields() { + // FIP-27: Hudi lake tables contain only user columns; the row type is the business columns + // and no system columns are written. RowType rowType = RowType.of( new BooleanType(), @@ -59,10 +61,7 @@ void testLogRecordFieldsAndSystemColumns() { new DecimalType(10, 2), new LocalZonedTimestampType(6), new TimestampType(6), - new BinaryType(), - new IntType(), - new BigIntType(), - new TimestampType(6)); + new BinaryType()); GenericRow row = new GenericRow(8); row.setField(0, true); row.setField(1, 1); @@ -73,11 +72,8 @@ void testLogRecordFieldsAndSystemColumns() { row.setField(6, TimestampNtz.fromMillis(1698235273182L, 5678)); row.setField(7, new byte[] {1, 2, 3}); - int bucket = 3; - long offset = 11L; - long timestamp = 1698235273999L; - LogRecord logRecord = new GenericRecord(offset, timestamp, APPEND_ONLY, row); - FlussRecordAsHudiRow hudiRow = new FlussRecordAsHudiRow(bucket, rowType); + LogRecord logRecord = new GenericRecord(11L, 1698235273999L, APPEND_ONLY, row); + FlussRecordAsHudiRow hudiRow = new FlussRecordAsHudiRow(rowType); hudiRow.setFlussRecord(logRecord); assertThat(hudiRow.getBoolean(0)).isTrue(); @@ -89,23 +85,17 @@ void testLogRecordFieldsAndSystemColumns() { assertThat(hudiRow.getTimestamp(5, 6).getNanoOfMillisecond()).isEqualTo(5678); assertThat(hudiRow.getTimestamp(6, 6).getMillisecond()).isEqualTo(1698235273182L); assertThat(hudiRow.getBinary(7)).containsExactly(1, 2, 3); - - assertThat(hudiRow.getInt(8)).isEqualTo(bucket); - assertThat(hudiRow.getLong(9)).isEqualTo(offset); - assertThat(hudiRow.getLong(10)).isEqualTo(timestamp); - assertThat(hudiRow.getTimestamp(10, 6).getMillisecond()).isEqualTo(timestamp); - assertThat(hudiRow.getArity()).isEqualTo(11); + // clean tables expose only the business columns; no system columns are appended + assertThat(hudiRow.getArity()).isEqualTo(8); assertThat(hudiRow.getRowKind()).isEqualTo(RowKind.INSERT); } @Test void testChangeTypeToRowKind() { - RowType rowType = - RowType.of( - new BooleanType(), new IntType(), new BigIntType(), new TimestampType(6)); + RowType rowType = RowType.of(new BooleanType()); GenericRow row = new GenericRow(1); row.setField(0, true); - FlussRecordAsHudiRow hudiRow = new FlussRecordAsHudiRow(0, rowType); + FlussRecordAsHudiRow hudiRow = new FlussRecordAsHudiRow(rowType); hudiRow.setFlussRecord(new GenericRecord(0, 1, UPDATE_BEFORE, row)); assertThat(hudiRow.getRowKind()).isEqualTo(RowKind.UPDATE_BEFORE); diff --git a/website/docs/streaming-lakehouse/datalake-formats/hudi.md b/website/docs/streaming-lakehouse/datalake-formats/hudi.md index 431b99fa3af..7292f82f946 100644 --- a/website/docs/streaming-lakehouse/datalake-formats/hudi.md +++ b/website/docs/streaming-lakehouse/datalake-formats/hudi.md @@ -133,15 +133,9 @@ When a Fluss table is created with the option `'table.datalake.enabled' = 'true' For DFS catalog mode, the Hudi table path is `${catalog.path}/${database_name}/${table_name}` unless the Hudi table path is explicitly set by Hudi options. For Hive Metastore catalog mode, the table path follows Hudi Hive catalog path inference. -The schema of the Hudi table matches the Fluss table schema, except for three system columns appended by Fluss: +The schema of the Hudi table matches the Fluss table schema, containing only the user-defined columns. Fluss does not add any system columns to the physical Hudi schema. -| Column | Type | Description | -|---------------|--------------|-----------------------------------------------| -| `__bucket` | INT | Fluss bucket identifier for data distribution | -| `__offset` | BIGINT | Fluss log offset for ordering and seeking | -| `__timestamp` | TIMESTAMP(6) | Fluss log timestamp | - -Do not use user columns named `__bucket`, `__offset`, or `__timestamp`. Hudi metadata column names starting with `_hoodie_` are also reserved. +The names `__bucket`, `__offset`, and `__timestamp` are reserved for Fluss internal use, so do not use user columns with these names. Hudi metadata column names starting with `_hoodie_` are also reserved. ### Primary Key Tables