danny0405 commented on code in PR #13679:
URL: https://github.com/apache/hudi/pull/13679#discussion_r2255756968
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java:
##########
@@ -171,24 +280,18 @@ private void init(CompactionOperation operation, String
partitionPath) {
public void doMerge() {
boolean usePosition =
config.getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
Option<InternalSchema> internalSchemaOption =
SerDeHelper.fromJson(config.getInternalSchema());
- TypedProperties props = TypedProperties.copy(config.getProps());
long maxMemoryPerCompaction =
IOUtils.getMaxMemoryPerCompaction(taskContextSupplier, config);
props.put(HoodieMemoryConfig.MAX_MEMORY_FOR_MERGE.key(),
String.valueOf(maxMemoryPerCompaction));
- Stream<HoodieLogFile> logFiles =
operation.getDeltaFileNames().stream().map(logFileName ->
+ Option<Stream<HoodieLogFile>> logFilesStreamOpt = operation.map(op ->
op.getDeltaFileNames().stream().map(logFileName ->
new HoodieLogFile(new StoragePath(FSUtils.constructAbsolutePath(
- config.getBasePath(), operation.getPartitionPath()),
logFileName)));
+ config.getBasePath(), op.getPartitionPath()), logFileName))));
// Initializes file group reader
- try (HoodieFileGroupReader<T> fileGroupReader =
HoodieFileGroupReader.<T>newBuilder().withReaderContext(readerContext).withHoodieTableMetaClient(hoodieTable.getMetaClient())
-
.withLatestCommitTime(maxInstantTime).withPartitionPath(partitionPath).withBaseFileOption(Option.ofNullable(baseFileToMerge)).withLogFiles(logFiles)
-
.withDataSchema(writeSchemaWithMetaFields).withRequestedSchema(writeSchemaWithMetaFields).withInternalSchema(internalSchemaOption).withProps(props)
-
.withShouldUseRecordPosition(usePosition).withSortOutput(hoodieTable.requireSortedRecords())
- .withFileGroupUpdateCallback(cdcLogger.map(logger -> new
CDCCallback(logger,
readerContext))).withEnableOptimizedLogBlockScan(config.enableOptimizedLogBlocksScan()).build())
{
+ try (HoodieFileGroupReader<T> fileGroupReader =
getFileGroupReader(usePosition, getNewSchema(), internalSchemaOption, props,
isCompaction, logFilesStreamOpt, incomingRecordsItr)) {
// Reads the records from the file slice
try (ClosableIterator<HoodieRecord<T>> recordIterator =
fileGroupReader.getClosableHoodieRecordIterator()) {
while (recordIterator.hasNext()) {
HoodieRecord<T> record = recordIterator.next();
- record.setCurrentLocation(newRecordLocation);
- record.setNewLocation(newRecordLocation);
+ setRecordLocations(record);
Review Comment:
the deletes are ignored by the reader by default, we might need to enable it
when RLI is enabled to delete the deleted keys.
--
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]
