Revert "[HUDI-9794] Convert the avro into file bytes in LegacyArchivedMetaEnt…" (#13854)

This reverts commit e7b360d.

Author: danny0405

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/versioning/v1/TimelineArchiverV1.java

@@ -71,7 +71,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -443,14 +442,9 @@ private void writeToFile(Schema wrapperSchema, List<IndexedRecord> records) thro
       Map<HeaderMetadataType, String> header = new HashMap<>();
       header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, wrapperSchema.toString());
       final String keyField = table.getMetaClient().getTableConfig().getRecordKeyFieldProp();
-      List<HoodieRecord> indexRecords = records.stream()
-          .filter(Objects::nonNull)
-          .map(HoodieAvroIndexedRecord::new)
-          .collect(Collectors.toList());
-      if (!indexRecords.isEmpty()) {
-        HoodieAvroDataBlock block = new HoodieAvroDataBlock(indexRecords, header, keyField);
-        writer.appendBlock(block);
-      }
+      List<HoodieRecord> indexRecords = records.stream().map(HoodieAvroIndexedRecord::new).collect(Collectors.toList());
+      HoodieAvroDataBlock block = new HoodieAvroDataBlock(indexRecords, header, keyField);
+      writer.appendBlock(block);
       records.clear();
     }
   }

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/LegacyArchivedMetaEntryReader.java

@@ -100,7 +100,7 @@ private Pair<HoodieInstant, Option<byte[]>> readInstant(GenericRecord record) {
       Object actionData = record.get(key);
       if (actionData != null) {
         if (actionData instanceof IndexedRecord) {
-          return HoodieAvroUtils.avroToFileBytes((IndexedRecord) actionData);
+          return HoodieAvroUtils.avroToBytes((IndexedRecord) actionData);
         } else {
           // should be json bytes.
           try {

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

@@ -40,7 +40,6 @@
 import org.apache.avro.LogicalTypes.Decimal;
 import org.apache.avro.Schema;
 import org.apache.avro.Schema.Field;
-import org.apache.avro.file.DataFileWriter;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.GenericData.Record;
 import org.apache.avro.generic.GenericDatumReader;
@@ -162,19 +161,6 @@ public static <T extends IndexedRecord> ByteArrayOutputStream indexedRecordToByt
     }
   }
 
-  public static byte[] avroToFileBytes(IndexedRecord record) {
-    try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
-      try (DataFileWriter<IndexedRecord> writer = new DataFileWriter<>(new GenericDatumWriter<>(record.getSchema()))) {
-        writer.create(record.getSchema(), baos);
-        writer.append(record);
-        writer.flush();
-        return baos.toByteArray();
-      }
-    } catch (IOException e) {
-      throw new HoodieIOException("Failed to convert IndexedRecord to Avro file format", e);
-    }
-  }
-
   /**
    * Convert a given avro record to json and return the string
    *

hudi-common/src/main/java/org/apache/hudi/common/table/timeline/MetadataConversionUtils.java

@@ -183,9 +183,7 @@ public static HoodieArchivedMetaEntry createMetaWrapper(
     switch (actionType) {
       case HoodieTimeline.CLEAN_ACTION: {
         archivedMetaWrapper.setHoodieCleanMetadata(CleanerUtils.getCleanerMetadata(metaClient, new ByteArrayInputStream(instantDetails.get())));
-        if (planBytes.isPresent()) {
-          archivedMetaWrapper.setHoodieCleanerPlan(CleanerUtils.getCleanerPlan(metaClient, new ByteArrayInputStream(planBytes.get())));
-        }
+        archivedMetaWrapper.setHoodieCleanerPlan(CleanerUtils.getCleanerPlan(metaClient, new ByteArrayInputStream(planBytes.get())));
         archivedMetaWrapper.setActionType(ActionType.clean.name());
         break;
       }

hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSevenToEightUpgrade.scala

@@ -25,9 +25,8 @@ import org.apache.hudi.common.config.{HoodieMetadataConfig, RecordMergeMode, Typ
 import org.apache.hudi.common.model.{DefaultHoodieRecordPayload, HoodieRecordMerger, HoodieRecordPayload, HoodieTableType, OverwriteWithLatestAvroPayload, TableServiceType}
 import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient, HoodieTableVersion}
 import org.apache.hudi.common.table.timeline.InstantComparison.{compareTimestamps, GREATER_THAN_OR_EQUALS}
-import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion
 import org.apache.hudi.common.util.{Option, StringUtils}
-import org.apache.hudi.config.{HoodieArchivalConfig, HoodieCleanConfig, HoodieCompactionConfig, HoodieLockConfig, HoodieWriteConfig}
+import org.apache.hudi.config.{HoodieCleanConfig, HoodieCompactionConfig, HoodieLockConfig, HoodieWriteConfig}
 import org.apache.hudi.keygen.NonpartitionedKeyGenerator
 import org.apache.hudi.keygen.constant.KeyGeneratorType
 import org.apache.hudi.table.upgrade.{SparkUpgradeDowngradeHelper, UpgradeDowngrade}
@@ -286,94 +285,6 @@ class TestSevenToEightUpgrade extends RecordLevelIndexTestBase {
     assertEquals(1, df2.filter("price = 35").count())
   }
 
-  @Test
-  def testV6TableUpgradeToV9ToV6(): Unit = {
-    val partitionFields = "partition:simple"
-
-    val hudiOptsV6 = commonOpts ++ Map(
-      TABLE_TYPE.key -> HoodieTableType.COPY_ON_WRITE.name(),
-      KEYGENERATOR_CLASS_NAME.key -> KeyGeneratorType.CUSTOM.getClassName,
-      PARTITIONPATH_FIELD.key -> partitionFields,
-      "hoodie.metadata.enable" -> "true",
-      PAYLOAD_CLASS_NAME.key -> classOf[OverwriteWithLatestAvroPayload].getName,
-      RECORD_MERGE_MODE.key -> RecordMergeMode.COMMIT_TIME_ORDERING.name,
-      HoodieWriteConfig.WRITE_TABLE_VERSION.key -> "6",
-      HoodieWriteConfig.TIMELINE_LAYOUT_VERSION_NUM.key() -> Integer.toString(TimelineLayoutVersion.VERSION_1),
-      HoodieWriteConfig.AUTO_UPGRADE_VERSION.key -> "false",
-      HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.key -> "15",
-      HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key -> "3",
-      HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key -> "4",
-      HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key -> "5"
-    )
-
-    doWriteAndValidateDataAndRecordIndex(hudiOptsV6,
-      operation = INSERT_OPERATION_OPT_VAL,
-      saveMode = SaveMode.Overwrite,
-      validate = false)
-
-    for (i <- 1 to 10) {
-      doWriteAndValidateDataAndRecordIndex(hudiOptsV6,
-        operation = INSERT_OPERATION_OPT_VAL,
-        saveMode = SaveMode.Append,
-        validate = false)
-    }
-    metaClient = getLatestMetaClient(true)
-
-    assertEquals(HoodieTableVersion.SIX, metaClient.getTableConfig.getTableVersion)
-    assertEquals("partition", HoodieTableConfig.getPartitionFieldPropForKeyGenerator(metaClient.getTableConfig).get())
-
-    val archivePath = new org.apache.hudi.storage.StoragePath(metaClient.getArchivePath, ".commits_.archive*")
-    val archivedFiles = metaClient.getStorage.globEntries(archivePath)
-    println(s"Archived files found ${archivedFiles.size()}")
-
-    metaClient = HoodieTableMetaClient.builder()
-      .setBasePath(basePath)
-      .setConf(storage.getConf())
-      .build()
-
-    val hudiOptsUpgrade = hudiOptsV6 ++ Map(
-      HoodieWriteConfig.WRITE_TABLE_VERSION.key -> HoodieTableVersion.current().versionCode().toString
-    ) - HoodieWriteConfig.AUTO_UPGRADE_VERSION.key
-
-    doWriteAndValidateDataAndRecordIndex(hudiOptsUpgrade,
-      operation = UPSERT_OPERATION_OPT_VAL,
-      saveMode = SaveMode.Append,
-      validate = false)
-
-    metaClient = HoodieTableMetaClient.builder()
-      .setBasePath(basePath)
-      .setConf(storage.getConf())
-      .build()
-
-    assertEquals(HoodieTableVersion.current(), metaClient.getTableConfig.getTableVersion)
-    assertEquals(partitionFields, HoodieTableConfig.getPartitionFieldPropForKeyGenerator(metaClient.getTableConfig).get())
-
-    val archivedFilesAfterUpgrade = metaClient.getStorage.globEntries(archivePath)
-
-    assertTrue(archivedFilesAfterUpgrade.size() > 0,
-      "Even after upgrade, fresh table with ~12 commits should have archived files")
-
-    val hudiOptsDowngrade = hudiOptsV6 ++ Map(
-      HoodieWriteConfig.WRITE_TABLE_VERSION.key -> HoodieTableVersion.SIX.versionCode().toString
-    )
-
-    new UpgradeDowngrade(metaClient, getWriteConfig(hudiOptsDowngrade, basePath), context, SparkUpgradeDowngradeHelper.getInstance)
-      .run(HoodieTableVersion.SIX, null)
-
-    metaClient = HoodieTableMetaClient.builder()
-      .setBasePath(basePath)
-      .setConf(storage.getConf())
-      .build()
-
-    assertEquals(HoodieTableVersion.SIX, metaClient.getTableConfig.getTableVersion)
-    assertEquals("partition", HoodieTableConfig.getPartitionFieldPropForKeyGenerator(metaClient.getTableConfig).get())
-
-    val v6ArchivePath = new org.apache.hudi.storage.StoragePath(metaClient.getArchivePath, ".commits_.archive*")
-    val v6ArchivedFiles = metaClient.getStorage.globEntries(v6ArchivePath)
-
-    assertTrue(v6ArchivedFiles.size() > 0, "Downgrade should have archived files in V6 format")
-  }
-
   private def getWriteConfig(hudiOpts: Map[String, String], basePath: String): HoodieWriteConfig = {
     val props = TypedProperties.fromMap(hudiOpts.asJava)
     HoodieWriteConfig.newBuilder()