Copilot commented on code in PR #3499: URL: https://github.com/apache/fluss/pull/3499#discussion_r3441726803
########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/UnifiedHudiTableReader.java: ########## @@ -0,0 +1,308 @@ +/* + * 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.getString(FlinkOptions.PARTITION_DEFAULT_NAME), + flinkHudiOptions.getString(FlinkOptions.PARTITION_PATH_FIELD), + flinkHudiOptions.getBoolean(FlinkOptions.HIVE_STYLE_PARTITIONING)); + + return RecordIterators.getParquetRecordIterator( + internalSchemaManager, + flinkHudiOptions.getBoolean(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.getString(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 (latestCommitTime == null) { + throw new IllegalArgumentException("latestCommitTime is required."); + } Review Comment: `selectedFields` is allowed to remain null, but it is passed into the reader and ultimately used for Parquet reads. If a caller forgets to set it, this will lead to a null-related failure deep in the read path. Default it to selecting all fields (or explicitly validate it) in `build()` to make the builder safer to use. ########## fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiRecordReaderTest.java: ########## @@ -0,0 +1,178 @@ +/* + * 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; + +/** 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")); + } Review Comment: This test creates a closable iterator but never closes it, so it doesn't verify resource cleanup (and could mask leaks if the underlying iterator holds file handles). Close the iterator and assert the wrapped Hudi iterator is closed, similar to the first test. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRecordReader.java: ########## @@ -0,0 +1,280 @@ +/* + * 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 -> schema.getField(systemColumn).pos()) + .toArray(); Review Comment: `schema.getField(systemColumn)` can return null when the Hudi schema doesn't contain Fluss-required system columns (e.g., reading an existing Hudi table). The current code would then throw a NullPointerException via `.pos()`, which is hard to diagnose. Validate presence and fail with a clear IllegalArgumentException instead. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
