linliu-code commented on code in PR #12597:
URL: https://github.com/apache/hudi/pull/12597#discussion_r1931350066
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/DefaultSparkRecordMerger.java:
##########
@@ -48,36 +49,28 @@ public Option<Pair<HoodieRecord, Schema>>
merge(HoodieRecord older, Schema oldSc
ValidationUtils.checkArgument(older.getRecordType() ==
HoodieRecordType.SPARK);
ValidationUtils.checkArgument(newer.getRecordType() ==
HoodieRecordType.SPARK);
- if (newer instanceof HoodieSparkRecord) {
- HoodieSparkRecord newSparkRecord = (HoodieSparkRecord) newer;
- if (newSparkRecord.isDelete(newSchema, props)) {
- // Delete record
- return Option.empty();
- }
- } else {
- if (newer.getData() == null) {
- // Delete record
- return Option.empty();
- }
+ if (newer instanceof HoodieEmptyRecord) {
+ return Option.empty();
+ }
+ HoodieSparkRecord newRecord = (HoodieSparkRecord) newer;
+ if (older instanceof HoodieEmptyRecord) {
+ return newRecord.isDelete(newSchema, props)
+ ? Option.empty() : Option.of(Pair.of(newer, newSchema));
}
+ HoodieSparkRecord oldRecord = (HoodieSparkRecord) older;
+ Comparable newOrderingVal = newRecord.getOrderingValue(newSchema, props);
+ Comparable oldOrderingVal = oldRecord.getOrderingValue(oldSchema, props);
- if (older instanceof HoodieSparkRecord) {
- HoodieSparkRecord oldSparkRecord = (HoodieSparkRecord) older;
- if (oldSparkRecord.isDelete(oldSchema, props)) {
- // use natural order for delete record
- return Option.of(Pair.of(newer, newSchema));
- }
- } else {
- if (older.getData() == null) {
- // use natural order for delete record
- return Option.of(Pair.of(newer, newSchema));
- }
+ // The same logic as fg reader.
+ if (newOrderingVal.equals(0) && newRecord.isDelete(newSchema, props)) {
Review Comment:
Done.
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/DefaultSparkRecordMerger.java:
##########
@@ -48,36 +49,28 @@ public Option<Pair<HoodieRecord, Schema>>
merge(HoodieRecord older, Schema oldSc
ValidationUtils.checkArgument(older.getRecordType() ==
HoodieRecordType.SPARK);
ValidationUtils.checkArgument(newer.getRecordType() ==
HoodieRecordType.SPARK);
- if (newer instanceof HoodieSparkRecord) {
- HoodieSparkRecord newSparkRecord = (HoodieSparkRecord) newer;
- if (newSparkRecord.isDelete(newSchema, props)) {
- // Delete record
- return Option.empty();
- }
- } else {
- if (newer.getData() == null) {
- // Delete record
- return Option.empty();
- }
+ if (newer instanceof HoodieEmptyRecord) {
+ return Option.empty();
+ }
+ HoodieSparkRecord newRecord = (HoodieSparkRecord) newer;
+ if (older instanceof HoodieEmptyRecord) {
+ return newRecord.isDelete(newSchema, props)
+ ? Option.empty() : Option.of(Pair.of(newer, newSchema));
}
+ HoodieSparkRecord oldRecord = (HoodieSparkRecord) older;
+ Comparable newOrderingVal = newRecord.getOrderingValue(newSchema, props);
+ Comparable oldOrderingVal = oldRecord.getOrderingValue(oldSchema, props);
- if (older instanceof HoodieSparkRecord) {
- HoodieSparkRecord oldSparkRecord = (HoodieSparkRecord) older;
- if (oldSparkRecord.isDelete(oldSchema, props)) {
- // use natural order for delete record
- return Option.of(Pair.of(newer, newSchema));
- }
- } else {
- if (older.getData() == null) {
- // use natural order for delete record
- return Option.of(Pair.of(newer, newSchema));
- }
+ // The same logic as fg reader.
+ if (newOrderingVal.equals(0) && newRecord.isDelete(newSchema, props)) {
+ return Option.empty();
}
- if (older.getOrderingValue(oldSchema,
props).compareTo(newer.getOrderingValue(newSchema, props)) > 0) {
- return Option.of(Pair.of(older, oldSchema));
- } else {
- return Option.of(Pair.of(newer, newSchema));
+ if (!oldOrderingVal.equals(0) && oldOrderingVal.compareTo(newOrderingVal)
> 0) {
Review Comment:
Done.
--
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]
