[HUDI-1255] Add new Payload(OverwriteNonDefaultsWithLatestAvroPayload) for updating specified fields in storage (#2056)

Add new Payload(OverwriteNonDefaultsWithLatestAvroPayload) for updating specified fields in storage

## Brief change log

update current value for several fields that you want to change.

The default payload OverwriteWithLatestAvroPayload overwrite the whole record when 

compared to `orderingVal`.This doesn't meet our need when we just want to change specified fields.
For example: (suppose Default value is null)
```
current Value 
Field:      name   age   gender
Value:     karl     20    male
```
```
insert Value
Field:      name   age   gender
Value:     null     30    null
```
```
After insert:
Field:      name   age   gender
Value:     karl     30    male
```
## Verify this pull request

Added TestOverwriteNonDefaultsWithLatestAvroPayload to verify the change.

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

@@ -0,0 +1,72 @@
+/*
+ * 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.hudi.common.util.Option;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * subclass of OverwriteWithLatestAvroPayload used for delta streamer.
+ * <p>
+ * 1. preCombine - Picks the latest delta record for a key, based on an ordering field.
+ * 2. combineAndGetUpdateValue/getInsertValue - overwrite storage for specified fields
+ * that doesn't equal defaultValue.
+ */
+public class OverwriteNonDefaultsWithLatestAvroPayload extends OverwriteWithLatestAvroPayload {
+
+  public OverwriteNonDefaultsWithLatestAvroPayload(GenericRecord record, Comparable orderingVal) {
+    super(record, orderingVal);
+  }
+
+  public OverwriteNonDefaultsWithLatestAvroPayload(Option<GenericRecord> record) {
+    super(record); // natural order
+  }
+
+  @Override
+  public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord currentValue, Schema schema) throws IOException {
+
+    Option<IndexedRecord> recordOption = getInsertValue(schema);
+    if (!recordOption.isPresent()) {
+      return Option.empty();
+    }
+
+    GenericRecord insertRecord = (GenericRecord) recordOption.get();
+    GenericRecord currentRecord = (GenericRecord) currentValue;
+
+    if (isDeleteRecord(insertRecord)) {
+      return Option.empty();
+    } else {
+      List<Schema.Field> fields = schema.getFields();
+      fields.forEach(field -> {
+        Object value = insertRecord.get(field.name());
+        Object defaultValue = field.defaultVal();
+        if (!overwriteField(value, defaultValue)) {
+          currentRecord.put(field.name(), value);
+        }
+      });
+      return Option.of(currentRecord);
+    }
+  }
+}

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

@@ -79,8 +79,15 @@ public class OverwriteWithLatestAvroPayload extends BaseAvroPayload
    * @param genericRecord instance of {@link GenericRecord} of interest.
    * @returns {@code true} if record represents a delete record. {@code false} otherwise.
    */
-  private boolean isDeleteRecord(GenericRecord genericRecord) {
+  protected boolean isDeleteRecord(GenericRecord genericRecord) {
     Object deleteMarker = genericRecord.get("_hoodie_is_deleted");
     return (deleteMarker instanceof Boolean && (boolean) deleteMarker);
   }
+
+  /**
+   * Return true if value equals defaultValue otherwise false.
+   */
+  public Boolean overwriteField(Object value, Object defaultValue) {
+    return defaultValue == null ? value == null : defaultValue.toString().equals(value.toString());
+  }
 }