hudi-agent commented on code in PR #19576:
URL: https://github.com/apache/hudi/pull/19576#discussion_r3818495901
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java:
##########
@@ -3724,6 +3730,9 @@ public Builder withWriteIgnoreFailed(boolean
ignoreFailedWriteData) {
protected void setDefaults() {
writeConfig.setDefaultValue(MARKERS_TYPE,
getDefaultMarkersType(engineType));
+ if (writeConfig.isLSMTreeStorageLayout()) {
Review Comment:
🤖 This PARTITION_SORT default is gated on
`writeConfig.isLSMTreeStorageLayout()`, which reads
`hoodie.table.storage.layout` from the write-config props. But the partitioner
factory (`BulkInsertInternalPartitionerFactory.get(table, config)`) and
`requireSortedRecords()` both source the LSM flag from
`getMetaClient().getTableConfig()`. If a write config is built for an existing
LSM table without the storage-layout key in its own props (which
`requireSortedRecords()` already tolerates by reading the table config
instead), this default won't apply, the sort mode stays `NONE`, and the factory
will then throw "NONE ... does not guarantee record ordering". Could you source
the flag from the table config here as well, or confirm this path always has
the prop populated?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/execution/bulkinsert/TestLSMBulkInsertPartitioner.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.execution.bulkinsert;
+
+import org.apache.hudi.common.model.HoodieEmptyRecord;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.table.BulkInsertPartitioner;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.testutils.HoodieSparkClientTestHarness;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import scala.Tuple2;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests LSM bulk-insert ordering without changing the configured built-in
sort mode. */
+public class TestLSMBulkInsertPartitioner extends HoodieSparkClientTestHarness
{
+
+ private HoodieTable lsmTable;
+
+ private static final Comparator<Tuple2<String, String>> KEY_COMPARATOR =
(left, right) -> {
+ int partitionComparison = StringUtils.compareUtf8Bytes(left._1, right._1);
+ return partitionComparison != 0
+ ? partitionComparison
+ : StringUtils.compareUtf8Bytes(left._2, right._2);
+ };
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ initSparkContexts("TestLSMBulkInsertPartitioner");
+ initPath();
+ initHoodieStorage();
+
+ lsmTable = mock(HoodieTable.class);
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
+ when(lsmTable.getMetaClient()).thenReturn(metaClient);
+ when(metaClient.getTableConfig()).thenReturn(tableConfig);
+ when(tableConfig.isLSMTreeStorageLayout()).thenReturn(true);
+ when(lsmTable.isPartitioned()).thenReturn(true);
+ }
+
+ @AfterEach
+ public void tearDown() throws Exception {
+ cleanupResources();
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = BulkInsertSortMode.class, names = {
+ "GLOBAL_SORT", "PARTITION_SORT", "PARTITION_PATH_REPARTITION_AND_SORT"})
+ void testHoodieRecordPartitionerSortsSupportedModes(BulkInsertSortMode
sortMode) {
+ JavaRDD<HoodieRecord<Object>> input = jsc.parallelize(createRecords(), 3);
+ BulkInsertPartitioner<JavaRDD<HoodieRecord<Object>>> partitioner =
+ BulkInsertInternalPartitionerFactory.get(
+ lsmTable, createWriteConfig(sortMode, true));
+
+ JavaRDD<HoodieRecord<Object>> actual =
partitioner.repartitionRecords(input, 4);
+
+ assertSortedSparkPartitions(actual.glom().collect(), record ->
+ new Tuple2<>(record.getPartitionPath(), record.getRecordKey()));
+ assertDistributionSemantics(sortMode, actual);
+ assertTrue(partitioner.arePartitionRecordsSorted());
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = BulkInsertSortMode.class, names = {
+ "GLOBAL_SORT", "PARTITION_SORT", "PARTITION_PATH_REPARTITION_AND_SORT"})
+ void
testRowPartitionerSortsSupportedModesWithoutChangingSchema(BulkInsertSortMode
sortMode) {
+ StructType schema = new StructType()
+ .add(HoodieRecord.PARTITION_PATH_METADATA_FIELD, DataTypes.StringType,
false)
+ .add(HoodieRecord.RECORD_KEY_METADATA_FIELD, DataTypes.StringType,
false)
+ .add("value", DataTypes.IntegerType, false);
+ Dataset<Row> input = sqlContext.createDataFrame(
+ jsc.parallelize(createRows(), 3), schema);
+ BulkInsertPartitioner<Dataset<Row>> partitioner =
+ BulkInsertInternalPartitionerWithRowsFactory.get(
+ lsmTable.getMetaClient().getTableConfig(),
createWriteConfig(sortMode, true), true);
+
+ Dataset<Row> actual = partitioner.repartitionRecords(input, 4);
+
+ assertEquals(schema, actual.schema(), "Sorting must not add temporary
columns");
+ assertSortedSparkPartitions(actual.javaRDD().glom().collect(), row -> new
Tuple2<>(
+ row.getAs(HoodieRecord.PARTITION_PATH_METADATA_FIELD),
+ row.getAs(HoodieRecord.RECORD_KEY_METADATA_FIELD)));
+ assertDistributionSemantics(sortMode, actual.javaRDD());
+ assertTrue(partitioner.arePartitionRecordsSorted());
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = BulkInsertSortMode.class, names = {
+ "GLOBAL_SORT", "PARTITION_SORT", "PARTITION_PATH_REPARTITION_AND_SORT"})
+ void testPartitionersRequireMetaFields(BulkInsertSortMode sortMode) {
+ HoodieWriteConfig config = createWriteConfig(sortMode, false);
+ BulkInsertPartitioner<JavaRDD<HoodieRecord<Object>>> recordPartitioner =
+ BulkInsertInternalPartitionerFactory.get(lsmTable, config);
+ BulkInsertPartitioner<Dataset<Row>> rowPartitioner =
+ BulkInsertInternalPartitionerWithRowsFactory.get(
+ lsmTable.getMetaClient().getTableConfig(), config, true);
+
+ HoodieException recordException = assertThrows(HoodieException.class,
+ () -> recordPartitioner.repartitionRecords(jsc.emptyRDD(), 1));
+ HoodieException rowException = assertThrows(HoodieException.class,
+ () -> rowPartitioner.repartitionRecords(sparkSession.emptyDataFrame(),
1));
+
+ String expectedMessage = sortMode.name() + " mode requires meta-fields to
be enabled";
+ assertEquals(expectedMessage, recordException.getMessage());
+ assertEquals(expectedMessage, rowException.getMessage());
+ }
+
+ @Test
+ void testRowPartitionerSelectionForLsmModes() {
+ assertRowPartitionerSelection(
+ BulkInsertSortMode.GLOBAL_SORT, GlobalSortPartitionerWithRows.class);
+ assertRowPartitionerSelection(
+ BulkInsertSortMode.PARTITION_SORT,
PartitionSortPartitionerWithRows.class);
+ assertRowPartitionerSelection(
+ BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT,
+ LSMPartitionPathRepartitionAndSortPartitionerWithRows.class);
+ }
+
+ @Test
+ void testHoodieRecordPartitionerSelectionForLsmModes() {
+ assertHoodieRecordPartitionerSelection(
+ BulkInsertSortMode.GLOBAL_SORT, LSMGlobalSortPartitioner.class);
+ assertHoodieRecordPartitionerSelection(
+ BulkInsertSortMode.PARTITION_SORT, LSMPartitionSortPartitioner.class);
+ assertHoodieRecordPartitionerSelection(
+ BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT,
+ LSMPartitionPathRepartitionAndSortPartitioner.class);
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = BulkInsertSortMode.class, names = {"NONE",
"PARTITION_PATH_REPARTITION"})
+ void testNonSortingModesAreRejected(BulkInsertSortMode sortMode) {
+ HoodieWriteConfig config = createWriteConfig(sortMode, true);
+ String expectedMessage = "The bulk insert sort mode \"" + sortMode.name()
+ + "\" does not guarantee record ordering and is not supported for LSM
tables.";
+
+ HoodieException recordException = assertThrows(HoodieException.class,
+ () -> BulkInsertInternalPartitionerFactory.get(lsmTable, config));
+ HoodieException rowException = assertThrows(HoodieException.class,
+ () -> BulkInsertInternalPartitionerWithRowsFactory.get(
+ lsmTable.getMetaClient().getTableConfig(), config, true));
+
+ assertEquals(expectedMessage, recordException.getMessage());
+ assertEquals(expectedMessage, rowException.getMessage());
+ }
+
+ private BulkInsertPartitioner<Dataset<Row>>
getRowPartitioner(BulkInsertSortMode sortMode) {
+ return BulkInsertInternalPartitionerWithRowsFactory.get(
+ lsmTable.getMetaClient().getTableConfig(), createWriteConfig(sortMode,
true), true);
+ }
+
+ private void assertRowPartitionerSelection(BulkInsertSortMode sortMode,
+ Class<?>
expectedPartitionerClass) {
+ HoodieWriteConfig config = createWriteConfig(sortMode, true);
+ assertEquals(expectedPartitionerClass,
+ BulkInsertInternalPartitionerWithRowsFactory.get(
+ lsmTable.getMetaClient().getTableConfig(), config,
true).getClass());
+ assertEquals(expectedPartitionerClass,
+ BulkInsertInternalPartitionerWithRowsFactory.get(
+ lsmTable.getMetaClient().getTableConfig(), config, true,
true).getClass());
+ }
+
+ private BulkInsertPartitioner<JavaRDD<HoodieRecord<Object>>>
getHoodieRecordPartitioner(
+ BulkInsertSortMode sortMode) {
+ return BulkInsertInternalPartitionerFactory.get(
+ lsmTable, createWriteConfig(sortMode, true));
+ }
+
+ private void assertHoodieRecordPartitionerSelection(BulkInsertSortMode
sortMode,
+ Class<?>
expectedPartitionerClass) {
+ HoodieWriteConfig config = createWriteConfig(sortMode, true);
+ assertEquals(expectedPartitionerClass,
+ BulkInsertInternalPartitionerFactory.get(lsmTable, config).getClass());
+ assertEquals(expectedPartitionerClass,
+ BulkInsertInternalPartitionerFactory.get(lsmTable, config,
true).getClass());
+ }
+
+ private HoodieWriteConfig createWriteConfig(BulkInsertSortMode sortMode,
boolean populateMetaFields) {
+ return HoodieWriteConfig.newBuilder()
+ .withPath(basePath)
+ .withBulkInsertSortMode(sortMode.name())
+ .withPopulateMetaFields(populateMetaFields)
+ .build();
+ }
+
+ private <T> void assertDistributionSemantics(BulkInsertSortMode sortMode,
+ JavaRDD<T> actual) {
+ if (sortMode == BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT) {
+ assertEquals(4, actual.getNumPartitions());
+ assertEachTablePartitionRoutesToOneSparkPartition(actual);
+ }
+ }
+
+ private <T> void
assertEachTablePartitionRoutesToOneSparkPartition(JavaRDD<T> records) {
+ List<Tuple2<String, Integer>> partitionLocations =
records.mapPartitionsWithIndex(
+ (sparkPartition, iterator) -> {
+ Set<String> tablePartitions = new HashSet<>();
+ while (iterator.hasNext()) {
+ Object record = iterator.next();
+ tablePartitions.add(record instanceof HoodieRecord
+ ? ((HoodieRecord<?>) record).getPartitionPath()
+ : ((Row)
record).getAs(HoodieRecord.PARTITION_PATH_METADATA_FIELD));
+ }
+ List<Tuple2<String, Integer>> locations = new ArrayList<>();
+ tablePartitions.forEach(path -> locations.add(new Tuple2<>(path,
sparkPartition)));
Review Comment:
🤖 nit: `getRowPartitioner` and `getHoodieRecordPartitioner` (just below)
appear to be unused — none of the test methods call them; they inline the
factory calls directly. Could you remove these to avoid dead code?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnLsmStorage.java:
##########
@@ -0,0 +1,364 @@
+/*
+ * 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.client.functional;
+
+import org.apache.hudi.client.HoodieWriteResult;
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.WriteClientTestUtils;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.config.HoodieStorageConfig;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieInsertException;
+import org.apache.hudi.execution.bulkinsert.NonSortPartitioner;
+import org.apache.hudi.testutils.HoodieClientTestBase;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.getCommitTimeAtUTC;
+import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors;
+import static
org.apache.hudi.testutils.HoodieClientTestBase.wrapRecordsGenFunctionForPreppedCalls;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@Tag("functional")
+public class TestHoodieClientOnLsmStorage extends HoodieClientTestBase {
+
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableType.class)
+ void testInsert(HoodieTableType tableType) throws IOException {
+ LsmTableTestContext testContext = createTestContext(tableType);
+ try (SparkRDDWriteClient client =
getHoodieWriteClient(testContext.writeConfig)) {
+ String instantTime = getCommitTimeAtUTC(1);
+ List<HoodieRecord> records = generateInserts(testContext.dataGenerator,
instantTime);
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ commitWrite(client, instantTime, client.insert(jsc.parallelize(records,
2), instantTime));
+ assertCompletedOperation(testContext.metaClient, instantTime,
WriteOperationType.INSERT);
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableType.class)
+ void testInsertPrepped(HoodieTableType tableType) throws IOException {
+ LsmTableTestContext testContext = createTestContext(tableType);
+ try (SparkRDDWriteClient client =
getHoodieWriteClient(testContext.writeConfig)) {
+ String instantTime = getCommitTimeAtUTC(1);
+ List<HoodieRecord> records = wrapRecordsGenFunctionForPreppedCalls(
+ testContext.tablePath, storageConf, context, testContext.writeConfig,
+ testContext.dataGenerator::generateInserts).apply(instantTime, 4);
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ commitWrite(client, instantTime,
client.insertPreppedRecords(jsc.parallelize(records, 2), instantTime));
+ assertCompletedOperation(testContext.metaClient, instantTime,
WriteOperationType.INSERT_PREPPED);
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableType.class)
+ void testBulkInsert(HoodieTableType tableType) throws IOException {
+ LsmTableTestContext testContext = createTestContext(tableType);
+ try (SparkRDDWriteClient client =
getHoodieWriteClient(testContext.writeConfig)) {
+ String instantTime = getCommitTimeAtUTC(1);
+ List<HoodieRecord> records = generateInserts(testContext.dataGenerator,
instantTime);
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ commitWrite(client, instantTime,
client.bulkInsert(jsc.parallelize(records, 2), instantTime));
+ assertCompletedOperation(testContext.metaClient, instantTime,
WriteOperationType.BULK_INSERT);
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableType.class)
+ void testBulkInsertPrepped(HoodieTableType tableType) throws IOException {
+ LsmTableTestContext testContext = createTestContext(tableType);
+ try (SparkRDDWriteClient client =
getHoodieWriteClient(testContext.writeConfig)) {
+ String instantTime = getCommitTimeAtUTC(1);
+ List<HoodieRecord> records = wrapRecordsGenFunctionForPreppedCalls(
+ testContext.tablePath, storageConf, context, testContext.writeConfig,
+ testContext.dataGenerator::generateInserts).apply(instantTime, 4);
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ commitWrite(client, instantTime, client.bulkInsertPreppedRecords(
+ jsc.parallelize(records, 2), instantTime, Option.empty()));
+ assertCompletedOperation(
+ testContext.metaClient, instantTime,
WriteOperationType.BULK_INSERT_PREPPED);
+ }
+ }
+
+ @Test
+ void testRejectsCustomBulkInsertPartitionerBeforeInflight() throws
IOException {
+ LsmTableTestContext testContext =
createTestContext(HoodieTableType.COPY_ON_WRITE);
+ try (SparkRDDWriteClient client =
getHoodieWriteClient(testContext.writeConfig)) {
+ String instantTime = getCommitTimeAtUTC(1);
+ List<HoodieRecord> records = generateInserts(testContext.dataGenerator,
instantTime);
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+
+ HoodieInsertException exception =
assertThrows(HoodieInsertException.class, () -> client.bulkInsert(
+ jsc.parallelize(records, 2), instantTime, Option.of(new
NonSortPartitioner<>())));
+ assertTrue(exception.getCause() instanceof IllegalArgumentException);
+ assertEquals(
+ "User-defined bulk insert partitioners are not supported for LSM
tables because their record-key ordering cannot be verified",
+ exception.getCause().getMessage());
+
+ HoodieInstant instant =
testContext.metaClient.reloadActiveTimeline().getInstants().stream()
+ .filter(candidate -> candidate.requestedTime().equals(instantTime))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("No instant " + instantTime));
+ assertEquals(HoodieInstant.State.REQUESTED, instant.getState());
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableType.class)
+ void testUpsert(HoodieTableType tableType) throws IOException {
+ LsmTableTestContext testContext = createTestContext(tableType);
+ try (SparkRDDWriteClient client =
getHoodieWriteClient(testContext.writeConfig)) {
+ bootstrapTable(testContext, client);
+ String instantTime = getCommitTimeAtUTC(2);
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ commitWrite(client, instantTime, client.upsert(jsc.parallelize(
+ testContext.dataGenerator.generateUniqueUpdates(instantTime, 4), 2),
instantTime));
+ assertCompletedOperation(testContext.metaClient, instantTime,
WriteOperationType.UPSERT);
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableType.class)
+ void testUpsertPrepped(HoodieTableType tableType) throws IOException {
+ LsmTableTestContext testContext = createTestContext(tableType);
+ try (SparkRDDWriteClient client =
getHoodieWriteClient(testContext.writeConfig)) {
+ bootstrapTable(testContext, client);
+ String instantTime = getCommitTimeAtUTC(2);
+ List<HoodieRecord> records = wrapRecordsGenFunctionForPreppedCalls(
+ testContext.tablePath, storageConf, context, testContext.writeConfig,
+ testContext.dataGenerator::generateUniqueUpdates).apply(instantTime,
4);
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ commitWrite(client, instantTime,
client.upsertPreppedRecords(jsc.parallelize(records, 2), instantTime));
+ assertCompletedOperation(testContext.metaClient, instantTime,
WriteOperationType.UPSERT_PREPPED);
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableType.class)
+ void testDelete(HoodieTableType tableType) throws IOException {
+ LsmTableTestContext testContext = createTestContext(tableType);
+ try (SparkRDDWriteClient client =
getHoodieWriteClient(testContext.writeConfig)) {
+ bootstrapTable(testContext, client);
+ String instantTime = getCommitTimeAtUTC(2);
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ commitWrite(client, instantTime, client.delete(
+ jsc.parallelize(testContext.dataGenerator.generateUniqueDeletes(2),
2), instantTime));
+ assertCompletedOperation(testContext.metaClient, instantTime,
WriteOperationType.DELETE);
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableType.class)
+ void testDeletePrepped(HoodieTableType tableType) throws IOException {
+ LsmTableTestContext testContext = createTestContext(tableType);
+ try (SparkRDDWriteClient client =
getHoodieWriteClient(testContext.writeConfig)) {
+ bootstrapTable(testContext, client);
+ String instantTime = getCommitTimeAtUTC(2);
+ List<HoodieRecord> records = wrapRecordsGenFunctionForPreppedCalls(
+ testContext.tablePath, storageConf, context, testContext.writeConfig,
+
testContext.dataGenerator::generateUniqueDeleteRecords).apply(instantTime, 2);
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ commitWrite(client, instantTime,
client.deletePrepped(jsc.parallelize(records, 2), instantTime));
+ assertCompletedOperation(testContext.metaClient, instantTime,
WriteOperationType.DELETE_PREPPED);
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableType.class)
+ void testInsertOverwrite(HoodieTableType tableType) throws IOException {
+ LsmTableTestContext testContext = createTestContext(tableType);
+ try (SparkRDDWriteClient client =
getHoodieWriteClient(testContext.writeConfig)) {
+ bootstrapTable(testContext, client);
+ String instantTime = getCommitTimeAtUTC(2);
+ List<HoodieRecord> records =
testContext.dataGenerator.generateInsertsForPartition(
+ instantTime, 3,
HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH);
+ WriteClientTestUtils.startCommitWithTime(client, instantTime,
HoodieTimeline.REPLACE_COMMIT_ACTION);
+ HoodieWriteResult result =
client.insertOverwrite(jsc.parallelize(records, 1), instantTime);
+ commitReplace(client, instantTime, result);
+ assertReplaceCommit(
+ testContext.metaClient, instantTime,
WriteOperationType.INSERT_OVERWRITE,
+ result.getPartitionToReplaceFileIds(),
HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH);
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableType.class)
+ void testInsertOverwriteTable(HoodieTableType tableType) throws IOException {
+ LsmTableTestContext testContext = createTestContext(tableType);
+ try (SparkRDDWriteClient client =
getHoodieWriteClient(testContext.writeConfig)) {
+ bootstrapTable(testContext, client);
+ String instantTime = getCommitTimeAtUTC(2);
+ List<HoodieRecord> records =
testContext.dataGenerator.generateInsertsForPartition(
+ instantTime, 3,
HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH);
+ WriteClientTestUtils.startCommitWithTime(client, instantTime,
HoodieTimeline.REPLACE_COMMIT_ACTION);
+ HoodieWriteResult result =
client.insertOverwriteTable(jsc.parallelize(records, 1), instantTime);
+ commitReplace(client, instantTime, result);
+ assertReplaceCommit(
+ testContext.metaClient, instantTime,
WriteOperationType.INSERT_OVERWRITE_TABLE,
+ result.getPartitionToReplaceFileIds(),
HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH);
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableType.class)
+ void testDeletePartition(HoodieTableType tableType) throws IOException {
+ LsmTableTestContext testContext = createTestContext(tableType);
+ try (SparkRDDWriteClient client =
getHoodieWriteClient(testContext.writeConfig)) {
+ bootstrapTable(testContext, client);
+ String instantTime = getCommitTimeAtUTC(2);
+ WriteClientTestUtils.startCommitWithTime(client, instantTime,
HoodieTimeline.REPLACE_COMMIT_ACTION);
+ HoodieWriteResult result = client.deletePartitions(
+
Collections.singletonList(HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH),
instantTime);
+ commitReplace(client, instantTime, result);
+ assertReplaceCommit(
+ testContext.metaClient, instantTime,
WriteOperationType.DELETE_PARTITION,
+ result.getPartitionToReplaceFileIds(),
HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH);
+ }
+ }
+
+ private LsmTableTestContext createTestContext(HoodieTableType tableType)
throws IOException {
+ String tablePath = basePath + "_" + tableType.name().toLowerCase() +
"_lsm";
+ Properties tableProperties = getPropertiesForKeyGen(true);
+ tableProperties.setProperty(
+ HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
+ HoodieTableConfig.TableStorageLayout.LSM_TREE.configValue());
+ HoodieTableMetaClient lsmMetaClient = HoodieTestUtils.init(storageConf,
tablePath, tableType, tableProperties);
+ assertEquals(
+ HoodieTableConfig.TableStorageLayout.LSM_TREE,
+ lsmMetaClient.getTableConfig().getTableStorageLayout());
+
+ Properties writeProperties = new Properties();
+ writeProperties.setProperty(
+ HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
+ HoodieTableConfig.TableStorageLayout.LSM_TREE.configValue());
+
writeProperties.setProperty(HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT.key(),
"parquet");
+ HoodieWriteConfig writeConfig = getConfigBuilder()
+ .withPath(tablePath)
+ .withEmbeddedTimelineServerEnabled(false)
+ .withProperties(writeProperties)
+ .build();
+ return new LsmTableTestContext(
+ tablePath, lsmMetaClient, writeConfig, new
HoodieTestDataGenerator(0x19437));
+ }
+
+ private void bootstrapTable(LsmTableTestContext testContext,
SparkRDDWriteClient client) {
+ String instantTime = getCommitTimeAtUTC(1);
+ List<HoodieRecord> records = generateInserts(testContext.dataGenerator,
instantTime);
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ commitWrite(client, instantTime, client.insert(jsc.parallelize(records,
2), instantTime));
+ }
+
+ private List<HoodieRecord> generateInserts(HoodieTestDataGenerator
dataGenerator, String instantTime) {
+ List<HoodieRecord> records = new ArrayList<>();
+ records.addAll(dataGenerator.generateInsertsForPartition(
+ instantTime, 6, HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH));
+ records.addAll(dataGenerator.generateInsertsForPartition(
+ instantTime, 6,
HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH));
+ return records;
+ }
+
+ private void commitWrite(SparkRDDWriteClient client, String instantTime,
JavaRDD<WriteStatus> writeStatuses) {
+ assertNoWriteErrors(writeStatuses.collect());
+ assertTrue(client.commit(instantTime, writeStatuses));
+ }
+
+ private void commitReplace(SparkRDDWriteClient client, String instantTime,
HoodieWriteResult writeResult) {
+ assertNoWriteErrors(writeResult.getWriteStatuses().collect());
+ assertTrue(client.commit(
+ instantTime,
+ writeResult.getWriteStatuses(),
+ Option.empty(),
+ HoodieTimeline.REPLACE_COMMIT_ACTION,
+ writeResult.getPartitionToReplaceFileIds()));
+ }
+
+ private void assertCompletedOperation(
+ HoodieTableMetaClient metaClient, String instantTime, WriteOperationType
operationType) throws IOException {
+ HoodieInstant instant = findCompletedInstant(metaClient, instantTime);
+ HoodieCommitMetadata commitMetadata =
metaClient.getActiveTimeline().readCommitMetadata(instant);
+ assertEquals(operationType, commitMetadata.getOperationType());
+ }
+
+ private void assertReplaceCommit(
+ HoodieTableMetaClient metaClient,
+ String instantTime,
+ WriteOperationType operationType,
+ Map<String, List<String>> expectedReplacedFileIds,
+ String expectedPartition) throws IOException {
+ HoodieInstant instant = findCompletedInstant(metaClient, instantTime);
+ assertEquals(HoodieTimeline.REPLACE_COMMIT_ACTION, instant.getAction());
+ HoodieReplaceCommitMetadata commitMetadata =
metaClient.getActiveTimeline().readReplaceCommitMetadata(instant);
+ assertEquals(operationType, commitMetadata.getOperationType());
+ assertEquals(expectedReplacedFileIds,
commitMetadata.getPartitionToReplaceFileIds());
+
assertTrue(commitMetadata.getPartitionToReplaceFileIds().containsKey(expectedPartition));
+
assertFalse(commitMetadata.getPartitionToReplaceFileIds().get(expectedPartition).isEmpty());
Review Comment:
🤖 nit: could you rename `LsmTableTestContext` to `LSMTableTestContext`?
Every other new type in this PR uses the all-caps `LSM` acronym
(`LSMGlobalSortPartitioner`, `LSMPartitionSortPartitioner`, etc.), so the
mixed-case `Lsm` here stands out.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
