[HUDI-3350][HUDI-3351] fix comments

wulei0302 · wulei0302 · commit 247df2328bfd · 2022-06-23T17:57:36.000+08:00
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieWriteHelper.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieWriteHelper.java
@@ -59,7 +59,8 @@ public HoodieData<HoodieRecord<T>> deduplicateRecords(
     }).reduceByKey((rec1, rec2) -> {
       @SuppressWarnings("unchecked")
       HoodieRecord<T> reducedRecord =  hoodieMerge.preCombine(rec1, rec2);
-      return reducedRecord.newInstance();
+      HoodieKey reducedKey = rec1.getData().equals(reducedRecord.getData()) ? rec1.getKey() : rec2.getKey();
+      return reducedRecord.newInstance(reducedKey);
     }, parallelism).map(Pair::getRight);
   }
 
diff --git a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaWriteHelper.java b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaWriteHelper.java
@@ -67,11 +67,10 @@ public List<HoodieRecord<T>> deduplicateRecords(
     return keyedRecords.values().stream().map(x -> x.stream().map(Pair::getRight).reduce((rec1, rec2) -> {
       @SuppressWarnings("unchecked")
       HoodieRecord<T> reducedRecord =  hoodieMerge.preCombine(rec1,rec2);
-
       // we cannot allow the user to change the key or partitionPath, since that will affect
       // everything
       // so pick it from one of the records.
-      return reducedRecord.newInstance();
+      return reducedRecord.newInstance(rec1.getKey());
     }).orElse(null)).filter(Objects::nonNull).collect(Collectors.toList());
   }
 }
diff --git a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java
@@ -25,6 +25,7 @@
 import org.apache.hudi.common.model.HoodieRecord.HoodieMetadataField;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.ValidationUtils;
 import org.apache.hudi.common.util.collection.Pair;
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.exception.HoodieIOException;
@@ -765,9 +766,7 @@ private static Object rewriteRecordWithNewSchema(Object oldRecord, Schema oldSch
     }
     switch (newSchema.getType()) {
       case RECORD:
-        if (!(oldRecord instanceof IndexedRecord)) {
-          throw new IllegalArgumentException("cannot rewrite record with different type");
-        }
+        ValidationUtils.checkArgument(oldRecord instanceof IndexedRecord, "cannot rewrite record with different type");
         IndexedRecord indexedRecord = (IndexedRecord) oldRecord;
         List<Schema.Field> fields = newSchema.getFields();
         Map<Integer, Object> helper = new HashMap<>();
@@ -806,9 +805,7 @@ private static Object rewriteRecordWithNewSchema(Object oldRecord, Schema oldSch
         }
         return newRecord;
       case ARRAY:
-        if (!(oldRecord instanceof Collection)) {
-          throw new IllegalArgumentException("cannot rewrite record with different type");
-        }
+        ValidationUtils.checkArgument(oldRecord instanceof Collection, "cannot rewrite record with different type");
         Collection array = (Collection)oldRecord;
         List<Object> newArray = new ArrayList();
         fieldNames.push("element");
@@ -818,9 +815,7 @@ private static Object rewriteRecordWithNewSchema(Object oldRecord, Schema oldSch
         fieldNames.pop();
         return newArray;
       case MAP:
-        if (!(oldRecord instanceof Map)) {
-          throw new IllegalArgumentException("cannot rewrite record with different type");
-        }
+        ValidationUtils.checkArgument(oldRecord instanceof Map, "cannot rewrite record with different type");
         Map<Object, Object> map = (Map<Object, Object>) oldRecord;
         Map<Object, Object> newMap = new HashMap<>();
         fieldNames.push("value");
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieAvroRecordMerge.java b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieAvroRecordMerge.java
@@ -34,7 +34,7 @@ public HoodieRecord preCombine(HoodieRecord older, HoodieRecord newer) {
     HoodieRecordPayload picked = unsafeCast(((HoodieAvroRecord) newer).getData().preCombine(((HoodieAvroRecord) older).getData()));
     if (picked instanceof HoodieMetadataPayload) {
       // NOTE: HoodieMetadataPayload return a new payload
-      return new HoodieAvroRecord(newer.getKey(), ((HoodieMetadataPayload) picked), newer.getOperation());
+      return new HoodieAvroRecord(newer.getKey(), picked, newer.getOperation());
     }
     return picked.equals(((HoodieAvroRecord) newer).getData()) ? newer : older;
   }
@@ -43,13 +43,9 @@ public HoodieRecord preCombine(HoodieRecord older, HoodieRecord newer) {
   public Option<HoodieRecord> combineAndGetUpdateValue(HoodieRecord older, HoodieRecord newer, Schema schema, Properties props) throws IOException {
     Option<IndexedRecord> previousRecordAvroPayload;
     if (older instanceof HoodieAvroIndexedRecord) {
-      previousRecordAvroPayload = Option.of(((HoodieAvroIndexedRecord) older).getData());
+      previousRecordAvroPayload = Option.ofNullable(((HoodieAvroIndexedRecord) older).getData());
     } else {
-      if (null == props) {
-        previousRecordAvroPayload = ((HoodieRecordPayload)older.getData()).getInsertValue(schema);
-      } else {
-        previousRecordAvroPayload = ((HoodieRecordPayload)older.getData()).getInsertValue(schema, props);
-      }
+      previousRecordAvroPayload = ((HoodieRecordPayload)older.getData()).getInsertValue(schema, props);
     }
     if (!previousRecordAvroPayload.isPresent()) {
       return Option.empty();
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieRecord.java b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieRecord.java
@@ -151,16 +151,15 @@ public HoodieRecord(HoodieKey key, T data, HoodieOperation operation, Comparable
     this.newLocation = null;
     this.sealed = false;
     this.operation = operation;
-    this.orderingVal = orderingVal;
+    // default natural order is 0
+    this.orderingVal = orderingVal == null ? 0 : orderingVal;
   }
 
   public HoodieRecord(HoodieRecord<T> record) {
-    this(record.key, record.data, record.orderingVal);
+    this(record.key, record.data, record.operation, record.orderingVal);
     this.currentLocation = record.currentLocation;
     this.newLocation = record.newLocation;
     this.sealed = record.sealed;
-    this.operation = record.operation;
-    this.orderingVal = record.orderingVal;
   }
 
   public HoodieRecord() {
@@ -181,10 +180,6 @@ public HoodieOperation getOperation() {
   }
 
   public Comparable getOrderingValue() {
-    if (null == orderingVal) {
-      // default natural order is 0
-      return 0;
-    }
     return orderingVal;
   }
 
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/ReflectionUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/util/ReflectionUtils.java
@@ -63,10 +63,6 @@ public static Class<?> getClass(String clazzName) {
     return CLAZZ_CACHE.get(clazzName);
   }
 
-  private static Object getInstance(String clazzName) {
-    return INSTANCE_CACHE.get(clazzName);
-  }
-
   public static <T> T loadClass(String fqcn) {
     try {
       return (T) getClass(fqcn).newInstance();
@@ -92,10 +88,10 @@ public static <T extends HoodieRecordPayload> T loadPayload(String recordPayload
    */
   public static HoodieMerge loadHoodieMerge(String mergeClass) {
     try {
-      HoodieMerge hoodieMerge = (HoodieMerge) getInstance(mergeClass);
+      HoodieMerge hoodieMerge = (HoodieMerge) INSTANCE_CACHE.get(mergeClass);
       if (null == hoodieMerge) {
         synchronized (HoodieMerge.class) {
-          hoodieMerge = (HoodieMerge) getInstance(mergeClass);
+          hoodieMerge = (HoodieMerge) INSTANCE_CACHE.get(mergeClass);
           if (null == hoodieMerge) {
             hoodieMerge = (HoodieMerge)loadClass(mergeClass, new Object[]{});
             INSTANCE_CACHE.put(mergeClass, hoodieMerge);
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputFormat.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputFormat.java
@@ -65,6 +65,7 @@
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Properties;
 import java.util.Set;
 import java.util.stream.IntStream;
 
@@ -761,8 +762,7 @@ private Option<IndexedRecord> mergeRowWithLog(
         String curKey) throws IOException {
       final HoodieAvroRecord<?> record = (HoodieAvroRecord) scanner.getRecords().get(curKey);
       GenericRecord historyAvroRecord = (GenericRecord) rowDataToAvroConverter.convert(tableSchema, curRow);
-      // TODO IndexedRecord to HoodieRecord
-      Option<HoodieRecord> resultRecord = hoodieMerge.combineAndGetUpdateValue(new HoodieAvroIndexedRecord(historyAvroRecord), record, tableSchema, null);
+      Option<HoodieRecord> resultRecord = hoodieMerge.combineAndGetUpdateValue(new HoodieAvroIndexedRecord(historyAvroRecord), record, tableSchema, new Properties());
       return ((HoodieAvroIndexedRecord) resultRecord.get()).toIndexedRecord();
     }
   }
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDD.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDD.scala
@@ -305,7 +305,6 @@ class HoodieMergeOnReadRDD(@transient sc: SparkContext,
     private def merge(curAvroRecord: GenericRecord, newRecord: HoodieRecord[_ <: HoodieRecordPayload[_]]): Option[IndexedRecord] = {
       // NOTE: We have to pass in Avro Schema used to read from Delta Log file since we invoke combining API
       //       on the record from the Delta Log
-      // TODO IndexedRecord to HoodieRecord
       if (hoodieMerge.combineAndGetUpdateValue(new HoodieAvroIndexedRecord(curAvroRecord), newRecord, logFileReaderAvroSchema, payloadProps).isPresent) {
         toScalaOption(hoodieMerge.combineAndGetUpdateValue(new HoodieAvroIndexedRecord(curAvroRecord), newRecord, logFileReaderAvroSchema, payloadProps)
           .get.asInstanceOf[HoodieAvroIndexedRecord].toIndexedRecord)
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieInternalRowUtils.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieInternalRowUtils.scala
@@ -23,40 +23,40 @@ import java.util
 import java.util.concurrent.ConcurrentHashMap
 import org.apache.avro.Schema
 import org.apache.hudi.AvroConversionUtils
+import org.apache.hudi.avro.HoodieAvroUtils
 import org.apache.hudi.avro.HoodieAvroUtils.{createFullName, fromJavaDate, toJavaDate}
 import org.apache.hudi.common.model.HoodieRecord.HoodieMetadataField
+import org.apache.hudi.common.util.ValidationUtils
 import org.apache.hudi.exception.HoodieException
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{GenericInternalRow, JoinedRow, MutableProjection, Projection}
 import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, ArrayData, GenericArrayData, MapData}
 import org.apache.spark.sql.hudi.ColumnStatsExpressionUtils.AllowedTransformationExpression.exprUtils.generateMutableProjection
 import org.apache.spark.sql.types._
-import scala.collection.mutable
 
+import scala.collection.mutable
 
+/**
+ * Helper class to do common stuff across Spark InternalRow.
+ * Provides common methods similar to {@link HoodieAvroUtils}.
+ */
 object HoodieInternalRowUtils {
 
   val projectionMap = new ConcurrentHashMap[(StructType, StructType), MutableProjection]
   val schemaMap = new ConcurrentHashMap[Schema, StructType]
   val SchemaPosMap = new ConcurrentHashMap[StructType, Map[String, (StructField, Int)]]
 
-  /**
-   * @see org.apache.hudi.avro.HoodieAvroUtils#stitchRecords(org.apache.avro.generic.GenericRecord, org.apache.avro.generic.GenericRecord, org.apache.avro.Schema)
-   */
   def stitchRecords(left: InternalRow, leftSchema: StructType, right: InternalRow, rightSchema: StructType, stitchedSchema: StructType): InternalRow = {
     val mergeSchema = StructType(leftSchema.fields ++ rightSchema.fields)
     val row = new JoinedRow(left, right)
-    val projection = getCacheProjection(mergeSchema, stitchedSchema)
+    val projection = getCachedProjection(mergeSchema, stitchedSchema)
     projection(row)
   }
 
-  /**
-   * @see org.apache.hudi.avro.HoodieAvroUtils#rewriteRecord(org.apache.avro.generic.GenericRecord, org.apache.avro.Schema)
-   */
   def rewriteRecord(oldRecord: InternalRow, oldSchema: StructType, newSchema: StructType): InternalRow = {
     val newRow = new GenericInternalRow(Array.fill(newSchema.fields.length)(null).asInstanceOf[Array[Any]])
 
-    val oldFieldMap = getCacheSchemaPosMap(oldSchema)
+    val oldFieldMap = getCachedSchemaPosMap(oldSchema)
     for ((field, pos) <- newSchema.fields.zipWithIndex) {
       var oldValue: AnyRef = null
       if (oldFieldMap.contains(field.name)) {
@@ -87,29 +87,21 @@ object HoodieInternalRowUtils {
     newRow
   }
 
-  /**
-   * @see org.apache.hudi.avro.HoodieAvroUtils#rewriteRecordWithNewSchema(org.apache.avro.generic.IndexedRecord, org.apache.avro.Schema, java.util.Map)
-   */
   def rewriteRecordWithNewSchema(oldRecord: InternalRow, oldSchema: StructType, newSchema: StructType, renameCols: util.Map[String, String]): InternalRow = {
     rewriteRecordWithNewSchema(oldRecord, oldSchema, newSchema, renameCols, new util.LinkedList[String]).asInstanceOf[InternalRow]
   }
 
-  /**
-   * @see org.apache.hudi.avro.HoodieAvroUtils#rewriteRecordWithNewSchema(java.lang.Object, org.apache.avro.Schema, org.apache.avro.Schema, java.util.Map, java.util.Deque)
-   */
   private def rewriteRecordWithNewSchema(oldRecord: Any, oldSchema: DataType, newSchema: DataType, renameCols: util.Map[String, String], fieldNames: util.Deque[String]): Any = {
     if (oldRecord == null) {
       null
     } else {
       newSchema match {
         case targetSchema: StructType =>
-          if (!oldRecord.isInstanceOf[InternalRow]) {
-            throw new IllegalArgumentException("cannot rewrite record with different type")
-          }
+          ValidationUtils.checkArgument(oldRecord.isInstanceOf[InternalRow], "cannot rewrite record with different type")
           val oldRow = oldRecord.asInstanceOf[InternalRow]
           val helper = mutable.Map[Integer, Any]()
 
-          val oldSchemaPos = getCacheSchemaPosMap(oldSchema.asInstanceOf[StructType])
+          val oldSchemaPos = getCachedSchemaPosMap(oldSchema.asInstanceOf[StructType])
           targetSchema.fields.zipWithIndex.foreach { case (field, i) =>
             fieldNames.push(field.name)
             if (oldSchemaPos.contains(field.name)) {
@@ -140,9 +132,7 @@ object HoodieInternalRowUtils {
 
           newRow
         case targetSchema: ArrayType =>
-          if (!oldRecord.isInstanceOf[ArrayData]) {
-            throw new IllegalArgumentException("cannot rewrite record with different type")
-          }
+          ValidationUtils.checkArgument(oldRecord.isInstanceOf[ArrayData], "cannot rewrite record with different type")
           val oldElementType = oldSchema.asInstanceOf[ArrayType].elementType
           val oldArray = oldRecord.asInstanceOf[ArrayData]
           val newElementType = targetSchema.elementType
@@ -153,9 +143,7 @@ object HoodieInternalRowUtils {
 
           newArray
         case targetSchema: MapType =>
-          if (!oldRecord.isInstanceOf[MapData]) {
-            throw new IllegalArgumentException("cannot rewrite record with different type")
-          }
+          ValidationUtils.checkArgument(oldRecord.isInstanceOf[MapData], "cannot rewrite record with different type")
           val oldValueType = oldSchema.asInstanceOf[MapType].valueType
           val oldKeyType = oldSchema.asInstanceOf[MapType].keyType
           val oldMap = oldRecord.asInstanceOf[MapData]
@@ -174,27 +162,21 @@ object HoodieInternalRowUtils {
     }
   }
 
-  /**
-   * @see org.apache.hudi.avro.HoodieAvroUtils#rewriteRecordWithMetadata(org.apache.avro.generic.GenericRecord, org.apache.avro.Schema, java.lang.String)
-   */
   def rewriteRecordWithMetadata(record: InternalRow, oldSchema: StructType, newSchema: StructType, fileName: String): InternalRow = {
     val newRecord = rewriteRecord(record, oldSchema, newSchema)
     newRecord.update(HoodieMetadataField.FILENAME_METADATA_FIELD.ordinal, fileName)
 
     newRecord
   }
 
-  /**
-   * @see org.apache.hudi.avro.HoodieAvroUtils#rewriteEvolutionRecordWithMetadata(org.apache.avro.generic.GenericRecord, org.apache.avro.Schema, java.lang.String)
-   */
   def rewriteEvolutionRecordWithMetadata(record: InternalRow, oldSchema: StructType, newSchema: StructType, fileName: String): InternalRow = {
     val newRecord = rewriteRecordWithNewSchema(record, oldSchema, newSchema, new util.HashMap[String, String]())
     newRecord.update(HoodieMetadataField.FILENAME_METADATA_FIELD.ordinal, fileName)
 
     newRecord
   }
 
-  def getCacheSchema(schema: Schema): StructType = {
+  def getCachedSchema(schema: Schema): StructType = {
     if (!schemaMap.contains(schema)) {
       schemaMap.synchronized {
         if (!schemaMap.contains(schema)) {
@@ -206,7 +188,7 @@ object HoodieInternalRowUtils {
     schemaMap.get(schema)
   }
 
-  private def getCacheProjection(from: StructType, to: StructType): Projection = {
+  private def getCachedProjection(from: StructType, to: StructType): Projection = {
     val schemaPair = (from, to)
     if (!projectionMap.contains(schemaPair)) {
       projectionMap.synchronized {
@@ -219,7 +201,7 @@ object HoodieInternalRowUtils {
     projectionMap.get(schemaPair)
   }
 
-  def getCacheSchemaPosMap(schema: StructType): Map[String, (StructField, Int)] = {
+  def getCachedSchemaPosMap(schema: StructType): Map[String, (StructField, Int)] = {
     if (!SchemaPosMap.contains(schema)) {
       SchemaPosMap.synchronized {
         if (!SchemaPosMap.contains(schema)) {
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieSparkRecord.java b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieSparkRecord.java
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestStructTypeSchemaEvolutionUtils.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestStructTypeSchemaEvolutionUtils.scala

Original file line number	Diff line number	Diff line change
`@@ -59,7 +59,8 @@ public HoodieData<HoodieRecord<T>> deduplicateRecords(`
`59`	`59`	`}).reduceByKey((rec1, rec2) -> {`
`60`	`60`	`@SuppressWarnings("unchecked")`
`61`	`61`	`HoodieRecord<T> reducedRecord = hoodieMerge.preCombine(rec1, rec2);`
`62`		`- return reducedRecord.newInstance();`
	`62`	`+ HoodieKey reducedKey = rec1.getData().equals(reducedRecord.getData()) ? rec1.getKey() : rec2.getKey();`
	`63`	`+ return reducedRecord.newInstance(reducedKey);`
`63`	`64`	`}, parallelism).map(Pair::getRight);`
`64`	`65`	`}`
`65`	`66`
Original file line number	Diff line number	Diff line change
`@@ -67,11 +67,10 @@ public List<HoodieRecord<T>> deduplicateRecords(`
`67`	`67`	`return keyedRecords.values().stream().map(x -> x.stream().map(Pair::getRight).reduce((rec1, rec2) -> {`
`68`	`68`	`@SuppressWarnings("unchecked")`
`69`	`69`	`HoodieRecord<T> reducedRecord = hoodieMerge.preCombine(rec1,rec2);`
`70`		`-`
`71`	`70`	`// we cannot allow the user to change the key or partitionPath, since that will affect`
`72`	`71`	`// everything`
`73`	`72`	`// so pick it from one of the records.`
`74`		`- return reducedRecord.newInstance();`
	`73`	`+ return reducedRecord.newInstance(rec1.getKey());`
`75`	`74`	`}).orElse(null)).filter(Objects::nonNull).collect(Collectors.toList());`
`76`	`75`	`}`
`77`	`76`	`}`