Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.Map;
import java.util.Set;

Expand Down Expand Up @@ -117,16 +118,14 @@ public HoodieMergeHandle(HoodieWriteConfig config, String instantTime, HoodieTab
Iterator<HoodieRecord<T>> recordItr, String partitionPath, String fileId,
TaskContextSupplier taskContextSupplier, Option<BaseKeyGenerator> keyGeneratorOpt) {
this(config, instantTime, hoodieTable, recordItr, partitionPath, fileId, taskContextSupplier,
hoodieTable.getBaseFileOnlyView().getLatestBaseFile(partitionPath, fileId).get(), keyGeneratorOpt);
getLatestBaseFile(hoodieTable, partitionPath, fileId), keyGeneratorOpt);
}

public HoodieMergeHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
Iterator<HoodieRecord<T>> recordItr, String partitionPath, String fileId,
TaskContextSupplier taskContextSupplier, HoodieBaseFile baseFile, Option<BaseKeyGenerator> keyGeneratorOpt) {
super(config, instantTime, partitionPath, fileId, hoodieTable, taskContextSupplier);
init(fileId, recordItr);
init(fileId, partitionPath, baseFile);
validateAndSetAndKeyGenProps(keyGeneratorOpt, config.populateMetaFields());
init(recordItr, baseFile, keyGeneratorOpt);
}

/**
Expand All @@ -139,15 +138,30 @@ public HoodieMergeHandle(HoodieWriteConfig config, String instantTime, HoodieTab
this.keyToNewRecords = keyToNewRecords;
this.useWriterSchemaForCompaction = true;
this.preserveMetadata = config.isPreserveHoodieCommitMetadataForCompaction();
init(fileId, this.partitionPath, dataFileToBeMerged);
validateAndSetAndKeyGenProps(keyGeneratorOpt, config.populateMetaFields());
init(null, dataFileToBeMerged, keyGeneratorOpt);
}

private void validateAndSetAndKeyGenProps(Option<BaseKeyGenerator> keyGeneratorOpt, boolean populateMetaFields) {
ValidationUtils.checkArgument(populateMetaFields == !keyGeneratorOpt.isPresent());
this.keyGeneratorOpt = keyGeneratorOpt;
}

private void init(Iterator<HoodieRecord<T>> recordItr, HoodieBaseFile baseFile, Option<BaseKeyGenerator> keyGeneratorOpt) {
if (recordItr != null) {
Copy link
Contributor

@danny0405 danny0405 Jun 14, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we revert this method init ? I don't think it makes much sense to wrap as a new method here. The two #init methods do totally different things here.

init(this.fileId, recordItr);
}
init(this.fileId, this.partitionPath, baseFile);
validateAndSetAndKeyGenProps(keyGeneratorOpt, this.config.populateMetaFields());
}

public static HoodieBaseFile getLatestBaseFile(HoodieTable<?, ?, ?, ?> hoodieTable, String partitionPath, String fileId) {
Option<HoodieBaseFile> baseFileOp = hoodieTable.getBaseFileOnlyView().getLatestBaseFile(partitionPath, fileId);
if (!baseFileOp.isPresent()) {
throw new NoSuchElementException(String.format("FileID %s of partition path %s does not exist.", fileId, partitionPath));
}
return baseFileOp.get();
}

@Override
public Schema getWriterSchemaWithMetaFields() {
return writeSchemaWithMetaFields;
Expand Down