yihua commented on code in PR #12843:
URL: https://github.com/apache/hudi/pull/12843#discussion_r1999268772
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/KeyBasedFileGroupRecordBuffer.java:
##########
@@ -123,6 +132,18 @@ public void processNextDeletedRecord(DeleteRecord
deleteRecord, Serializable rec
}
}
+ protected void processCustomDeleteRecord(T record, Map<String, Object>
metadata) {
+ DeleteRecord deleteRecord = DeleteRecord.create(
+ new HoodieKey(
+ (String) metadata.get(INTERNAL_META_RECORD_KEY),
+ (String) metadata.get(INTERNAL_META_PARTITION_PATH)),
Review Comment:
It looks like `INTERNAL_META_PARTITION_PATH` is not filled by
`readerContext.generateMetadataForRecord(nextRecord, schema)` and now this
logic uses it. We should fix that.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/PositionBasedFileGroupRecordBuffer.java:
##########
@@ -135,13 +137,13 @@ public void processDataBlock(HoodieDataBlock dataBlock,
Option<KeySpec> keySpecO
}
long recordPosition = recordPositions.get(recordIndex++);
-
T evolvedNextRecord =
schemaTransformerWithEvolvedSchema.getLeft().apply(nextRecord);
- processNextDataRecord(
- evolvedNextRecord,
- readerContext.generateMetadataForRecord(evolvedNextRecord, schema),
- recordPosition
- );
+ Map<String, Object> metadata =
readerContext.generateMetadataForRecord(evolvedNextRecord, schema);
+ if (shouldCheckCustomDeleteMarker &&
isCustomDeleteRecord(evolvedNextRecord)) {
+ processCustomDeleteRecord(evolvedNextRecord, metadata,
recordPosition);
Review Comment:
Also, check CUSTOM merge to make sure deletes are handled properly in the
new flow.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/PositionBasedFileGroupRecordBuffer.java:
##########
@@ -202,6 +204,16 @@ public void processDeleteBlock(HoodieDeleteBlock
deleteBlock) throws IOException
}
}
+ protected void processCustomDeleteRecord(T record, Map<String, Object>
metadata, long recordPosition) {
+ DeleteRecord deleteRecord = DeleteRecord.create(
+ new HoodieKey(
+ (String) metadata.get(INTERNAL_META_RECORD_KEY),
+ (String) metadata.get(INTERNAL_META_PARTITION_PATH)),
Review Comment:
Similar here to revisit the partition path fetching.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/PositionBasedFileGroupRecordBuffer.java:
##########
@@ -135,13 +137,13 @@ public void processDataBlock(HoodieDataBlock dataBlock,
Option<KeySpec> keySpecO
}
long recordPosition = recordPositions.get(recordIndex++);
-
T evolvedNextRecord =
schemaTransformerWithEvolvedSchema.getLeft().apply(nextRecord);
- processNextDataRecord(
- evolvedNextRecord,
- readerContext.generateMetadataForRecord(evolvedNextRecord, schema),
- recordPosition
- );
+ Map<String, Object> metadata =
readerContext.generateMetadataForRecord(evolvedNextRecord, schema);
+ if (shouldCheckCustomDeleteMarker &&
isCustomDeleteRecord(evolvedNextRecord)) {
+ processCustomDeleteRecord(evolvedNextRecord, metadata,
recordPosition);
Review Comment:
Since all deletes now go through `doProcessNextDeletedRecord`, could you
simplify `doProcessNextDataRecord` to only consider non-delete records?
--
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]
