nsivabalan commented on code in PR #13603:
URL: https://github.com/apache/hudi/pull/13603#discussion_r2232001499
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndex.java:
##########
@@ -193,8 +193,15 @@ public enum IndexType {
@EnumFieldDescription("Index which saves the record key to location
mappings in the "
+ "HUDI Metadata Table. Record index is a global index, enforcing key
uniqueness across all "
- + "partitions in the table. Supports sharding to achieve very high
scale.")
- RECORD_INDEX
+ + "partitions in the table. Supports sharding to achieve very high
scale. For a table with "
+ + "keys that are only unique inside each partition, use
`PARTITIONED_RECORD_INDEX` instead.")
+ RECORD_INDEX,
+
+ @EnumFieldDescription("Index which saves the record key to location
mappings in the "
+ + "HUDI Metadata Table. This is a non global index, keys only need to
be unique inside each "
+ + "partition in the table. Supports sharding to achieve very high
scale. If a table has keys "
Review Comment:
minor.
`Index which saves the record key to location mappings in the HUDI Metadata
Table. Supports sharding to achieve very high scale. This is a non global
index, where keys can be replicated across partitions, since a pair of
partition path and record keys will uniquely map to a location using this
index. If users expect record keys to be unique across all partitions, use
`RECORD_INDEX` instead.`
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -1291,10 +1364,52 @@ protected Pair<List<HoodieFileGroupId>,
HoodieData<HoodieRecord>> tagRecordsWith
@Override
public void update(HoodieCommitMetadata commitMetadata, String instantTime) {
mayBeReinitMetadataReader();
+ maybeInitializeNewFileGroupsForPartitionedRLI(commitMetadata, instantTime);
processAndCommit(instantTime, new
BatchMetadataConversionFunction(instantTime, commitMetadata,
getMetadataPartitionsToUpdate()));
closeInternal();
}
+ /**
+ * This method is used to initialize new file groups for partitioned record
index during the in-flight commit.
+ * It will initialize new file groups for partitions that are newly added in
the inflight commit.
+ *
+ * @param commitMetadata metadata for the inflight commit
+ * @param instantTime Timestamp for the mdt commit
+ */
+ private void
maybeInitializeNewFileGroupsForPartitionedRLI(HoodieCommitMetadata
commitMetadata, String instantTime) {
+ if (dataWriteConfig.isPartitionedRecordIndexEnabled()) {
+ Set<String> partitionsTouchedByInflightCommit =
commitMetadata.getPartitionToWriteStats().keySet();
+
initializeNewFileGroupsForPartitionedRLIHelper(partitionsTouchedByInflightCommit,
instantTime);
+ }
+ }
+
+ private void
maybeInitializeNewFileGroupsForPartitionedRLI(HoodieData<WriteStatus>
writeStatus, String instantTime) {
+ if (dataWriteConfig.isPartitionedRecordIndexEnabled()) {
+ Set<String> partitionsTouchedByInflightCommit = new
HashSet<>(writeStatus.map(WriteStatus::getPartitionPath).collectAsList());
+
initializeNewFileGroupsForPartitionedRLIHelper(partitionsTouchedByInflightCommit,
instantTime);
+ }
+ }
+
+ private void initializeNewFileGroupsForPartitionedRLIHelper(Set<String>
partitionsTouchedByInflightCommit, String instantTime) {
+ try {
+ Set<String> partitionsFromFilesIndex = new
HashSet<>(metadata.getAllPartitionPaths());
+ if
(!partitionsFromFilesIndex.containsAll(partitionsTouchedByInflightCommit)) {
+ for (String partitionToWrite : partitionsTouchedByInflightCommit) {
+ if (!partitionsFromFilesIndex.contains(partitionToWrite)) {
Review Comment:
can we do
`partitionsTouchedByInflightCommit.removeAll(partitionsFromFilesIndex)` and
then do a for loop from the diff set
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -454,29 +457,26 @@ private boolean initializeFromFilesystem(String
dataTableInstantTime, List<Metad
String instantTimeForPartition =
generateUniqueInstantTime(dataTableInstantTime);
String partitionTypeName = partitionType.name();
LOG.info("Initializing MDT partition {} at instant {}",
partitionTypeName, instantTimeForPartition);
- String partitionName;
+ String relativePartitionPath;
Pair<Integer, HoodieData<HoodieRecord>> fileGroupCountAndRecordsPair;
- List<String> columnsToIndex = new ArrayList<>();
Lazy<Option<Schema>> tableSchema = Lazy.lazily(() ->
HoodieTableMetadataUtil.tryResolveSchemaForTable(dataMetaClient));
try {
switch (partitionType) {
case FILES:
fileGroupCountAndRecordsPair =
initializeFilesPartition(partitionIdToAllFilesMap);
- partitionName = FILES.getPartitionPath();
+ initializeFilegroupsAndCommitToMetadataPartition(partitionType,
FILES.getPartitionPath(), fileGroupCountAndRecordsPair,
instantTimeForPartition, Collections.emptyList());
Review Comment:
little too long of a name.
can we just keep it as `initializeFilegroupsAndCommit`
Anyways, entire class is meant to write to mdt :)
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -721,30 +700,126 @@ private Lazy<List<Pair<String, FileSlice>>>
getLazyLatestMergedPartitionFileSlic
});
}
+ void initializeFilegroupsAndCommitToMetadataPartition(MetadataPartitionType
partitionType,
Review Comment:
can we add a overloaded method which can avoid taking in the last arg which
is only required to be set for col stats partition
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/SparkMetadataTablePartitionedRecordIndex.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.index;
+
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.data.HoodieListData;
+import org.apache.hudi.common.data.HoodiePairData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
+import org.apache.hudi.common.util.Either;
+import org.apache.hudi.common.util.HoodieDataUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.data.HoodieJavaPairRDD;
+import org.apache.hudi.data.HoodieJavaRDD;
+import org.apache.hudi.metadata.HoodieTableMetadataUtil;
+import org.apache.hudi.metadata.MetadataPartitionType;
+import org.apache.hudi.metadata.BucketizedMetadataTableFileGroupIndexParser;
+import org.apache.hudi.table.HoodieTable;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.function.PairFlatMapFunction;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import scala.Tuple2;
+
+/**
+ * Index to be used with partitioned RLI. Queries the record index for tables
with non-global record keys
+ */
+public class SparkMetadataTablePartitionedRecordIndex extends
SparkMetadataTableRecordIndex {
+
+ public SparkMetadataTablePartitionedRecordIndex(HoodieWriteConfig config) {
+ super(config);
+ }
+
+ @Override
+ public boolean isGlobal() {
+ return false;
+ }
+
+ @Override
+ protected HoodieIndex.IndexType getFallbackIndexType() {
+ return IndexType.SIMPLE;
+ }
+
+ @Override
+ protected <R> HoodiePairData<String, HoodieRecordGlobalLocation>
lookupRecords(HoodieData<HoodieRecord<R>> records, HoodieEngineContext context,
+
HoodieTable hoodieTable, Either<Integer, Map<String, Integer>> fileGroupSize)
{
+ Map<String, Integer> fileGroupCountPerDataPartition =
fileGroupSize.asRight();
+ int numFileGroups = getTotalFileGroupCount(fileGroupSize);
+ Map<String, Integer> partitionOffsetIndexes =
BucketizedMetadataTableFileGroupIndexParser.generatePartitionToBaseIndexOffsets(fileGroupCountPerDataPartition);
+
+ // Partition the record keys to lookup such that each partition looks up
one record index shard
+ JavaRDD<Pair<String, String>> partitionedKeyRDD =
HoodieJavaRDD.getJavaRDD(records)
+ .filter(record ->
partitionOffsetIndexes.containsKey(record.getPartitionPath()))
+ .map(record -> Pair.of(record.getPartitionPath(),
record.getRecordKey()))
+ // get offset from partitionOffsetIndexes then add the hash of the key
+ .keyBy(k -> partitionOffsetIndexes.get(k.getLeft()) +
HoodieTableMetadataUtil.mapRecordKeyToFileGroupIndex(k.getRight(),
fileGroupCountPerDataPartition.get(k.getLeft())))
+ .partitionBy(new PartitionIdPassthrough(numFileGroups))
+ .map(t -> t._2);
+ ValidationUtils.checkState(partitionedKeyRDD.getNumPartitions() <=
numFileGroups);
+ // Lookup the keys in the record index
+ return HoodieJavaPairRDD.of(partitionedKeyRDD.mapPartitionsToPair(new
PartitionedRecordIndexFileGroupLookupFunction(hoodieTable)));
+ }
+
+ @Override
+ protected Either<Integer, Map<String, Integer>>
fetchFileGroupSize(HoodieTable hoodieTable) {
+ Map<String, Integer> partitionSizes = new HashMap<>();
+ Map<String, List<FileSlice>> fileGroups =
hoodieTable.getMetadataTable().getBucketizedFileGroupsForPartitionedRLI(MetadataPartitionType.RECORD_INDEX);
+ fileGroups.keySet().forEach(k -> partitionSizes.put(k,
fileGroups.get(k).size()));
+ return Either.right(partitionSizes);
+ }
+
+ @Override
+ protected int getTotalFileGroupCount(Either<Integer, Map<String, Integer>>
fileGroupSize) {
+ return
BucketizedMetadataTableFileGroupIndexParser.calculateNumberOfFileGroups(fileGroupSize.asRight());
+ }
+
+ @Override
+ protected boolean shouldUpdatePartitionPath(HoodieTable hoodieTable) {
+ return false;
+ }
+
+ /**
+ * Function that lookups a list of keys in a single shard of the record index
+ */
+ private static class PartitionedRecordIndexFileGroupLookupFunction
implements PairFlatMapFunction<Iterator<Pair<String,String>>, String,
HoodieRecordGlobalLocation> {
+ private final HoodieTable hoodieTable;
+
+ public PartitionedRecordIndexFileGroupLookupFunction(HoodieTable
hoodieTable) {
+ this.hoodieTable = hoodieTable;
+ }
+
+ @Override
+ public Iterator<Tuple2<String, HoodieRecordGlobalLocation>>
call(Iterator<Pair<String, String>> partitionPathRecordKeyIterator) throws
Exception {
+ //Needs to be final, so we must use 1 element array to store the value
+ final String[] partitionName = {null};
+ List<String> keysToLookup = new ArrayList<>();
+ partitionPathRecordKeyIterator.forEachRemaining(p -> {
+ keysToLookup.add(p.getRight());
+ if (partitionName[0] == null) {
+ partitionName[0] = p.getLeft();
+ }
+ });
+
+ // recordIndexInfo object only contains records that are present in
record_index.
+ assert partitionName[0] != null || keysToLookup.isEmpty();
+ Map<String, HoodieRecordGlobalLocation> recordIndexInfo =
HoodieDataUtils.dedupeAndCollectAsMap(
+
hoodieTable.getMetadataTable().readRecordIndex(HoodieListData.eager(keysToLookup),
Option.ofNullable(partitionName[0])));
Review Comment:
May I know why do we need to dedup?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionedRecordLevelIndex.scala:
##########
@@ -0,0 +1,461 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.functional
+
+import org.apache.hudi.DataSourceWriteOptions
+import org.apache.hudi.DataSourceWriteOptions._
+import org.apache.hudi.client.SparkRDDWriteClient
+import org.apache.hudi.client.common.HoodieSparkEngineContext
+import org.apache.hudi.common.config.{HoodieMetadataConfig, TypedProperties}
+import org.apache.hudi.common.data.HoodieListData
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.{HoodieRecordGlobalLocation,
HoodieTableType}
+import org.apache.hudi.common.table.TableSchemaResolver
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator
+import org.apache.hudi.common.testutils.RawTripTestPayload.recordsToStrings
+import org.apache.hudi.common.util.{Option => HOption}
+import org.apache.hudi.config.{HoodieCompactionConfig, HoodieIndexConfig,
HoodieWriteConfig}
+import
org.apache.hudi.functional.TestPartitionedRecordLevelIndex.TestPartitionedRecordLevelIndexTestCase
+import org.apache.hudi.index.HoodieIndex.IndexType.PARTITIONED_RECORD_INDEX
+import org.apache.hudi.metadata.HoodieBackedTableMetadata
+import org.apache.hudi.storage.StoragePath
+import
org.apache.hudi.table.action.compact.strategy.UnBoundedCompactionStrategy
+
+import org.apache.spark.sql.{Row, SaveMode}
+import org.apache.spark.sql.functions.lit
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse,
assertTrue, fail}
+import org.junit.jupiter.api.Tag
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.{Arguments, MethodSource, ValueSource}
+
+import java.util.stream.Collectors
+
+import scala.collection.JavaConverters
+import scala.collection.JavaConverters._
+
+@Tag("functional")
+class TestPartitionedRecordLevelIndex extends RecordLevelIndexTestBase {
+ private class testPartitionedRecordLevelIndexHolder {
+ var bulkRecordKeys: java.util.List[String] = null
+ var options: Map[String, String] = null
+ var recordKeys: java.util.List[String] = null
+ var newRecordKeys: java.util.List[String] = null
+ }
+
+ def testPartitionedRecordLevelIndex(tableType: HoodieTableType,
streamingWriteEnabled: Boolean, holder: testPartitionedRecordLevelIndexHolder):
Unit = {
+ val dataGen = new HoodieTestDataGenerator();
+ val inserts = dataGen.generateInserts("001", 5)
+ val latestBatch = recordsToStrings(inserts).asScala.toSeq
+ val latestBatchDf =
spark.read.json(spark.sparkContext.parallelize(latestBatch, 1))
+ val insertDf = latestBatchDf.withColumn("data_partition_path",
lit("partition1")).union(latestBatchDf.withColumn("data_partition_path",
lit("partition2")))
+ val options = Map(HoodieWriteConfig.TBL_NAME.key -> "hoodie_test",
+ DataSourceWriteOptions.TABLE_TYPE.key -> tableType.name(),
+ RECORDKEY_FIELD.key -> "_row_key",
+ PARTITIONPATH_FIELD.key -> "data_partition_path",
+ PRECOMBINE_FIELD.key -> "timestamp",
+ HoodieMetadataConfig.RECORD_INDEX_ENABLE_PROP.key()-> "false",
+ HoodieMetadataConfig.PARTITIONED_RECORD_INDEX_ENABLE_PROP.key() ->
"true",
+ HoodieMetadataConfig.STREAMING_WRITE_ENABLED.key() ->
streamingWriteEnabled.toString,
+ HoodieCompactionConfig.INLINE_COMPACT.key() -> "false",
+ HoodieIndexConfig.INDEX_TYPE.key() -> PARTITIONED_RECORD_INDEX.name())
+ holder.options = options
+ insertDf.write.format("org.apache.hudi")
+ .options(options)
+ .mode(SaveMode.Overwrite)
+ .save(basePath)
+ assertEquals(10, spark.read.format("hudi").load(basePath).count())
+ val props =
TypedProperties.fromMap(JavaConverters.mapAsJavaMapConverter(options).asJava)
+ val writeConfig = HoodieWriteConfig.newBuilder()
+ .withProps(props)
+ .withPath(basePath)
+ .build()
+ var metadata = metadataWriter(writeConfig).getTableMetadata
+ val recordKeys = inserts.asScala.map(i =>
i.getRecordKey).asJava.stream().collect(Collectors.toList())
+ holder.recordKeys = recordKeys
+ var partition1Locations = readRecordIndex(metadata, recordKeys,
HOption.of("partition1"))
+ assertEquals(5, partition1Locations.size)
+ var partition2Locations = readRecordIndex(metadata, recordKeys,
HOption.of("partition2"))
+ assertEquals(5, partition2Locations.size)
+ var df = spark.read.format("hudi").load(basePath).collect()
+ validateDFWithLocations(df, partition1Locations, "partition1")
+ validateDFWithLocations(df, partition2Locations, "partition2")
+
+ val newDeletes = dataGen.generateUpdates("004", 1)
+ val updates = dataGen.generateUniqueUpdates("002", 3)
+ val nextBatch = recordsToStrings(updates).asScala.toSeq
+ val nextBatchDf =
spark.read.json(spark.sparkContext.parallelize(nextBatch, 1))
+ val updateDf = nextBatchDf.withColumn("data_partition_path",
lit("partition1"))
+
+ updateDf.write.format("org.apache.hudi")
+ .options(options)
+ .option(DataSourceWriteOptions.OPERATION.key(), UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+ assertEquals(10, spark.read.format("hudi").load(basePath).count())
+ partition1Locations = readRecordIndex(metadata, recordKeys,
HOption.of("partition1"))
+ assertEquals(5, partition1Locations.size)
+ partition2Locations = readRecordIndex(metadata, recordKeys,
HOption.of("partition2"))
+ assertEquals(5, partition2Locations.size)
+ df = spark.read.format("hudi").load(basePath).collect()
+ validateDFWithLocations(df, partition1Locations, "partition1")
+ validateDFWithLocations(df, partition2Locations, "partition2")
+
+ val newInserts = dataGen.generateInserts("003", 3)
+ val newInsertBatch = recordsToStrings(newInserts).asScala.toSeq
+ val newInsertBatchDf =
spark.read.json(spark.sparkContext.parallelize(newInsertBatch, 1))
+ val newInsertDf = newInsertBatchDf.withColumn("data_partition_path",
lit("partition2")).union(newInsertBatchDf.withColumn("data_partition_path",
lit("partition3")))
+ newInsertDf.write.format("org.apache.hudi")
+ .options(options)
+ .option(DataSourceWriteOptions.OPERATION.key(), UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+ assertEquals(16, spark.read.format("hudi").load(basePath).count())
+ metadata = metadataWriter(writeConfig).getTableMetadata
+ partition1Locations = readRecordIndex(metadata, recordKeys,
HOption.of("partition1"))
+ assertEquals(5, partition1Locations.size)
+ partition2Locations = readRecordIndex(metadata, recordKeys,
HOption.of("partition2"))
+ assertEquals(5, partition2Locations.size)
+ df = spark.read.format("hudi").load(basePath).collect()
+ validateDFWithLocations(df, partition1Locations, "partition1")
+ validateDFWithLocations(df, partition2Locations, "partition2")
+
+ val newRecordKeys = newInserts.asScala.map(i =>
i.getRecordKey).asJava.stream().collect(Collectors.toList())
+ holder.newRecordKeys = newRecordKeys
+ partition1Locations = readRecordIndex(metadata, newRecordKeys,
HOption.of("partition1"))
+ assertEquals(0, partition1Locations.size)
+ partition2Locations = readRecordIndex(metadata, newRecordKeys,
HOption.of("partition2"))
+ assertEquals(3, partition2Locations.size)
+ var partition3Locations = readRecordIndex(metadata, newRecordKeys,
HOption.of("partition3"))
+ assertEquals(3, partition3Locations.size)
+ validateDFWithLocations(df, partition3Locations, "partition3")
+
+ val newDeletesBatch = recordsToStrings(newDeletes).asScala.toSeq
+ val newDeletesBatchDf =
spark.read.json(spark.sparkContext.parallelize(newDeletesBatch, 1))
+ val newDeletesDf = newDeletesBatchDf.withColumn("data_partition_path",
lit("partition1"))
+ newDeletesDf.write.format("org.apache.hudi")
+ .options(options)
+ .option(DataSourceWriteOptions.OPERATION.key(), DELETE_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+ assertEquals(15, spark.read.format("hudi").load(basePath).count())
+ metadata = metadataWriter(writeConfig).getTableMetadata
+ partition1Locations = readRecordIndex(metadata, recordKeys,
HOption.of("partition1"))
+ assertEquals(4, partition1Locations.size)
+ partition2Locations = readRecordIndex(metadata, recordKeys,
HOption.of("partition2"))
+ assertEquals(5, partition2Locations.size)
+ df = spark.read.format("hudi").load(basePath).collect()
+ validateDFWithLocations(df, partition1Locations, "partition1")
+ validateDFWithLocations(df, partition2Locations, "partition2")
+
+ assertFalse(partition1Locations.contains(newDeletes.get(0).getRecordKey))
+ assertTrue(partition2Locations.contains(newDeletes.get(0).getRecordKey))
+ partition1Locations = readRecordIndex(metadata, newRecordKeys,
HOption.of("partition1"))
+ assertEquals(0, partition1Locations.size)
+ partition2Locations = readRecordIndex(metadata, newRecordKeys,
HOption.of("partition2"))
+ assertEquals(3, partition2Locations.size)
+ partition3Locations = readRecordIndex(metadata, newRecordKeys,
HOption.of("partition3"))
+ assertEquals(3, partition3Locations.size)
+ validateDFWithLocations(df, partition2Locations, "partition2")
+ validateDFWithLocations(df, partition3Locations, "partition3")
+
+ val bulkInserts = dataGen.generateInserts("005", 5)
+ val bulkInsertBatch = recordsToStrings(bulkInserts).asScala.toSeq
+ val bulkInsertDf =
spark.read.json(spark.sparkContext.parallelize(bulkInsertBatch, 1))
+ val bulkInsertPartitionedDf =
bulkInsertDf.withColumn("data_partition_path", lit("partition0"))
+
+ // Use bulk_insert operation explicitly
+ bulkInsertPartitionedDf.write.format("org.apache.hudi")
+ .options(options)
+ .option(DataSourceWriteOptions.OPERATION.key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+
+ val bulkRecordKeys = bulkInserts.asScala.map(_.getRecordKey).asJava
+ holder.bulkRecordKeys = bulkRecordKeys
+ metadata = metadataWriter(writeConfig).getTableMetadata
+ val partition0Locations = readRecordIndex(metadata, bulkRecordKeys,
HOption.of("partition0"))
+ assertEquals(5, partition0Locations.size)
+ df = spark.read.format("hudi").load(basePath).collect()
+ validateDFWithLocations(df, partition0Locations, "partition0")
+ }
+
+ @ParameterizedTest
+ @MethodSource(Array("testArgsForPartitionedRecordLevelIndex"))
+ def testPartitionedRecordLevelIndexRollback(testCase:
TestPartitionedRecordLevelIndexTestCase): Unit = {
+ val holder = new testPartitionedRecordLevelIndexHolder
+ testPartitionedRecordLevelIndex(testCase.tableType,
testCase.streamingWriteEnabled, holder)
+ val writeConfig = getWriteConfig(holder.options)
+ new SparkRDDWriteClient(new HoodieSparkEngineContext(jsc), writeConfig)
+
.rollback(metaClient.getActiveTimeline.lastInstant().get().requestedTime())
+ val metadata = metadataWriter(writeConfig).getTableMetadata
+ try {
+ val partition0Locations = readRecordIndex(metadata,
holder.bulkRecordKeys, HOption.of("partition0"))
+ fail("rollback happened, so partition should be deleted")
+ } catch {
+ case t: Throwable => assertTrue(t.isInstanceOf[ArithmeticException])
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = Array(true, false))
+ def testPartitionedRecordLevelIndexCompact(streamingWriteEnabled: Boolean):
Unit = {
+ val holder = new testPartitionedRecordLevelIndexHolder
+ testPartitionedRecordLevelIndex(HoodieTableType.MERGE_ON_READ,
streamingWriteEnabled, holder)
+ assertEquals("deltacommit",
metaClient.getActiveTimeline.lastInstant().get().getAction)
+ val writeConfig = getWriteConfig(holder.options)
+ var metadata = metadataWriter(writeConfig).getTableMetadata
+ doAllAssertions(holder, metadata)
+ val writeClient = new SparkRDDWriteClient(new
HoodieSparkEngineContext(jsc), writeConfig)
+ val timeOpt = writeClient.scheduleCompaction(HOption.empty())
+ assertTrue(timeOpt.isPresent)
+ writeClient.compact(timeOpt.get())
+ metaClient.reloadActiveTimeline()
+ assertEquals("compaction",
metaClient.getActiveTimeline.lastInstant().get().getAction)
+ metadata = metadataWriter(writeConfig).getTableMetadata
+ doAllAssertions(holder, metadata)
+ }
+
+ @ParameterizedTest
+ @MethodSource(Array("testArgsForPartitionedRecordLevelIndex"))
+ def testPartitionedRecordLevelIndexCluster(testCase:
TestPartitionedRecordLevelIndexTestCase): Unit = {
+ val holder = new testPartitionedRecordLevelIndexHolder
+ testPartitionedRecordLevelIndex(testCase.tableType,
testCase.streamingWriteEnabled, holder)
+ assertEquals(if (testCase.tableType.equals(HoodieTableType.MERGE_ON_READ))
"deltacommit" else "commit",
+ metaClient.getActiveTimeline.lastInstant().get().getAction)
+ val writeConfig = getWriteConfig(holder.options ++
Map(HoodieWriteConfig.AVRO_SCHEMA_STRING.key() ->
HoodieTestDataGenerator.AVRO_SCHEMA.toString))
+ var metadata = metadataWriter(writeConfig).getTableMetadata
+ doAllAssertions(holder, metadata)
+ val writeClient = new SparkRDDWriteClient(new
HoodieSparkEngineContext(jsc), writeConfig)
+ val timeOpt = writeClient.scheduleClustering(HOption.empty())
+ assertTrue(timeOpt.isPresent)
+ writeClient.cluster(timeOpt.get())
+ metaClient.reloadActiveTimeline()
+ assertEquals("replacecommit",
metaClient.getActiveTimeline.lastInstant().get().getAction)
+ metadata = metadataWriter(writeConfig).getTableMetadata
+ doAllAssertions(holder, metadata)
+ }
+
+ private def validateDFWithLocations(df: Array[Row], locations: Map[String,
HoodieRecordGlobalLocation],
+ partition: String): Unit = {
+ var count: Int = 0
+ for (row <- df) {
+ val recordKey = row.getString(2)
+ locations.get(recordKey).foreach { loc =>
+ if (partition == row.getString(3)) {
+ count += 1
+ assertEquals(row.getString(3), loc.getPartitionPath)
+ assertEquals(FSUtils.getFileId(row.getString(4)), loc.getFileId)
+ }
+ }
+ }
+ assertEquals(locations.size, count)
+ }
+
+ private def doAllAssertions(holder: testPartitionedRecordLevelIndexHolder,
metadata: HoodieBackedTableMetadata): Unit = {
+ val df = spark.read.format("hudi").load(basePath).collect()
+ var partition0Locations = readRecordIndex(metadata, holder.recordKeys,
HOption.of("partition0"))
+ assertEquals(0, partition0Locations.size)
+ var partition1Locations = readRecordIndex(metadata, holder.recordKeys,
HOption.of("partition1"))
+ assertEquals(4, partition1Locations.size)
+ validateDFWithLocations(df, partition1Locations, "partition1")
+ var partition2Locations = readRecordIndex(metadata, holder.recordKeys,
HOption.of("partition2"))
+ assertEquals(5, partition2Locations.size)
+ validateDFWithLocations(df, partition2Locations, "partition2")
+ var partition3Locations = readRecordIndex(metadata, holder.recordKeys,
HOption.of("partition3"))
+ assertEquals(0, partition3Locations.size)
+
+ partition0Locations = readRecordIndex(metadata, holder.newRecordKeys,
HOption.of("partition0"))
+ assertEquals(0, partition0Locations.size)
+ partition1Locations = readRecordIndex(metadata, holder.newRecordKeys,
HOption.of("partition1"))
+ assertEquals(0, partition1Locations.size)
+ partition2Locations = readRecordIndex(metadata, holder.newRecordKeys,
HOption.of("partition2"))
+ assertEquals(3, partition2Locations.size)
+ validateDFWithLocations(df, partition2Locations, "partition2")
+ partition3Locations = readRecordIndex(metadata, holder.newRecordKeys,
HOption.of("partition3"))
+ assertEquals(3, partition3Locations.size)
+ validateDFWithLocations(df, partition3Locations, "partition3")
+
+ partition0Locations = readRecordIndex(metadata, holder.bulkRecordKeys,
HOption.of("partition0"))
+ assertEquals(5, partition0Locations.size)
+ validateDFWithLocations(df, partition0Locations, "partition0")
+ partition1Locations = readRecordIndex(metadata, holder.bulkRecordKeys,
HOption.of("partition1"))
+ assertEquals(0, partition1Locations.size)
+ partition2Locations = readRecordIndex(metadata, holder.bulkRecordKeys,
HOption.of("partition2"))
+ assertEquals(0, partition2Locations.size)
+ partition3Locations = readRecordIndex(metadata, holder.bulkRecordKeys,
HOption.of("partition3"))
+ assertEquals(0, partition3Locations.size)
+ }
+
+ @ParameterizedTest
+ @MethodSource(Array("testArgsForPartitionedRecordLevelIndex"))
+ def testPartitionedRecordLevelIndexInitializationBasic(testCase:
TestPartitionedRecordLevelIndexTestCase): Unit = {
+ testPartitionedRecordLevelIndexInitialization(testCase.tableType,
testCase.streamingWriteEnabled, failAndDoRollback = false, compact = false,
cluster = false)
+ }
+
+ @ParameterizedTest
+ @MethodSource(Array("testArgsForPartitionedRecordLevelIndex"))
+ def testPartitionedRecordLevelIndexInitializationRollback(testCase:
TestPartitionedRecordLevelIndexTestCase): Unit = {
+ testPartitionedRecordLevelIndexInitialization(testCase.tableType,
testCase.streamingWriteEnabled, failAndDoRollback = true, compact = false,
cluster = false)
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = Array(true, false))
+ def
testPartitionedRecordLevelIndexInitializationCompact(streamingWriteEnabled:
Boolean): Unit = {
+
testPartitionedRecordLevelIndexInitialization(HoodieTableType.MERGE_ON_READ,
streamingWriteEnabled, failAndDoRollback = false, compact = true, cluster =
false)
+ }
+
+ @ParameterizedTest
+ @MethodSource(Array("testArgsForPartitionedRecordLevelIndex"))
+ def testPartitionedRecordLevelIndexInitializationCluster(testCase:
TestPartitionedRecordLevelIndexTestCase): Unit = {
+ testPartitionedRecordLevelIndexInitialization(testCase.tableType,
testCase.streamingWriteEnabled, failAndDoRollback = false, compact = false,
cluster = true)
+ }
+
+ def testPartitionedRecordLevelIndexInitialization(tableType: HoodieTableType,
+ streamingWriteEnabled:
Boolean,
+ failAndDoRollback: Boolean,
+ compact: Boolean,
+ cluster: Boolean): Unit = {
+ initMetaClient(tableType)
+ val dataGen = new HoodieTestDataGenerator()
+ val inserts = dataGen.generateInserts("001", 5)
+ val latestBatch = recordsToStrings(inserts).asScala.toSeq
+ val latestBatchDf =
spark.read.json(spark.sparkContext.parallelize(latestBatch, 1))
+ val insertDf = latestBatchDf.withColumn("data_partition_path",
lit("partition1")).union(latestBatchDf.withColumn("data_partition_path",
lit("partition2")))
+ val options = Map(HoodieWriteConfig.TBL_NAME.key -> "hoodie_test",
+ DataSourceWriteOptions.TABLE_TYPE.key -> tableType.name(),
+ RECORDKEY_FIELD.key -> "_row_key",
+ PARTITIONPATH_FIELD.key -> "data_partition_path",
+ PRECOMBINE_FIELD.key -> "timestamp",
+ HoodieMetadataConfig.RECORD_INDEX_ENABLE_PROP.key()-> "false",
+ HoodieMetadataConfig.SECONDARY_INDEX_ENABLE_PROP.key() -> "false",
+ HoodieMetadataConfig.STREAMING_WRITE_ENABLED.key() ->
streamingWriteEnabled.toString,
+ HoodieCompactionConfig.INLINE_COMPACT.key() -> "false",
+ HoodieIndexConfig.INDEX_TYPE.key() -> PARTITIONED_RECORD_INDEX.name())
+ insertDf.write.format("org.apache.hudi")
+ .options(options)
+ .mode(SaveMode.Overwrite)
+ .save(basePath)
+
+ assertEquals(10, spark.read.format("hudi").load(basePath).count())
+
+ val updates = dataGen.generateUniqueUpdates("002", 3)
+ val nextBatch = recordsToStrings(updates).asScala.toSeq
+ val nextBatchDf =
spark.read.json(spark.sparkContext.parallelize(nextBatch, 1))
+ val updateDf = nextBatchDf.withColumn("data_partition_path",
lit("partition1"))
+ updateDf.write.format("org.apache.hudi")
+ .options(options)
+ .option(DataSourceWriteOptions.OPERATION.key(), UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+ assertEquals(10, spark.read.format("hudi").load(basePath).count())
+ metaClient.reloadActiveTimeline()
+ val tableSchemaResolver = new TableSchemaResolver(metaClient)
+ val latestTableSchemaFromCommitMetadata =
tableSchemaResolver.getTableAvroSchemaFromLatestCommit(false)
+
+ if (failAndDoRollback) {
+ val updatesToFail = dataGen.generateUniqueUpdates("003", 3)
+ val batchToFail = recordsToStrings(updatesToFail).asScala.toSeq
+ val batchToFailDf =
spark.read.json(spark.sparkContext.parallelize(batchToFail, 1))
+ val failDf = batchToFailDf.withColumn("data_partition_path",
lit("partition1")).union(batchToFailDf.withColumn("data_partition_path",
lit("partition3")))
+ failDf.write.format("org.apache.hudi")
+ .options(options)
+ .option(DataSourceWriteOptions.OPERATION.key(),
UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+ assertEquals(13, spark.read.format("hudi").load(basePath).count())
+
+ metaClient.reloadActiveTimeline()
+ val lastInstant = metaClient.getActiveTimeline.lastInstant().get()
+ assertTrue(storage.deleteFile(new
StoragePath(metaClient.getTimelinePath,
metaClient.getInstantFileNameGenerator.getFileName(lastInstant))))
+ assertEquals(10, spark.read.format("hudi").load(basePath).count())
+
+ // rollback
+ val writeConfig = HoodieWriteConfig.newBuilder()
+ .withProps(TypedProperties.fromMap(JavaConverters
+ .mapAsJavaMapConverter(options ++
Map(HoodieWriteConfig.AVRO_SCHEMA_STRING.key() ->
latestTableSchemaFromCommitMetadata.get().toString)).asJava))
+ .withPath(basePath)
+ .build()
+ new SparkRDDWriteClient(new HoodieSparkEngineContext(jsc), writeConfig)
+ .rollback(lastInstant.requestedTime())
+ }
+
+ if (compact) {
+ assertEquals("deltacommit",
metaClient.getActiveTimeline.lastInstant().get().getAction)
+ val writeConfig = getWriteConfig(options ++
+ Map(HoodieCompactionConfig.COMPACTION_STRATEGY.key() ->
classOf[UnBoundedCompactionStrategy].getName,
+ HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key() ->
"1"))
+ val writeClient = new SparkRDDWriteClient(new
HoodieSparkEngineContext(jsc), writeConfig)
+ val timeOpt = writeClient.scheduleCompaction(HOption.empty())
+ assertTrue(timeOpt.isPresent)
+ writeClient.compact(timeOpt.get())
+ metaClient.reloadActiveTimeline()
+ assertEquals("compaction",
metaClient.getActiveTimeline.lastInstant().get().getAction)
+ }
+
+ if (cluster) {
+ assertEquals(if (tableType.equals(HoodieTableType.MERGE_ON_READ))
"deltacommit" else "commit",
+ metaClient.getActiveTimeline.lastInstant().get().getAction)
+ val writeConfig = getWriteConfig(options ++
Map(HoodieWriteConfig.AVRO_SCHEMA_STRING.key() ->
HoodieTestDataGenerator.AVRO_SCHEMA.toString))
+ val writeClient = new SparkRDDWriteClient(new
HoodieSparkEngineContext(jsc), writeConfig)
+ val timeOpt = writeClient.scheduleClustering(HOption.empty())
+ assertTrue(timeOpt.isPresent)
+ writeClient.cluster(timeOpt.get())
+ metaClient.reloadActiveTimeline()
+ assertEquals("replacecommit",
metaClient.getActiveTimeline.lastInstant().get().getAction)
+ }
+
+ //init mdt
+ val updateOptions = options ++
Map(HoodieMetadataConfig.PARTITIONED_RECORD_INDEX_ENABLE_PROP.key() -> "true",
+ HoodieWriteConfig.AVRO_SCHEMA_STRING.key() ->
latestTableSchemaFromCommitMetadata.get().toString)
+ val props =
TypedProperties.fromMap(JavaConverters.mapAsJavaMapConverter(updateOptions).asJava)
+ val writeConfig = HoodieWriteConfig.newBuilder()
+ .withProps(props)
+ .withPath(basePath)
+ .build()
+ val metadata = metadataWriter(writeConfig).getTableMetadata
+ val recordKeys = inserts.asScala.map(i =>
i.getRecordKey).asJava.stream().collect(Collectors.toList())
+ val partition1Locations = readRecordIndex(metadata, recordKeys,
HOption.of("partition1"))
+ assertEquals(5, partition1Locations.size)
+ val partition2Locations = readRecordIndex(metadata, recordKeys,
HOption.of("partition2"))
+ assertEquals(5, partition2Locations.size)
+ val df = spark.read.format("hudi").load(basePath).collect()
+ validateDFWithLocations(df, partition1Locations, "partition1")
+ validateDFWithLocations(df, partition2Locations, "partition2")
+ }
+
+ def readRecordIndex(metadata: HoodieBackedTableMetadata, recordKeys:
java.util.List[String], dataTablePartition: HOption[String]): Map[String,
HoodieRecordGlobalLocation] = {
+ metadata.readRecordIndex(HoodieListData.eager(recordKeys),
dataTablePartition)
+ .collectAsList().asScala.map(p => p.getKey -> p.getValue).toMap
+ }
+}
+
+object TestPartitionedRecordLevelIndex {
+
+ case class TestPartitionedRecordLevelIndexTestCase(tableType:
HoodieTableType, streamingWriteEnabled: Boolean)
+
+ def testArgsForPartitionedRecordLevelIndex:
java.util.stream.Stream[Arguments] = {
+ java.util.stream.Stream.of(
+
Arguments.arguments(TestPartitionedRecordLevelIndexTestCase(HoodieTableType.COPY_ON_WRITE,
streamingWriteEnabled = true)),
+
Arguments.arguments(TestPartitionedRecordLevelIndexTestCase(HoodieTableType.COPY_ON_WRITE,
streamingWriteEnabled = false)),
+
Arguments.arguments(TestPartitionedRecordLevelIndexTestCase(HoodieTableType.MERGE_ON_READ,
streamingWriteEnabled = true)),
+
Arguments.arguments(TestPartitionedRecordLevelIndexTestCase(HoodieTableType.MERGE_ON_READ,
streamingWriteEnabled = false))
+ )
Review Comment:
I see we end up writing lot of end to end functional tests.
Can you take a look test classes in here
https://github.com/apache/hudi/tree/93e3df5af813290d0b60beb780b2791c40fb80a4/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io
Here we try to directly use write handles to write and then use metadata
writer to commit to mdt.
If we can use something similar, might save us test run time and we could
test more combinations
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
