This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 862f4b9a5c92 fix(metadata): key RLI and SI write stats by file group,
not file id (#20188)
862f4b9a5c92 is described below
commit 862f4b9a5c92b82047ed40c8880708e4351bc5f2
Author: Vinish Reddy <[email protected]>
AuthorDate: Tue Oct 6 09:45:13 2026 +0530
fix(metadata): key RLI and SI write stats by file group, not file id
(#20188)
* fix(metadata): key RLI and SI write stats by file group, not file id
The record level index and the secondary index grouped a commit's write
stats by file id alone. A file group is identified by its partition and
file id, and for files written outside Hudi the file id is only unique
within a partition: one task of a Spark partitionBy write leaves the same
part-<task>-<job uuid> name in every partition it writes. The secondary
index then saw several base files under one key and failed with "Only one
new parquet file expected per file group per commit".
Group the write stats by HoodieFileGroupId in both indexers, and match the
replaced file groups the secondary index skips by HoodieFileGroupId too.
* Address review feedback
---
.../SecondaryIndexRecordGenerationUtils.java | 22 ++++---
.../metadata/index/record/BaseRecordIndexer.java | 18 +++---
.../TestExternalFileRecordAndSecondaryIndex.java | 36 +++++++++++
.../TestMetadataUtilRLIandSIRecordGeneration.java | 48 +++++++++++++++
.../TestSparkNonBlockingConcurrencyControl.java | 70 +++++++++++++++++++++-
5 files changed, 176 insertions(+), 18 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java
index 388e14d0d5a8..54cb8004fa90 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java
@@ -30,6 +30,7 @@ import org.apache.hudi.common.model.FileSlice;
import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.HoodieIndexDefinition;
import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodieRecord;
@@ -118,14 +119,17 @@ public class SecondaryIndexRecordGenerationUtils {
}
HoodieSchema tableSchema = resolveTableSchema(dataMetaClient,
commitMetadata);
- Map<String, List<HoodieWriteStat>> writeStatsByFileId =
allWriteStats.stream().collect(Collectors.groupingBy(HoodieWriteStat::getFileId));
- int parallelism = Math.max(Math.min(writeStatsByFileId.size(),
metadataConfig.getSecondaryIndexParallelism()), 1);
+ // a file group is identified by its partition and file id, the file id
alone is only unique within a partition for files
+ // written outside Hudi, e.g. Spark names a file part-00048-<job uuid> in
every partition its task writes
+ Map<HoodieFileGroupId, List<HoodieWriteStat>> writeStatsByFileGroupId =
allWriteStats.stream()
+ .collect(Collectors.groupingBy(writeStat -> new
HoodieFileGroupId(writeStat.getPartitionPath(), writeStat.getFileId())));
+ int parallelism = Math.max(Math.min(writeStatsByFileGroupId.size(),
metadataConfig.getSecondaryIndexParallelism()), 1);
ReaderContextFactory<T> readerContextFactory =
engineContext.getReaderContextFactory(dataMetaClient);
- HoodieData<HoodieRecord> secondaryIndexRecords =
engineContext.parallelize(new ArrayList<>(writeStatsByFileId.entrySet()),
parallelism).flatMap(writeStatsByFileIdEntry -> {
- String fileId = writeStatsByFileIdEntry.getKey();
- List<HoodieWriteStat> writeStats = writeStatsByFileIdEntry.getValue();
- String partition = writeStats.get(0).getPartitionPath();
+ HoodieData<HoodieRecord> secondaryIndexRecords =
engineContext.parallelize(new ArrayList<>(writeStatsByFileGroupId.entrySet()),
parallelism).flatMap(writeStatsByFileGroupIdEntry -> {
+ String fileId = writeStatsByFileGroupIdEntry.getKey().getFileId();
+ List<HoodieWriteStat> writeStats =
writeStatsByFileGroupIdEntry.getValue();
+ String partition =
writeStatsByFileGroupIdEntry.getKey().getPartitionPath();
StoragePath basePath = dataMetaClient.getBasePath();
// validate that for a given fileId, either we have 1 parquet file or N
log files.
@@ -209,7 +213,7 @@ public class SecondaryIndexRecordGenerationUtils {
// a replace commit without a known operation type registers files
written outside Hudi and drops the replaced
// file groups without rewriting their records under the same key
secondaryIndexRecords =
secondaryIndexRecords.union(convertReplacedFileGroupsToSecondaryIndexRecords(
- (HoodieReplaceCommitMetadata) commitMetadata,
writeStatsByFileId.keySet(), instantTime, indexDefinition, metadataConfig,
+ (HoodieReplaceCommitMetadata) commitMetadata,
writeStatsByFileGroupId.keySet(), instantTime, indexDefinition, metadataConfig,
dataMetaClient, engineContext, writeConfig, tableSchema));
}
@@ -253,7 +257,7 @@ public class SecondaryIndexRecordGenerationUtils {
* A table without record keys never writes a file group it replaces,
because the record index rejects such a commit.
*/
private static <T> HoodieData<HoodieRecord>
convertReplacedFileGroupsToSecondaryIndexRecords(HoodieReplaceCommitMetadata
replaceCommitMetadata,
-
Set<String> writtenFileIds,
+
Set<HoodieFileGroupId> writtenFileGroupIds,
String instantTime,
HoodieIndexDefinition indexDefinition,
HoodieMetadataConfig metadataConfig,
@@ -263,7 +267,7 @@ public class SecondaryIndexRecordGenerationUtils {
HoodieSchema tableSchema) {
List<Pair<String, String>> replacedFileGroups =
replaceCommitMetadata.getPartitionToReplaceFileIds().entrySet().stream()
.flatMap(partitionAndFileIds -> partitionAndFileIds.getValue().stream()
- .filter(fileId -> !writtenFileIds.contains(fileId))
+ .filter(fileId -> !writtenFileGroupIds.contains(new
HoodieFileGroupId(partitionAndFileIds.getKey(), fileId)))
.map(fileId -> Pair.of(partitionAndFileIds.getKey(), fileId)))
.collect(Collectors.toList());
if (replacedFileGroups.isEmpty()) {
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java
index e6904c843153..0ecec1ec15fc 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java
@@ -32,6 +32,7 @@ import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieDeltaWriteStat;
import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
@@ -392,8 +393,11 @@ public abstract class BaseRecordIndexer extends
BaseIndexer {
}
try {
- Map<String, List<HoodieWriteStat>> writeStatsByFileId =
allWriteStats.stream().collect(Collectors.groupingBy(HoodieWriteStat::getFileId));
- int parallelism = Math.max(Math.min(writeStatsByFileId.size(),
metadataConfig.getRecordIndexMaxParallelism()), 1);
+ // a file group is identified by its partition and file id, the file id
alone is only unique within a partition for files
+ // written outside Hudi
+ Map<HoodieFileGroupId, List<HoodieWriteStat>> writeStatsByFileGroupId =
allWriteStats.stream()
+ .collect(Collectors.groupingBy(writeStat -> new
HoodieFileGroupId(writeStat.getPartitionPath(), writeStat.getFileId())));
+ int parallelism = Math.max(Math.min(writeStatsByFileGroupId.size(),
metadataConfig.getRecordIndexMaxParallelism()), 1);
String basePath = dataTableMetaClient.getBasePath().toString();
StorageConfiguration storageConfiguration =
dataTableMetaClient.getStorageConf();
// a table without record keys, e.g. one that registers files written
outside Hudi, keys every row by file path and position
@@ -401,10 +405,10 @@ public abstract class BaseRecordIndexer extends
BaseIndexer {
Option<HoodieSchema> writerSchemaOpt =
HoodieTableMetadataUtil.tryResolveSchemaForTable(dataTableMetaClient);
Option<HoodieSchema> finalWriterSchemaOpt = writerSchemaOpt;
ReaderContextFactory<T> readerContextFactory =
engineContext.getReaderContextFactory(dataTableMetaClient);
- HoodieData<HoodieRecord> recordIndexRecords =
engineContext.parallelize(new ArrayList<>(writeStatsByFileId.entrySet()),
parallelism)
- .flatMap(writeStatsByFileIdEntry -> {
- String fileId = writeStatsByFileIdEntry.getKey();
- List<HoodieWriteStat> writeStats =
writeStatsByFileIdEntry.getValue();
+ HoodieData<HoodieRecord> recordIndexRecords =
engineContext.parallelize(new ArrayList<>(writeStatsByFileGroupId.entrySet()),
parallelism)
+ .flatMap(writeStatsByFileGroupIdEntry -> {
+ String fileId = writeStatsByFileGroupIdEntry.getKey().getFileId();
+ List<HoodieWriteStat> writeStats =
writeStatsByFileGroupIdEntry.getValue();
// Partition the write stats into base file and log file write
stats
List<HoodieWriteStat> baseFileWriteStats = writeStats.stream()
.filter(writeStat -> FSUtils.isBaseFile(new
StoragePath(writeStat.getPath())))
@@ -428,7 +432,7 @@ public abstract class BaseRecordIndexer extends BaseIndexer
{
}
// Process log file write stats
if (!logFileWriteStats.isEmpty()) {
- String partitionPath =
logFileWriteStats.get(0).getPartitionPath();
+ String partitionPath =
writeStatsByFileGroupIdEntry.getKey().getPartitionPath();
List<String> currentLogFilePaths = logFileWriteStats.stream()
.map(writeStat -> new
StoragePath(dataTableMetaClient.getBasePath(), writeStat.getPath()).toString())
.collect(Collectors.toList());
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestExternalFileRecordAndSecondaryIndex.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestExternalFileRecordAndSecondaryIndex.java
index 244e5aff8745..a39c9f947484 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestExternalFileRecordAndSecondaryIndex.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestExternalFileRecordAndSecondaryIndex.java
@@ -54,6 +54,7 @@ import org.apache.spark.sql.Row;
import org.apache.spark.sql.execution.FileSourceScanExec;
import org.apache.spark.sql.execution.SparkPlan;
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
@@ -145,6 +146,41 @@ public class TestExternalFileRecordAndSecondaryIndex
extends HoodieClientTestBas
}
}
+ @Test
+ public void testFilesWithTheSameNameInDifferentPartitions() throws Exception
{
+ // Spark names a file part-<task>-<job uuid>, so one task of a partitioned
write leaves the same file name in every
+ // partition it writes, and the file ids of those files are only unique
within their partition
+ initExternalTable(PARTITION);
+ HoodieWriteConfig writeConfig = writeConfig(true, false);
+ writeClient = getHoodieWriteClient(writeConfig);
+ writeClient.setOperationType(WriteOperationType.UNKNOWN);
+
+ ExternalFile brazil = new ExternalFile(PARTITION, Option.empty(),
"part-00000.parquet");
+ ExternalFile chile = new ExternalFile("americas/chile", Option.empty(),
"part-00000.parquet");
+ commitReplace(Arrays.asList(
+ Pair.of(brazil, rows(1, "alice", 2, "bob")),
+ Pair.of(chile, rows(3, "alice", 4, "carol", 5, "dave"))),
Collections.emptyMap());
+
+ HoodieBackedTableMetadata tableMetadata = tableMetadata(writeConfig);
+ assertRecordIndex(tableMetadata, Option.empty(), brazil, 2);
+ assertRecordIndex(tableMetadata, Option.empty(), chile, 3);
+ assertEquals(mapOf("alice", setOf(brazil.key(0), chile.key(0)), "bob",
setOf(brazil.key(1)), "carol", setOf(chile.key(1))),
+ readSecondaryIndex(tableMetadata, secondaryIndexPartition(),
Arrays.asList("alice", "bob", "carol")));
+
+ // the second commit drops the file in brazil and writes a file with the
same name in a third partition, which must
+ // not keep the replaced file group in brazil in the secondary index
+ ExternalFile peru = new ExternalFile("americas/peru", Option.empty(),
"part-00000.parquet");
+ commitReplace(Collections.singletonList(Pair.of(peru, rows(6, "erin"))),
+ Collections.singletonMap(PARTITION,
Collections.singletonList(brazil.fileId())));
+
+ tableMetadata = tableMetadata(writeConfig);
+ assertTrue(readRecordIndex(tableMetadata, Option.empty(),
brazil.keys(2)).isEmpty());
+ assertRecordIndex(tableMetadata, Option.empty(), chile, 3);
+ assertRecordIndex(tableMetadata, Option.empty(), peru, 1);
+ assertEquals(mapOf("alice", setOf(chile.key(0)), "carol",
setOf(chile.key(1)), "erin", setOf(peru.key(0))),
+ readSecondaryIndex(tableMetadata, secondaryIndexPartition(),
Arrays.asList("alice", "bob", "carol", "erin")));
+ }
+
@ParameterizedTest
@MethodSource("partitionsAndPrefixes")
public void testIndexesAreBuiltFromRegisteredFilesWhenEnabledLater(String
partitionPath, Option<String> prefix) throws Exception {
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestMetadataUtilRLIandSIRecordGeneration.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestMetadataUtilRLIandSIRecordGeneration.java
index e87d55653171..e227a17516d1 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestMetadataUtilRLIandSIRecordGeneration.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestMetadataUtilRLIandSIRecordGeneration.java
@@ -32,6 +32,7 @@ import org.apache.hudi.common.model.EmptyHoodieRecordPayload;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
import org.apache.hudi.common.model.HoodieIndexDefinition;
+import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTableType;
@@ -53,6 +54,7 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.metadata.BaseFileRecordParsingUtils;
+import org.apache.hudi.metadata.EmptyHoodieRecordPayloadWithPartition;
import org.apache.hudi.metadata.HoodieIndexVersion;
import org.apache.hudi.metadata.HoodieMetadataPayload;
import org.apache.hudi.metadata.HoodieTableMetadata;
@@ -301,6 +303,52 @@ public class TestMetadataUtilRLIandSIRecordGeneration
extends HoodieClientTestBa
}
}
+ /**
+ * A file id is only unique within a partition for files written outside
Hudi and for the buckets of a table under
+ * non-blocking concurrency control. The deletes written to log files of two
partitions under one file id must each
+ * carry their own partition in a partitioned record index.
+ */
+ @Test
+ public void
testRecordIndexRecordsForLogFilesWithTheSameFileIdInDifferentPartitions()
throws IOException {
+ cleanupClients();
+ Properties props = new Properties();
+ props.put(HoodieTableConfig.ORDERING_FIELDS.key(), "timestamp");
+ initMetaClient(HoodieTableType.MERGE_ON_READ, props);
+ cleanupTimelineService();
+ initTimelineService();
+
+ HoodieWriteConfig writeConfig =
getConfigBuilder(HoodieFailedWritesCleaningPolicy.EAGER).build();
+ try (SparkRDDWriteClient client = new SparkRDDWriteClient(new
HoodieSparkEngineContext(jsc), writeConfig)) {
+ String commitTime = client.startCommit();
+ List<HoodieRecord> inserts = dataGen.generateInserts(commitTime, 30);
+ List<WriteStatus> insertStatuses =
client.insert(jsc.parallelize(inserts, 1), commitTime).collect();
+ assertNoWriteErrors(insertStatuses);
+ client.commit(commitTime, jsc.parallelize(insertStatuses),
Option.empty(), DELTA_COMMIT_ACTION, Collections.emptyMap(), Option.empty());
+
+ // delete one record in each of two partitions, which writes one log
file in each
+ List<HoodieKey> deletedKeys = inserts.stream().map(HoodieRecord::getKey)
+ .collect(Collectors.toMap(HoodieKey::getPartitionPath, key -> key,
(first, second) -> first)).values().stream()
+ .limit(2).collect(Collectors.toList());
+ commitTime = client.startCommit();
+ List<WriteStatus> deleteStatuses =
client.delete(jsc.parallelize(deletedKeys, 1), commitTime).collect();
+ assertNoWriteErrors(deleteStatuses);
+ client.commit(commitTime, jsc.parallelize(deleteStatuses),
Option.empty(), DELTA_COMMIT_ACTION, Collections.emptyMap(), Option.empty());
+
+ // register both log files under one file id
+ List<HoodieWriteStat> writeStats =
deleteStatuses.stream().map(WriteStatus::getStat).collect(Collectors.toList());
+ assertEquals(2, writeStats.size());
+ writeStats.forEach(writeStat -> assertTrue(FSUtils.isLogFile(new
StoragePath(writeStat.getPath()))));
+ String sharedFileId = writeStats.get(0).getFileId();
+ writeStats.forEach(writeStat -> writeStat.setFileId(sharedFileId));
+
+ HoodieMetadataConfig partitionedRecordIndexConfig =
HoodieMetadataConfig.newBuilder().enable(true).withEnableRecordLevelIndex(true).build();
+ Map<String, String> deletedKeyToPartition =
convertMetadataToRecordIndexRecords(context, toCommitMetadata(writeStats),
partitionedRecordIndexConfig,
+ metaClient, writeConfig.getWritesFileIdEncoding(),
commitTime).collectAsList().stream()
+ .collect(Collectors.toMap(HoodieRecord::getRecordKey, record ->
((EmptyHoodieRecordPayloadWithPartition) record.getData()).getPartitionPath()));
+
assertEquals(deletedKeys.stream().collect(Collectors.toMap(HoodieKey::getRecordKey,
HoodieKey::getPartitionPath)), deletedKeyToPartition);
+ }
+ }
+
/**
* Tests methods used for Secondary Index record generation flows with MOR
table during index initialization as well as update.
*/
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestSparkNonBlockingConcurrencyControl.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestSparkNonBlockingConcurrencyControl.java
index 1cf246c5a664..be609b137fc3 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestSparkNonBlockingConcurrencyControl.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestSparkNonBlockingConcurrencyControl.java
@@ -23,6 +23,7 @@ import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.WriteClientTestUtils;
import org.apache.hudi.client.WriteStatus;
import
org.apache.hudi.client.transaction.BucketIndexConcurrentFileWritesConflictResolutionStrategy;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.config.RecordMergeMode;
import org.apache.hudi.common.fs.FSUtils;
@@ -41,9 +42,11 @@ import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.marker.MarkerType;
import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.CollectionUtils;
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.config.HoodieCleanConfig;
import org.apache.hudi.config.HoodieCompactionConfig;
import org.apache.hudi.config.HoodieIndexConfig;
@@ -52,8 +55,11 @@ import org.apache.hudi.config.HoodieLockConfig;
import org.apache.hudi.config.HoodiePayloadConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.core.transaction.lock.InProcessLockProvider;
+import org.apache.hudi.data.HoodieJavaRDD;
import org.apache.hudi.exception.HoodieWriteConflictException;
import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.metadata.HoodieBackedTableMetadata;
+import org.apache.hudi.metadata.HoodieTableMetadataUtil;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.action.HoodieWriteMetadata;
import org.apache.hudi.table.action.commit.SparkBucketIndexPartitioner;
@@ -603,11 +609,72 @@ public class TestSparkNonBlockingConcurrencyControl
extends SparkClientFunctiona
client2.close();
}
+ /**
+ * Every bucket keeps one file id in all partitions under non-blocking
concurrency control, so the record index and
+ * the secondary index must key the write stats of a commit by file group,
not by file id.
+ */
+ @Test
+ public void
testRecordAndSecondaryIndexWithTheSameBucketInDifferentPartitions() throws
Exception {
+ HoodieWriteConfig config = createHoodieWriteConfigBuilder(true)
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder()
+ .enable(true)
+ // the indexes are updated from the commit metadata
+ .withStreamingWriteEnabled(false)
+ .withEnableGlobalRecordLevelIndex(true)
+ .withSecondaryIndexEnabled(true)
+ .withSecondaryIndexForColumn("name")
+ .build())
+ .build();
+ metaClient = getHoodieMetaClient(HoodieTableType.MERGE_ON_READ,
config.getProps());
+
+ try (SparkRDDWriteClient client = getHoodieWriteClient(config)) {
+ // bulk insert writes a base file for the bucket in each partition,
under the same file id. The partitions are
+ // written in separate commits, because the bucket bulk insert
partitioner keys its tasks by file id prefix.
+ List<WriteStatus> writeStatuses = new ArrayList<>(writeData(client,
WriteClientTestUtils.createNewInstantTime(),
+ Collections.singletonList("id1,Danny,23,1,par1"), true,
WriteOperationType.BULK_INSERT, true));
+ writeStatuses.addAll(writeData(client,
WriteClientTestUtils.createNewInstantTime(),
+ Collections.singletonList("id2,Stephen,33,1,par2"), true,
WriteOperationType.BULK_INSERT, true));
+ assertEquals(1,
writeStatuses.stream().map(WriteStatus::getFileId).distinct().count());
+ assertRecordAndSecondaryIndex(config, "Danny", "Stephen");
+
+ // one upsert writes a log file for the bucket in both partitions, under
the same file id
+ writeStatuses = writeData(client,
WriteClientTestUtils.createNewInstantTime(),
+ Arrays.asList("id1,Julian,23,2,par1", "id2,Fabian,33,2,par2"), true,
WriteOperationType.UPSERT, true);
+ assertEquals(2, writeStatuses.size());
+ assertEquals(1,
writeStatuses.stream().map(WriteStatus::getFileId).distinct().count());
+ assertRecordAndSecondaryIndex(config, "Julian", "Fabian");
+ }
+ }
+
+ /** Asserts that id1 in par1 and id2 in par2 are in the record index, and
only under the given names in the secondary index. */
+ private void assertRecordAndSecondaryIndex(HoodieWriteConfig config, String
nameOfId1, String nameOfId2) {
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieBackedTableMetadata tableMetadata = new HoodieBackedTableMetadata(
+ context(), metaClient.getStorage(), config.getMetadataConfig(),
config.getBasePath(), true);
+ Map<String, String> recordKeyToPartition =
tableMetadata.readRecordIndexLocationsWithKeys(
+ HoodieJavaRDD.of(jsc().parallelize(Arrays.asList("id1", "id2"),
1))).collectAsList().stream()
+ .collect(Collectors.toMap(Pair::getKey, pair ->
pair.getValue().getPartitionPath()));
+ assertEquals(CollectionUtils.createImmutableMap(Pair.of("id1", "par1"),
Pair.of("id2", "par2")), recordKeyToPartition);
+
+ String secondaryIndexPartition =
metaClient.getIndexMetadata().get().getIndexDefinitions().keySet().stream()
+ .filter(indexName ->
indexName.startsWith(HoodieTableMetadataUtil.PARTITION_NAME_SECONDARY_INDEX_PREFIX))
+ .findFirst().get();
+ Map<String, String> secondaryKeyToRecordKey =
tableMetadata.readSecondaryIndexDataTableRecordKeysWithKeys(
+ HoodieJavaRDD.of(jsc().parallelize(Arrays.asList("Danny",
"Stephen", "Julian", "Fabian"), 1)), secondaryIndexPartition)
+ .collectAsList().stream()
+ .collect(Collectors.toMap(Pair::getKey, Pair::getValue));
+ assertEquals(CollectionUtils.createImmutableMap(Pair.of(nameOfId1, "id1"),
Pair.of(nameOfId2, "id2")), secondaryKeyToRecordKey);
+ }
+
private HoodieWriteConfig createHoodieWriteConfig() {
return createHoodieWriteConfig(false);
}
private HoodieWriteConfig createHoodieWriteConfig(boolean fullUpdate) {
+ return createHoodieWriteConfigBuilder(fullUpdate).build();
+ }
+
+ private HoodieWriteConfig.Builder createHoodieWriteConfigBuilder(boolean
fullUpdate) {
String payloadClassName = PartialUpdateAvroPayload.class.getName();
if (fullUpdate) {
payloadClassName = OverwriteWithLatestAvroPayload.class.getName();
@@ -647,8 +714,7 @@ public class TestSparkNonBlockingConcurrencyControl extends
SparkClientFunctiona
.withLockConfig(HoodieLockConfig.newBuilder()
.withLockProvider(InProcessLockProvider.class)
.withConflictResolutionStrategy(new
BucketIndexConcurrentFileWritesConflictResolutionStrategy())
- .build())
- .build();
+ .build());
}
private void checkWrittenData(
