yihua commented on code in PR #9593:
URL: https://github.com/apache/hudi/pull/9593#discussion_r1334697266
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java:
##########
@@ -270,14 +272,46 @@ protected boolean writeUpdateRecord(HoodieRecord<T>
newRecord, HoodieRecord<T> o
if (combineRecordOpt.isPresent()) {
if (oldRecord.getData() != combineRecordOpt.get().getData()) {
// the incoming record is chosen
- isDelete = HoodieOperation.isDelete(newRecord.getOperation());
+ isDelete = is_delete_record(newRecord, writerSchema,
config.getProps());
} else {
// the incoming record is dropped
return false;
}
updatedRecordsWritten++;
}
- return writeRecord(newRecord, combineRecordOpt, writerSchema,
config.getPayloadConfig().getProps(), isDelete);
+
+ // Do delete since the newRecord is a delete record.
+ if (isDelete) {
+ recordsDeleted++;
+ newRecord.unseal();
+ newRecord.clearNewLocation();
+ newRecord.seal();
+ newRecord.deflate();
+ return true;
+ }
+
+ // Inject custom insert/abort logic.
+ Option<Pair<HoodieRecord, Schema>> processedRecord = recordMerger.merge(
+ Option.empty(), writerSchema, combineRecordOpt, writerSchema,
config.getProps());
+ if (!processedRecord.isPresent()
+ || !is_valid_record(processedRecord.get().getLeft(), writerSchema,
config.getProps())) {
+ return false;
+ }
+
+ // Write the record finally.
+ // TODO: remove delete logic from writeRecord function.
+ return writeRecord(newRecord, Option.of(processedRecord.get().getLeft()),
writerSchema, config.getPayloadConfig().getProps(), isDelete);
+ }
+
+ protected boolean is_delete_record(HoodieRecord record, Schema schema,
TypedProperties props) throws IOException {
+ return record.isDelete(schema, props)
+ || record instanceof HoodieEmptyRecord
+ || (record.getData() != null && record.getData() instanceof
EmptyHoodieRecordPayload)
+ || HoodieOperation.isDelete(record.getOperation());
+ }
+
+ protected boolean is_valid_record(HoodieRecord record, Schema schema,
TypedProperties props) throws IOException {
Review Comment:
```suggestion
protected boolean isValidRecord(HoodieRecord record, Schema schema,
TypedProperties props) throws IOException {
```
##########
hudi-common/src/main/java/org/apache/hudi/common/model/HoodieAvroRecordMerger.java:
##########
@@ -56,7 +76,8 @@ public HoodieRecordType getRecordType() {
private Option<IndexedRecord> combineAndGetUpdateValue(HoodieRecord older,
HoodieRecord newer, Schema schema, Properties props) throws IOException {
Option<IndexedRecord> previousAvroData = older.toIndexedRecord(schema,
props).map(HoodieAvroIndexedRecord::getData);
if (!previousAvroData.isPresent()) {
- return Option.empty();
+ Option<IndexedRecord> newData = newer.toIndexedRecord(schema,
props).map(HoodieAvroIndexedRecord::getData);
+ return newData;
Review Comment:
Is this a bug before?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java:
##########
@@ -149,6 +150,9 @@ public static <R> HoodieRecord<R>
tagAsNewRecordIfNeeded(HoodieRecord<R> record,
// currentLocation 2 times and it will fail the second time. So creating
a new in memory
// copy of the hoodie record.
HoodieRecord<R> newRecord = record.newInstance();
+ if (record instanceof HoodieEmptyRecord) {
+ newRecord = record.newInstance(record.getKey(), record.getOperation());
+ }
Review Comment:
Can this be before L152 `HoodieRecord<R> newRecord = record.newInstance();`?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java:
##########
@@ -270,14 +272,46 @@ protected boolean writeUpdateRecord(HoodieRecord<T>
newRecord, HoodieRecord<T> o
if (combineRecordOpt.isPresent()) {
if (oldRecord.getData() != combineRecordOpt.get().getData()) {
// the incoming record is chosen
- isDelete = HoodieOperation.isDelete(newRecord.getOperation());
+ isDelete = is_delete_record(newRecord, writerSchema,
config.getProps());
} else {
// the incoming record is dropped
return false;
}
updatedRecordsWritten++;
}
- return writeRecord(newRecord, combineRecordOpt, writerSchema,
config.getPayloadConfig().getProps(), isDelete);
+
+ // Do delete since the newRecord is a delete record.
+ if (isDelete) {
+ recordsDeleted++;
+ newRecord.unseal();
+ newRecord.clearNewLocation();
+ newRecord.seal();
+ newRecord.deflate();
+ return true;
+ }
+
+ // Inject custom insert/abort logic.
+ Option<Pair<HoodieRecord, Schema>> processedRecord = recordMerger.merge(
+ Option.empty(), writerSchema, combineRecordOpt, writerSchema,
config.getProps());
+ if (!processedRecord.isPresent()
+ || !is_valid_record(processedRecord.get().getLeft(), writerSchema,
config.getProps())) {
+ return false;
+ }
+
+ // Write the record finally.
+ // TODO: remove delete logic from writeRecord function.
+ return writeRecord(newRecord, Option.of(processedRecord.get().getLeft()),
writerSchema, config.getPayloadConfig().getProps(), isDelete);
+ }
+
+ protected boolean is_delete_record(HoodieRecord record, Schema schema,
TypedProperties props) throws IOException {
Review Comment:
```suggestion
protected boolean isDeleteRecord(HoodieRecord record, Schema schema,
TypedProperties props) throws IOException {
```
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java:
##########
@@ -270,14 +272,46 @@ protected boolean writeUpdateRecord(HoodieRecord<T>
newRecord, HoodieRecord<T> o
if (combineRecordOpt.isPresent()) {
if (oldRecord.getData() != combineRecordOpt.get().getData()) {
// the incoming record is chosen
- isDelete = HoodieOperation.isDelete(newRecord.getOperation());
+ isDelete = is_delete_record(newRecord, writerSchema,
config.getProps());
} else {
// the incoming record is dropped
return false;
}
updatedRecordsWritten++;
}
- return writeRecord(newRecord, combineRecordOpt, writerSchema,
config.getPayloadConfig().getProps(), isDelete);
+
+ // Do delete since the newRecord is a delete record.
+ if (isDelete) {
+ recordsDeleted++;
+ newRecord.unseal();
+ newRecord.clearNewLocation();
+ newRecord.seal();
+ newRecord.deflate();
+ return true;
+ }
+
+ // Inject custom insert/abort logic.
+ Option<Pair<HoodieRecord, Schema>> processedRecord = recordMerger.merge(
+ Option.empty(), writerSchema, combineRecordOpt, writerSchema,
config.getProps());
+ if (!processedRecord.isPresent()
+ || !is_valid_record(processedRecord.get().getLeft(), writerSchema,
config.getProps())) {
+ return false;
+ }
+
+ // Write the record finally.
+ // TODO: remove delete logic from writeRecord function.
Review Comment:
Is this already done?
##########
hudi-common/src/main/java/org/apache/hudi/common/model/HoodiePreCombineAvroRecordMerger.java:
##########
@@ -35,8 +35,16 @@ public class HoodiePreCombineAvroRecordMerger extends
HoodieAvroRecordMerger {
public static final HoodiePreCombineAvroRecordMerger INSTANCE = new
HoodiePreCombineAvroRecordMerger();
Review Comment:
I assume this merger will be removed once the merging stage (e.g.,
`PRE_COMBINE`, `COMBINE`) is added to the merge API.
--
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]
