[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
@@ -284,7 +298,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/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;
+  }
+}

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

@@ -27,6 +27,7 @@
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.StringUtils;
 import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.keygen.BaseKeyGenerator;
 
 import org.apache.avro.Schema;
@@ -51,6 +52,15 @@ public HoodieAvroRecord(HoodieRecord<T> record) {
     super(record);
   }
 
+  public HoodieAvroRecord(
+      HoodieKey key,
+      T data,
+      HoodieOperation operation,
+      HoodieRecordLocation currentLocation,
+      HoodieRecordLocation newLocation) {
+    super(key, data, operation, currentLocation, newLocation);
+  }
+
   public HoodieAvroRecord() {
   }
 
@@ -113,14 +123,14 @@ public HoodieRecord rewriteRecord(Schema recordSchema, Properties props, Schema
     Option<IndexedRecord> avroRecordPayloadOpt = getData().getInsertValue(recordSchema, props);
     GenericRecord avroPayloadInNewSchema =
         HoodieAvroUtils.rewriteRecord((GenericRecord) avroRecordPayloadOpt.get(), targetSchema);
-    return new HoodieAvroRecord<>(getKey(), new RewriteAvroPayload(avroPayloadInNewSchema), getOperation());
+    return new HoodieAvroRecord<>(getKey(), new RewriteAvroPayload(avroPayloadInNewSchema), getOperation(), this.currentLocation, this.newLocation);
   }
 
   @Override
   public HoodieRecord rewriteRecordWithNewSchema(Schema recordSchema, Properties props, Schema newSchema, Map<String, String> renameCols) throws IOException {
     GenericRecord oldRecord = (GenericRecord) getData().getInsertValue(recordSchema, props).get();
     GenericRecord rewriteRecord = HoodieAvroUtils.rewriteRecordWithNewSchema(oldRecord, newSchema, renameCols);
-    return new HoodieAvroRecord<>(getKey(), new RewriteAvroPayload(rewriteRecord), getOperation());
+    return new HoodieAvroRecord<>(getKey(), new RewriteAvroPayload(rewriteRecord), getOperation(), this.currentLocation, this.newLocation);
   }
 
   @Override
@@ -133,23 +143,29 @@ public HoodieRecord updateMetadataValues(Schema recordSchema, Properties props,
       }
     });
 
-    return new HoodieAvroRecord<>(getKey(), new RewriteAvroPayload(avroRecordPayload), getOperation());
+    return new HoodieAvroRecord<>(getKey(), new RewriteAvroPayload(avroRecordPayload), getOperation(), this.currentLocation, this.newLocation);
   }
 
   @Override
   public HoodieRecord truncateRecordKey(Schema recordSchema, Properties props, String keyFieldName) throws IOException {
     GenericRecord avroRecordPayload = (GenericRecord) getData().getInsertValue(recordSchema, props).get();
     avroRecordPayload.put(keyFieldName, StringUtils.EMPTY_STRING);
-    return new HoodieAvroRecord<>(getKey(), new RewriteAvroPayload(avroRecordPayload), getOperation());
+    return new HoodieAvroRecord<>(getKey(), new RewriteAvroPayload(avroRecordPayload), getOperation(), this.currentLocation, this.newLocation);
   }
 
   @Override
   public boolean isDelete(Schema recordSchema, Properties props) throws IOException {
+    if (!(data instanceof HoodieAvroInsertValuePayload) && !(data instanceof RewriteAvroPayload)) {
+      throw new HoodieException("We should deserialization before calling isDelete");
+    }
     return !getData().getInsertValue(recordSchema, props).isPresent();
   }
 
   @Override
   public boolean shouldIgnore(Schema recordSchema, Properties props) throws IOException {
+    if (!(data instanceof HoodieAvroInsertValuePayload) && !(data instanceof RewriteAvroPayload)) {
+      throw new HoodieException("We should deserialization before calling shouldIgnore");
+    }
     Option<IndexedRecord> insertRecord = getData().getInsertValue(recordSchema, props);
     // just skip the ignored record
     if (insertRecord.isPresent() && insertRecord.get().equals(SENTINEL)) {
@@ -197,6 +213,15 @@ public Option<HoodieAvroIndexedRecord> toIndexedRecord(Schema recordSchema, Prop
     }
   }
 
+  @Override
+  public HoodieRecord deserialization(Schema recordSchema, Properties props) throws IOException {
+    if (this.data instanceof RewriteAvroPayload || this.data instanceof HoodieAvroInsertValuePayload) {
+      return this;
+    }
+    Option<IndexedRecord> data = this.data.getInsertValue(recordSchema, props);
+    return new HoodieAvroRecord<>(getKey(), new HoodieAvroInsertValuePayload(data), getOperation(), this.currentLocation, this.newLocation);
+  }
+
   @Override
   protected final void writeRecordPayload(T payload, Kryo kryo, Output output) {
     // NOTE: Since [[orderingVal]] is polymorphic we have to write out its class

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

@@ -124,12 +124,12 @@ public String getFieldName() {
   /**
    * Current location of record on storage. Filled in by looking up index
    */
-  private HoodieRecordLocation currentLocation;
+  protected HoodieRecordLocation currentLocation;
 
   /**
    * New location of record on storage, after written.
    */
-  private HoodieRecordLocation newLocation;
+  protected HoodieRecordLocation newLocation;
 
   /**
    * Indicates whether the object is sealed.
@@ -154,6 +154,19 @@ public HoodieRecord(HoodieKey key, T data, HoodieOperation operation) {
     this.operation = operation;
   }
 
+  public HoodieRecord(
+      HoodieKey key,
+      T data,
+      HoodieOperation operation,
+      HoodieRecordLocation currentLocation,
+      HoodieRecordLocation newLocation) {
+    this.key = key;
+    this.data = data;
+    this.currentLocation = currentLocation;
+    this.newLocation = newLocation;
+    this.operation = operation;
+  }
+
   public HoodieRecord(HoodieRecord<T> record) {
     this(record.key, record.data, record.operation);
     this.currentLocation = record.currentLocation;
@@ -368,6 +381,11 @@ public HoodieRecord rewriteRecordWithNewSchema(Schema recordSchema, Properties p
 
   public abstract Option<Map<String, String>> getMetadata();
 
+  @Override
+  public HoodieRecord deserialization(Schema recordSchema, Properties props) throws IOException {
+    return this;
+  }
+
   public static String generateSequenceId(String instantTime, int partitionId, long recordIndex) {
     return instantTime + "_" + partitionId + "_" + recordIndex;
   }

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

@@ -49,4 +49,6 @@ HoodieRecord wrapIntoHoodieRecordPayloadWithParams(
   HoodieRecord truncateRecordKey(Schema recordSchema, Properties props, String keyFieldName) throws IOException;
 
   Option<HoodieAvroIndexedRecord> toIndexedRecord(Schema recordSchema, Properties props) throws IOException;
+
+  HoodieRecord deserialization(Schema recordSchema, Properties props) throws IOException;
 }