linliu-code commented on code in PR #13580:
URL: https://github.com/apache/hudi/pull/13580#discussion_r2225382683
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java:
##########
@@ -67,55 +75,72 @@
/**
* A merge handle implementation based on the {@link HoodieFileGroupReader}.
* <p>
- * This merge handle is used for compaction, which passes a file slice from the
- * compaction operation of a single file group to a file group reader, get an
iterator of
- * the records, and writes the records to a new base file.
+ * This merge handle is used for
+ * 1. compaction, which passes a file slice from the compaction operation of a
single file group
+ * to a file group reader, get an iterator of the records, and writes the
records to a new base file.
+ * 2. cow write, which takes an iterator of hoodie records, and a base file,
and merge them,
+ * and write to a new base file.
*/
@NotThreadSafe
public class FileGroupReaderBasedMergeHandle<T, I, K, O> extends
HoodieWriteMergeHandle<T, I, K, O> {
private static final Logger LOG =
LoggerFactory.getLogger(FileGroupReaderBasedMergeHandle.class);
-
private final HoodieReaderContext<T> readerContext;
- private final CompactionOperation operation;
private final String maxInstantTime;
- private HoodieReadStats readStats;
private final HoodieRecord.HoodieRecordType recordType;
private final Option<HoodieCDCLogger> cdcLogger;
+ private final Option<CompactionOperation> operation;
+ private final Option<Iterator<HoodieRecord<T>>> recordIterator;
+ private HoodieReadStats readStats;
+ /**
+ * For compactor.
+ */
public FileGroupReaderBasedMergeHandle(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
CompactionOperation operation,
TaskContextSupplier taskContextSupplier,
+ Option<BaseKeyGenerator>
keyGeneratorOpt,
HoodieReaderContext<T> readerContext,
String maxInstantTime,
HoodieRecord.HoodieRecordType
enginRecordType) {
- super(config, instantTime, operation.getPartitionPath(),
operation.getFileId(), hoodieTable, taskContextSupplier);
+ this(config, instantTime, hoodieTable, null, operation.getPartitionPath(),
operation.getFileId(),
+ taskContextSupplier, keyGeneratorOpt, readerContext, maxInstantTime,
enginRecordType, Option.of(operation));
+ }
+
+ /**
+ * For generic FG reader based merge handle.
+ */
+ public FileGroupReaderBasedMergeHandle(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ Iterator<HoodieRecord<T>> recordItr,
String partitionPath, String fileId,
+ TaskContextSupplier
taskContextSupplier, Option<BaseKeyGenerator> keyGeneratorOpt,
+ HoodieReaderContext<T> readerContext,
String maxInstantTime,
+ HoodieRecord.HoodieRecordType
enginRecordType, Option<CompactionOperation> operation) {
+ super(config, instantTime, hoodieTable, recordItr, partitionPath, fileId,
taskContextSupplier, keyGeneratorOpt);
+ this.recordIterator = Option.ofNullable(recordItr);
+ this.operation = operation;
+
+ // Common attributes.
this.maxInstantTime = maxInstantTime;
this.keyToNewRecords = Collections.emptyMap();
this.readerContext = readerContext;
- this.operation = operation;
- // If the table is a metadata table or the base file is an HFile, we use
AVRO record type, otherwise we use the engine record type.
- this.recordType = hoodieTable.isMetadataTable() ||
HFILE.getFileExtension().equals(hoodieTable.getBaseFileExtension()) ?
HoodieRecord.HoodieRecordType.AVRO : enginRecordType;
- if (hoodieTable.getMetaClient().getTableConfig().isCDCEnabled()) {
- this.cdcLogger = Option.of(new HoodieCDCLogger(
- instantTime,
- config,
- hoodieTable.getMetaClient().getTableConfig(),
- partitionPath,
- storage,
- getWriterSchema(),
- createLogWriter(instantTime, HoodieCDCUtils.CDC_LOGFILE_SUFFIX,
Option.empty()),
- IOUtils.getMaxMemoryPerPartitionMerge(taskContextSupplier, config)));
- } else {
- this.cdcLogger = Option.empty();
- }
+ this.recordType = hoodieTable.isMetadataTable() ||
HFILE.getFileExtension().equals(hoodieTable.getBaseFileExtension())
+ ? HoodieRecord.HoodieRecordType.AVRO : enginRecordType;
+ this.cdcLogger =
hoodieTable.getMetaClient().getTableConfig().isCDCEnabled()
+ ? Option.of(new HoodieCDCLogger(instantTime, config,
hoodieTable.getMetaClient().getTableConfig(),
+ partitionPath, storage, getWriterSchema(),
+ createLogWriter(instantTime, HoodieCDCUtils.CDC_LOGFILE_SUFFIX,
Option.empty()),
+ IOUtils.getMaxMemoryPerPartitionMerge(taskContextSupplier,
config)))
+ : Option.empty();
init(operation, this.partitionPath);
}
- private void init(CompactionOperation operation, String partitionPath) {
+ private void init(Option<CompactionOperation> operation, String
partitionPath) {
LOG.info("partitionPath:{}, fileId to be merged:{}", partitionPath,
fileId);
- this.baseFileToMerge = operation.getBaseFile(config.getBasePath(),
operation.getPartitionPath()).orElse(null);
+
Review Comment:
from the first `HoodieWriteMergeHandle` constructor.
--
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]
