This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new aa7101080 [lake/hudi] Introduce Hudi source reader (#3499)
aa7101080 is described below
commit aa7101080a474e5688bd2696e252c7cc9cbcebe7
Author: fhan <[email protected]>
AuthorDate: Sat Jun 20 10:16:07 2026 +0800
[lake/hudi] Introduce Hudi source reader (#3499)
---
.../lake/hudi/source/HudiArrayAsFlussArray.java | 224 +++++++++++++++
.../fluss/lake/hudi/source/HudiLakeSource.java | 12 +-
.../fluss/lake/hudi/source/HudiMapAsFlussMap.java | 48 ++++
.../fluss/lake/hudi/source/HudiRecordReader.java | 291 +++++++++++++++++++
.../fluss/lake/hudi/source/HudiRowAsFlussRow.java | 175 ++++++++++++
.../lake/hudi/source/UnifiedHudiTableReader.java | 310 +++++++++++++++++++++
.../fluss/lake/hudi/utils/HudiConversions.java | 17 ++
.../fluss/lake/hudi/source/HudiLakeSourceTest.java | 7 +-
.../lake/hudi/source/HudiRecordReaderTest.java | 208 ++++++++++++++
.../lake/hudi/source/HudiRowAsFlussRowTest.java | 104 +++++++
10 files changed, 1390 insertions(+), 6 deletions(-)
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
new file mode 100644
index 000000000..9e2569957
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiArrayAsFlussArray.java
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.hudi.source;
+
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.Decimal;
+import org.apache.fluss.row.InternalArray;
+import org.apache.fluss.row.InternalMap;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+
+/** Wraps a Hudi/Flink {@link ArrayData} as a Fluss {@link InternalArray}. */
+public class HudiArrayAsFlussArray implements InternalArray {
+
+ private final ArrayData hudiArray;
+
+ public HudiArrayAsFlussArray(ArrayData hudiArray) {
+ this.hudiArray = hudiArray;
+ }
+
+ @Override
+ public int size() {
+ return hudiArray.size();
+ }
+
+ @Override
+ public boolean isNullAt(int pos) {
+ return hudiArray.isNullAt(pos);
+ }
+
+ @Override
+ public boolean getBoolean(int pos) {
+ return hudiArray.getBoolean(pos);
+ }
+
+ @Override
+ public byte getByte(int pos) {
+ return hudiArray.getByte(pos);
+ }
+
+ @Override
+ public short getShort(int pos) {
+ return hudiArray.getShort(pos);
+ }
+
+ @Override
+ public int getInt(int pos) {
+ return hudiArray.getInt(pos);
+ }
+
+ @Override
+ public long getLong(int pos) {
+ return hudiArray.getLong(pos);
+ }
+
+ @Override
+ public float getFloat(int pos) {
+ return hudiArray.getFloat(pos);
+ }
+
+ @Override
+ public double getDouble(int pos) {
+ return hudiArray.getDouble(pos);
+ }
+
+ @Override
+ public BinaryString getChar(int pos, int length) {
+ return BinaryString.fromBytes(hudiArray.getString(pos).toBytes());
+ }
+
+ @Override
+ public BinaryString getString(int pos) {
+ return BinaryString.fromBytes(hudiArray.getString(pos).toBytes());
+ }
+
+ @Override
+ public Decimal getDecimal(int pos, int precision, int scale) {
+ DecimalData decimalData = hudiArray.getDecimal(pos, precision, scale);
+ if (decimalData.isCompact()) {
+ return Decimal.fromUnscaledLong(decimalData.toUnscaledLong(),
precision, scale);
+ }
+ return Decimal.fromBigDecimal(decimalData.toBigDecimal(), precision,
scale);
+ }
+
+ @Override
+ public TimestampNtz getTimestampNtz(int pos, int precision) {
+ TimestampData timestamp = hudiArray.getTimestamp(pos, precision);
+ if (TimestampNtz.isCompact(precision)) {
+ return TimestampNtz.fromMillis(timestamp.getMillisecond());
+ }
+ return TimestampNtz.fromMillis(
+ timestamp.getMillisecond(), timestamp.getNanoOfMillisecond());
+ }
+
+ @Override
+ public TimestampLtz getTimestampLtz(int pos, int precision) {
+ TimestampData timestamp = hudiArray.getTimestamp(pos, precision);
+ if (TimestampLtz.isCompact(precision)) {
+ return TimestampLtz.fromEpochMillis(timestamp.getMillisecond());
+ }
+ return TimestampLtz.fromEpochMillis(
+ timestamp.getMillisecond(), timestamp.getNanoOfMillisecond());
+ }
+
+ @Override
+ public byte[] getBinary(int pos, int length) {
+ return hudiArray.getBinary(pos);
+ }
+
+ @Override
+ public byte[] getBytes(int pos) {
+ return hudiArray.getBinary(pos);
+ }
+
+ @Override
+ public InternalArray getArray(int pos) {
+ ArrayData nestedArray = hudiArray.getArray(pos);
+ return nestedArray == null ? null : new
HudiArrayAsFlussArray(nestedArray);
+ }
+
+ @Override
+ public InternalMap getMap(int pos) {
+ MapData nestedMap = hudiArray.getMap(pos);
+ return nestedMap == null ? null : new HudiMapAsFlussMap(nestedMap);
+ }
+
+ @Override
+ public InternalRow getRow(int pos, int numFields) {
+ RowData nestedRow = hudiArray.getRow(pos, numFields);
+ return nestedRow == null ? null : new HudiRowAsFlussRow(nestedRow,
false);
+ }
+
+ @Override
+ public boolean[] toBooleanArray() {
+ int arraySize = hudiArray.size();
+ boolean[] result = new boolean[arraySize];
+ for (int i = 0; i < arraySize; i++) {
+ result[i] = hudiArray.getBoolean(i);
+ }
+ return result;
+ }
+
+ @Override
+ public byte[] toByteArray() {
+ int arraySize = hudiArray.size();
+ byte[] result = new byte[arraySize];
+ for (int i = 0; i < arraySize; i++) {
+ result[i] = hudiArray.getByte(i);
+ }
+ return result;
+ }
+
+ @Override
+ public short[] toShortArray() {
+ int arraySize = hudiArray.size();
+ short[] result = new short[arraySize];
+ for (int i = 0; i < arraySize; i++) {
+ result[i] = hudiArray.getShort(i);
+ }
+ return result;
+ }
+
+ @Override
+ public int[] toIntArray() {
+ int arraySize = hudiArray.size();
+ int[] result = new int[arraySize];
+ for (int i = 0; i < arraySize; i++) {
+ result[i] = hudiArray.getInt(i);
+ }
+ return result;
+ }
+
+ @Override
+ public long[] toLongArray() {
+ int arraySize = hudiArray.size();
+ long[] result = new long[arraySize];
+ for (int i = 0; i < arraySize; i++) {
+ result[i] = hudiArray.getLong(i);
+ }
+ return result;
+ }
+
+ @Override
+ public float[] toFloatArray() {
+ int arraySize = hudiArray.size();
+ float[] result = new float[arraySize];
+ for (int i = 0; i < arraySize; i++) {
+ result[i] = hudiArray.getFloat(i);
+ }
+ return result;
+ }
+
+ @Override
+ public double[] toDoubleArray() {
+ int arraySize = hudiArray.size();
+ double[] result = new double[arraySize];
+ for (int i = 0; i < arraySize; i++) {
+ result[i] = hudiArray.getDouble(i);
+ }
+ return result;
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiLakeSource.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiLakeSource.java
index c491ddfa6..87f9ee0de 100644
---
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiLakeSource.java
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiLakeSource.java
@@ -25,6 +25,8 @@ import org.apache.fluss.lake.source.RecordReader;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.predicate.Predicate;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
@@ -37,6 +39,7 @@ public class HudiLakeSource implements LakeSource<HudiSplit> {
private final Configuration hudiConfig;
private final TablePath tablePath;
+ private @Nullable int[][] project;
public HudiLakeSource(Configuration hudiConfig, TablePath tablePath) {
this.hudiConfig = hudiConfig;
@@ -45,7 +48,7 @@ public class HudiLakeSource implements LakeSource<HudiSplit> {
@Override
public void withProject(int[][] project) {
- // Projection is applied by the Hudi record reader, which is not
implemented yet.
+ this.project = project;
}
@Override
@@ -65,8 +68,11 @@ public class HudiLakeSource implements LakeSource<HudiSplit>
{
@Override
public RecordReader createRecordReader(ReaderContext<HudiSplit> context)
throws IOException {
- throw new UnsupportedOperationException(
- "Hudi lake source does not support record reading yet.");
+ try {
+ return new HudiRecordReader(hudiConfig, tablePath,
context.lakeSplit(), project);
+ } catch (Exception e) {
+ throw new IOException("Fail to create Hudi record reader for " +
tablePath + ".", e);
+ }
}
@Override
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiMapAsFlussMap.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiMapAsFlussMap.java
new file mode 100644
index 000000000..3ada5e5cf
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiMapAsFlussMap.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.hudi.source;
+
+import org.apache.fluss.row.InternalArray;
+import org.apache.fluss.row.InternalMap;
+
+import org.apache.flink.table.data.MapData;
+
+/** Wraps a Hudi/Flink {@link MapData} as a Fluss {@link InternalMap}. */
+public class HudiMapAsFlussMap implements InternalMap {
+
+ private final MapData hudiMap;
+
+ public HudiMapAsFlussMap(MapData hudiMap) {
+ this.hudiMap = hudiMap;
+ }
+
+ @Override
+ public int size() {
+ return hudiMap.size();
+ }
+
+ @Override
+ public InternalArray keyArray() {
+ return new HudiArrayAsFlussArray(hudiMap.keyArray());
+ }
+
+ @Override
+ public InternalArray valueArray() {
+ return new HudiArrayAsFlussArray(hudiMap.valueArray());
+ }
+}
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
new file mode 100644
index 000000000..9cd52a15b
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRecordReader.java
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.hudi.source;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.lake.hudi.utils.HudiTableInfo;
+import org.apache.fluss.lake.source.RecordReader;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.ChangeType;
+import org.apache.fluss.record.GenericRecord;
+import org.apache.fluss.record.LogRecord;
+import org.apache.fluss.row.ProjectedRow;
+import org.apache.fluss.utils.CloseableIterator;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.hudi.avro.AvroSchemaCache;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.configuration.HadoopConfigurations;
+import org.apache.hudi.org.apache.avro.Schema;
+import org.apache.hudi.source.ExpressionPredicates;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+import org.apache.hudi.table.format.InternalSchemaManager;
+import org.apache.hudi.util.AvroSchemaConverter;
+import org.apache.hudi.util.StreamerUtil;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+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 {
+
+ private static final String HUDI_METADATA_COLUMN_PREFIX = "_hoodie_";
+
+ private final HudiRecordAsFlussRecordIterator iterator;
+
+ public HudiRecordReader(
+ Configuration hudiConfig,
+ TablePath tablePath,
+ HudiSplit hudiSplit,
+ @Nullable int[][] project)
+ throws Exception {
+ this(hudiConfig, tablePath, hudiSplit, project,
java.util.Collections.emptyList());
+ }
+
+ public HudiRecordReader(
+ Configuration hudiConfig,
+ TablePath tablePath,
+ HudiSplit hudiSplit,
+ @Nullable int[][] project,
+ List<ExpressionPredicates.Predicate> predicates)
+ throws Exception {
+ FileSlice fileSlice = hudiSplit.getFileSlice();
+ try (HudiTableInfo hudiTableInfo = HudiTableInfo.create(tablePath,
hudiConfig)) {
+ Schema avroSchema =
+
StreamerUtil.getTableAvroSchema(hudiTableInfo.getMetaClient(), true);
+ org.apache.flink.configuration.Configuration flinkHudiOptions =
+ buildFlinkHudiOptions(hudiTableInfo, tablePath,
avroSchema);
+
+ int metadataFieldCount = metadataFieldCount(avroSchema);
+ int[] selectedFields = selectedFields(avroSchema, project,
metadataFieldCount);
+ Schema requiredSchema = createRequiredSchema(avroSchema,
selectedFields);
+
+ InternalSchemaManager internalSchemaManager =
+ InternalSchemaManager.get(
+ new HadoopStorageConfiguration(
+
HadoopConfigurations.getHadoopConf(flinkHudiOptions)),
+ hudiTableInfo.getMetaClient());
+
+ boolean emitDelete =
+ hudiTableInfo.getTableType() ==
HoodieTableType.MERGE_ON_READ
+ && (flinkHudiOptions
+ .getString(FlinkOptions.QUERY_TYPE)
+
.equals(FlinkOptions.QUERY_TYPE_SNAPSHOT)
+ || flinkHudiOptions
+ .getString(FlinkOptions.QUERY_TYPE)
+
.equals(FlinkOptions.QUERY_TYPE_INCREMENTAL));
+
+ UnifiedHudiTableReader unifiedHudiTableReader =
+ UnifiedHudiTableReader.newBuilder()
+ .withMetaClient(hudiTableInfo.getMetaClient())
+ .withInternalSchemaManager(internalSchemaManager)
+ .withProps(flinkHudiOptions)
+ .withTableSchema(avroSchema)
+ .withRequiredSchema(requiredSchema)
+ .withSelectedFields(selectedFields)
+ .withPredicates(predicates)
+ .withEmitDelete(emitDelete)
+
.withLatestCommitTime(fileSlice.getLatestInstantTime())
+ .build();
+
+ ClosableIterator<RowData> hudiRecordIterator =
+ unifiedHudiTableReader.readFileSlice(fileSlice);
+ this.iterator =
+ new HudiRecordAsFlussRecordIterator(
+ hudiRecordIterator,
+ requiredSchema,
+ metadataFieldCount,
+ SYSTEM_COLUMNS.size());
+ }
+ }
+
+ @Override
+ public CloseableIterator<LogRecord> read() throws IOException {
+ return iterator;
+ }
+
+ private static org.apache.flink.configuration.Configuration
buildFlinkHudiOptions(
+ HudiTableInfo hudiTableInfo, TablePath tablePath, Schema
avroSchema) {
+ Map<String, String> hudiOptions = new
HashMap<>(hudiTableInfo.getTableOptions());
+ hudiOptions.put(FlinkOptions.PATH.key(), hudiTableInfo.getBasePath());
+ hudiOptions.put(FlinkOptions.TABLE_NAME.key(),
tablePath.getTableName());
+ hudiOptions.put(FlinkOptions.SOURCE_AVRO_SCHEMA.key(),
avroSchema.toString());
+ return
org.apache.flink.configuration.Configuration.fromMap(hudiOptions);
+ }
+
+ private static int metadataFieldCount(Schema schema) {
+ int metadataFieldCount = 0;
+ for (Schema.Field field : schema.getFields()) {
+ if (!field.name().startsWith(HUDI_METADATA_COLUMN_PREFIX)) {
+ break;
+ }
+ metadataFieldCount++;
+ }
+ return metadataFieldCount;
+ }
+
+ private static int[] selectedFields(
+ Schema schema, @Nullable int[][] project, int metadataFieldCount) {
+ if (project == null) {
+ return IntStream.range(0, schema.getFields().size()).toArray();
+ }
+
+ 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))
+ .toArray();
+ }
+
+ private static Schema createRequiredSchema(Schema avroSchema, int[]
selectedFields) {
+ DataType rowDataType =
AvroSchemaConverter.convertToDataType(avroSchema);
+ RowType rowType = (RowType) rowDataType.getLogicalType();
+ return AvroSchemaCache.intern(
+ new Schema.Parser()
+ .parse(
+ AvroSchemaConverter.convertToSchema(
+ producedDataType(rowType,
selectedFields)
+ .notNull()
+ .getLogicalType())
+ .toString()));
+ }
+
+ private static DataType producedDataType(RowType rowType, int[]
selectedFields) {
+ List<String> fieldNames = rowType.getFieldNames();
+ List<LogicalType> fieldLogicalTypes =
+ rowType.getFields().stream()
+ .map(RowType.RowField::getType)
+ .collect(Collectors.toList());
+ List<DataType> fieldDataTypes =
+
fieldLogicalTypes.stream().map(DataTypes::of).collect(Collectors.toList());
+
+ return DataTypes.ROW(
+ Arrays.stream(selectedFields)
+ .mapToObj(
+ pos ->
+ DataTypes.FIELD(
+ fieldNames.get(pos),
+
fieldDataTypes.get(pos)))
+ .toArray(DataTypes.Field[]::new))
+ .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<LogRecord> {
+
+ private final ClosableIterator<RowData> hudiRecordIterator;
+ private final ProjectedRow projectedRow;
+ private final HudiRowAsFlussRow hudiRowAsFlussRow;
+ private final int logOffsetColIndex;
+ private final int timestampColIndex;
+
+ private boolean closed;
+
+ public HudiRecordAsFlussRecordIterator(
+ ClosableIterator<RowData> hudiRecordIterator,
+ Schema schema,
+ int metadataFieldCount,
+ int systemFieldCount) {
+ this.hudiRecordIterator = hudiRecordIterator;
+ this.logOffsetColIndex = requiredFieldPosition(schema,
OFFSET_COLUMN_NAME);
+ this.timestampColIndex = requiredFieldPosition(schema,
TIMESTAMP_COLUMN_NAME);
+ this.hudiRowAsFlussRow = new HudiRowAsFlussRow();
+ this.projectedRow =
+ ProjectedRow.from(
+ IntStream.range(
+ metadataFieldCount,
+ schema.getFields().size() -
systemFieldCount)
+ .toArray());
+ }
+
+ @Override
+ public void close() {
+ if (closed) {
+ return;
+ }
+ try {
+ hudiRecordIterator.close();
+ } catch (Exception e) {
+ throw new RuntimeException("Fail to close iterator.", e);
+ } finally {
+ closed = true;
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ return !closed && hudiRecordIterator.hasNext();
+ }
+
+ @Override
+ public LogRecord next() {
+ RowData rowData = hudiRecordIterator.next();
+ ChangeType changeType = toChangeType(rowData.getRowKind());
+ long offset = rowData.getLong(logOffsetColIndex);
+ long timestamp = rowData.getTimestamp(timestampColIndex,
6).getMillisecond();
+
+ return new GenericRecord(
+ offset,
+ timestamp,
+ 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
new file mode 100644
index 000000000..059dd317f
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRowAsFlussRow.java
@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.hudi.source;
+
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.Decimal;
+import org.apache.fluss.row.InternalArray;
+import org.apache.fluss.row.InternalMap;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.MapData;
+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(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) {
+ this.rowData = rowData;
+ return this;
+ }
+
+ @Override
+ public int getFieldCount() {
+ return stripSystemColumns ? rowData.getArity() - SYSTEM_COLUMNS.size()
: rowData.getArity();
+ }
+
+ @Override
+ public boolean isNullAt(int pos) {
+ return rowData.isNullAt(pos);
+ }
+
+ @Override
+ public boolean getBoolean(int pos) {
+ return rowData.getBoolean(pos);
+ }
+
+ @Override
+ public byte getByte(int pos) {
+ return rowData.getByte(pos);
+ }
+
+ @Override
+ public short getShort(int pos) {
+ return rowData.getShort(pos);
+ }
+
+ @Override
+ public int getInt(int pos) {
+ return rowData.getInt(pos);
+ }
+
+ @Override
+ public long getLong(int pos) {
+ return rowData.getLong(pos);
+ }
+
+ @Override
+ public float getFloat(int pos) {
+ return rowData.getFloat(pos);
+ }
+
+ @Override
+ public double getDouble(int pos) {
+ return rowData.getDouble(pos);
+ }
+
+ @Override
+ public BinaryString getChar(int pos, int length) {
+ return BinaryString.fromBytes(rowData.getString(pos).toBytes());
+ }
+
+ @Override
+ public BinaryString getString(int pos) {
+ return BinaryString.fromBytes(rowData.getString(pos).toBytes());
+ }
+
+ @Override
+ public Decimal getDecimal(int pos, int precision, int scale) {
+ DecimalData decimalData = rowData.getDecimal(pos, precision, scale);
+ if (decimalData.isCompact()) {
+ return Decimal.fromUnscaledLong(decimalData.toUnscaledLong(),
precision, scale);
+ }
+ return Decimal.fromBigDecimal(decimalData.toBigDecimal(), precision,
scale);
+ }
+
+ @Override
+ public TimestampNtz getTimestampNtz(int pos, int precision) {
+ TimestampData timestamp = rowData.getTimestamp(pos, precision);
+ if (TimestampNtz.isCompact(precision)) {
+ return TimestampNtz.fromMillis(timestamp.getMillisecond());
+ }
+ return TimestampNtz.fromMillis(
+ timestamp.getMillisecond(), timestamp.getNanoOfMillisecond());
+ }
+
+ @Override
+ public TimestampLtz getTimestampLtz(int pos, int precision) {
+ TimestampData timestamp = rowData.getTimestamp(pos, precision);
+ if (TimestampLtz.isCompact(precision)) {
+ return TimestampLtz.fromEpochMillis(timestamp.getMillisecond());
+ }
+ return TimestampLtz.fromEpochMillis(
+ timestamp.getMillisecond(), timestamp.getNanoOfMillisecond());
+ }
+
+ @Override
+ public byte[] getBinary(int pos, int length) {
+ return rowData.getBinary(pos);
+ }
+
+ @Override
+ public byte[] getBytes(int pos) {
+ return rowData.getBinary(pos);
+ }
+
+ @Override
+ public InternalArray getArray(int pos) {
+ ArrayData value = rowData.getArray(pos);
+ return value == null ? null : new HudiArrayAsFlussArray(value);
+ }
+
+ @Override
+ public InternalMap getMap(int pos) {
+ MapData value = rowData.getMap(pos);
+ return value == null ? null : new HudiMapAsFlussMap(value);
+ }
+
+ @Override
+ public InternalRow getRow(int pos, int numFields) {
+ RowData value = rowData.getRow(pos, numFields);
+ return value == null ? null : new HudiRowAsFlussRow(value, false);
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/UnifiedHudiTableReader.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/UnifiedHudiTableReader.java
new file mode 100644
index 000000000..8c4160685
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/UnifiedHudiTableReader.java
@@ -0,0 +1,310 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.hudi.source;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.common.config.HoodieReaderConfig;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.log.InstantRange;
+import org.apache.hudi.common.table.read.HoodieFileGroupReader;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.configuration.HadoopConfigurations;
+import org.apache.hudi.org.apache.avro.Schema;
+import org.apache.hudi.source.ExpressionPredicates.Predicate;
+import org.apache.hudi.table.format.FilePathUtils;
+import org.apache.hudi.table.format.FlinkRowDataReaderContext;
+import org.apache.hudi.table.format.InternalSchemaManager;
+import org.apache.hudi.table.format.RecordIterators;
+import org.apache.hudi.util.AvroSchemaConverter;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static
org.apache.hudi.configuration.HadoopConfigurations.getParquetConf;
+
+/** Unified reader for both Copy-On-Write and Merge-On-Read Hudi file slices.
*/
+public class UnifiedHudiTableReader implements AutoCloseable {
+
+ private final HoodieTableMetaClient metaClient;
+ private final org.apache.flink.configuration.Configuration
flinkHudiOptions;
+ private final Schema tableSchema;
+ private final Schema requiredSchema;
+ private final int[] selectedFields;
+ private final InternalSchemaManager internalSchemaManager;
+ private final boolean caseSensitive;
+ private final int batchSize;
+ private final boolean emitDelete;
+ private final List<Predicate> predicates;
+ private final String latestCommitTime;
+ private final Option<InstantRange> instantRangeOpt;
+
+ private ClosableIterator<RowData> currentIterator;
+ private boolean closed;
+
+ private UnifiedHudiTableReader(
+ HoodieTableMetaClient metaClient,
+ org.apache.flink.configuration.Configuration flinkHudiOptions,
+ Schema tableSchema,
+ Schema requiredSchema,
+ int[] selectedFields,
+ InternalSchemaManager internalSchemaManager,
+ boolean caseSensitive,
+ int batchSize,
+ boolean emitDelete,
+ List<Predicate> predicates,
+ String latestCommitTime) {
+ this.metaClient = metaClient;
+ this.flinkHudiOptions = flinkHudiOptions;
+ this.tableSchema = tableSchema;
+ this.requiredSchema = requiredSchema;
+ this.selectedFields = selectedFields;
+ this.internalSchemaManager = internalSchemaManager;
+ this.caseSensitive = caseSensitive;
+ this.batchSize = batchSize;
+ this.emitDelete = emitDelete;
+ this.predicates = predicates;
+ this.latestCommitTime = latestCommitTime;
+ this.instantRangeOpt = Option.empty();
+ }
+
+ public ClosableIterator<RowData> readFileSlice(FileSlice fileSlice) throws
IOException {
+ if (closed) {
+ throw new IllegalStateException("Reader is already closed.");
+ }
+ closeCurrentIterator();
+ if (fileSlice.getLogFiles().findAny().isPresent()) {
+ currentIterator = readMorFileSlice(fileSlice);
+ } else {
+ currentIterator = readCowFileSlice(fileSlice);
+ }
+ return currentIterator;
+ }
+
+ private ClosableIterator<RowData> readCowFileSlice(FileSlice fileSlice)
throws IOException {
+ if (!fileSlice.getBaseFile().isPresent()) {
+ throw new IllegalArgumentException("COW file slice must have a
base file.");
+ }
+
+ HoodieBaseFile baseFile = fileSlice.getBaseFile().get();
+ String filePath = baseFile.getPath();
+ Path hadoopPath = new Path(filePath);
+
+ List<String> fieldNames =
+ tableSchema.getFields().stream()
+ .map(Schema.Field::name)
+ .collect(Collectors.toList());
+ List<DataType> fieldTypes =
+ tableSchema.getFields().stream()
+ .map(field ->
AvroSchemaConverter.convertToDataType(field.schema()))
+ .collect(Collectors.toList());
+ LinkedHashMap<String, Object> partitionSpec =
+ FilePathUtils.generatePartitionSpecs(
+ filePath,
+ fieldNames,
+ fieldTypes,
+
flinkHudiOptions.get(FlinkOptions.PARTITION_DEFAULT_NAME),
+
flinkHudiOptions.get(FlinkOptions.PARTITION_PATH_FIELD),
+
flinkHudiOptions.get(FlinkOptions.HIVE_STYLE_PARTITIONING));
+
+ return RecordIterators.getParquetRecordIterator(
+ internalSchemaManager,
+ flinkHudiOptions.get(FlinkOptions.READ_UTC_TIMEZONE),
+ caseSensitive,
+ getParquetConf(
+ flinkHudiOptions,
HadoopConfigurations.getHadoopConf(flinkHudiOptions)),
+ fieldNames.toArray(new String[0]),
+ fieldTypes.toArray(new DataType[0]),
+ partitionSpec,
+ selectedFields,
+ batchSize,
+ new org.apache.flink.core.fs.Path(hadoopPath.toUri()),
+ 0L,
+ baseFile.getFileLen(),
+ predicates);
+ }
+
+ private ClosableIterator<RowData> readMorFileSlice(FileSlice fileSlice)
throws IOException {
+ FlinkRowDataReaderContext readerContext =
+ new FlinkRowDataReaderContext(
+ metaClient.getStorageConf(),
+ () -> internalSchemaManager,
+ predicates,
+ metaClient.getTableConfig(),
+ instantRangeOpt);
+
+ TypedProperties typedProperties = getReadProps(metaClient,
flinkHudiOptions);
+ typedProperties.put(
+ HoodieReaderConfig.MERGE_TYPE.key(),
flinkHudiOptions.get(FlinkOptions.MERGE_TYPE));
+
+ return HoodieFileGroupReader.<RowData>newBuilder()
+ .withReaderContext(readerContext)
+ .withHoodieTableMetaClient(metaClient)
+ .withLatestCommitTime(latestCommitTime)
+ .withFileSlice(fileSlice)
+ .withDataSchema(tableSchema)
+ .withRequestedSchema(requiredSchema)
+
.withInternalSchema(Option.ofNullable(internalSchemaManager.getQuerySchema()))
+ .withProps(typedProperties)
+ .withShouldUseRecordPosition(false)
+ .withEmitDelete(emitDelete)
+ .build()
+ .getClosableIterator();
+ }
+
+ public static TypedProperties getReadProps(
+ HoodieTableMetaClient metaClient,
+ org.apache.flink.configuration.Configuration flinkHudiOptions) {
+ TypedProperties properties = new TypedProperties();
+ properties.putAll(metaClient.getTableConfig().getProps());
+ properties.putAll(flinkHudiOptions.toMap());
+ return properties;
+ }
+
+ private void closeCurrentIterator() {
+ if (currentIterator != null) {
+ currentIterator.close();
+ currentIterator = null;
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closeCurrentIterator();
+ closed = true;
+ }
+
+ public static Builder newBuilder() {
+ return new Builder();
+ }
+
+ /** Builder for {@link UnifiedHudiTableReader}. */
+ public static class Builder {
+ private HoodieTableMetaClient metaClient;
+ private org.apache.flink.configuration.Configuration flinkHudiOptions;
+ private Schema tableSchema;
+ private Schema requiredSchema;
+ private int[] selectedFields;
+ private InternalSchemaManager internalSchemaManager =
InternalSchemaManager.DISABLED;
+ private boolean caseSensitive = true;
+ private int batchSize = 2048;
+ private boolean emitDelete;
+ private List<Predicate> predicates = Collections.emptyList();
+ private String latestCommitTime;
+
+ public Builder withMetaClient(HoodieTableMetaClient metaClient) {
+ this.metaClient = metaClient;
+ return this;
+ }
+
+ public Builder withProps(org.apache.flink.configuration.Configuration
flinkHudiOptions) {
+ this.flinkHudiOptions = flinkHudiOptions;
+ return this;
+ }
+
+ public Builder withTableSchema(Schema tableSchema) {
+ this.tableSchema = tableSchema;
+ return this;
+ }
+
+ public Builder withRequiredSchema(Schema requiredSchema) {
+ this.requiredSchema = requiredSchema;
+ return this;
+ }
+
+ public Builder withSelectedFields(int[] selectedFields) {
+ this.selectedFields = selectedFields;
+ return this;
+ }
+
+ public Builder withInternalSchemaManager(InternalSchemaManager
internalSchemaManager) {
+ this.internalSchemaManager = internalSchemaManager;
+ return this;
+ }
+
+ public Builder withCaseSensitive(boolean caseSensitive) {
+ this.caseSensitive = caseSensitive;
+ return this;
+ }
+
+ public Builder withBatchSize(int batchSize) {
+ this.batchSize = batchSize;
+ return this;
+ }
+
+ public Builder withEmitDelete(boolean emitDelete) {
+ this.emitDelete = emitDelete;
+ return this;
+ }
+
+ public Builder withPredicates(List<Predicate> predicates) {
+ this.predicates = predicates;
+ return this;
+ }
+
+ public Builder withLatestCommitTime(String latestCommitTime) {
+ this.latestCommitTime = latestCommitTime;
+ return this;
+ }
+
+ public UnifiedHudiTableReader build() {
+ if (metaClient == null) {
+ throw new IllegalArgumentException("metaClient is required.");
+ }
+ if (flinkHudiOptions == null) {
+ throw new IllegalArgumentException("flinkHudiOptions is
required.");
+ }
+ if (tableSchema == null) {
+ throw new IllegalArgumentException("tableSchema is required.");
+ }
+ if (requiredSchema == null) {
+ requiredSchema = tableSchema;
+ }
+ if (selectedFields == null) {
+ throw new IllegalArgumentException("selectedFields is
required.");
+ }
+ if (latestCommitTime == null) {
+ throw new IllegalArgumentException("latestCommitTime is
required.");
+ }
+ return new UnifiedHudiTableReader(
+ metaClient,
+ flinkHudiOptions,
+ tableSchema,
+ requiredSchema,
+ selectedFields,
+ internalSchemaManager,
+ caseSensitive,
+ batchSize,
+ emitDelete,
+ predicates,
+ latestCommitTime);
+ }
+ }
+}
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 2f90124e5..5c4a66214 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
@@ -24,6 +24,7 @@ import org.apache.fluss.lake.hudi.FlussDataTypeToHudiDataType;
import org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.ChangeType;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.catalog.CatalogTable;
@@ -33,6 +34,7 @@ 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;
import org.apache.hudi.index.HoodieIndex;
@@ -316,4 +318,19 @@ public class HudiConversions {
}
return primaryKeys;
}
+
+ public static ChangeType toChangeType(RowKind rowKind) {
+ switch (rowKind) {
+ case INSERT:
+ return ChangeType.INSERT;
+ case UPDATE_BEFORE:
+ return ChangeType.UPDATE_BEFORE;
+ case UPDATE_AFTER:
+ return ChangeType.UPDATE_AFTER;
+ case DELETE:
+ return ChangeType.DELETE;
+ default:
+ throw new IllegalArgumentException("Unsupported change type: "
+ rowKind);
+ }
+ }
}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiLakeSourceTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiLakeSourceTest.java
index 1b0e2fe89..e8446ee85 100644
---
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiLakeSourceTest.java
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiLakeSourceTest.java
@@ -27,6 +27,7 @@ import org.apache.fluss.types.RowType;
import org.junit.jupiter.api.Test;
+import java.io.IOException;
import java.util.Collections;
import static org.assertj.core.api.Assertions.assertThat;
@@ -75,12 +76,12 @@ class HudiLakeSourceTest {
}
@Test
- void testCreateRecordReaderIsExplicitlyUnsupported() {
+ void testCreateRecordReaderWrapsReaderErrors() {
HudiLakeSource source =
new HudiLakeSource(new Configuration(), TablePath.of("db1",
"table1"));
assertThatThrownBy(() -> source.createRecordReader(() -> null))
- .isInstanceOf(UnsupportedOperationException.class)
- .hasMessageContaining("record reading");
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Fail to create Hudi record reader");
}
}
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
new file mode 100644
index 000000000..64f802a9a
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiRecordReaderTest.java
@@ -0,0 +1,208 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.hudi.source;
+
+import org.apache.fluss.record.ChangeType;
+import org.apache.fluss.record.LogRecord;
+import org.apache.fluss.row.BinaryString;
+
+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;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Iterator;
+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() {
+ TestingClosableIterator hudiIterator =
+ new TestingClosableIterator(
+ rowData(RowKind.UPDATE_AFTER, 11, "value", 3, 42L,
1234L));
+ HudiRecordReader.HudiRecordAsFlussRecordIterator iterator =
+ new HudiRecordReader.HudiRecordAsFlussRecordIterator(
+ hudiIterator, fullSchema(), 2, 3);
+
+ 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.getRow().getFieldCount()).isEqualTo(2);
+ assertThat(logRecord.getRow().getInt(0)).isEqualTo(11);
+
assertThat(logRecord.getRow().getString(1)).isEqualTo(BinaryString.fromString("value"));
+
+ iterator.close();
+ iterator.close();
+ assertThat(iterator.hasNext()).isFalse();
+ assertThat(hudiIterator.getCloseCount()).isEqualTo(1);
+ }
+
+ @Test
+ void testIteratorSkipsMetadataAndAllSystemColumnsForProjectedSchema() {
+ TestingClosableIterator hudiIterator =
+ new TestingClosableIterator(
+ projectedRowData(RowKind.DELETE, "projected", 5, 7L,
999L));
+ HudiRecordReader.HudiRecordAsFlussRecordIterator iterator =
+ new HudiRecordReader.HudiRecordAsFlussRecordIterator(
+ hudiIterator, projectedSchema(), 2, 3);
+
+ LogRecord logRecord = iterator.next();
+
+ assertThat(logRecord.getChangeType()).isEqualTo(ChangeType.DELETE);
+ assertThat(logRecord.logOffset()).isEqualTo(7L);
+ assertThat(logRecord.timestamp()).isEqualTo(999L);
+ assertThat(logRecord.getRow().getFieldCount()).isEqualTo(1);
+
assertThat(logRecord.getRow().getString(0)).isEqualTo(BinaryString.fromString("projected"));
+
+ iterator.close();
+ 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);
+ 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);
+ 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;
+ }
+
+ private static Schema fullSchema() {
+ return parseSchema(
+ "["
+ +
"{\"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\"}"
+ + "]");
+ }
+
+ private static Schema projectedSchema() {
+ return parseSchema(
+ "["
+ +
"{\"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\"}"
+ + "]");
+ }
+
+ private static Schema parseSchema(String fields) {
+ return new Schema.Parser()
+ .parse(
+ "{"
+ + "\"type\":\"record\","
+ + "\"name\":\"hudi_record\","
+ + "\"fields\":"
+ + fields
+ + "}");
+ }
+
+ private static class TestingClosableIterator implements
ClosableIterator<RowData> {
+ private final Iterator<RowData> iterator;
+ private int closeCount;
+
+ private TestingClosableIterator(RowData rowData) {
+ this.iterator = Collections.singletonList(rowData).iterator();
+ }
+
+ @Override
+ public void close() {
+ closeCount++;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public RowData next() {
+ if (!iterator.hasNext()) {
+ throw new NoSuchElementException();
+ }
+ return iterator.next();
+ }
+
+ private int getCloseCount() {
+ return closeCount;
+ }
+ }
+}
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
new file mode 100644
index 000000000..814c89651
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiRowAsFlussRowTest.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.hudi.source;
+
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.Decimal;
+import org.apache.fluss.row.InternalArray;
+import org.apache.fluss.row.InternalMap;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.GenericMapData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link HudiRowAsFlussRow}. */
+class HudiRowAsFlussRowTest {
+
+ @Test
+ void testAccessPrimitiveAndNestedValues() {
+ GenericRowData nestedRow = new GenericRowData(1);
+ nestedRow.setField(0, StringData.fromString("nested"));
+
+ GenericRowData rowData = new GenericRowData(19);
+ rowData.setField(0, true);
+ rowData.setField(1, (byte) 1);
+ rowData.setField(2, (short) 2);
+ rowData.setField(3, 3);
+ rowData.setField(4, 4L);
+ rowData.setField(5, 1.5f);
+ rowData.setField(6, 2.5d);
+ rowData.setField(7, StringData.fromString("string"));
+ rowData.setField(8, DecimalData.fromUnscaledLong(1234, 10, 2));
+ rowData.setField(9, DecimalData.fromBigDecimal(new
BigDecimal("1234567890"), 20, 0));
+ rowData.setField(10, TimestampData.fromEpochMillis(1000L));
+ rowData.setField(11, TimestampData.fromEpochMillis(2000L, 123456));
+ rowData.setField(12, new byte[] {1, 2});
+ rowData.setField(13, new GenericArrayData(new Object[]
{StringData.fromString("a")}));
+ rowData.setField(
+ 14,
+ new GenericMapData(
+ 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);
+
+ assertThat(row.getFieldCount()).isEqualTo(16);
+ assertThat(row.getBoolean(0)).isTrue();
+ assertThat(row.getByte(1)).isEqualTo((byte) 1);
+ assertThat(row.getShort(2)).isEqualTo((short) 2);
+ assertThat(row.getInt(3)).isEqualTo(3);
+ assertThat(row.getLong(4)).isEqualTo(4L);
+ assertThat(row.getFloat(5)).isEqualTo(1.5f);
+ assertThat(row.getDouble(6)).isEqualTo(2.5d);
+
assertThat(row.getString(7)).isEqualTo(BinaryString.fromString("string"));
+ assertThat(row.getDecimal(8, 10,
2)).isEqualTo(Decimal.fromUnscaledLong(1234, 10, 2));
+ assertThat(row.getDecimal(9, 20, 0))
+ .isEqualTo(Decimal.fromBigDecimal(new
BigDecimal("1234567890"), 20, 0));
+ assertThat(row.getTimestampNtz(10,
3)).isEqualTo(TimestampNtz.fromMillis(1000L));
+ assertThat(row.getTimestampLtz(11, 6))
+ .isEqualTo(TimestampLtz.fromEpochMillis(2000L, 123456));
+ assertThat(row.getBytes(12)).containsExactly(1, 2);
+
+ InternalArray array = row.getArray(13);
+ assertThat(array.getString(0)).isEqualTo(BinaryString.fromString("a"));
+
+ InternalMap map = row.getMap(14);
+
assertThat(map.keyArray().getString(0)).isEqualTo(BinaryString.fromString("key"));
+
assertThat(map.valueArray().getString(0)).isEqualTo(BinaryString.fromString("value"));
+
+ InternalRow nested = row.getRow(15, 1);
+ assertThat(nested.getFieldCount()).isEqualTo(1);
+
assertThat(nested.getString(0)).isEqualTo(BinaryString.fromString("nested"));
+ }
+}
