This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 7bc38af1be0bad97bec6780f055c000dc3779fcd Author: voonhous <[email protected]> AuthorDate: Tue Jun 2 19:08:47 2026 +0800 refactor: Add Lombok Builders (#17781) - Add Lombok Builder to HoodieFileGroupReader - Add Lombok Builder to InputSplit - Add Lombok Builder to ReaderParameters (cherry picked from commit 7964202a751dc24d65fa0bcd1f300278a453e6bd) --- .../hudi/cli/commands/HoodieLogFileCommand.java | 6 +- .../org/apache/hudi/index/HoodieIndexUtils.java | 8 +- .../hudi/io/FileGroupReaderBasedAppendHandle.java | 20 +- .../hudi/io/FileGroupReaderBasedMergeHandle.java | 6 +- .../metadata/HoodieBackedTableMetadataWriter.java | 8 +- .../SecondaryIndexRecordGenerationUtils.java | 10 +- .../strategy/ClusteringExecutionStrategy.java | 17 +- .../hudi/client/TestJavaHoodieBackedMetadata.java | 2 +- .../client/utils/SparkMetadataWriterUtils.java | 2 +- .../client/TestSparkRDDMetadataWriteClient.java | 6 +- .../functional/TestHoodieBackedTableMetadata.java | 2 +- .../common/table/read/HoodieFileGroupReader.java | 296 +++++++-------------- .../apache/hudi/common/table/read/InputSplit.java | 64 +++-- .../hudi/common/table/read/ReaderParameters.java | 82 ++---- .../buffer/DefaultFileGroupRecordBufferLoader.java | 6 +- .../read/buffer/LogScanningRecordBufferLoader.java | 2 +- .../ReusableFileGroupRecordBufferLoader.java | 2 +- .../StreamingFileGroupRecordBufferLoader.java | 4 +- .../hudi/metadata/HoodieBackedTableMetadata.java | 6 +- .../hudi/metadata/HoodieTableMetadataUtil.java | 8 +- .../table/read/TestHoodieFileGroupReaderBase.java | 8 +- .../read/buffer/BaseTestFileGroupRecordBuffer.java | 2 +- .../buffer/TestFileGroupRecordBufferLoader.java | 4 +- .../TestSortedKeyBasedFileGroupRecordBuffer.java | 2 +- .../org/apache/hudi/table/format/FormatUtils.java | 8 +- .../reader/HoodieFileGroupReaderTestHarness.java | 7 +- .../HoodieFileGroupReaderBasedRecordReader.java | 6 +- .../org/apache/hudi/HoodieMergeOnReadRDDV2.scala | 8 +- .../org/apache/hudi/cdc/CDCFileGroupIterator.scala | 8 +- .../HoodieFileGroupReaderBasedFileFormat.scala | 8 +- .../procedures/PartitionBucketIndexManager.scala | 8 +- .../TestMetadataUtilRLIandSIRecordGeneration.java | 2 +- .../hudi/functional/TestHoodieBackedMetadata.java | 2 +- 33 files changed, 279 insertions(+), 351 deletions(-) diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/HoodieLogFileCommand.java b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/HoodieLogFileCommand.java index 6138e1fb3839..d6d2546e2cb1 100644 --- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/HoodieLogFileCommand.java +++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/HoodieLogFileCommand.java @@ -242,10 +242,12 @@ public class HoodieLogFileCommand { Option.empty(), Option.empty(), fileGroupReaderProperties); - try (HoodieFileGroupReader<IndexedRecord> fileGroupReader = HoodieFileGroupReader.<IndexedRecord>newBuilder() + try (HoodieFileGroupReader<IndexedRecord> fileGroupReader = HoodieFileGroupReader.<IndexedRecord>builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(HoodieCLI.getTableMetaClient()) - .withFileSlice(fileSlice) + .withBaseFileOption(fileSlice.getBaseFile()) + .withLogFiles(fileSlice.getLogFiles()) + .withPartitionPath(fileSlice.getPartitionPath()) .withDataSchema(readerSchema) .withRequestedSchema(readerSchema) .withLatestCommitTime(client.getActiveTimeline().getCommitAndReplaceTimeline().lastInstant().map(HoodieInstant::requestedTime).orElse(HoodieInstantTimeGenerator.getCurrentInstantTimeStr())) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java index d9fe1068e421..d2cc530295f2 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java @@ -319,14 +319,16 @@ public class HoodieIndexUtils { Option<InternalSchema> internalSchemaOption = SerDeHelper.fromJson(config.getInternalSchema()); FileSlice fileSlice = fileSliceOption.get(); HoodieReaderContext<R> readerContext = readerContextFactory.getContext(); - HoodieFileGroupReader<R> fileGroupReader = HoodieFileGroupReader.<R>newBuilder() + HoodieFileGroupReader<R> fileGroupReader = HoodieFileGroupReader.<R>builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(metaClient) .withLatestCommitTime(instantTime.get()) - .withFileSlice(fileSlice) + .withBaseFileOption(fileSlice.getBaseFile()) + .withLogFiles(fileSlice.getLogFiles()) + .withPartitionPath(fileSlice.getPartitionPath()) .withDataSchema(dataSchema) .withRequestedSchema(dataSchema) - .withInternalSchema(internalSchemaOption) + .withInternalSchemaOpt(internalSchemaOption) .withProps(metaClient.getTableConfig().getProps()) .build(); try { diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedAppendHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedAppendHandle.java index a081709f6fc2..c40ce0158a3c 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedAppendHandle.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedAppendHandle.java @@ -82,10 +82,20 @@ public class FileGroupReaderBasedAppendHandle<T, I, K, O> extends HoodieAppendHa new HoodieLogFile(new StoragePath(FSUtils.constructAbsolutePath( config.getBasePath(), operation.getPartitionPath()), logFileName))); // Initializes the record iterator, log compaction requires writing the deletes into the delete block of the resulting log file. - try (HoodieFileGroupReader<T> fileGroupReader = HoodieFileGroupReader.<T>newBuilder().withReaderContext(readerContext).withHoodieTableMetaClient(hoodieTable.getMetaClient()) - .withLatestCommitTime(instantTime).withPartitionPath(partitionPath).withLogFiles(logFiles).withBaseFileOption(Option.empty()).withDataSchema(writeSchemaWithMetaFields) - .withRequestedSchema(writeSchemaWithMetaFields).withInternalSchema(internalSchemaOption).withProps(props).withEmitDelete(true) - .withShouldUseRecordPosition(usePosition).withSortOutput(hoodieTable.requireSortedRecords()) + try (HoodieFileGroupReader<T> fileGroupReader = HoodieFileGroupReader.<T>builder() + .withReaderContext(readerContext) + .withHoodieTableMetaClient(hoodieTable.getMetaClient()) + .withLatestCommitTime(instantTime) + .withPartitionPath(partitionPath) + .withLogFiles(logFiles) + .withBaseFileOption(Option.empty()) + .withDataSchema(writeSchemaWithMetaFields) + .withRequestedSchema(writeSchemaWithMetaFields) + .withInternalSchemaOpt(internalSchemaOption) + .withProps(props) + .withEmitDelete(true) + .withShouldUseRecordPosition(usePosition) + .withSortOutput(hoodieTable.requireSortedRecords()) // instead of using config.enableOptimizedLogBlocksScan(), we set to true as log compaction blocks only supported in scanV2 .build()) { recordItr = new CloseableMappingIterator<>(fileGroupReader.getLogRecordsOnly(), record -> { @@ -96,7 +106,7 @@ public class FileGroupReaderBasedAppendHandle<T, I, K, O> extends HoodieAppendHa header.put(HoodieLogBlock.HeaderMetadataType.COMPACTED_BLOCK_TIMES, StringUtils.join(fileGroupReader.getValidBlockInstants(), ",")); super.doAppend(); - this.readStats = fileGroupReader.getStats(); + this.readStats = fileGroupReader.getReadStats(); } catch (IOException e) { throw new HoodieIOException("Failed to initialize file group reader for " + fileId, e); } diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java index d708c15f3384..2893913a0d13 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java @@ -301,7 +301,7 @@ public class FileGroupReaderBasedMergeHandle<T, I, K, O> extends HoodieWriteMerg // The stats of inserts, updates, and deletes are updated once at the end // These will be set in the write stat when closing the merge handle - this.readStats = fileGroupReader.getStats(); + this.readStats = fileGroupReader.getReadStats(); this.insertRecordsWritten = readStats.getNumInserts(); this.updatedRecordsWritten = readStats.getNumUpdates(); this.recordsDeleted = readStats.getNumDeletes(); @@ -318,10 +318,10 @@ public class FileGroupReaderBasedMergeHandle<T, I, K, O> extends HoodieWriteMerg private HoodieFileGroupReader<T> getFileGroupReader(boolean usePosition, Option<InternalSchema> internalSchemaOption, TypedProperties props, Option<Stream<HoodieLogFile>> logFileStreamOpt, Iterator<HoodieRecord<T>> incomingRecordsItr) { - HoodieFileGroupReader.Builder<T> fileGroupBuilder = HoodieFileGroupReader.<T>newBuilder().withReaderContext(readerContext).withHoodieTableMetaClient(hoodieTable.getMetaClient()) + HoodieFileGroupReader.HoodieFileGroupReaderBuilder<T> fileGroupBuilder = HoodieFileGroupReader.<T>builder().withReaderContext(readerContext).withHoodieTableMetaClient(hoodieTable.getMetaClient()) .withLatestCommitTime(maxInstantTime).withPartitionPath(partitionPath).withBaseFileOption(Option.ofNullable(baseFileToMerge)) .withDataSchema(writeSchemaWithMetaFields).withRequestedSchema(writeSchemaWithMetaFields) - .withInternalSchema(internalSchemaOption).withProps(props) + .withInternalSchemaOpt(internalSchemaOption).withProps(props) .withShouldUseRecordPosition(usePosition).withSortOutput(hoodieTable.requireSortedRecords()) .withFileGroupUpdateCallback(createCallback()); diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java index 7035c40c9b42..9ed26c586ef1 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java @@ -997,14 +997,16 @@ public abstract class HoodieBackedTableMetadataWriter<I, O> implements HoodieTab HoodieSchema requestedSchema = metaClient.getTableConfig().populateMetaFields() ? getRecordKeySchema() : HoodieSchemaUtils.projectSchema(dataSchema, Arrays.asList(metaClient.getTableConfig().getRecordKeyFields().orElse(new String[0]))); Option<InternalSchema> internalSchemaOption = SerDeHelper.fromJson(dataWriteConfig.getInternalSchema()); - HoodieFileGroupReader<T> fileGroupReader = HoodieFileGroupReader.<T>newBuilder() + HoodieFileGroupReader<T> fileGroupReader = HoodieFileGroupReader.<T>builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(metaClient) - .withFileSlice(fileSlice) + .withBaseFileOption(fileSlice.getBaseFile()) + .withLogFiles(fileSlice.getLogFiles()) + .withPartitionPath(fileSlice.getPartitionPath()) .withLatestCommitTime(instantTime.get()) .withDataSchema(dataSchema) .withRequestedSchema(requestedSchema) - .withInternalSchema(internalSchemaOption) + .withInternalSchemaOpt(internalSchemaOption) .withShouldUseRecordPosition(false) .withProps(metaClient.getTableConfig().getProps()) .build(); diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java index 9cb9b8ca7df5..b86688b81e60 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java @@ -123,7 +123,7 @@ public class SecondaryIndexRecordGenerationUtils { // validate that for a given fileId, either we have 1 parquet file or N log files. AtomicInteger totalParquetFiles = new AtomicInteger(); AtomicInteger totalLogFiles = new AtomicInteger(); - writeStats.stream().forEach(writeStat -> { + writeStats.forEach(writeStat -> { if (FSUtils.isLogFile(new StoragePath(basePath, writeStat.getPath()))) { totalLogFiles.getAndIncrement(); } else { @@ -156,7 +156,7 @@ public class SecondaryIndexRecordGenerationUtils { } else { // log files are added in current commit // add new log files to existing latest file slice and compute the secondary index to primary key mapping. FileSlice latestFileSlice = fileSliceOption.get(); - writeStats.stream().forEach(writeStat -> { + writeStats.forEach(writeStat -> { StoragePathInfo logFile = new StoragePathInfo(new StoragePath(basePath, writeStat.getPath()), writeStat.getFileSizeInBytes(), false, (short) 0, 0, 0); latestFileSlice.addLogFile(new HoodieLogFile(logFile)); }); @@ -286,9 +286,11 @@ public class SecondaryIndexRecordGenerationUtils { boolean allowInflightInstants) throws IOException { String secondaryKeyField = indexDefinition.getSourceFieldsKey(); HoodieSchema requestedSchema = getRequestedSchemaForSecondaryIndex(metaClient, tableSchema, secondaryKeyField); - HoodieFileGroupReader<T> fileGroupReader = HoodieFileGroupReader.<T>newBuilder() + HoodieFileGroupReader<T> fileGroupReader = HoodieFileGroupReader.<T>builder() .withReaderContext(readerContext) - .withFileSlice(fileSlice) + .withBaseFileOption(fileSlice.getBaseFile()) + .withLogFiles(fileSlice.getLogFiles()) + .withPartitionPath(fileSlice.getPartitionPath()) .withHoodieTableMetaClient(metaClient) .withProps(props) .withLatestCommitTime(instantTime) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java index d34a85a97bd4..cc3c4a02eb22 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/ClusteringExecutionStrategy.java @@ -142,9 +142,18 @@ public abstract class ClusteringExecutionStrategy<T, I, K, O> implements Seriali ReaderContextFactory<R> readerContextFactory, String instantTime, TypedProperties properties, boolean usePosition) { HoodieReaderContext<R> readerContext = readerContextFactory.getContext(); - return HoodieFileGroupReader.<R>newBuilder() - .withReaderContext(readerContext).withHoodieTableMetaClient(metaClient).withLatestCommitTime(instantTime) - .withFileSlice(fileSlice).withDataSchema(readerSchema).withRequestedSchema(readerSchema).withInternalSchema(internalSchemaOption) - .withShouldUseRecordPosition(usePosition).withProps(properties).build(); + return HoodieFileGroupReader.<R>builder() + .withReaderContext(readerContext) + .withHoodieTableMetaClient(metaClient) + .withLatestCommitTime(instantTime) + .withBaseFileOption(fileSlice.getBaseFile()) + .withLogFiles(fileSlice.getLogFiles()) + .withPartitionPath(fileSlice.getPartitionPath()) + .withDataSchema(readerSchema) + .withRequestedSchema(readerSchema) + .withInternalSchemaOpt(internalSchemaOption) + .withShouldUseRecordPosition(usePosition) + .withProps(properties) + .build(); } } diff --git a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java index 9563439c39eb..feda056472dc 100644 --- a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java +++ b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java @@ -946,7 +946,7 @@ public class TestJavaHoodieBackedMetadata extends TestHoodieMetadataBase { HoodieSchema schema = HoodieSchemaUtils.addMetadataFields(HoodieSchema.fromAvroSchema(HoodieMetadataRecord.getClassSchema())); HoodieAvroReaderContext readerContext = new HoodieAvroReaderContext(metadataMetaClient.getStorageConf(), metadataMetaClient.getTableConfig(), Option.empty(), Option.empty(), new TypedProperties()); - HoodieFileGroupReader<IndexedRecord> fileGroupReader = HoodieFileGroupReader.<IndexedRecord>newBuilder() + HoodieFileGroupReader<IndexedRecord> fileGroupReader = HoodieFileGroupReader.<IndexedRecord>builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(metadataMetaClient) .withLogFiles(logFiles.stream()) diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkMetadataWriterUtils.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkMetadataWriterUtils.java index 47155899e204..2184c24ad136 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkMetadataWriterUtils.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkMetadataWriterUtils.java @@ -335,7 +335,7 @@ public class SparkMetadataWriterUtils { baseFileOption = Option.empty(); logFileStream = Stream.of(new HoodieLogFile(filePath)); } - HoodieFileGroupReader<InternalRow> fileGroupReader = HoodieFileGroupReader.<InternalRow>newBuilder() + HoodieFileGroupReader<InternalRow> fileGroupReader = HoodieFileGroupReader.<InternalRow>builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(metaClient) .withDataSchema(tableSchema) diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestSparkRDDMetadataWriteClient.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestSparkRDDMetadataWriteClient.java index 8a88dd813588..65e45ce96f36 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestSparkRDDMetadataWriteClient.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestSparkRDDMetadataWriteClient.java @@ -218,10 +218,12 @@ public class TestSparkRDDMetadataWriteClient extends HoodieClientTestBase { HoodieSchema schema = HoodieSchemaUtils.addMetadataFields(metadataSchema); HoodieAvroReaderContext readerContext = new HoodieAvroReaderContext(metadataMetaClient.getStorageConf(), metadataMetaClient.getTableConfig(), Option.of(instantRange), Option.of(predicate)); - HoodieFileGroupReader<IndexedRecord> fileGroupReader = HoodieFileGroupReader.<IndexedRecord>newBuilder() + HoodieFileGroupReader<IndexedRecord> fileGroupReader = HoodieFileGroupReader.<IndexedRecord>builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(metadataMetaClient) - .withFileSlice(fileSlice) + .withBaseFileOption(fileSlice.getBaseFile()) + .withLogFiles(fileSlice.getLogFiles()) + .withPartitionPath(fileSlice.getPartitionPath()) .withLatestCommitTime(validMetadataInstant) .withRequestedSchema(metadataSchema) .withDataSchema(schema) diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedTableMetadata.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedTableMetadata.java index 6bb71f07de61..fcbd689f7447 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedTableMetadata.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedTableMetadata.java @@ -560,7 +560,7 @@ public class TestHoodieBackedTableMetadata extends TestHoodieMetadataBase { private void verifyMetadataMergedRecords(HoodieTableMetaClient metadataMetaClient, List<HoodieLogFile> logFiles, String latestCommitTimestamp, HoodieWriteConfig metadataTableWriteConfig) { HoodieSchema schema = HoodieSchemaUtils.addMetadataFields(HoodieSchema.fromAvroSchema(HoodieMetadataRecord.getClassSchema())); HoodieAvroReaderContext readerContext = new HoodieAvroReaderContext(metadataMetaClient.getStorageConf(), metadataMetaClient.getTableConfig(), Option.empty(), Option.empty()); - HoodieFileGroupReader<IndexedRecord> fileGroupReader = HoodieFileGroupReader.<IndexedRecord>newBuilder() + HoodieFileGroupReader<IndexedRecord> fileGroupReader = HoodieFileGroupReader.<IndexedRecord>builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(metadataMetaClient) .withLogFiles(logFiles.stream()) diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodieFileGroupReader.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodieFileGroupReader.java index 2b35eaff4f30..0b536b511db1 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodieFileGroupReader.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodieFileGroupReader.java @@ -23,7 +23,6 @@ import org.apache.hudi.common.config.HoodieReaderConfig; import org.apache.hudi.common.config.TypedProperties; import org.apache.hudi.common.engine.HoodieReaderContext; import org.apache.hudi.common.model.BaseFile; -import org.apache.hudi.common.model.FileSlice; import org.apache.hudi.common.model.HoodieBaseFile; import org.apache.hudi.common.model.HoodieLogFile; import org.apache.hudi.common.model.HoodieRecord; @@ -35,7 +34,6 @@ import org.apache.hudi.common.table.PartitionPathParser; import org.apache.hudi.common.table.read.buffer.FileGroupRecordBufferLoader; import org.apache.hudi.common.table.read.buffer.HoodieFileGroupRecordBuffer; import org.apache.hudi.common.util.ConfigUtils; -import org.apache.hudi.common.util.Either; import org.apache.hudi.common.util.HoodieRecordUtils; import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.ValidationUtils; @@ -49,6 +47,10 @@ import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.StoragePath; import org.apache.hudi.storage.StoragePathInfo; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Getter; + import java.io.Closeable; import java.io.IOException; import java.util.ArrayList; @@ -67,7 +69,9 @@ import java.util.stream.Stream; * @param <T> The type of engine-specific record representation, e.g.,{@code InternalRow} * in Spark and {@code RowData} in Flink. */ +@AllArgsConstructor public final class HoodieFileGroupReader<T> implements Closeable { + private final HoodieReaderContext<T> readerContext; private final HoodieTableMetaClient metaClient; private final InputSplit inputSplit; @@ -81,25 +85,113 @@ public final class HoodieFileGroupReader<T> implements Closeable { private HoodieFileGroupRecordBuffer<T> recordBuffer; private ClosableIterator<T> baseFileIterator; private final Option<UnaryOperator<T>> outputConverter; + @Getter private final HoodieReadStats readStats; // Callback to run custom logic on updates to the base files for the file group private final Option<BaseFileUpdateCallback<T>> fileGroupUpdateCallback; // The list of instant times read from the log blocks, this value is used by the log-compaction to allow optimized log-block scans + @Getter private List<String> validBlockInstants = Collections.emptyList(); private BufferedRecordConverter<T> bufferedRecordConverter; - private HoodieFileGroupReader(HoodieReaderContext<T> readerContext, HoodieStorage storage, String tablePath, - String latestCommitTime, HoodieSchema dataSchema, HoodieSchema requestedSchema, - Option<InternalSchema> internalSchemaOpt, HoodieTableMetaClient hoodieTableMetaClient, TypedProperties props, - ReaderParameters readerParameters, InputSplit inputSplit, Option<BaseFileUpdateCallback<T>> updateCallback, - FileGroupRecordBufferLoader<T> recordBufferLoader) { + @Builder(setterPrefix = "with") + private HoodieFileGroupReader( + HoodieReaderContext<T> readerContext, + String latestCommitTime, + HoodieSchema dataSchema, + HoodieSchema requestedSchema, + Option<InternalSchema> internalSchemaOpt, + HoodieTableMetaClient hoodieTableMetaClient, + TypedProperties props, + Option<HoodieBaseFile> baseFileOption, + Stream<HoodieLogFile> logFiles, + String partitionPath, + Long start, + Long length, + Iterator<? extends HoodieRecord> recordIterator, + Boolean shouldUseRecordPosition, + Boolean allowInflightInstants, + Boolean emitDelete, + Boolean sortOutput, + Option<BaseFileUpdateCallback<T>> fileGroupUpdateCallback, + FileGroupRecordBufferLoader<T> recordBufferLoader) { + + // Validations + ValidationUtils.checkArgument(readerContext != null, "Reader context is required"); + ValidationUtils.checkArgument(hoodieTableMetaClient != null, "Hoodie table meta client is required"); + ValidationUtils.checkArgument(latestCommitTime != null, "Latest commit time is required"); + ValidationUtils.checkArgument(dataSchema != null, "Data schema is required"); + ValidationUtils.checkArgument(requestedSchema != null, "Requested schema is required"); + ValidationUtils.checkArgument(props != null, "Props is required"); + ValidationUtils.checkArgument(partitionPath != null, "Partition path is required"); + + // Handle defaults + if (internalSchemaOpt == null) { + internalSchemaOpt = Option.empty(); + } + if (baseFileOption == null) { + baseFileOption = Option.empty(); + } + if (start == null) { + start = 0L; + } + if (length == null) { + length = Long.MAX_VALUE; + } + if (shouldUseRecordPosition == null) { + shouldUseRecordPosition = false; + } + if (allowInflightInstants == null) { + allowInflightInstants = false; + } + if (emitDelete == null) { + emitDelete = false; + } + if (sortOutput == null) { + sortOutput = false; + } + if (fileGroupUpdateCallback == null) { + fileGroupUpdateCallback = Option.empty(); + } + + // Derive tablePath + String tablePath = hoodieTableMetaClient.getBasePath().toString(); + + // Set the storage with the readerContext's storage configuration + HoodieStorage storage = hoodieTableMetaClient.getStorage().newInstance(new StoragePath(tablePath), readerContext.getStorageConfiguration()); + + // Handle recordBufferLoader default + if (recordBufferLoader == null) { + if (recordIterator != null) { + recordBufferLoader = FileGroupRecordBufferLoader.createStreamingRecordsBufferLoader(); + } else { + recordBufferLoader = FileGroupRecordBufferLoader.createDefault(); + } + } + + // Build composite objects using static helpers + this.readerParameters = ReaderParameters.builder() + .shouldUseRecordPosition(shouldUseRecordPosition) + .emitDeletes(emitDelete) + .sortOutputs(sortOutput) + .inflightInstantsAllowed(allowInflightInstants) + .build(); + this.inputSplit = InputSplit.builder() + .baseFileOption(baseFileOption) + .logFileStream(logFiles) + .recordIterator((Iterator<HoodieRecord>) recordIterator) + .partitionPath(partitionPath) + .start(start) + .length(length) + .build(); + + // Initialize fields this.readerContext = readerContext; this.recordBufferLoader = recordBufferLoader; - this.fileGroupUpdateCallback = updateCallback; + this.fileGroupUpdateCallback = fileGroupUpdateCallback; this.metaClient = hoodieTableMetaClient; this.storage = storage; - this.readerParameters = readerParameters; - this.inputSplit = inputSplit; + readerContext.setHasLogFiles(this.inputSplit.hasLogFiles()); readerContext.getRecordContext().setPartitionPath(inputSplit.getPartitionPath()); if (readerContext.getHasLogFiles() && inputSplit.getStart() != 0) { @@ -112,7 +204,7 @@ public final class HoodieFileGroupReader<T> implements Closeable { readerContext.setTablePath(tablePath); readerContext.setLatestCommitTime(latestCommitTime); boolean isSkipMerge = ConfigUtils.getStringWithAltKeys(props, HoodieReaderConfig.MERGE_TYPE, true).equalsIgnoreCase(HoodieReaderConfig.REALTIME_SKIP_MERGE); - readerContext.setShouldMergeUseRecordPosition(readerParameters.useRecordPosition() && !isSkipMerge && readerContext.getHasLogFiles() && inputSplit.isParquetBaseFile()); + readerContext.setShouldMergeUseRecordPosition(readerParameters.shouldUseRecordPosition() && !isSkipMerge && readerContext.getHasLogFiles() && inputSplit.isParquetBaseFile()); readerContext.setHasBootstrapBaseFile(inputSplit.getBaseFileOption().flatMap(HoodieBaseFile::getBootstrapBaseFile).isPresent()); readerContext.setSchemaHandler(readerContext.getRecordContext().supportsParquetRowIndex() ? new ParquetRowIndexBasedSchemaHandler<>(readerContext, dataSchema, requestedSchema, internalSchemaOpt, props, metaClient) @@ -248,13 +340,6 @@ public final class HoodieFileGroupReader<T> implements Closeable { } } - /** - * @return statistics of reading a file group. - */ - public HoodieReadStats getStats() { - return readStats; - } - /** * @return The next record after calling {@link #hasNext}. */ @@ -266,10 +351,6 @@ public final class HoodieFileGroupReader<T> implements Closeable { return nextVal; } - public List<String> getValidBlockInstants() { - return validBlockInstants; - } - /** * Notifies a write failure with the given record key. */ @@ -355,175 +436,4 @@ public final class HoodieFileGroupReader<T> implements Closeable { } } } - - public static <T> Builder<T> newBuilder() { - return new Builder<>(); - } - - public static class Builder<T> { - private HoodieReaderContext<T> readerContext; - private HoodieStorage storage; - private String tablePath; - private String latestCommitTime; - private HoodieSchema dataSchema; - private HoodieSchema requestedSchema; - private Option<InternalSchema> internalSchemaOpt = Option.empty(); - private HoodieTableMetaClient hoodieTableMetaClient; - private TypedProperties props; - private Option<HoodieBaseFile> baseFileOption; - private Stream<HoodieLogFile> logFiles; - private String partitionPath; - private long start = 0; - private long length = Long.MAX_VALUE; - private Iterator<HoodieRecord> recordIterator; - private boolean shouldUseRecordPosition = false; - private boolean allowInflightInstants = false; - private boolean emitDelete; - private boolean sortOutput = false; - private Option<BaseFileUpdateCallback<T>> fileGroupUpdateCallback = Option.empty(); - private FileGroupRecordBufferLoader<T> recordBufferLoader; - - public Builder<T> withReaderContext(HoodieReaderContext<T> readerContext) { - this.readerContext = readerContext; - return this; - } - - public Builder<T> withLatestCommitTime(String latestCommitTime) { - this.latestCommitTime = latestCommitTime; - return this; - } - - public Builder<T> withFileSlice(FileSlice fileSlice) { - this.baseFileOption = fileSlice.getBaseFile(); - this.logFiles = fileSlice.getLogFiles(); - this.partitionPath = fileSlice.getPartitionPath(); - return this; - } - - public Builder<T> withBaseFileOption(Option<HoodieBaseFile> baseFileOption) { - this.baseFileOption = baseFileOption; - return this; - } - - public Builder<T> withLogFiles(Stream<HoodieLogFile> logFiles) { - this.logFiles = logFiles; - return this; - } - - public Builder<T> withRecordIterator(Iterator<? extends HoodieRecord> recordIterator) { - this.recordIterator = (Iterator<HoodieRecord>) recordIterator; - this.recordBufferLoader = FileGroupRecordBufferLoader.createStreamingRecordsBufferLoader(); - return this; - } - - public Builder<T> withPartitionPath(String partitionPath) { - this.partitionPath = partitionPath; - return this; - } - - public Builder<T> withDataSchema(HoodieSchema dataSchema) { - this.dataSchema = dataSchema; - return this; - } - - public Builder<T> withRequestedSchema(HoodieSchema requestedSchema) { - this.requestedSchema = requestedSchema; - return this; - } - - public Builder<T> withInternalSchema(Option<InternalSchema> internalSchemaOpt) { - this.internalSchemaOpt = internalSchemaOpt; - return this; - } - - public Builder<T> withHoodieTableMetaClient(HoodieTableMetaClient hoodieTableMetaClient) { - this.hoodieTableMetaClient = hoodieTableMetaClient; - this.tablePath = hoodieTableMetaClient.getBasePath().toString(); - return this; - } - - public Builder<T> withProps(TypedProperties props) { - this.props = props; - return this; - } - - public Builder<T> withStart(long start) { - this.start = start; - return this; - } - - public Builder<T> withLength(long length) { - this.length = length; - return this; - } - - public Builder<T> withShouldUseRecordPosition(boolean shouldUseRecordPosition) { - this.shouldUseRecordPosition = shouldUseRecordPosition; - return this; - } - - public Builder<T> withAllowInflightInstants(boolean allowInflightInstants) { - this.allowInflightInstants = allowInflightInstants; - return this; - } - - public Builder<T> withEmitDelete(boolean emitDelete) { - this.emitDelete = emitDelete; - return this; - } - - public Builder<T> withFileGroupUpdateCallback(Option<BaseFileUpdateCallback<T>> fileGroupUpdateCallback) { - this.fileGroupUpdateCallback = fileGroupUpdateCallback; - return this; - } - - /** - * If true, the output of the merge will be sorted instead of appending log records to end of the iterator if they do not have matching keys in the base file. - * This assumes that the base file is already sorted by key. - * @param sortOutput whether to sort the output iterator - * @return this builder instance - */ - public Builder<T> withSortOutput(boolean sortOutput) { - this.sortOutput = sortOutput; - return this; - } - - public Builder<T> withRecordBufferLoader(FileGroupRecordBufferLoader<T> recordBufferLoader) { - this.recordBufferLoader = recordBufferLoader; - return this; - } - - public HoodieFileGroupReader<T> build() { - ValidationUtils.checkArgument(readerContext != null, "Reader context is required"); - ValidationUtils.checkArgument(hoodieTableMetaClient != null, "Hoodie table meta client is required"); - ValidationUtils.checkArgument(tablePath != null, "Table path is required"); - // set the storage with the readerContext's storage configuration - this.storage = hoodieTableMetaClient.getStorage().newInstance(new StoragePath(tablePath), readerContext.getStorageConfiguration()); - - ValidationUtils.checkArgument(storage != null, "Storage is required"); - ValidationUtils.checkArgument(latestCommitTime != null, "Latest commit time is required"); - ValidationUtils.checkArgument(dataSchema != null, "Data schema is required"); - ValidationUtils.checkArgument(requestedSchema != null, "Requested schema is required"); - ValidationUtils.checkArgument(props != null, "Props is required"); - ValidationUtils.checkArgument(baseFileOption != null, "Base file option is required"); - ValidationUtils.checkArgument(partitionPath != null, "Partition path is required"); - - if (recordBufferLoader == null) { - recordBufferLoader = FileGroupRecordBufferLoader.createDefault(); - } - - ReaderParameters readerParameters = ReaderParameters.builder() - .shouldUseRecordPosition(shouldUseRecordPosition) - .emitDeletes(emitDelete) - .sortOutputs(sortOutput) - .allowInflightInstants(allowInflightInstants) - .build(); - InputSplit inputSplit = new InputSplit(baseFileOption, recordIterator != null ? Either.right(recordIterator) : Either.left(logFiles == null ? Stream.empty() : logFiles), - partitionPath, start, length); - return new HoodieFileGroupReader<>( - readerContext, storage, tablePath, latestCommitTime, dataSchema, requestedSchema, internalSchemaOpt, hoodieTableMetaClient, - props, readerParameters, inputSplit, fileGroupUpdateCallback, recordBufferLoader); - } - } - } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/InputSplit.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/InputSplit.java index 6d7b578e15b3..a0f1a3d04e90 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/InputSplit.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/InputSplit.java @@ -24,10 +24,13 @@ import org.apache.hudi.common.model.HoodieFileFormat; import org.apache.hudi.common.model.HoodieLogFile; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.table.cdc.HoodieCDCUtils; -import org.apache.hudi.common.util.Either; import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.ValidationUtils; +import lombok.AccessLevel; +import lombok.Builder; +import lombok.Getter; + import java.util.Collections; import java.util.Iterator; import java.util.List; @@ -38,9 +41,13 @@ import java.util.stream.Stream; * Represents a split of input data for reading, which includes the partition path the data belongs to along with an optional base file and a list of log files. * If there is only a base file, it is possible for the reader to specify a particular range of the file with the start and length parameters. */ +@Getter public class InputSplit { + private final Option<HoodieBaseFile> baseFileOption; + @Getter(AccessLevel.NONE) private final List<HoodieLogFile> logFiles; + @Getter(AccessLevel.NONE) private final Option<Iterator<HoodieRecord>> recordIterator; private final String partitionPath; // Byte offset to start reading from the base file @@ -48,26 +55,41 @@ public class InputSplit { // Length of bytes to read from the base file private final long length; - InputSplit(Option<HoodieBaseFile> baseFileOption, - Either<Stream<HoodieLogFile>, Iterator<HoodieRecord>> recordsToMerge, - String partitionPath, long start, long length) { - this.baseFileOption = baseFileOption; - if (recordsToMerge.isLeft()) { - this.logFiles = recordsToMerge.asLeft().sorted(HoodieLogFile.getLogFileComparator()) + @Builder + private InputSplit( + Option<HoodieBaseFile> baseFileOption, + Stream<HoodieLogFile> logFileStream, + Iterator<HoodieRecord> recordIterator, + String partitionPath, + long start, + long length) { + + // Ensure we do not have both sources of data to merge + // i.e. logFileStream and recordIterator cannot be both non-null + ValidationUtils.checkArgument(!(logFileStream != null && recordIterator != null), + "Cannot provide both logFileStream and recordIterator"); + + this.baseFileOption = Option.ofNullable(baseFileOption).orElse(Option.empty()); + this.partitionPath = partitionPath; + this.start = start; + this.length = length; + + if (logFileStream != null) { + // Process Log Files (if provided) + this.logFiles = logFileStream + .sorted(HoodieLogFile.getLogFileComparator()) .filter(logFile -> !logFile.getFileName().endsWith(HoodieCDCUtils.CDC_LOGFILE_SUFFIX)) .collect(Collectors.toList()); this.recordIterator = Option.empty(); + } else if (recordIterator != null) { + // Process Record Iterator (if provided) + this.logFiles = Collections.emptyList(); + this.recordIterator = Option.of(recordIterator); } else { + // Handle Case with neither (Base file only read) this.logFiles = Collections.emptyList(); - this.recordIterator = Option.of(recordsToMerge.asRight()); + this.recordIterator = Option.empty(); } - this.partitionPath = partitionPath; - this.start = start; - this.length = length; - } - - public Option<HoodieBaseFile> getBaseFileOption() { - return baseFileOption; } public List<HoodieLogFile> getLogFiles() { @@ -79,18 +101,6 @@ public class InputSplit { return !logFiles.isEmpty(); } - public String getPartitionPath() { - return partitionPath; - } - - public long getStart() { - return start; - } - - public long getLength() { - return length; - } - public boolean isParquetBaseFile() { return baseFileOption.map(baseFile -> HoodieFileFormat.fromFileExtension(baseFile.getStoragePath().getFileExtension()) == HoodieFileFormat.PARQUET).orElse(false); } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/ReaderParameters.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/ReaderParameters.java index bfb791fe5fa7..d19ad39418b3 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/ReaderParameters.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/ReaderParameters.java @@ -19,78 +19,38 @@ package org.apache.hudi.common.table.read; +import lombok.AccessLevel; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Getter; + /** * Parameters for how the reader should process the FileGroup while reading. */ +@AllArgsConstructor(access = AccessLevel.PRIVATE) +@Getter +@Builder public class ReaderParameters { + // Rely on the position of the record in the file instead of the record keys while merging data between base and log files - private final boolean useRecordPosition; + @Getter(AccessLevel.NONE) + @Builder.Default + private final boolean shouldUseRecordPosition = false; // Whether to emit delete records while reading - private final boolean emitDelete; + @Builder.Default + private final boolean emitDeletes = false; // Whether to sort the output records while reading, this implicitly requires the base file to be sorted - private final boolean sortOutput; + @Builder.Default + private final boolean sortOutputs = false; // Allows to consider inflight instants while merging log records using HoodieMergedLogRecordReader // The inflight instants need to be considered while updating RLI records. RLI needs to fetch the revived // and deleted keys from the log files written as part of active data commit. During the RLI update, - // the allowInflightInstants flag would need to be set to true. This would ensure the HoodieMergedLogRecordReader + // the inflightInstantsAllowed flag would need to be set to true. This would ensure the HoodieMergedLogRecordReader // considers the log records which are inflight. - private final boolean allowInflightInstants; - - private ReaderParameters(boolean useRecordPosition, boolean emitDelete, boolean sortOutput, boolean allowInflightInstants) { - this.useRecordPosition = useRecordPosition; - this.emitDelete = emitDelete; - this.sortOutput = sortOutput; - this.allowInflightInstants = allowInflightInstants; - } - - public boolean useRecordPosition() { - return useRecordPosition; - } - - public boolean emitDeletes() { - return emitDelete; - } - - public boolean sortOutputs() { - return sortOutput; - } - - public boolean allowInflightInstants() { - return allowInflightInstants; - } - - static Builder builder() { - return new Builder(); - } - - static class Builder { - private boolean shouldUseRecordPosition = false; - private boolean emitDelete = false; - private boolean sortOutput = false; - private boolean allowInflightInstants = false; - - public Builder shouldUseRecordPosition(boolean shouldUseRecordPosition) { - this.shouldUseRecordPosition = shouldUseRecordPosition; - return this; - } - - public Builder emitDeletes(boolean emitDelete) { - this.emitDelete = emitDelete; - return this; - } - - public Builder sortOutputs(boolean sortOutput) { - this.sortOutput = sortOutput; - return this; - } - - public Builder allowInflightInstants(boolean allowInflightInstants) { - this.allowInflightInstants = allowInflightInstants; - return this; - } + @Builder.Default + private final boolean inflightInstantsAllowed = false; - public ReaderParameters build() { - return new ReaderParameters(shouldUseRecordPosition, emitDelete, sortOutput, allowInflightInstants); - } + public boolean shouldUseRecordPosition() { + return shouldUseRecordPosition; } } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/DefaultFileGroupRecordBufferLoader.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/DefaultFileGroupRecordBufferLoader.java index 3e0609003ad3..e84921c080ae 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/DefaultFileGroupRecordBufferLoader.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/DefaultFileGroupRecordBufferLoader.java @@ -62,15 +62,15 @@ class DefaultFileGroupRecordBufferLoader<T> extends LogScanningRecordBufferLoade Option<BaseFileUpdateCallback<T>> fileGroupUpdateCallback) { boolean isSkipMerge = ConfigUtils.getStringWithAltKeys(props, HoodieReaderConfig.MERGE_TYPE, true).equalsIgnoreCase(HoodieReaderConfig.REALTIME_SKIP_MERGE); Option<PartialUpdateMode> partialUpdateModeOpt = hoodieTableMetaClient.getTableConfig().getPartialUpdateMode(); - UpdateProcessor<T> updateProcessor = UpdateProcessor.create(readStats, readerContext, readerParameters.emitDeletes(), fileGroupUpdateCallback, props); + UpdateProcessor<T> updateProcessor = UpdateProcessor.create(readStats, readerContext, readerParameters.isEmitDeletes(), fileGroupUpdateCallback, props); FileGroupRecordBuffer<T> recordBuffer; if (isSkipMerge) { recordBuffer = new UnmergedFileGroupRecordBuffer<>( readerContext, hoodieTableMetaClient, readerContext.getMergeMode(), partialUpdateModeOpt, props, readStats); - } else if (readerParameters.sortOutputs()) { + } else if (readerParameters.isSortOutputs()) { recordBuffer = new SortedKeyBasedFileGroupRecordBuffer<>( readerContext, hoodieTableMetaClient, readerContext.getMergeMode(), partialUpdateModeOpt, props, orderingFieldNames, updateProcessor); - } else if (readerParameters.useRecordPosition() && inputSplit.getBaseFileOption().isPresent()) { + } else if (readerParameters.shouldUseRecordPosition() && inputSplit.getBaseFileOption().isPresent()) { recordBuffer = new PositionBasedFileGroupRecordBuffer<>( readerContext, hoodieTableMetaClient, readerContext.getMergeMode(), partialUpdateModeOpt, inputSplit.getBaseFileOption().get().getCommitTime(), props, orderingFieldNames, updateProcessor); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/LogScanningRecordBufferLoader.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/LogScanningRecordBufferLoader.java index b6638b1ecd04..1c4f6082e4f1 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/LogScanningRecordBufferLoader.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/LogScanningRecordBufferLoader.java @@ -48,7 +48,7 @@ abstract class LogScanningRecordBufferLoader { .withInstantRange(readerContext.getInstantRange()) .withPartition(inputSplit.getPartitionPath()) .withRecordBuffer(recordBuffer) - .withAllowInflightInstants(readerParameters.allowInflightInstants()) + .withAllowInflightInstants(readerParameters.isInflightInstantsAllowed()) .withMetaClient(hoodieTableMetaClient) .build()) { readStats.setTotalLogReadTimeMs(logRecordReader.getTotalTimeTakenToReadAndMergeBlocks()); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/ReusableFileGroupRecordBufferLoader.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/ReusableFileGroupRecordBufferLoader.java index aa3fbd22cf94..bd3b3ec55bb3 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/ReusableFileGroupRecordBufferLoader.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/ReusableFileGroupRecordBufferLoader.java @@ -58,7 +58,7 @@ public class ReusableFileGroupRecordBufferLoader<T> extends LogScanningRecordBuf ReaderParameters readerParameters, HoodieReadStats readStats, Option<BaseFileUpdateCallback<T>> fileGroupUpdateCallback) { - UpdateProcessor<T> updateProcessor = UpdateProcessor.create(readStats, readerContext, readerParameters.emitDeletes(), fileGroupUpdateCallback, props); + UpdateProcessor<T> updateProcessor = UpdateProcessor.create(readStats, readerContext, readerParameters.isEmitDeletes(), fileGroupUpdateCallback, props); Option<PartialUpdateMode> partialUpdateModeOpt = hoodieTableMetaClient.getTableConfig().getPartialUpdateMode(); if (cachedResults == null) { // Create an initial buffer to process the log files diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/StreamingFileGroupRecordBufferLoader.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/StreamingFileGroupRecordBufferLoader.java index 02c1d5b4cb9c..2087b7dceaf8 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/StreamingFileGroupRecordBufferLoader.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/StreamingFileGroupRecordBufferLoader.java @@ -67,9 +67,9 @@ public class StreamingFileGroupRecordBufferLoader<T> implements FileGroupRecordB readerContext.getSchemaHandler().setSchemaForUpdates(recordSchema); HoodieTableConfig tableConfig = hoodieTableMetaClient.getTableConfig(); Option<PartialUpdateMode> partialUpdateModeOpt = tableConfig.getPartialUpdateMode(); - UpdateProcessor<T> updateProcessor = UpdateProcessor.create(readStats, readerContext, readerParameters.emitDeletes(), fileGroupUpdateCallback, props); + UpdateProcessor<T> updateProcessor = UpdateProcessor.create(readStats, readerContext, readerParameters.isEmitDeletes(), fileGroupUpdateCallback, props); FileGroupRecordBuffer<T> recordBuffer; - if (readerParameters.sortOutputs()) { + if (readerParameters.isSortOutputs()) { recordBuffer = new SortedKeyBasedFileGroupRecordBuffer<>( readerContext, hoodieTableMetaClient, readerContext.getMergeMode(), partialUpdateModeOpt, props, orderingFieldNames, updateProcessor); } else { diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java index 1fabe8c97786..70b136f2a646 100644 --- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java +++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java @@ -601,11 +601,13 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata { baseFileReaders, fileGroupReaderProps); - HoodieFileGroupReader<IndexedRecord> fileGroupReader = HoodieFileGroupReader.<IndexedRecord>newBuilder() + HoodieFileGroupReader<IndexedRecord> fileGroupReader = HoodieFileGroupReader.<IndexedRecord>builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(metadataMetaClient) .withLatestCommitTime(latestMetadataInstantTime) - .withFileSlice(fileSlice) + .withBaseFileOption(fileSlice.getBaseFile()) + .withLogFiles(fileSlice.getLogFiles()) + .withPartitionPath(fileSlice.getPartitionPath()) .withDataSchema(SCHEMA) .withRequestedSchema(SCHEMA) .withProps(fileGroupReaderProps) diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java index 59e2d6dcd83f..0ede3ae5015e 100644 --- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java +++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java @@ -1809,7 +1809,7 @@ public class HoodieTableMetadataUtil { properties.setProperty(HoodieReaderConfig.MERGE_TYPE.key(), REALTIME_SKIP_MERGE); // Currently only avro is fully supported for extracting column ranges (see HUDI-8585) HoodieReaderContext readerContext = new HoodieAvroReaderContext(datasetMetaClient.getStorageConf(), datasetMetaClient.getTableConfig(), Option.empty(), Option.empty()); - HoodieFileGroupReader fileGroupReader = HoodieFileGroupReader.newBuilder() + HoodieFileGroupReader fileGroupReader = HoodieFileGroupReader.builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(datasetMetaClient) .withLogFiles(Stream.of(logFile)) @@ -2568,10 +2568,12 @@ public class HoodieTableMetadataUtil { final String partition = partitionAndBaseFile.getKey(); final FileSlice fileSlice = partitionAndBaseFile.getValue(); if (!fileSlice.getBaseFile().isPresent()) { - HoodieFileGroupReader fileGroupReader = HoodieFileGroupReader.<T>newBuilder() + HoodieFileGroupReader fileGroupReader = HoodieFileGroupReader.<T>builder() .withReaderContext(readerContextFactory.getContext()) .withHoodieTableMetaClient(metaClient) - .withFileSlice(fileSlice) + .withBaseFileOption(fileSlice.getBaseFile()) + .withLogFiles(fileSlice.getLogFiles()) + .withPartitionPath(fileSlice.getPartitionPath()) .withDataSchema(tableSchema) .withRequestedSchema(HoodieSchemaUtils.getRecordKeySchema()) .withLatestCommitTime(latestCommitTime) diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderBase.java b/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderBase.java index ffcda26a3796..158ef475d843 100644 --- a/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderBase.java +++ b/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderBase.java @@ -1040,15 +1040,17 @@ public abstract class TestHoodieFileGroupReaderBase<T> { HoodieSchema schema, FileSlice fileSlice, int start, TypedProperties props, boolean sortOutput) { - return HoodieFileGroupReader.<T>newBuilder() + return HoodieFileGroupReader.<T>builder() .withReaderContext(getHoodieReaderContext(tablePath, schema, storageConf, metaClient)) .withHoodieTableMetaClient(metaClient) .withLatestCommitTime(metaClient.getActiveTimeline().lastInstant().get().requestedTime()) - .withFileSlice(fileSlice) + .withBaseFileOption(fileSlice.getBaseFile()) + .withLogFiles(fileSlice.getLogFiles()) + .withPartitionPath(fileSlice.getPartitionPath()) .withDataSchema(schema) .withRequestedSchema(schema) .withProps(props) - .withStart(start) + .withStart((long) start) .withLength(fileSlice.getTotalFileSize()) .withShouldUseRecordPosition(false) .withAllowInflightInstants(false) diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/BaseTestFileGroupRecordBuffer.java b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/BaseTestFileGroupRecordBuffer.java index e6ee09b3e6ad..b2869df3310d 100644 --- a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/BaseTestFileGroupRecordBuffer.java +++ b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/BaseTestFileGroupRecordBuffer.java @@ -149,7 +149,7 @@ public class BaseTestFileGroupRecordBuffer { when(inputSplit.hasNoRecordsToMerge()).thenReturn(false); when(inputSplit.getRecordIterator()).thenReturn(fileGroupRecordBufferItrOpt.get()); ReaderParameters readerParameters = mock(ReaderParameters.class); - when(readerParameters.sortOutputs()).thenReturn(false); + when(readerParameters.isSortOutputs()).thenReturn(false); return (KeyBasedFileGroupRecordBuffer<IndexedRecord>) recordBufferLoader.getRecordBuffer(readerContext, mockMetaClient.getStorage(), inputSplit, orderingFieldNames, mockMetaClient, props, readerParameters, readStats, Option.empty()).getKey(); } diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestFileGroupRecordBufferLoader.java b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestFileGroupRecordBufferLoader.java index 51dad1c752af..e1c810c56910 100644 --- a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestFileGroupRecordBufferLoader.java +++ b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestFileGroupRecordBufferLoader.java @@ -90,12 +90,12 @@ public class TestFileGroupRecordBufferLoader extends BaseTestFileGroupRecordBuff } ReaderParameters readerParameters = mock(ReaderParameters.class); if (fileGroupRecordBufferType.contains("Sorted")) { - when(readerParameters.sortOutputs()).thenReturn(true); + when(readerParameters.isSortOutputs()).thenReturn(true); } if (fileGroupRecordBufferType.contains("Position")) { HoodieBaseFile baseFile = mock(HoodieBaseFile.class); when(inputSplit.getBaseFileOption()).thenReturn(Option.of(baseFile)); - when(readerParameters.useRecordPosition()).thenReturn(true); + when(readerParameters.shouldUseRecordPosition()).thenReturn(true); } Option<BaseFileUpdateCallback> fileGroupUpdateCallback = Option.empty(); diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java index 0d43c1d463ca..611128a96309 100644 --- a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java +++ b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java @@ -141,7 +141,7 @@ class TestSortedKeyBasedFileGroupRecordBuffer extends BaseTestFileGroupRecordBuf when(inputSplit.hasNoRecordsToMerge()).thenReturn(false); when(inputSplit.getRecordIterator()).thenReturn(inputRecords.iterator()); ReaderParameters readerParameters = mock(ReaderParameters.class); - when(readerParameters.sortOutputs()).thenReturn(true); + when(readerParameters.isSortOutputs()).thenReturn(true); SortedKeyBasedFileGroupRecordBuffer fileGroupRecordBuffer = (SortedKeyBasedFileGroupRecordBuffer<IndexedRecord>) recordBufferLoader .getRecordBuffer(readerContext, mockMetaClient.getStorage(), inputSplit, Collections.singletonList("ts"), mockMetaClient, properties, readerParameters, readStats, Option.empty()).getKey(); diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java index d62763ef64af..0bd7f53b611f 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java @@ -132,14 +132,16 @@ public class FormatUtils { final TypedProperties typedProps = FlinkClientUtil.getReadProps(metaClient.getTableConfig(), writeConfig); typedProps.put(HoodieReaderConfig.MERGE_TYPE.key(), mergeType); - return HoodieFileGroupReader.<RowData>newBuilder() + return HoodieFileGroupReader.<RowData>builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(metaClient) .withLatestCommitTime(latestInstant) - .withFileSlice(fileSlice) + .withBaseFileOption(fileSlice.getBaseFile()) + .withLogFiles(fileSlice.getLogFiles()) + .withPartitionPath(fileSlice.getPartitionPath()) .withDataSchema(tableSchema) .withRequestedSchema(requiredSchema) - .withInternalSchema(Option.ofNullable(internalSchemaManager.getQuerySchema())) + .withInternalSchemaOpt(Option.ofNullable(internalSchemaManager.getQuerySchema())) .withProps(typedProps) .withShouldUseRecordPosition(false) .withEmitDelete(emitDelete) diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/reader/HoodieFileGroupReaderTestHarness.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/reader/HoodieFileGroupReaderTestHarness.java index b7c63d3dd81c..9259d75adb24 100644 --- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/reader/HoodieFileGroupReaderTestHarness.java +++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/reader/HoodieFileGroupReaderTestHarness.java @@ -137,11 +137,14 @@ public class HoodieFileGroupReaderTestHarness extends HoodieCommonTestHarness { properties.setProperty(HoodieMemoryConfig.SPILLABLE_MAP_BASE_PATH.key(), basePath + "/" + HoodieTableMetaClient.TEMPFOLDER_NAME); properties.setProperty(HoodieCommonConfig.SPILLABLE_DISK_MAP_TYPE.key(), ExternalSpillableMap.DiskMapType.ROCKS_DB.name()); properties.setProperty(HoodieCommonConfig.DISK_MAP_BITCASK_COMPRESSION_ENABLED.key(), "false"); - HoodieFileGroupReader<IndexedRecord> fileGroupReader = HoodieFileGroupReader.<IndexedRecord>newBuilder() + FileSlice fileSlice = fileSliceOpt.orElseThrow(() -> new IllegalArgumentException("FileSlice is not present")); + HoodieFileGroupReader<IndexedRecord> fileGroupReader = HoodieFileGroupReader.<IndexedRecord>builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(metaClient) .withLatestCommitTime("1000") // Not used internally. - .withFileSlice(fileSliceOpt.orElseThrow(() -> new IllegalArgumentException("FileSlice is not present"))) + .withBaseFileOption(fileSlice.getBaseFile()) + .withLogFiles(fileSlice.getLogFiles()) + .withPartitionPath(fileSlice.getPartitionPath()) .withDataSchema(HOODIE_SCHEMA) .withRequestedSchema(HOODIE_SCHEMA) .withProps(properties) diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieFileGroupReaderBasedRecordReader.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieFileGroupReaderBasedRecordReader.java index 1fbb6ad4a982..22d5ba701711 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieFileGroupReaderBasedRecordReader.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieFileGroupReaderBasedRecordReader.java @@ -145,11 +145,13 @@ public class HoodieFileGroupReaderBasedRecordReader implements RecordReader<Null LOG.debug("Creating HoodieFileGroupReaderRecordReader with tableBasePath={}, latestCommitTime={}, fileSplit={}", tableBasePath, latestCommitTime, fileSplit.getPath()); FileSlice fileSlice = getFileSliceFromSplit(fileSplit, getFs(tableBasePath, jobConfCopy), tableBasePath); this.containsBaseFile = fileSlice.getBaseFile().isPresent(); - this.recordIterator = HoodieFileGroupReader.<ArrayWritable>newBuilder() + this.recordIterator = HoodieFileGroupReader.<ArrayWritable>builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(metaClient) .withLatestCommitTime(latestCommitTime) - .withFileSlice(fileSlice) + .withBaseFileOption(fileSlice.getBaseFile()) + .withLogFiles(fileSlice.getLogFiles()) + .withPartitionPath(fileSlice.getPartitionPath()) .withDataSchema(tableSchema) .withRequestedSchema(requestedSchema) .withProps(props) diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala index 3d9b908c4d31..eadbdd80bd31 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala @@ -171,7 +171,7 @@ class HoodieMergeOnReadRDDV2(@transient sc: SparkContext, val requestedSchema = requiredSchema.schema val instantRange = InstantRange.builder().rangeType(RangeType.EXACT_MATCH).explicitInstants(validInstants.value).build() val readerContext = new HoodieAvroReaderContext(storageConf, metaClient.getTableConfig, HOption.of(instantRange), HOption.empty().asInstanceOf[HOption[HPredicate]]) - val fileGroupReader: HoodieFileGroupReader[IndexedRecord] = HoodieFileGroupReader.newBuilder() + val fileGroupReader: HoodieFileGroupReader[IndexedRecord] = HoodieFileGroupReader.builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(metaClient) .withLatestCommitTime(tableState.latestCommitTimestamp.orNull) @@ -181,13 +181,13 @@ class HoodieMergeOnReadRDDV2(@transient sc: SparkContext, .withProps(properties) .withDataSchema(tableSchema.schema) .withRequestedSchema(requestedSchema) - .withInternalSchema(HOption.ofNullable(tableSchema.internalSchema.orNull)) + .withInternalSchemaOpt(HOption.ofNullable(tableSchema.internalSchema.orNull)) .build() convertAvroToRowIterator(fileGroupReader.getClosableIterator, requestedSchema) } else { val readerContext = new SparkFileFormatInternalRowReaderContext(fileGroupBaseFileReader.value, optionalFilters, Seq.empty, storageConf, metaClient.getTableConfig) - val fileGroupReader = HoodieFileGroupReader.newBuilder() + val fileGroupReader = HoodieFileGroupReader.builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(metaClient) .withLatestCommitTime(tableState.latestCommitTimestamp.orNull) @@ -197,7 +197,7 @@ class HoodieMergeOnReadRDDV2(@transient sc: SparkContext, .withProps(properties) .withDataSchema(tableSchema.schema) .withRequestedSchema(requiredSchema.schema) - .withInternalSchema(HOption.ofNullable(tableSchema.internalSchema.orNull)) + .withInternalSchemaOpt(HOption.ofNullable(tableSchema.internalSchema.orNull)) .build() convertCloseableIterator(fileGroupReader.getClosableIterator) } diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala index 7a97367bb3c5..4970be3c249b 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala @@ -513,13 +513,15 @@ class CDCFileGroupIterator(split: HoodieCDCFileGroupSplit, } private def loadFileSlice(fileSlice: FileSlice, readerContext: SparkFileFormatInternalRowReaderContext): Iterator[BufferedRecord[InternalRow]] = { - val fileGroupReader = HoodieFileGroupReader.newBuilder() + val fileGroupReader = HoodieFileGroupReader.builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(metaClient) - .withFileSlice(fileSlice) + .withBaseFileOption(fileSlice.getBaseFile) + .withLogFiles(fileSlice.getLogFiles) + .withPartitionPath(fileSlice.getPartitionPath) .withDataSchema(schema) .withRequestedSchema(schema) - .withInternalSchema(toJavaOption(originTableSchema.internalSchema)) + .withInternalSchemaOpt(toJavaOption(originTableSchema.internalSchema)) .withProps(readerProperties) .withLatestCommitTime(split.changes.last.getInstant) .build() diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala index de4ffb400d4c..da5bef9f785c 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala @@ -313,14 +313,16 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath: String, } else { 0 } - val reader = HoodieFileGroupReader.newBuilder() + val reader = HoodieFileGroupReader.builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(metaClient) .withLatestCommitTime(queryTimestamp) - .withFileSlice(fileSlice) + .withBaseFileOption(fileSlice.getBaseFile) + .withLogFiles(fileSlice.getLogFiles) + .withPartitionPath(fileSlice.getPartitionPath) .withDataSchema(dataSchema) .withRequestedSchema(requestedSchema) - .withInternalSchema(internalSchemaOpt) + .withInternalSchemaOpt(internalSchemaOpt) .withProps(props) .withStart(file.start) .withLength(baseFileLength) diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/PartitionBucketIndexManager.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/PartitionBucketIndexManager.scala index 6eaade123a64..4e044ecb6786 100644 --- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/PartitionBucketIndexManager.scala +++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/PartitionBucketIndexManager.scala @@ -225,14 +225,16 @@ class PartitionBucketIndexManager extends BaseProcedure // instantiate other supporting cast val internalSchemaOption: Option[InternalSchema] = Option.empty() // instantiate FG reader - val fileGroupReader = HoodieFileGroupReader.newBuilder() + val fileGroupReader = HoodieFileGroupReader.builder() .withReaderContext(readerContextFactory.getContext) .withHoodieTableMetaClient(metaClient) .withLatestCommitTime(latestInstantTime.requestedTime()) - .withFileSlice(fileSlice) + .withBaseFileOption(fileSlice.getBaseFile) + .withLogFiles(fileSlice.getLogFiles) + .withPartitionPath(fileSlice.getPartitionPath) .withDataSchema(tableSchemaWithMetaFields) .withRequestedSchema(tableSchemaWithMetaFields) - .withInternalSchema(internalSchemaOption) // not support evolution of schema for now + .withInternalSchemaOpt(internalSchemaOption) // not support evolution of schema for now .withProps(metaClient.getTableConfig.getProps) .withShouldUseRecordPosition(false) .build() diff --git a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestMetadataUtilRLIandSIRecordGeneration.java b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestMetadataUtilRLIandSIRecordGeneration.java index b35da0450cfd..7dbd7e7aaf96 100644 --- a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestMetadataUtilRLIandSIRecordGeneration.java +++ b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestMetadataUtilRLIandSIRecordGeneration.java @@ -712,7 +712,7 @@ public class TestMetadataUtilRLIandSIRecordGeneration extends HoodieClientTestBa TypedProperties properties = new TypedProperties(); // configure un-merged log file reader HoodieReaderContext readerContext = context.getReaderContextFactory(metaClient).getContext(); - HoodieFileGroupReader reader = HoodieFileGroupReader.newBuilder() + HoodieFileGroupReader reader = HoodieFileGroupReader.builder() .withReaderContext(readerContext) .withDataSchema(writerSchemaOpt.get()) .withRequestedSchema(writerSchemaOpt.get()) diff --git a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHoodieBackedMetadata.java b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHoodieBackedMetadata.java index b3beca5fa18c..b0a4aab2e789 100644 --- a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHoodieBackedMetadata.java +++ b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHoodieBackedMetadata.java @@ -1664,7 +1664,7 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase { String latestCommitTimestamp) { HoodieSchema schema = HoodieSchemaUtils.addMetadataFields(HoodieSchema.fromAvroSchema(HoodieMetadataRecord.getClassSchema())); HoodieAvroReaderContext readerContext = new HoodieAvroReaderContext(metadataMetaClient.getStorageConf(), metadataMetaClient.getTableConfig(), Option.empty(), Option.empty()); - HoodieFileGroupReader<IndexedRecord> fileGroupReader = HoodieFileGroupReader.<IndexedRecord>newBuilder() + HoodieFileGroupReader<IndexedRecord> fileGroupReader = HoodieFileGroupReader.<IndexedRecord>builder() .withReaderContext(readerContext) .withHoodieTableMetaClient(metadataMetaClient) .withLogFiles(logFiles.stream())
