[HUDI-6909] Add logic to check operation metadata field for Spark Record (apache#11930)

Author: linliu-code

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/common/model/HoodieSparkRecord.java

@@ -237,14 +237,25 @@ public HoodieRecord truncateRecordKey(Schema recordSchema, Properties props, Str
   }
 
   @Override
-  public boolean isDelete(Schema recordSchema, Properties props) throws IOException {
+  public boolean isDelete(Schema recordSchema, Properties props) {
     if (null == data) {
       return true;
     }
-    if (recordSchema.getField(HoodieRecord.HOODIE_IS_DELETED_FIELD) == null) {
+
+    // Use metadata filed to decide.
+    Schema.Field operationField = recordSchema.getField(OPERATION_METADATA_FIELD);
+    if (null != operationField
+        && HoodieOperation.isDeleteRecord((String) data.get(operationField.pos(), StringType))) {
+      return true;
+    }
+
+    // Use data field to decide.
+    if (recordSchema.getField(HOODIE_IS_DELETED_FIELD) == null) {
       return false;
     }
-    Object deleteMarker = data.get(recordSchema.getField(HoodieRecord.HOODIE_IS_DELETED_FIELD).pos(), BooleanType);
+
+    Object deleteMarker = data.get(
+        recordSchema.getField(HOODIE_IS_DELETED_FIELD).pos(), BooleanType);
     return deleteMarker instanceof Boolean && (boolean) deleteMarker;
   }
 

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieOperation.java

@@ -122,4 +122,8 @@ public static boolean isUpdateAfter(HoodieOperation operation) {
   public static boolean isDelete(HoodieOperation operation) {
     return operation == DELETE;
   }
+
+  public static boolean isDeleteRecord(String name) {
+    return isDelete(fromName(name));
+  }
 }