fix

wangzixuan.wzxuan · wangzixuan.wzxuan · commit 33d151a10411 · 2022-09-19T11:13:41.000+08:00
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/RDDConsistentBucketPartitioner.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/RDDConsistentBucketPartitioner.java
@@ -22,7 +22,6 @@
 import org.apache.hudi.common.config.SerializableSchema;
 import org.apache.hudi.common.fs.FSUtils;
 import org.apache.hudi.common.model.ConsistentHashingNode;
-import org.apache.hudi.common.model.HoodieAvroRecord;
 import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
 import org.apache.hudi.common.model.HoodieKey;
 import org.apache.hudi.common.model.HoodieRecord;
@@ -235,8 +234,8 @@ private JavaRDD<HoodieRecord<T>> doPartitionAndCustomColumnSort(JavaRDD<HoodieRe
     final String[] sortColumns = sortColumnNames;
     final SerializableSchema schema = new SerializableSchema(HoodieAvroUtils.addMetadataFields((new Schema.Parser().parse(table.getConfig().getSchema()))));
     Comparator<HoodieRecord<T>> comparator = (Comparator<HoodieRecord<T>> & Serializable) (t1, t2) -> {
-      Object obj1 = HoodieAvroUtils.getRecordColumnValues((HoodieAvroRecord) t1, sortColumns, schema, consistentLogicalTimestampEnabled);
-      Object obj2 = HoodieAvroUtils.getRecordColumnValues((HoodieAvroRecord)t2, sortColumns, schema, consistentLogicalTimestampEnabled);
+      Object obj1 = t1.getRecordColumnValues(schema.get(), sortColumns, consistentLogicalTimestampEnabled);
+      Object obj2 = t2.getRecordColumnValues(schema.get(), sortColumns, consistentLogicalTimestampEnabled);
       return ((Comparable) obj1).compareTo(obj2);
     };
 
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReader.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReader.java
@@ -44,7 +44,6 @@ default ClosableIterator<HoodieRecord<InternalRow>> getRecordIterator(Schema rea
 
   @Override
   default ClosableIterator<HoodieRecord<InternalRow>> getRecordIterator(Schema readerSchema, Schema requestedSchema) throws IOException {
-    // TODO used in HoodieParquetDataBlock
-    return getRecordIterator(readerSchema);
+    return getRecordIterator(readerSchema, requestedSchema);
   }
 }
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/clustering/plan/strategy/TestSparkConsistentBucketClusteringPlanStrategy.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/clustering/plan/strategy/TestSparkConsistentBucketClusteringPlanStrategy.java
@@ -29,7 +29,7 @@
 import org.apache.hudi.common.testutils.HoodieTestUtils;
 import org.apache.hudi.common.util.collection.Triple;
 import org.apache.hudi.config.HoodieIndexConfig;
-import org.apache.hudi.config.HoodieStorageConfig;
+import org.apache.hudi.common.config.HoodieStorageConfig;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.index.HoodieIndex;
 import org.apache.hudi.index.bucket.ConsistentBucketIdentifier;
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestSparkConsistentBucketClustering.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestSparkConsistentBucketClustering.java
@@ -36,7 +36,7 @@
 import org.apache.hudi.config.HoodieClusteringConfig;
 import org.apache.hudi.config.HoodieCompactionConfig;
 import org.apache.hudi.config.HoodieIndexConfig;
-import org.apache.hudi.config.HoodieStorageConfig;
+import org.apache.hudi.common.config.HoodieStorageConfig;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.execution.bulkinsert.BulkInsertSortMode;
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
@@ -102,10 +102,6 @@ public String getFieldName() {
       IntStream.range(0, HOODIE_META_COLUMNS.size()).mapToObj(idx -> Pair.of(HOODIE_META_COLUMNS.get(idx), idx))
           .collect(Collectors.toMap(Pair::getKey, Pair::getValue));
 
-  public static final Map<String, Integer> HOODIE_META_COLUMNS_NAME_TO_POS_WITH_OPERATION =
-      IntStream.range(0, HOODIE_META_COLUMNS_WITH_OPERATION.size()).mapToObj(idx -> Pair.of(HOODIE_META_COLUMNS_WITH_OPERATION.get(idx), idx))
-          .collect(Collectors.toMap(Pair::getKey, Pair::getValue));
-
   public static int RECORD_KEY_META_FIELD_ORD = HOODIE_META_COLUMNS_NAME_TO_POS.get(RECORD_KEY_METADATA_FIELD);
   public static int PARTITION_PATH_META_FIELD_ORD = HOODIE_META_COLUMNS_NAME_TO_POS.get(PARTITION_PATH_METADATA_FIELD);
   public static int FILENAME_META_FIELD_ORD = HOODIE_META_COLUMNS_NAME_TO_POS.get(FILENAME_METADATA_FIELD);
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java
@@ -33,8 +33,14 @@
 import org.apache.hudi.avro.HoodieAvroUtils;
 import org.apache.hudi.common.fs.inline.InLineFSUtils;
 import org.apache.hudi.common.fs.inline.InLineFileSystem;
+import org.apache.hudi.common.model.HoodieAvroIndexedRecord;
 import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock.HoodieLogBlockContentLocation;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock.HoodieLogBlockType;
 import org.apache.hudi.common.util.ClosableIterator;
+import org.apache.hudi.common.util.MappingIterator;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.StringUtils;
 import org.apache.hudi.common.util.ValidationUtils;
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieParquetDataBlock.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieParquetDataBlock.java
@@ -151,7 +151,7 @@ protected <T> ClosableIterator<HoodieRecord<T>> readRecordsFromBlockPayload(Hood
         blockContentLoc.getBlockSize());
 
     ClosableIterator<HoodieRecord<T>> iterator = HoodieFileReaderFactory.getReaderFactory(type).getFileReader(inlineConf, inlineLogFilePath, PARQUET)
-        .getRecordIterator(readerSchema, readerSchema);
+        .getRecordIterator(readerSchema);
     return iterator;
   }
 
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java
@@ -31,7 +31,7 @@
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.StringUtils;
 import org.apache.hudi.common.util.collection.Pair;
-import org.apache.hudi.config.HoodieStorageConfig;
+import org.apache.hudi.common.config.HoodieStorageConfig;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.configuration.FlinkOptions;
 import org.apache.hudi.configuration.OptionsResolver;
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
@@ -48,7 +48,6 @@
 import org.apache.hudi.config.HoodieLockConfig;
 import org.apache.hudi.config.HoodieMemoryConfig;
 import org.apache.hudi.config.HoodiePayloadConfig;
-import org.apache.hudi.config.HoodieStorageConfig;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.configuration.FlinkOptions;
 import org.apache.hudi.configuration.HadoopConfigurations;
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/HoodieOptionConfig.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/HoodieOptionConfig.scala
@@ -77,7 +77,7 @@ object HoodieOptionConfig {
     .build()
 
   val SQL_MERGER_STRATEGY: HoodieSQLOption[String] = buildConf()
-    .withSqlKey("mergerSTRATEGY")
+    .withSqlKey("mergerStrategy")
     .withHoodieKey(DataSourceWriteOptions.MERGER_STRATEGY.key)
     .withTableConfigKey(HoodieTableConfig.MERGER_STRATEGY.key)
     .defaultValue(StringUtils.DEFAULT_MERGER_STRATEGY_UUID)
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/execution/benchmark/ReadAndWriteWithoutAvroBenchmark.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/execution/benchmark/ReadAndWriteWithoutAvroBenchmark.scala
@@ -121,10 +121,10 @@ object ReadAndWriteWithoutAvroBenchmark extends HoodieBenchmarkBase {
   private def upsertThenReadBenchmark(): Unit = {
     val avroMerger = classOf[HoodieAvroRecordMerger].getName
     val sparkMerger = classOf[HoodieSparkRecordMerger].getName
-    val df = createComplexDataFrame(100, 5)
+    val df = createComplexDataFrame(1000000, 1)
     withTempDir { avroPath =>
       withTempDir { sparkPath =>
-        val upsertBenchmark = new HoodieBenchmark("pref upsert", 100, 3)
+        val upsertBenchmark = new HoodieBenchmark("pref upsert", 1000000, 3)
         prepareHoodieTable(avroTable, new Path(avroPath.getCanonicalPath, avroTable).toUri.toString, "mor", avroMerger, df)
         prepareHoodieTable(sparkTable, new Path(sparkPath.getCanonicalPath, sparkTable).toUri.toString, "mor", sparkMerger, df)
         df.createOrReplaceTempView("input_df")
@@ -133,9 +133,9 @@ object ReadAndWriteWithoutAvroBenchmark extends HoodieBenchmarkBase {
             spark.sql(s"update $tableName set s1 = 's1_new' where id > 0")
           }
         }
-        // upsertBenchmark.run()
+        upsertBenchmark.run()
 
-        val readBenchmark = new HoodieBenchmark("pref read", 100000, 3)
+        val readBenchmark = new HoodieBenchmark("pref read", 1000000, 3)
         Seq(avroMerger, sparkMerger).zip(Seq(avroTable, sparkTable)).foreach {
           case (merger, tableName) => readBenchmark.addCase(merger) { _ =>
             spark.sql(s"select * from $tableName").collect()
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestHoodieOptionConfig.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestHoodieOptionConfig.scala
@@ -36,7 +36,6 @@ class TestHoodieOptionConfig extends SparkClientFunctionalTestHarness {
     assertTrue(with1("primaryKey") == "id")
     assertTrue(with1("type") == "cow")
     assertTrue(with1("payloadClass") == classOf[OverwriteWithLatestAvroPayload].getName)
-    assertTrue(with1("mergerImpls") == classOf[HoodieAvroRecordMerger].getName)
     assertTrue(with1("mergerStrategy") == StringUtils.DEFAULT_MERGER_STRATEGY_UUID)
 
     val ops2 = Map("primaryKey" -> "id",

Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,6 @@ default ClosableIterator<HoodieRecord<InternalRow>> getRecordIterator(Schema rea`
`44`	`44`
`45`	`45`	`@Override`
`46`	`46`	`default ClosableIterator<HoodieRecord<InternalRow>> getRecordIterator(Schema readerSchema, Schema requestedSchema) throws IOException {`
`47`		`- // TODO used in HoodieParquetDataBlock`
`48`		`- return getRecordIterator(readerSchema);`
	`47`	`+ return getRecordIterator(readerSchema, requestedSchema);`
`49`	`48`	`}`
`50`	`49`	`}`
Original file line number	Diff line number	Diff line change
`@@ -151,7 +151,7 @@ protected <T> ClosableIterator<HoodieRecord<T>> readRecordsFromBlockPayload(Hood`
`151`	`151`	`blockContentLoc.getBlockSize());`
`152`	`152`
`153`	`153`	`ClosableIterator<HoodieRecord<T>> iterator = HoodieFileReaderFactory.getReaderFactory(type).getFileReader(inlineConf, inlineLogFilePath, PARQUET)`
`154`		`- .getRecordIterator(readerSchema, readerSchema);`
	`154`	`+ .getRecordIterator(readerSchema);`
`155`	`155`	`return iterator;`
`156`	`156`	`}`
`157`	`157`