hudi-agent commented on code in PR #19576:
URL: https://github.com/apache/hudi/pull/19576#discussion_r3756552426
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkBulkInsertHelper.java:
##########
@@ -69,13 +71,18 @@ public HoodieWriteMetadata<HoodieData<WriteStatus>>
bulkInsert(final HoodieData<
final
Option<BulkInsertPartitioner> userDefinedBulkInsertPartitioner) {
HoodieWriteMetadata result = new HoodieWriteMetadata();
+ boolean isLsmTable =
table.getMetaClient().getTableConfig().isLSMTreeStorageLayout();
+ checkArgument(!isLsmTable || userDefinedBulkInsertPartitioner.isEmpty(),
Review Comment:
🤖 This guards the partitioner choice for LSM, but not duplicate record keys.
Flink's HoodieTableFactory.checkStorageLayout hard-rejects INSERT on LSM
"because they allow duplicate record keys", whereas this PR enables
INSERT/bulk_insert on Spark LSM with no equivalent guard. The updated
TestMORDataSource shows dup keys written via INSERT survive only until the
first merge, then silently collapse to one row. Is allowing INSERT here
intentional, and if so how do we avoid that silent data loss? @danny0405
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerFactory.java:
##########
@@ -45,12 +45,27 @@ public static BulkInsertPartitioner get(HoodieTable table,
return new RDDSimpleBucketBulkInsertPartitioner(table);
}
}
+ if (table.getMetaClient().getTableConfig().isLSMTreeStorageLayout()) {
Review Comment:
🤖 The BUCKET-index branch above returns before this LSM check, so an LSM
table configured with hoodie.index.type=BUCKET would get the bucket partitioner
and skip the LSM UTF-8 ordering entirely. Is that combination prevented at
table creation, or could it silently produce base files that aren't record-key
sorted and break the LSM reader? Same ordering applies in the WithRows factory.
<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);
Review Comment:
🤖 nit: the inner class name `LsmTableTestContext` uses mixed-case "Lsm"
while everywhere else in this PR — and in the outer class name — the acronym is
all-caps "LSM". Could you rename it to `LSMTableTestContext` for consistency?
<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,306 @@
+/*
+ * 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, 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, 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,
config, true));
+
+ assertEquals(expectedMessage, recordException.getMessage());
+ assertEquals(expectedMessage, rowException.getMessage());
+ }
+
+ private BulkInsertPartitioner<Dataset<Row>>
getRowPartitioner(BulkInsertSortMode sortMode) {
+ return BulkInsertInternalPartitionerWithRowsFactory.get(
+ lsmTable, createWriteConfig(sortMode, true), true);
+ }
+
+ private void assertRowPartitionerSelection(BulkInsertSortMode sortMode,
+ Class<?>
expectedPartitionerClass) {
+ HoodieWriteConfig config = createWriteConfig(sortMode, true);
+ assertEquals(expectedPartitionerClass,
+ BulkInsertInternalPartitionerWithRowsFactory.get(lsmTable, config,
true).getClass());
+ assertEquals(expectedPartitionerClass,
+ BulkInsertInternalPartitionerWithRowsFactory.get(lsmTable, 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,
Review Comment:
🤖 nit: `getRowPartitioner` and `getHoodieRecordPartitioner` look unused —
`assertRowPartitionerSelection` and `assertHoodieRecordPartitionerSelection`
call the factories directly. Could you remove them to avoid dead code?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMPartitionPathRepartitionAndSortPartitionerWithRows.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.HoodieRecord;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.functions;
+
+import static
org.apache.hudi.execution.bulkinsert.BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT;
+
+/**
+ * LSM row partitioner for {@link
BulkInsertSortMode#PARTITION_PATH_REPARTITION_AND_SORT}.
+ *
+ * <p>The LSM ordering is stronger than the default mode's partition-path-only
ordering, so this
+ * implementation reuses partition-path distribution and applies the LSM sort
once afterward.
+ */
+public class LSMPartitionPathRepartitionAndSortPartitionerWithRows
+ extends PartitionPathRepartitionPartitionerWithRows {
+
+ private final boolean shouldPopulateMetaFields;
+
+ public LSMPartitionPathRepartitionAndSortPartitionerWithRows(boolean
isTablePartitioned,
+
HoodieWriteConfig config) {
+ super(isTablePartitioned, config);
+ this.shouldPopulateMetaFields = config.populateMetaFields();
+ }
+
+ @Override
+ public Dataset<Row> repartitionRecords(Dataset<Row> rows, int
outputSparkPartitions) {
+ if (!shouldPopulateMetaFields) {
+ throw new HoodieException(
+ PARTITION_PATH_REPARTITION_AND_SORT.name() + " mode requires
meta-fields to be enabled");
+ }
+ return super.repartitionRecords(rows, outputSparkPartitions)
+
.sortWithinPartitions(functions.col(HoodieRecord.PARTITION_PATH_METADATA_FIELD),
functions.col(HoodieRecord.RECORD_KEY_METADATA_FIELD));
+ }
Review Comment:
🤖 nit: this line is quite long — could you break the two
`functions.col(...)` arguments onto separate lines to keep it within the usual
column limit?
<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]
