[HUDI-1284] preCombine all HoodieRecords and update all fields according to orderingVal

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java

@@ -19,6 +19,7 @@
 package org.apache.hudi.client;
 
 import com.codahale.metrics.Timer;
+import org.apache.avro.Schema;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hudi.avro.model.HoodieCleanMetadata;
 import org.apache.hudi.avro.model.HoodieCompactionPlan;
@@ -259,7 +260,7 @@ protected void rollBackInflightBootstrap() {
    * @param instantTime Instant time of the commit
    * @return WriteStatus to inspect errors and counts
    */
-  public abstract O upsert(I records, final String instantTime);
+  public abstract O upsert(I records, final String instantTime, Schema schema);
 
   /**
    * Upserts the given prepared records into the Hoodie table, at the supplied instantTime.

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java

@@ -123,7 +123,7 @@ private synchronized FileSystemViewManager getViewManager() {
    * @return HoodieWriteMetadata
    */
   public abstract HoodieWriteMetadata<O> upsert(HoodieEngineContext context, String instantTime,
-      I records);
+      I records, Schema schema);
 
   /**
    * Insert a batch of new records into Hoodie table at the supplied instantTime.

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

@@ -19,6 +19,7 @@
 package org.apache.hudi.table.action.commit;
 
 import org.apache.hudi.client.common.HoodieEngineContext;
+import org.apache.hudi.common.config.SerializableSchema;
 import org.apache.hudi.common.model.HoodieRecordPayload;
 import org.apache.hudi.exception.HoodieUpsertException;
 import org.apache.hudi.index.HoodieIndex;
@@ -39,10 +40,23 @@ public HoodieWriteMetadata<O> write(String instantTime,
                                       int shuffleParallelism,
                                       BaseCommitActionExecutor<T, I, K, O, R> executor,
                                       boolean performTagging) {
+    return write(instantTime, inputRecords, context, table, shouldCombine, shuffleParallelism,
+            null, executor, performTagging);
+  }
+
+  public HoodieWriteMetadata<O> write(String instantTime,
+                                      I inputRecordsRDD,
+                                      HoodieEngineContext context,
+                                      HoodieTable<T, I, K, O> table,
+                                      boolean shouldCombine,
+                                      int shuffleParallelism,
+                                      SerializableSchema schema,
+                                      BaseCommitActionExecutor<T, I, K, O, R> executor,
+                                      boolean performTagging) {
     try {
       // De-dupe/merge if needed
       I dedupedRecords =
-          combineOnCondition(shouldCombine, inputRecords, shuffleParallelism, table);
+          combineOnCondition(shouldCombine, inputRecordsRDD, shuffleParallelism, table, schema);
 
       Instant lookupBegin = Instant.now();
       I taggedRecords = dedupedRecords;
@@ -70,8 +84,8 @@ private I tag(
   }
 
   public I combineOnCondition(
-      boolean condition, I records, int parallelism, HoodieTable<T, I, K, O> table) {
-    return condition ? deduplicateRecords(records, table, parallelism) : records;
+      boolean condition, I records, int parallelism, HoodieTable<T, I, K, O> table, SerializableSchema schema) {
+    return condition ? deduplicateRecords(records, table, parallelism, schema) : records;
   }
 
   /**
@@ -82,10 +96,10 @@ public I combineOnCondition(
    * @return Collection of HoodieRecord already be deduplicated
    */
   public I deduplicateRecords(
-      I records, HoodieTable<T, I, K, O> table, int parallelism) {
-    return deduplicateRecords(records, table.getIndex(), parallelism);
+      I records, HoodieTable<T, I, K, O> table, int parallelism, SerializableSchema schema) {
+    return deduplicateRecords(records, table.getIndex(), parallelism, schema);
   }
 
   public abstract I deduplicateRecords(
-      I records, HoodieIndex<T, I, K, O> index, int parallelism);
+      I records, HoodieIndex<T, I, K, O> index, int parallelism, SerializableSchema schema);
 }

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java

@@ -18,6 +18,7 @@
 
 package org.apache.hudi.client;
 
+import org.apache.avro.Schema;
 import org.apache.hudi.client.common.HoodieEngineContext;
 import org.apache.hudi.client.common.HoodieSparkEngineContext;
 import org.apache.hudi.client.embedded.EmbeddedTimelineService;
@@ -128,14 +129,18 @@ public void bootstrap(Option<Map<String, String>> extraMetadata) {
     getTableAndInitCtx(WriteOperationType.UPSERT, HoodieTimeline.METADATA_BOOTSTRAP_INSTANT_TS).bootstrap(context, extraMetadata);
   }
 
-  @Override
   public JavaRDD<WriteStatus> upsert(JavaRDD<HoodieRecord<T>> records, String instantTime) {
+    return upsert(records, instantTime, null);
+  }
+
+  @Override
+  public JavaRDD<WriteStatus> upsert(JavaRDD<HoodieRecord<T>> records, String instantTime, Schema schema) {
     HoodieTable<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<WriteStatus>> table =
         getTableAndInitCtx(WriteOperationType.UPSERT, instantTime);
     table.validateUpsertSchema();
     setOperationType(WriteOperationType.UPSERT);
     this.asyncCleanerService = AsyncCleanerService.startAsyncCleaningIfEnabled(this, instantTime);
-    HoodieWriteMetadata<JavaRDD<WriteStatus>> result = table.upsert(context, instantTime, records);
+    HoodieWriteMetadata<JavaRDD<WriteStatus>> result = table.upsert(context, instantTime, records, schema);
     if (result.getIndexLookupDuration().isPresent()) {
       metrics.updateIndexMetrics(LOOKUP_STR, result.getIndexLookupDuration().get().toMillis());
     }

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkCopyOnWriteTable.java

@@ -18,6 +18,7 @@
 
 package org.apache.hudi.table;
 
+import org.apache.avro.Schema;
 import org.apache.hudi.avro.model.HoodieCleanMetadata;
 import org.apache.hudi.avro.model.HoodieCompactionPlan;
 import org.apache.hudi.avro.model.HoodieRestoreMetadata;
@@ -84,8 +85,9 @@ public HoodieSparkCopyOnWriteTable(HoodieWriteConfig config, HoodieEngineContext
   }
 
   @Override
-  public HoodieWriteMetadata<JavaRDD<WriteStatus>> upsert(HoodieEngineContext context, String instantTime, JavaRDD<HoodieRecord<T>> records) {
-    return new SparkUpsertCommitActionExecutor<>((HoodieSparkEngineContext) context, config, this, instantTime, records).execute();
+  public HoodieWriteMetadata<JavaRDD<WriteStatus>> upsert(HoodieEngineContext context, String instantTime,
+                                                          JavaRDD<HoodieRecord<T>> records, Schema schema) {
+    return new SparkUpsertCommitActionExecutor<>((HoodieSparkEngineContext) context, config, this, instantTime, records, schema).execute();
   }
 
   @Override

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkMergeOnReadTable.java

@@ -18,6 +18,7 @@
 
 package org.apache.hudi.table;
 
+import org.apache.avro.Schema;
 import org.apache.hudi.avro.model.HoodieCompactionPlan;
 import org.apache.hudi.avro.model.HoodieRestoreMetadata;
 import org.apache.hudi.avro.model.HoodieRollbackMetadata;
@@ -78,8 +79,10 @@ public class HoodieSparkMergeOnReadTable<T extends HoodieRecordPayload> extends
   }
 
   @Override
-  public HoodieWriteMetadata<JavaRDD<WriteStatus>> upsert(HoodieEngineContext context, String instantTime, JavaRDD<HoodieRecord<T>> records) {
-    return new SparkUpsertDeltaCommitActionExecutor<>((HoodieSparkEngineContext) context, config, this, instantTime, records).execute();
+  public HoodieWriteMetadata<JavaRDD<WriteStatus>> upsert(HoodieEngineContext context, String instantTime,
+                                                          JavaRDD<HoodieRecord<T>> records, Schema schema) {
+    return new SparkUpsertDeltaCommitActionExecutor<>((HoodieSparkEngineContext) context, config, this, instantTime,
+            records, schema).execute();
   }
 
   @Override

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkBulkInsertHelper.java

@@ -73,7 +73,7 @@ public HoodieWriteMetadata<JavaRDD<WriteStatus>> bulkInsert(JavaRDD<HoodieRecord
 
     if (performDedupe) {
       dedupedRecords = (JavaRDD<HoodieRecord<T>>) SparkWriteHelper.newInstance().combineOnCondition(config.shouldCombineBeforeInsert(), inputRecords,
-          config.getBulkInsertShuffleParallelism(), table);
+          config.getBulkInsertShuffleParallelism(), table, null);
     }
 
     final JavaRDD<HoodieRecord<T>> repartitionedRecords;

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkUpsertCommitActionExecutor.java

@@ -18,8 +18,10 @@
 
 package org.apache.hudi.table.action.commit;
 
+import org.apache.avro.Schema;
 import org.apache.hudi.client.WriteStatus;
 import org.apache.hudi.client.common.HoodieSparkEngineContext;
+import org.apache.hudi.common.config.SerializableSchema;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieRecordPayload;
 import org.apache.hudi.common.model.WriteOperationType;
@@ -33,17 +35,21 @@ public class SparkUpsertCommitActionExecutor<T extends HoodieRecordPayload<T>>
     extends BaseSparkCommitActionExecutor<T> {
 
   private JavaRDD<HoodieRecord<T>> inputRecordsRDD;
+  private SerializableSchema schema;
 
   public SparkUpsertCommitActionExecutor(HoodieSparkEngineContext context,
                                          HoodieWriteConfig config, HoodieTable table,
-                                         String instantTime, JavaRDD<HoodieRecord<T>> inputRecordsRDD) {
+                                         String instantTime, JavaRDD<HoodieRecord<T>> inputRecordsRDD,
+                                         Schema schema) {
     super(context, config, table, instantTime, WriteOperationType.UPSERT);
     this.inputRecordsRDD = inputRecordsRDD;
+    this.schema = new SerializableSchema(schema);
   }
 
   @Override
   public HoodieWriteMetadata<JavaRDD<WriteStatus>> execute() {
     return SparkWriteHelper.newInstance().write(instantTime, inputRecordsRDD, context, table,
-        config.shouldCombineBeforeUpsert(), config.getUpsertShuffleParallelism(), this, true);
+            config.shouldCombineBeforeUpsert(), config.getUpsertShuffleParallelism(),
+                    schema, this, true);
   }
 }

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkWriteHelper.java

@@ -19,9 +19,11 @@
 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.model.HoodieKey;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.model.UpdatePrecombineAvroPayload;
 import org.apache.hudi.index.HoodieIndex;
 
 import org.apache.spark.api.java.JavaRDD;
@@ -33,8 +35,8 @@
  *
  * @param <T>
  */
-public class SparkWriteHelper<T extends HoodieRecordPayload,R> extends AbstractWriteHelper<T, JavaRDD<HoodieRecord<T>>,
-    JavaRDD<HoodieKey>, JavaRDD<WriteStatus>, R> {
+public class SparkWriteHelper<T extends HoodieRecordPayload, R> extends AbstractWriteHelper<T, JavaRDD<HoodieRecord<T>>,
+        JavaRDD<HoodieKey>, JavaRDD<WriteStatus>, R> {
   private SparkWriteHelper() {
   }
 
@@ -49,7 +51,7 @@ public static SparkWriteHelper newInstance() {
   @Override
   public JavaRDD<HoodieRecord<T>> deduplicateRecords(JavaRDD<HoodieRecord<T>> records,
                                                      HoodieIndex<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<WriteStatus>> index,
-                                                     int parallelism) {
+                                                     int parallelism, SerializableSchema schema) {
     boolean isIndexingGlobal = index.isGlobal();
     return records.mapToPair(record -> {
       HoodieKey hoodieKey = record.getKey();
@@ -58,7 +60,14 @@ public JavaRDD<HoodieRecord<T>> deduplicateRecords(JavaRDD<HoodieRecord<T>> reco
       return new Tuple2<>(key, record);
     }).reduceByKey((rec1, rec2) -> {
       @SuppressWarnings("unchecked")
-      T reducedData = (T) rec1.getData().preCombine(rec2.getData());
+      T reducedData;
+      //To prevent every records from parsing schema
+      if (rec2.getData() instanceof UpdatePrecombineAvroPayload) {
+        reducedData = schema.getSchema() != null ? (T) rec1.getData().preCombine(rec2.getData(), schema.getSchema())
+                : (T) rec1.getData().preCombine(rec2.getData());
+      } else {
+        reducedData = (T) rec1.getData().preCombine(rec2.getData());
+      }
       // 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-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/SparkUpsertDeltaCommitActionExecutor.java

@@ -18,7 +18,9 @@
 
 package org.apache.hudi.table.action.deltacommit;
 
+import org.apache.avro.Schema;
 import org.apache.hudi.client.common.HoodieSparkEngineContext;
+import org.apache.hudi.common.config.SerializableSchema;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieRecordPayload;
 import org.apache.hudi.common.model.WriteOperationType;
@@ -33,17 +35,21 @@ public class SparkUpsertDeltaCommitActionExecutor<T extends HoodieRecordPayload<
     extends AbstractSparkDeltaCommitActionExecutor<T> {
 
   private JavaRDD<HoodieRecord<T>> inputRecordsRDD;
+  private SerializableSchema schema;
 
   public SparkUpsertDeltaCommitActionExecutor(HoodieSparkEngineContext context,
                                               HoodieWriteConfig config, HoodieTable table,
-                                              String instantTime, JavaRDD<HoodieRecord<T>> inputRecordsRDD) {
+                                              String instantTime, JavaRDD<HoodieRecord<T>> inputRecordsRDD,
+                                              Schema schema) {
     super(context, config, table, instantTime, WriteOperationType.UPSERT);
     this.inputRecordsRDD = inputRecordsRDD;
+    this.schema = new SerializableSchema(schema);
   }
 
   @Override
   public HoodieWriteMetadata execute() {
     return SparkWriteHelper.newInstance().write(instantTime, inputRecordsRDD, context, table,
-        config.shouldCombineBeforeUpsert(), config.getUpsertShuffleParallelism(),this, true);
+            config.shouldCombineBeforeUpsert(), config.getUpsertShuffleParallelism(),
+                    schema, this, true);
   }
 }

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestHoodieClientOnCopyOnWriteStorage.java

@@ -239,14 +239,15 @@ private void testDeduplication(
     // Global dedup should be done based on recordKey only
     HoodieIndex index = mock(HoodieIndex.class);
     when(index.isGlobal()).thenReturn(true);
-    List<HoodieRecord<RawTripTestPayload>> dedupedRecs = SparkWriteHelper.newInstance().deduplicateRecords(records, index, 1).collect();
+    List<HoodieRecord<RawTripTestPayload>> dedupedRecs = SparkWriteHelper.newInstance().deduplicateRecords(records, index, 1,
+             null).collect();
     assertEquals(1, dedupedRecs.size());
     assertNodupesWithinPartition(dedupedRecs);
 
     // non-Global dedup should be done based on both recordKey and partitionPath
     index = mock(HoodieIndex.class);
     when(index.isGlobal()).thenReturn(false);
-    dedupedRecs = SparkWriteHelper.newInstance().deduplicateRecords(records, index, 1).collect();
+    dedupedRecs = SparkWriteHelper.newInstance().deduplicateRecords(records, index, 1, null).collect();
     assertEquals(2, dedupedRecs.size());
     assertNodupesWithinPartition(dedupedRecs);
 

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestCopyOnWriteActionExecutor.java

@@ -385,7 +385,7 @@ public void testFileSizeUpsertRecords() throws Exception {
 
     // Insert new records
     BaseSparkCommitActionExecutor actionExecutor = new SparkUpsertCommitActionExecutor(context, config, table,
-        instantTime, jsc.parallelize(records));
+        instantTime, jsc.parallelize(records), null);
     jsc.parallelize(Arrays.asList(1))
         .map(i -> actionExecutor.handleInsert(FSUtils.createNewFileIdPfx(), records.iterator()))
         .map(Transformations::flatten).collect();
@@ -426,7 +426,7 @@ public void testInsertUpsertWithHoodieAvroPayload() throws Exception {
     String partitionPath = writeStatus.getPartitionPath();
     long numRecordsInPartition = updates.stream().filter(u -> u.getPartitionPath().equals(partitionPath)).count();
     BaseSparkCommitActionExecutor newActionExecutor = new SparkUpsertCommitActionExecutor(context, config, table,
-        instantTime, jsc.parallelize(updates));
+        instantTime, jsc.parallelize(updates), null);
     final List<List<WriteStatus>> updateStatus = jsc.parallelize(Arrays.asList(1)).map(x -> {
       return newActionExecutor.handleUpdate(partitionPath, fileId, updates.iterator());
     }).map(Transformations::flatten).collect();