optimize multiple partial update code

Author: jerryyue

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseWriteHelper.java

@@ -43,7 +43,7 @@ public HoodieWriteMetadata<O> write(String instantTime,
     try {
       // De-dupe/merge if needed
       I dedupedRecords =
-          combineOnCondition(shouldCombine, inputRecords, shuffleParallelism, table, executor.config.getSchema());
+          combineOnCondition(shouldCombine, inputRecords, shuffleParallelism, table);
 
       Instant lookupBegin = Instant.now();
       I taggedRecords = dedupedRecords;
@@ -69,8 +69,8 @@ protected abstract I tag(
       I dedupedRecords, HoodieEngineContext context, HoodieTable<T, I, K, O> table);
 
   public I combineOnCondition(
-          boolean condition, I records, int parallelism, HoodieTable<T, I, K, O> table, String schemaString) {
-    return condition ? deduplicateRecords(records, table, parallelism, schemaString) : records;
+          boolean condition, I records, int parallelism, HoodieTable<T, I, K, O> table) {
+    return condition ? deduplicateRecords(records, table, parallelism) : records;
   }
 
   /**
@@ -81,8 +81,8 @@ public I combineOnCondition(
    * @return Collection of HoodieRecord already be deduplicated
    */
   public I deduplicateRecords(
-      I records, HoodieTable<T, I, K, O> table, int parallelism, String schemaString) {
-    return deduplicateRecords(records, table.getIndex(), parallelism, schemaString);
+      I records, HoodieTable<T, I, K, O> table, int parallelism) {
+    return deduplicateRecords(records, table.getIndex(), parallelism, table.getConfig().getSchema());
   }
 
   public abstract I deduplicateRecords(

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieWriteHelper.java

@@ -19,6 +19,7 @@
 package org.apache.hudi.table.action.commit;
 
 import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.config.SerializableSchema;
 import org.apache.hudi.common.data.HoodieData;
 import org.apache.hudi.common.engine.HoodieEngineContext;
 import org.apache.hudi.common.model.HoodieAvroRecord;
@@ -55,16 +56,15 @@ protected HoodieData<HoodieRecord<T>> tag(HoodieData<HoodieRecord<T>> dedupedRec
   public HoodieData<HoodieRecord<T>> deduplicateRecords(
       HoodieData<HoodieRecord<T>> records, HoodieIndex<?, ?> index, int parallelism, String schemaString) {
     boolean isIndexingGlobal = index.isGlobal();
-    Properties properties = new Properties();
-    properties.put("schema", schemaString);
+    final SerializableSchema schema = new SerializableSchema(schemaString);
     return records.mapToPair(record -> {
       HoodieKey hoodieKey = record.getKey();
       // If index used is global, then records are expected to differ in their partitionPath
       Object key = isIndexingGlobal ? hoodieKey.getRecordKey() : hoodieKey;
       return Pair.of(key, record);
     }).reduceByKey((rec1, rec2) -> {
       @SuppressWarnings("unchecked")
-      T reducedData = (T) rec2.getData().preCombine(rec1.getData(), properties);
+      T reducedData = (T) rec2.getData().preCombine(rec1.getData(), schema.get(), new  Properties());
       HoodieKey reducedKey = rec1.getData().equals(reducedData) ? rec1.getKey() : rec2.getKey();
 
       return new HoodieAvroRecord<>(reducedKey, reducedData);

hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/FlinkWriteHelper.java

@@ -18,6 +18,7 @@
 
 package org.apache.hudi.table.action.commit;
 
+import org.apache.avro.Schema;
 import org.apache.hudi.client.WriteStatus;
 import org.apache.hudi.common.data.HoodieListData;
 import org.apache.hudi.common.engine.HoodieEngineContext;
@@ -98,9 +99,8 @@ public List<HoodieRecord<T>> deduplicateRecords(
       final T data1 = rec1.getData();
       final T data2 = rec2.getData();
 
-      Properties properties = new Properties();
-      properties.put("schema", schemaString);
-      @SuppressWarnings("unchecked") final T reducedData = (T) data2.preCombine(data1, properties);
+      final Schema schema = new Schema.Parser().parse(schemaString);
+      @SuppressWarnings("unchecked") final T reducedData = (T) data2.preCombine(data1, schema, new Properties());
       // we cannot allow the user to change the key or partitionPath, since that will affect
       // everything
       // so pick it from one of the records.

hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/table/action/commit/FlinkWriteHelperTest.java

@@ -22,7 +22,6 @@
 import org.apache.hudi.common.model.HoodieOperation;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieRecordPayload;
-import org.apache.hudi.common.model.MultiplePartialUpdateUnit;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.StringUtils;
 import org.apache.hudi.common.util.collection.Pair;
@@ -497,17 +496,6 @@ public static Object getNestedFieldValAsString(GenericRecord record, String fiel
     return obj == null ? "" : StringUtils.objToString(obj);
   }
 
-  public static Object getMultipleNestedFieldVals(GenericRecord record, String fieldMappings, boolean consistentLogicalTimestampEnabled) {
-    MultiplePartialUpdateUnit multipleOrderingVal2ColsInfo = new MultiplePartialUpdateUnit(fieldMappings);
-    multipleOrderingVal2ColsInfo.getMultiplePartialUpdateUnits().forEach(orderingVal2ColsInfo -> {
-      Object val = getNestedFieldVal(record, orderingVal2ColsInfo.getOrderingField(), true, consistentLogicalTimestampEnabled);
-      if (Objects.nonNull(val)) {
-        orderingVal2ColsInfo.setOrderingValue(val.toString());
-      }
-    });
-    return multipleOrderingVal2ColsInfo.toString();
-  }
-
   /**
    * Obtain value of the provided field, denoted by dot notation. e.g: a.b.c
    */

hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java

@@ -38,6 +38,10 @@ public SerializableSchema() {
   public SerializableSchema(Schema schema) {
     this.schema = newCopy(schema);
   }
+
+  public SerializableSchema(String schemaStr) {
+    this.schema = new Schema.Parser().parse(schemaStr);
+  }
 
   public SerializableSchema(SerializableSchema serializableSchema) {
     this(serializableSchema.schema);

hudi-common/src/main/java/org/apache/hudi/common/config/SerializableSchema.java

@@ -58,6 +58,20 @@ default T preCombine(T oldValue, Properties properties) {
     return preCombine(oldValue);
   }
 
+  /**
+   * When more than one HoodieRecord have the same HoodieKey in the incoming batch, this function combines them before attempting to insert/upsert by taking in a schema.
+   * Implementation can leverage the schema to decide their business logic to do preCombine.
+   *
+   * @param oldValue   instance of the old {@link HoodieRecordPayload} to be combined with.
+   * @param schema     Payload related schema. For example use schema to overwrite old instance for specified fields that doesn't equal to default value.
+   * @param properties Payload related properties. For example pass the ordering field(s) name to extract from value in storage.
+   * @return the combined value
+   */
+  @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+  default T preCombine(T oldValue, Schema schema, Properties properties) {
+    return preCombine(oldValue, properties);
+  }
+
   /**
    * This methods is deprecated. Please refer to {@link #combineAndGetUpdateValue(IndexedRecord, Schema, Properties)} for java docs.
    */

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

@@ -22,7 +22,6 @@
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.generic.GenericRecordBuilder;
 import org.apache.avro.generic.IndexedRecord;
-
 import org.apache.hudi.common.util.Option;
 
 import java.io.IOException;
@@ -58,26 +57,43 @@ public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord currentValue
     GenericRecord insertRecord = (GenericRecord) recordOption.get();
     GenericRecord currentRecord = (GenericRecord) currentValue;
 
-    return getMergedIndexedRecordOption(schema, insertRecord, currentRecord);
+    return mergeRecords(schema, insertRecord, currentRecord);
   }
 
-  protected Option<IndexedRecord> getMergedIndexedRecordOption(Schema schema, GenericRecord insertRecord, GenericRecord currentRecord) {
-    if (isDeleteRecord(insertRecord)) {
+  /**
+   * Merges the given records into one.
+   * The fields in {@code baseRecord} has higher priority:
+   * it is set up into the merged record if it is not null or equals to the default.
+   *
+   * @param schema       The record schema
+   * @param baseRecord   The base record to merge with
+   * @param mergedRecord The record to be merged
+   *
+   * @return the merged record option
+   */
+  protected Option<IndexedRecord> mergeRecords(Schema schema, GenericRecord baseRecord, GenericRecord mergedRecord) {
+    if (isDeleteRecord(baseRecord)) {
       return Option.empty();
     } else {
       final GenericRecordBuilder builder = new GenericRecordBuilder(schema);
       List<Schema.Field> fields = schema.getFields();
-      fields.forEach(field -> {
-        Object value = insertRecord.get(field.name());
-        value = field.schema().getType().equals(Schema.Type.STRING) && value != null ? value.toString() : value;
-        Object defaultValue = field.defaultVal();
-        if (!overwriteField(value, defaultValue)) {
-          builder.set(field, value);
-        } else {
-          builder.set(field, currentRecord.get(field.pos()));
-        }
-      });
+      fields.forEach(field -> setField(baseRecord, mergedRecord, builder, field));
       return Option.of(builder.build());
     }
   }
+
+  protected void setField(
+      GenericRecord baseRecord,
+      GenericRecord mergedRecord,
+      GenericRecordBuilder builder,
+      Schema.Field field) {
+    Object value = baseRecord.get(field.name());
+    value = field.schema().getType().equals(Schema.Type.STRING) && value != null ? value.toString() : value;
+    Object defaultValue = field.defaultVal();
+    if (!overwriteField(value, defaultValue)) {
+      builder.set(field, value);
+    } else {
+      builder.set(field, mergedRecord.get(field.name()));
+    }
+  }
 }