danny0405 commented on code in PR #9819:
URL: https://github.com/apache/hudi/pull/9819#discussion_r1359178414
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordReader.java:
##########
@@ -204,75 +202,19 @@ public static Builder newBuilder() {
return new Builder();
}
- @Override
- public void processNextRecord(T record, Map<String, Object> metadata) throws
IOException {
- String key = (String)
metadata.get(HoodieReaderContext.INTERNAL_META_RECORD_KEY);
- Pair<Option<T>, Map<String, Object>> existingRecordMetadataPair =
records.get(key);
-
- if (existingRecordMetadataPair != null) {
- // Merge and store the combined record
- // Note that the incoming `record` is from an older commit, so it should
be put as
- // the `older` in the merge API
- HoodieRecord<T> combinedRecord = (HoodieRecord<T>) recordMerger.merge(
- readerContext.constructHoodieRecord(Option.of(record), metadata,
readerSchema), readerSchema,
-
readerContext.constructHoodieRecord(existingRecordMetadataPair.getLeft(),
existingRecordMetadataPair.getRight(), readerSchema),
- readerSchema, this.getPayloadProps()).get().getLeft();
- // If pre-combine returns existing record, no need to update it
- if (combinedRecord.getData() !=
existingRecordMetadataPair.getLeft().get()) {
- records.put(key,
Pair.of(Option.ofNullable(readerContext.seal(combinedRecord.getData())),
metadata));
- }
- } else {
- // Put the record as is
- // NOTE: Record have to be cloned here to make sure if it holds
low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we
get a clean copy of
- // it since these records will be put into records(Map).
- records.put(key, Pair.of(Option.of(readerContext.seal(record)),
metadata));
- }
- }
-
- @Override
- protected void processNextDeletedRecord(DeleteRecord deleteRecord) {
- String key = deleteRecord.getRecordKey();
- Pair<Option<T>, Map<String, Object>> existingRecordMetadataPair =
records.get(key);
- if (existingRecordMetadataPair != null) {
- // Merge and store the merged record. The ordering val is taken to
decide whether the same key record
- // should be deleted or be kept. The old record is kept only if the
DELETE record has smaller ordering val.
- // For same ordering values, uses the natural order(arrival time
semantics).
-
- Comparable existingOrderingVal = readerContext.getOrderingValue(
- existingRecordMetadataPair.getLeft(),
existingRecordMetadataPair.getRight(), readerSchema,
- this.hoodieTableMetaClient.getTableConfig().getProps());
- Comparable deleteOrderingVal = deleteRecord.getOrderingValue();
- // Checks the ordering value does not equal to 0
- // because we use 0 as the default value which means natural order
- boolean chooseExisting = !deleteOrderingVal.equals(0)
- && ReflectionUtils.isSameClass(existingOrderingVal,
deleteOrderingVal)
- && existingOrderingVal.compareTo(deleteOrderingVal) > 0;
- if (chooseExisting) {
- // The DELETE message is obsolete if the old message has greater
orderingVal.
- return;
- }
- }
- // Put the DELETE record
- records.put(key, Pair.of(Option.empty(),
- readerContext.generateMetadataForRecord(key,
deleteRecord.getPartitionPath(), deleteRecord.getOrderingValue())));
- }
-
public long getTotalTimeTakenToReadAndMergeBlocks() {
return totalTimeTakenToReadAndMergeBlocks;
}
@Override
public void close() {
- if (records != null) {
- records.clear();
- }
+ // No op.
Review Comment:
maybe release the resource for record buffer?
--
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]
