[Minor] fix multi deser avro payload

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java

@@ -35,6 +35,7 @@
 import org.apache.hudi.common.model.HoodiePartitionMetadata;
 import org.apache.hudi.common.model.HoodiePayloadProps;
 import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType;
 import org.apache.hudi.common.model.HoodieRecordLocation;
 import org.apache.hudi.common.model.HoodieWriteStat.RuntimeStats;
 import org.apache.hudi.common.model.IOType;
@@ -215,18 +216,16 @@ private Option<HoodieRecord> prepareRecord(HoodieRecord<T> hoodieRecord) {
       // If the format can not record the operation field, nullify the DELETE payload manually.
       boolean nullifyPayload = HoodieOperation.isDelete(hoodieRecord.getOperation()) && !config.allowOperationMetadataField();
       recordProperties.put(HoodiePayloadProps.PAYLOAD_IS_UPDATE_RECORD_FOR_MOR, String.valueOf(isUpdateRecord));
-      Option<HoodieRecord> finalRecord = Option.empty();
-      if (!nullifyPayload && !hoodieRecord.isDelete(tableSchema, recordProperties)) {
-        if (hoodieRecord.shouldIgnore(tableSchema, recordProperties)) {
-          return Option.of(hoodieRecord);
+      Option<HoodieRecord> finalRecord = nullifyPayload ? Option.empty() : Option.of(hoodieRecord.deserialization(tableSchema, recordProperties));
+      // Check for delete
+      if (finalRecord.isPresent() && !finalRecord.get().isDelete(tableSchema, recordProperties)) {
+        // Check for ignore ExpressionPayload
+        if (finalRecord.get().shouldIgnore(tableSchema, recordProperties)) {
+          return finalRecord;
         }
         // Convert GenericRecord to GenericRecord with hoodie commit metadata in schema
-        HoodieRecord rewrittenRecord;
-        if (schemaOnReadEnabled) {
-          rewrittenRecord = hoodieRecord.rewriteRecordWithNewSchema(tableSchema, recordProperties, writeSchemaWithMetaFields);
-        } else {
-          rewrittenRecord = hoodieRecord.rewriteRecord(tableSchema, recordProperties, writeSchemaWithMetaFields);
-        }
+        HoodieRecord rewrittenRecord = schemaOnReadEnabled ? finalRecord.get().rewriteRecordWithNewSchema(tableSchema, recordProperties, writeSchemaWithMetaFields)
+            : finalRecord.get().rewriteRecord(tableSchema, recordProperties, writeSchemaWithMetaFields);
         HoodieRecord populatedRecord = populateMetadataFields(rewrittenRecord, writeSchemaWithMetaFields, recordProperties);
         finalRecord = Option.of(populatedRecord);
         if (isUpdateRecord) {
@@ -236,6 +235,7 @@ private Option<HoodieRecord> prepareRecord(HoodieRecord<T> hoodieRecord) {
         }
         recordsWritten++;
       } else {
+        finalRecord = Option.empty();
         recordsDeleted++;
       }
 
@@ -364,7 +364,8 @@ private void processAppendResult(AppendResult result, List<HoodieRecord> recordL
       updateWriteStatus(stat, result);
     }
 
-    if (config.isMetadataColumnStatsIndexEnabled()) {
+    // TODO MetadataColumnStatsIndex for spark record
+    if (config.isMetadataColumnStatsIndexEnabled() && recordMerger.getRecordType() == HoodieRecordType.AVRO) {
       final List<Schema.Field> fieldsToIndex;
       // If column stats index is enabled but columns not configured then we assume that
       // all columns should be indexed
@@ -511,7 +512,7 @@ private void writeToBuffer(HoodieRecord<T> record) {
       record.seal();
     }
     // fetch the ordering val first in case the record was deflated.
-    final Comparable<?> orderingVal = record.getOrderingValue(tableSchema, config.getProps());
+    final Comparable<?> orderingVal = record.getOrderingValue(tableSchema, recordProperties);
     Option<HoodieRecord> indexedRecord = prepareRecord(record);
     if (indexedRecord.isPresent()) {
       // Skip the ignored record.

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCreateHandle.java

@@ -132,6 +132,7 @@ public boolean canWrite(HoodieRecord record) {
   protected void doWrite(HoodieRecord record, Schema schema, TypedProperties props) {
     Option<Map<String, String>> recordMetadata = record.getMetadata();
     try {
+      record = record.deserialization(schema, config.getProps());
       if (!HoodieOperation.isDelete(record.getOperation()) && !record.isDelete(schema, config.getProps())) {
         if (record.shouldIgnore(schema, config.getProps())) {
           return;

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java

@@ -40,6 +40,7 @@
 import org.apache.hudi.common.util.collection.Pair;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieCorruptedDataException;
+import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.exception.HoodieIOException;
 import org.apache.hudi.exception.HoodieUpsertException;
 import org.apache.hudi.io.storage.HoodieFileReader;
@@ -284,6 +285,7 @@ protected boolean writeUpdateRecord(HoodieRecord<T> newRecord, HoodieRecord<T> o
   }
 
   protected void writeInsertRecord(HoodieRecord<T> newRecord, Schema schema) throws IOException {
+    newRecord = newRecord.deserialization(schema, config.getProps());
     // just skip the ignored record
     if (newRecord.shouldIgnore(schema, config.getProps())) {
       return;
@@ -346,8 +348,14 @@ public void write(HoodieRecord<T> oldRecord) {
       try {
         Option<Pair<HoodieRecord, Schema>> mergeResult = recordMerger.merge(oldRecord, oldSchema, newRecord, newSchema, props);
         Schema combineRecordSchema = mergeResult.map(Pair::getRight).orElse(null);
-        Option<HoodieRecord> combinedRecord = mergeResult.map(Pair::getLeft);
-
+        Option<HoodieRecord> combinedRecord = mergeResult.map(Pair::getLeft)
+            .map(record -> {
+              try {
+                return record.deserialization(combineRecordSchema, props);
+              } catch (IOException e) {
+                throw new HoodieException(e);
+              }
+            });
         if (combinedRecord.isPresent() && combinedRecord.get().shouldIgnore(combineRecordSchema, props)) {
           // If it is an IGNORE_RECORD, just copy the old record, and do not update the new record.
           copyOldRecord = true;

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java

@@ -92,6 +92,7 @@ public void write(HoodieRecord oldRecord) {
         throw new HoodieUpsertException("Insert/Update not in sorted order");
       }
       try {
+        hoodieRecord = hoodieRecord.deserialization(newSchema, config.getProps());
         writeRecord(hoodieRecord, Option.of(hoodieRecord), newSchema, config.getProps());
         insertRecordsWritten++;
         writtenRecordKeys.add(keyToPreWrite);
@@ -112,8 +113,10 @@ public List<WriteStatus> close() {
         HoodieRecord<T> hoodieRecord = keyToNewRecords.get(key);
         if (!writtenRecordKeys.contains(hoodieRecord.getRecordKey())) {
           if (useWriterSchemaForCompaction) {
+            hoodieRecord = hoodieRecord.deserialization(tableSchemaWithMetaFields, config.getProps());
             writeRecord(hoodieRecord, Option.of(hoodieRecord), tableSchemaWithMetaFields, config.getProps());
           } else {
+            hoodieRecord = hoodieRecord.deserialization(tableSchema, config.getProps());
             writeRecord(hoodieRecord, Option.of(hoodieRecord), tableSchema, config.getProps());
           }
           insertRecordsWritten++;

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

@@ -30,6 +30,7 @@
 import org.apache.hudi.common.model.HoodieKey;
 import org.apache.hudi.common.model.HoodieOperation;
 import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordLocation;
 import org.apache.hudi.common.model.MetadataValues;
 import org.apache.hudi.common.util.ConfigUtils;
 import org.apache.hudi.common.util.Option;
@@ -122,6 +123,19 @@ private HoodieSparkRecord(HoodieKey key, InternalRow data, StructType schema, Ho
     this.schema = schema;
   }
 
+  public HoodieSparkRecord(
+      HoodieKey key,
+      InternalRow data,
+      StructType schema,
+      HoodieOperation operation,
+      HoodieRecordLocation currentLocation,
+      HoodieRecordLocation newLocation,
+      boolean copy) {
+    super(key, data, operation, currentLocation, newLocation);
+    this.copy = copy;
+    this.schema = schema;
+  }
+
   @Override
   public HoodieSparkRecord newInstance() {
     return new HoodieSparkRecord(this.key, this.data, this.schema, this.operation, this.copy);
@@ -175,7 +189,7 @@ public HoodieRecord joinWith(HoodieRecord other, Schema targetSchema) {
     InternalRow mergeRow = new JoinedRow(data, (InternalRow) other.getData());
     UnsafeProjection projection =
         HoodieInternalRowUtils.getCachedUnsafeProjection(targetStructType, targetStructType);
-    return new HoodieSparkRecord(getKey(), projection.apply(mergeRow), targetStructType, getOperation(), copy);
+    return new HoodieSparkRecord(getKey(), projection.apply(mergeRow), targetStructType, getOperation(), this.currentLocation, this.newLocation, copy);
   }
 
   @Override
@@ -189,7 +203,7 @@ public HoodieRecord rewriteRecord(Schema recordSchema, Properties props, Schema
     // TODO add actual rewriting
     InternalRow finalRow = new HoodieInternalRow(metaFields, data, containMetaFields);
 
-    return new HoodieSparkRecord(getKey(), finalRow, targetStructType, getOperation(), copy);
+    return new HoodieSparkRecord(getKey(), finalRow, targetStructType, getOperation(), this.currentLocation, this.newLocation, copy);
   }
 
   @Override
@@ -204,7 +218,7 @@ public HoodieRecord rewriteRecordWithNewSchema(Schema recordSchema, Properties p
         HoodieInternalRowUtils.rewriteRecordWithNewSchema(data, structType, newStructType, renameCols);
     HoodieInternalRow finalRow = new HoodieInternalRow(metaFields, rewrittenRow, containMetaFields);
 
-    return new HoodieSparkRecord(getKey(), finalRow, newStructType, getOperation(), copy);
+    return new HoodieSparkRecord(getKey(), finalRow, newStructType, getOperation(), this.currentLocation, this.newLocation, copy);
   }
 
   @Override
@@ -219,7 +233,7 @@ public HoodieRecord updateMetadataValues(Schema recordSchema, Properties props,
       }
     });
 
-    return new HoodieSparkRecord(getKey(), updatableRow, structType, getOperation(), copy);
+    return new HoodieSparkRecord(getKey(), updatableRow, structType, getOperation(), this.currentLocation, this.newLocation, copy);
   }
 
   @Override
@@ -244,11 +258,7 @@ public boolean isDelete(Schema recordSchema, Properties props) throws IOExceptio
 
   @Override
   public boolean shouldIgnore(Schema recordSchema, Properties props) throws IOException {
-    if (data != null && data.equals(SENTINEL)) {
-      return true;
-    } else {
-      return false;
-    }
+    return false;
   }
 
   @Override
@@ -284,7 +294,7 @@ public HoodieRecord wrapIntoHoodieRecordPayloadWithKeyGen(Schema recordSchema, P
       partition = data.get(HoodieMetadataField.PARTITION_PATH_METADATA_FIELD.ordinal(), StringType).toString();
     }
     HoodieKey hoodieKey = new HoodieKey(key, partition);
-    return new HoodieSparkRecord(hoodieKey, data, structType, getOperation(), copy);
+    return new HoodieSparkRecord(hoodieKey, data, structType, getOperation(), this.currentLocation, this.newLocation, copy);
   }
 
   @Override

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

@@ -32,12 +32,14 @@ public abstract class BaseAvroPayload implements Serializable {
   /**
    * Avro data extracted from the source converted to bytes.
    */
-  public final byte[] recordBytes;
+  protected final byte[] recordBytes;
 
   /**
    * For purposes of preCombining.
    */
-  public final Comparable orderingVal;
+  protected final Comparable orderingVal;
+
+  protected final boolean isDeletedRecord;
 
   /**
    * Instantiate {@link BaseAvroPayload}.
@@ -48,8 +50,34 @@ public abstract class BaseAvroPayload implements Serializable {
   public BaseAvroPayload(GenericRecord record, Comparable orderingVal) {
     this.recordBytes = record != null ? HoodieAvroUtils.avroToBytes(record) : new byte[0];
     this.orderingVal = orderingVal;
+    this.isDeletedRecord = record == null || isDeleteRecord(record);
+
     if (orderingVal == null) {
       throw new HoodieException("Ordering value is null for record: " + record);
     }
   }
+
+  /**
+   * Defines whether this implementation of {@link HoodieRecordPayload} could produce
+   * {@link HoodieRecord#SENTINEL}
+   */
+  public boolean canProduceSentinel() {
+    return false;
+  }
+
+  /**
+   * @param genericRecord instance of {@link GenericRecord} of interest.
+   * @returns {@code true} if record represents a delete record. {@code false} otherwise.
+   */
+  protected static boolean isDeleteRecord(GenericRecord genericRecord) {
+    final String isDeleteKey = HoodieRecord.HOODIE_IS_DELETED_FIELD;
+    // Modify to be compatible with new version Avro.
+    // The new version Avro throws for GenericRecord.get if the field name
+    // does not exist in the schema.
+    if (genericRecord.getSchema().getField(isDeleteKey) == null) {
+      return false;
+    }
+    Object deleteMarker = genericRecord.get(isDeleteKey);
+    return (deleteMarker instanceof Boolean && (boolean) deleteMarker);
+  }
 }

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

@@ -51,7 +51,7 @@ public DefaultHoodieRecordPayload(Option<GenericRecord> record) {
 
   @Override
   public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord currentValue, Schema schema, Properties properties) throws IOException {
-    if (recordBytes.length == 0) {
+    if (recordBytes.length == 0 || isDeletedRecord) {
       return Option.empty();
     }
 
@@ -71,18 +71,18 @@ public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord currentValue
     /*
      * Now check if the incoming record is a delete record.
      */
-    return isDeleteRecord(incomingRecord) ? Option.empty() : Option.of(incomingRecord);
+    return Option.of(incomingRecord);
   }
 
   @Override
   public Option<IndexedRecord> getInsertValue(Schema schema, Properties properties) throws IOException {
-    if (recordBytes.length == 0) {
+    if (recordBytes.length == 0 || isDeletedRecord) {
       return Option.empty();
     }
     GenericRecord incomingRecord = HoodieAvroUtils.bytesToAvro(recordBytes, schema);
     eventTime = updateEventTime(incomingRecord, properties);
 
-    return isDeleteRecord(incomingRecord) ? Option.empty() : Option.of(incomingRecord);
+    return Option.of(incomingRecord);
   }
 
   private static Option<Object> updateEventTime(GenericRecord record, Properties properties) {

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

@@ -46,7 +46,7 @@ public EventTimeAvroPayload(Option<GenericRecord> record) {
 
   @Override
   public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord currentValue, Schema schema, Properties properties) throws IOException {
-    if (recordBytes.length == 0) {
+    if (recordBytes.length == 0 || isDeletedRecord) {
       return Option.empty();
     }
 
@@ -61,17 +61,16 @@ public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord currentValue
     /*
      * Now check if the incoming record is a delete record.
      */
-    return isDeleteRecord(incomingRecord) ? Option.empty() : Option.of(incomingRecord);
+    return Option.of(incomingRecord);
   }
 
   @Override
   public Option<IndexedRecord> getInsertValue(Schema schema, Properties properties) throws IOException {
-    if (recordBytes.length == 0) {
+    if (recordBytes.length == 0 || isDeletedRecord) {
       return Option.empty();
     }
-    GenericRecord incomingRecord = bytesToAvro(recordBytes, schema);
 
-    return isDeleteRecord(incomingRecord) ? Option.empty() : Option.of(incomingRecord);
+    return Option.of(bytesToAvro(recordBytes, schema));
   }
 
   @Override

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

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.model;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.hudi.common.util.Option;
+
+import java.io.IOException;
+
+/**
+ * Default payload used in HoodieAvroRecord. We dont need to serialize/deserialize avro record in payload multiple times.
+ */
+public class HoodieAvroInsertValuePayload implements HoodieRecordPayload<RewriteAvroPayload> {
+
+  private Option<IndexedRecord> record;
+
+  public HoodieAvroInsertValuePayload(Option<IndexedRecord> record) {
+    this.record = record;
+  }
+
+  @Override
+  public RewriteAvroPayload preCombine(RewriteAvroPayload another) {
+    throw new UnsupportedOperationException("precombine is not expected for HoodieAvroInsertValuePayload");
+  }
+
+  @Override
+  public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord currentValue, Schema schema) throws IOException {
+    throw new UnsupportedOperationException("combineAndGetUpdateValue is not expected for HoodieAvroInsertValuePayload");
+  }
+
+  @Override
+  public Option<IndexedRecord> getInsertValue(Schema schema) throws IOException {
+    return record;
+  }
+}