danny0405 commented on code in PR #13411:
URL: https://github.com/apache/hudi/pull/13411#discussion_r2157859992
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java:
##########
@@ -139,23 +137,65 @@ private void init(CompactionOperation operation, String
partitionPath, Option<Ho
// Create the writer for writing the new version file
fileWriter = HoodieFileWriterFactory.getFileWriter(instantTime,
newFilePath, hoodieTable.getStorage(),
- config, writeSchemaWithMetaFields, taskContextSupplier,
getRecordType());
+ config, writeSchemaWithMetaFields, taskContextSupplier, recordType);
} catch (IOException io) {
- LOG.error("Error in update task at commit " + instantTime, io);
+ LOG.error("Error in update task at commit {}", instantTime, io);
writeStatus.setGlobalError(io);
throw new HoodieUpsertException("Failed to initialize HoodieUpdateHandle
for FileId: " + fileId + " on commit "
+ instantTime + " on path " +
hoodieTable.getMetaClient().getBasePath(), io);
}
}
- protected abstract HoodieRecord.HoodieRecordType getRecordType();
-
/**
* Reads the file slice of a compaction operation using a file group reader,
- * by getting an iterator of the records; then writes the records to a new
base file
- * using parquet writer.
+ * by getting an iterator of the records; then writes the records to a new
base file.
*/
- protected abstract void write();
+ public void write() {
+ boolean usePosition =
config.getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
+ Option<InternalSchema> internalSchemaOption =
SerDeHelper.fromJson(config.getInternalSchema());
+ TypedProperties props = TypedProperties.copy(config.getProps());
Review Comment:
I see it here in `HoodieFileGroupReader` constructor:
```java
this.orderingFieldName = recordMergeMode ==
RecordMergeMode.COMMIT_TIME_ORDERING
? Option.empty()
: Option.ofNullable(ConfigUtils.getOrderingField(props))
.or(() -> {
String preCombineField =
hoodieTableMetaClient.getTableConfig().getPreCombineField();
if (StringUtils.isNullOrEmpty(preCombineField)) {
return Option.empty();
}
return Option.of(preCombineField);
});
```
So it only affects the old payload and also the merger based merging.
--
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]
