fix new comments

Author: xiarixiaoyao

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

@@ -264,6 +264,7 @@ protected void commit(HoodieTable table, String commitActionType, String instant
         InternalSchema evolvedSchema = AvroSchemaEvolutionUtils.evolveSchemaFromNewAvroSchema(avroSchema, internalSchema);
         if (evolvedSchema.equals(internalSchema)) {
           metadata.addMetadata(SerDeHelper.LATEST_SCHEMA, SerDeHelper.toJson(evolvedSchema));
+          //TODO save history schema by metaTable
           schemasManager.persistHistorySchemaStr(instantTime, historySchemaStr);
         } else {
           evolvedSchema.setSchemaId(Long.parseLong(instantTime));

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java

@@ -155,6 +155,8 @@ public List<WriteStatus> compact(HoodieCompactionHandler compactionHandler,
     if (!StringUtils.isNullOrEmpty(config.getInternalSchema())) {
       readerSchema = new Schema.Parser().parse(config.getSchema());
       internalSchemaOption = SerDeHelper.fromJson(config.getInternalSchema());
+      // its safe to modify config here, since we running in task side.
+      ((HoodieTable) compactionHandler).getConfig().setDefault(config);
     } else {
       readerSchema = HoodieAvroUtils.addMetadataFields(
           new Schema.Parser().parse(config.getSchema()), config.allowOperationMetadataField());

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/RunCompactionActionExecutor.java

@@ -97,6 +97,7 @@ public HoodieWriteMetadata<HoodieData<WriteStatus>> execute() {
       metadata.addMetadata(HoodieCommitMetadata.SCHEMA_KEY, config.getSchema());
       if (schemaPair.getLeft().isPresent()) {
         metadata.addMetadata(SerDeHelper.LATEST_SCHEMA, schemaPair.getLeft().get());
+        metadata.addMetadata(HoodieCommitMetadata.SCHEMA_KEY, schemaPair.getRight().get());
       }
       compactionMetadata.setWriteStatuses(statuses);
       compactionMetadata.setCommitted(false);

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkInternalSchemaConverter.java

@@ -79,6 +79,7 @@ private SparkInternalSchemaConverter() {
   public static final String HOODIE_QUERY_SCHEMA = "hoodie.schema.internal.querySchema";
   public static final String HOODIE_TABLE_PATH = "hoodie.tablePath";
   public static final String HOODIE_VALID_COMMITS_LIST = "hoodie.valid.commits.list";
+
   /**
    * Converts a spark schema to an hudi internal schema. Fields without IDs are kept and assigned fallback IDs.
    *

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

@@ -935,7 +935,7 @@ public boolean hasNext() {
 
       @Override
       public GenericRecord next() {
-        return rewriteRecord(oldRecords.next(), newSchema);
+        return rewriteRecordWithNewSchema(oldRecords.next(), newSchema);
       }
     };
   }

hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java

@@ -559,7 +559,7 @@ public Option<InternalSchema> getTableInternalSchemaFromCommitMetadata() {
    */
   private Option<InternalSchema> getTableInternalSchemaFromCommitMetadata(HoodieInstant instant) {
     try {
-      HoodieTimeline timeline = metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
+      HoodieTimeline timeline = metaClient.getActiveTimeline().filterCompletedInstants();
       byte[] data = timeline.getInstantDetails(instant).get();
       HoodieCommitMetadata metadata = HoodieCommitMetadata.fromBytes(data, HoodieCommitMetadata.class);
       String latestInternalSchemaStr = metadata.getMetadata(SerDeHelper.LATEST_SCHEMA);

hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordReader.java

@@ -116,7 +116,7 @@ public abstract class AbstractHoodieLogRecordReader {
   // Total log files read - for metrics
   private AtomicLong totalLogFiles = new AtomicLong(0);
   // Internal schema
-  private InternalSchema internalSchema = InternalSchema.getEmptyInternalSchema();
+  private InternalSchema internalSchema;
   private final String path;
   // Total log blocks read - for metrics
   private AtomicLong totalLogBlocks = new AtomicLong(0);

hudi-common/src/main/java/org/apache/hudi/internal/schema/InternalSchema.java

@@ -26,6 +26,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -68,10 +69,9 @@ public InternalSchema(Field... columns) {
   public InternalSchema(long versionId, List<Field> cols) {
     this.versionId = versionId;
     this.record = RecordType.get(cols);
-    if (versionId >= 0) {
-      buildIdToName();
-      maxColumnId = idToName.keySet().stream().max(Comparator.comparing(Integer::valueOf)).get();
-    }
+    idToName = cols.isEmpty() ? new HashMap<>() : InternalSchemaBuilder.getBuilder().buildIdToName(record);
+    nameToId = cols.isEmpty() ? new HashMap<>() : idToName.entrySet().stream().collect(Collectors.toMap(Map.Entry::getValue, Map.Entry::getKey));
+    maxColumnId = idToName.isEmpty() ? -1 : idToName.keySet().stream().max(Comparator.comparing(Integer::valueOf)).get();
   }
 
   public InternalSchema(long versionId, int maxColumnId, List<Field> cols) {
@@ -90,20 +90,10 @@ public RecordType getRecord() {
   }
 
   private Map<Integer, String> buildIdToName() {
-    if (idToName == null) {
-      idToName = InternalSchemaBuilder.getBuilder().buildIdToName(record);
-    }
     return idToName;
   }
 
   private Map<String, Integer> buildNameToId() {
-    if (nameToId == null) {
-      if (idToName != null && !idToName.isEmpty()) {
-        nameToId = idToName.entrySet().stream().collect(Collectors.toMap(Map.Entry::getValue, Map.Entry::getKey));
-        return nameToId;
-      }
-      nameToId = InternalSchemaBuilder.getBuilder().buildNameToId(record);
-    }
     return nameToId;
   }
 

hudi-common/src/main/java/org/apache/hudi/internal/schema/action/InternalSchemaMerger.java

@@ -39,6 +39,13 @@ public class InternalSchemaMerger {
   // if mergeRequiredFiledForce is true, we will ignore the col's required attribute.
   private final boolean ignoreRequiredAttribute;
   // Whether to use column Type from file schema to read files when we find some column type has changed.
+  // spark parquetReader need the original column type to read data, otherwise the parquetReader will failed.
+  // eg: current column type is StringType, now we changed it to decimalType,
+  // we should not pass decimalType to parquetReader, we must pass StringType to it; when we read out the data, we convert data from String to Decimal, everything is ok.
+  // for log reader
+  // since our reWriteRecordWithNewSchema function support rewrite directly, so we no need this parameter
+  // eg: current column type is StringType, now we changed it to decimalType,
+  // we can pass decimalType to reWriteRecordWithNewSchema directly, everything is ok.
   private boolean useColumnTypeFromFileSchema = true;
 
   public InternalSchemaMerger(InternalSchema fileSchema, InternalSchema querySchema, boolean ignoreRequiredAttribute, boolean useColumnTypeFromFileSchema) {

hudi-common/src/main/java/org/apache/hudi/internal/schema/io/AbstractInternalSchemaStorageManager.java

@@ -37,6 +37,7 @@ abstract class AbstractInternalSchemaStorageManager {
   /**
    * Get latest history schema string.
    * Using give validCommits to validate all legal histroy Schema files, and return the latest one.
+   * If the passed valid commits is null or empty, valid instants will be fetched from the file-system and used.
    */
   public abstract String getHistorySchemaStrByGivenValidCommits(List<String> validCommits);