This is an automated email from the ASF dual-hosted git repository.
danny0405 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 287ea4f0c980 feat(spark): support standard writes and compaction for
LSM tables (#19576)
287ea4f0c980 is described below
commit 287ea4f0c980f0b215b09bca1e7f78e816545bb1
Author: Shuo Cheng <[email protected]>
AuthorDate: Fri Aug 21 10:33:25 2026 +0800
feat(spark): support standard writes and compaction for LSM tables (#19576)
* feat(spark): support standard writes and compaction for LSM tables
---
.../org/apache/hudi/config/HoodieWriteConfig.java | 9 +
.../java/org/apache/hudi/io/BaseCreateHandle.java | 4 +-
.../apache/hudi/config/TestHoodieWriteConfig.java | 35 ++
.../MultipleSparkJobExecutionStrategy.java | 4 +-
.../BulkInsertInternalPartitionerFactory.java | 21 +-
...lkInsertInternalPartitionerWithRowsFactory.java | 31 +-
.../bulkinsert/LSMBulkInsertRecordSorter.java | 98 +++++
.../bulkinsert/LSMGlobalSortPartitioner.java | 67 ++++
...PartitionPathRepartitionAndSortPartitioner.java | 88 ++++
...nPathRepartitionAndSortPartitionerWithRows.java | 82 ++++
.../bulkinsert/LSMPartitionSortPartitioner.java | 64 +++
.../commit/BaseSparkCommitActionExecutor.java | 4 +-
.../table/action/commit/SparkBulkInsertHelper.java | 11 +-
.../BaseDatasetBulkInsertCommitActionExecutor.java | 16 +-
.../functional/TestHoodieClientOnLsmStorage.java | 364 +++++++++++++++++
.../bulkinsert/TestLSMBulkInsertPartitioner.java | 310 +++++++++++++++
.../apache/hudi/functional/TestLSMDataSource.scala | 441 +++++++++++++++++++++
.../apache/hudi/functional/TestMORDataSource.scala | 122 +-----
.../hudi/dml/insert/TestInsertWithLSMLayout.scala | 238 +++++++++++
19 files changed, 1891 insertions(+), 118 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
index 73bbffb749d9..46a09baebc71 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
@@ -1775,6 +1775,12 @@ public class HoodieWriteConfig extends HoodieConfig {
return BulkInsertSortMode.valueOf(sortMode.toUpperCase());
}
+ public boolean isLSMTreeStorageLayout() {
+ return HoodieTableConfig.TableStorageLayout.fromConfigValue(
+ getStringOrDefault(HoodieTableConfig.TABLE_STORAGE_LAYOUT))
+ == HoodieTableConfig.TableStorageLayout.LSM_TREE;
+ }
+
public boolean isMergeDataValidationCheckEnabled() {
return getBoolean(MERGE_DATA_VALIDATION_CHECK_ENABLE);
}
@@ -3802,6 +3808,9 @@ public class HoodieWriteConfig extends HoodieConfig {
protected void setDefaults() {
writeConfig.setDefaultValue(MARKERS_TYPE,
getDefaultMarkersType(engineType));
+ if (writeConfig.isLSMTreeStorageLayout()) {
+ writeConfig.setDefaultValue(BULK_INSERT_SORT_MODE,
BulkInsertSortMode.PARTITION_SORT.name());
+ }
// Check for mandatory properties
writeConfig.setDefaults(HoodieWriteConfig.class.getName());
// Make sure the props is propagated
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java
index 2decb5f2ecdf..6eaba960606b 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java
@@ -141,8 +141,8 @@ public abstract class BaseCreateHandle<T, I, K, O> extends
HoodieWriteHandle<T,
Iterator<String> keyIterator;
if (hoodieTable.requireSortedRecords()) {
// Sorting the keys limits the amount of extra memory required for
writing sorted records.
- // requireSortedRecords() is true only for HFile base files, which order
keys by UTF-8 bytes,
- // not String (UTF-16) order, so sort with the matching comparator.
+ // HFile base files and LSM tables order keys by UTF-8 bytes, not String
(UTF-16) order,
+ // so sort with the matching comparator.
keyIterator =
recordMap.keySet().stream().sorted(StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR).iterator();
} else {
keyIterator = recordMap.keySet().stream().iterator();
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfig.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfig.java
index 8d50ec397ccc..80e0e84d4f52 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfig.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfig.java
@@ -38,6 +38,7 @@ import org.apache.hudi.config.HoodieWriteConfig.Builder;
import org.apache.hudi.core.transaction.lock.InProcessLockProvider;
import org.apache.hudi.core.transaction.lock.NoopLockProvider;
import org.apache.hudi.exception.HoodieIndexException;
+import org.apache.hudi.execution.bulkinsert.BulkInsertSortMode;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
@@ -155,6 +156,40 @@ public class TestHoodieWriteConfig {
EngineType.JAVA, HoodieIndex.IndexType.SIMPLE));
}
+ @Test
+ public void testDefaultBulkInsertSortModeForLsmLayout() {
+ HoodieWriteConfig defaultLayoutConfig = HoodieWriteConfig.newBuilder()
+ .withPath("/tmp/default-layout")
+ .build();
+ assertFalse(defaultLayoutConfig.isLSMTreeStorageLayout());
+ assertEquals(BulkInsertSortMode.NONE,
defaultLayoutConfig.getBulkInsertSortMode());
+
+ Properties lsmLayoutProps = new Properties();
+ lsmLayoutProps.setProperty(
+ HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
+ HoodieTableConfig.TableStorageLayout.LSM_TREE.configValue());
+ HoodieWriteConfig lsmLayoutConfig = HoodieWriteConfig.newBuilder()
+ .withPath("/tmp/lsm-layout")
+ .withProperties(lsmLayoutProps)
+ .build();
+ assertTrue(lsmLayoutConfig.isLSMTreeStorageLayout());
+ assertEquals(BulkInsertSortMode.PARTITION_SORT,
lsmLayoutConfig.getBulkInsertSortMode());
+
+ HoodieWriteConfig javaLsmLayoutConfig = HoodieWriteConfig.newBuilder()
+ .withPath("/tmp/java-lsm-layout")
+ .withEngineType(EngineType.JAVA)
+ .withProperties(lsmLayoutProps)
+ .build();
+ assertEquals(BulkInsertSortMode.PARTITION_SORT,
javaLsmLayoutConfig.getBulkInsertSortMode());
+
+ HoodieWriteConfig explicitlyUnsortedLsmConfig =
HoodieWriteConfig.newBuilder()
+ .withPath("/tmp/explicitly-unsorted-lsm-layout")
+ .withProperties(lsmLayoutProps)
+ .withBulkInsertSortMode(BulkInsertSortMode.NONE.name())
+ .build();
+ assertEquals(BulkInsertSortMode.NONE,
explicitlyUnsortedLsmConfig.getBulkInsertSortMode());
+ }
+
@Test
public void testDefaultClusteringPlanStrategyClassAccordingToEngineType() {
testEngineSpecificConfig(HoodieWriteConfig::getClusteringPlanStrategyClass,
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java
index befa4b925e8d..d085f2b29c1b 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java
@@ -215,7 +215,9 @@ public abstract class MultipleSparkJobExecutionStrategy<T>
throw new UnsupportedOperationException(String.format("Layout
optimization strategy '%s' is not supported", layoutOptStrategy));
}
}).orElseGet(() -> isRowPartitioner
- ? BulkInsertInternalPartitionerWithRowsFactory.get(getWriteConfig(),
getHoodieTable().isPartitioned(), true)
+ ? BulkInsertInternalPartitionerWithRowsFactory.get(
+ getHoodieTable().getMetaClient().getTableConfig(),
getWriteConfig(),
+ getHoodieTable().isPartitioned(), true)
: BulkInsertInternalPartitionerFactory.get(getHoodieTable(),
getWriteConfig(), true));
}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerFactory.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerFactory.java
index aa2d26f67c7c..74160a0a0efc 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerFactory.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerFactory.java
@@ -45,12 +45,27 @@ public abstract class BulkInsertInternalPartitionerFactory {
return new RDDSimpleBucketBulkInsertPartitioner(table);
}
}
+ if (table.getMetaClient().getTableConfig().isLSMTreeStorageLayout()) {
+ switch (config.getBulkInsertSortMode()) {
+ case GLOBAL_SORT:
+ return new LSMGlobalSortPartitioner<>(config);
+ case PARTITION_SORT:
+ return new LSMPartitionSortPartitioner<>(config);
+ case PARTITION_PATH_REPARTITION_AND_SORT:
+ return new LSMPartitionPathRepartitionAndSortPartitioner<>(
+ table.isPartitioned(), config);
+ default:
+ throw new HoodieException(
+ "The bulk insert sort mode \"" +
config.getBulkInsertSortMode().name()
+ + "\" does not guarantee record ordering and is not
supported for LSM tables.");
+ }
+ }
return get(config, table.isPartitioned(), enforceNumOutputPartitions);
}
- public static BulkInsertPartitioner get(HoodieWriteConfig config,
- boolean isTablePartitioned,
- boolean enforceNumOutputPartitions) {
+ static BulkInsertPartitioner get(HoodieWriteConfig config,
+ boolean isTablePartitioned,
+ boolean enforceNumOutputPartitions) {
BulkInsertSortMode sortMode = config.getBulkInsertSortMode();
switch (sortMode) {
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerWithRowsFactory.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerWithRowsFactory.java
index 07995e50d6a3..804cc1e3f3be 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerWithRowsFactory.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertInternalPartitionerWithRowsFactory.java
@@ -18,7 +18,9 @@
package org.apache.hudi.execution.bulkinsert;
+import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.spark.sql.Dataset;
@@ -30,14 +32,37 @@ import org.apache.spark.sql.Row;
*/
public abstract class BulkInsertInternalPartitionerWithRowsFactory {
- public static BulkInsertPartitioner<Dataset<Row>> get(HoodieWriteConfig
config,
+ public static BulkInsertPartitioner<Dataset<Row>> get(HoodieTableConfig
tableConfig,
+ HoodieWriteConfig
config,
boolean
isTablePartitioned) {
- return get(config, isTablePartitioned, false);
+ return get(tableConfig, config, isTablePartitioned, false);
}
- public static BulkInsertPartitioner<Dataset<Row>> get(HoodieWriteConfig
config,
+ public static BulkInsertPartitioner<Dataset<Row>> get(HoodieTableConfig
tableConfig,
+ HoodieWriteConfig
config,
boolean
isTablePartitioned,
boolean
enforceNumOutputPartitions) {
+ if (tableConfig.isLSMTreeStorageLayout()) {
+ switch (config.getBulkInsertSortMode()) {
+ case GLOBAL_SORT:
+ return new GlobalSortPartitionerWithRows(config);
+ case PARTITION_SORT:
+ return new PartitionSortPartitionerWithRows(config);
+ case PARTITION_PATH_REPARTITION_AND_SORT:
+ return new LSMPartitionPathRepartitionAndSortPartitionerWithRows(
+ isTablePartitioned, config);
+ default:
+ throw new HoodieException(
+ "The bulk insert sort mode \"" +
config.getBulkInsertSortMode().name()
+ + "\" does not guarantee record ordering and is not
supported for LSM tables.");
+ }
+ }
+ return get(config, isTablePartitioned, enforceNumOutputPartitions);
+ }
+
+ static BulkInsertPartitioner<Dataset<Row>> get(HoodieWriteConfig config,
+ boolean isTablePartitioned,
+ boolean
enforceNumOutputPartitions) {
BulkInsertSortMode sortMode = config.getBulkInsertSortMode();
switch (sortMode) {
case NONE:
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMBulkInsertRecordSorter.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMBulkInsertRecordSorter.java
new file mode 100644
index 000000000000..7e4cd4571abe
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMBulkInsertRecordSorter.java
@@ -0,0 +1,98 @@
+/*
+ * 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.common.util.StringUtils;
+
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaRDD;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+
+import scala.Tuple2;
+
+/**
+ * Shared sorting operations for LSM RDD bulk-insert partitioners.
+ *
+ * <p>LSM base files must be ordered by partition path and record key using
their UTF-8 byte
+ * representation. This helper exposes two variants of that ordering:
+ *
+ * <ul>
+ * <li>{@link #keyByPartitionAndRecordKey(JavaRDD)} attaches the LSM sort
key to records that
+ * will subsequently be repartitioned and sorted by a caller.</li>
+ * <li>{@link #sortWithinPartitions(JavaRDD)} sorts records without changing
which Spark
+ * partition they belong to. This is used when a bulk-insert partitioner
has already
+ * established the desired record distribution.</li>
+ * </ul>
+ */
+final class LSMBulkInsertRecordSorter {
+
+ /** Comparator for the externally visible (partition path, record key) LSM
sort key. */
+ static final Comparator<Tuple2<String, String>> KEY_COMPARATOR =
+ (Comparator<Tuple2<String, String>> & Serializable) (left, right) ->
+ comparePartitionAndRecordKeys(left._1, left._2, right._1, right._2);
+
+ private LSMBulkInsertRecordSorter() {
+ }
+
+ /**
+ * Keys records by the LSM physical ordering columns.
+ *
+ * <p>The returned pair RDD is intended for callers that need to choose
their own Spark
+ * partitioner and invoke {@code repartitionAndSortWithinPartitions}.
+ */
+ static <T> JavaPairRDD<Tuple2<String, String>, HoodieRecord<T>>
keyByPartitionAndRecordKey(
+ JavaRDD<HoodieRecord<T>> records) {
+ return records.mapToPair(record -> new Tuple2<>(
+ new Tuple2<>(record.getPartitionPath(), record.getRecordKey()),
record));
+ }
+
+ /**
+ * Sorts each existing Spark partition by the LSM physical ordering without
changing record
+ * distribution between partitions.
+ *
+ * <p>This follows the existing {@link RDDPartitionSortPartitioner}
execution model: each Spark
+ * partition is materialized into a list and sorted locally without
introducing another shuffle.
+ * Memory usage is therefore proportional to the largest input partition.
+ */
+ static <T> JavaRDD<HoodieRecord<T>> sortWithinPartitions(
+ JavaRDD<HoodieRecord<T>> records) {
+ return records.mapPartitions(iterator -> {
+ List<HoodieRecord<T>> recordList = new ArrayList<>();
+ iterator.forEachRemaining(recordList::add);
+ recordList.sort((left, right) -> comparePartitionAndRecordKeys(
+ left.getPartitionPath(), left.getRecordKey(),
+ right.getPartitionPath(), right.getRecordKey()));
+ return recordList.iterator();
+ });
+ }
+
+ private static int comparePartitionAndRecordKeys(
+ String leftPartitionPath, String leftRecordKey,
+ String rightPartitionPath, String rightRecordKey) {
+ int partitionComparison = StringUtils.compareUtf8Bytes(leftPartitionPath,
rightPartitionPath);
+ return partitionComparison != 0
+ ? partitionComparison
+ : StringUtils.compareUtf8Bytes(leftRecordKey, rightRecordKey);
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMGlobalSortPartitioner.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMGlobalSortPartitioner.java
new file mode 100644
index 000000000000..416e23127627
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMGlobalSortPartitioner.java
@@ -0,0 +1,67 @@
+/*
+ * 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.hudi.table.BulkInsertPartitioner;
+
+import org.apache.spark.api.java.JavaRDD;
+
+import static
org.apache.hudi.execution.bulkinsert.BulkInsertSortMode.GLOBAL_SORT;
+import static
org.apache.hudi.execution.bulkinsert.LSMBulkInsertRecordSorter.KEY_COMPARATOR;
+import static
org.apache.hudi.execution.bulkinsert.LSMBulkInsertRecordSorter.keyByPartitionAndRecordKey;
+
+/**
+ * LSM RDD bulk-insert partitioner for {@link BulkInsertSortMode#GLOBAL_SORT}.
+ *
+ * <p>Like {@link GlobalSortPartitioner}, this partitioner globally sorts the
input and range
+ * partitions it across the requested number of Spark partitions. The sort key
and comparator are
+ * intentionally different: {@code GlobalSortPartitioner} concatenates the
partition path and
+ * record key and relies on the natural Java String ordering, while this
implementation keeps the
+ * two key components separate and compares each component by its UTF-8 bytes.
This preserves key
+ * boundaries and produces the physical ordering required by LSM base files.
+ */
+public class LSMGlobalSortPartitioner<T>
+ implements BulkInsertPartitioner<JavaRDD<HoodieRecord<T>>> {
+
+ private final boolean shouldPopulateMetaFields;
+
+ public LSMGlobalSortPartitioner(HoodieWriteConfig config) {
+ this.shouldPopulateMetaFields = config.populateMetaFields();
+ }
+
+ @Override
+ public JavaRDD<HoodieRecord<T>> repartitionRecords(JavaRDD<HoodieRecord<T>>
records,
+ int
outputSparkPartitions) {
+ if (!shouldPopulateMetaFields) {
+ throw new HoodieException(GLOBAL_SORT.name() + " mode requires
meta-fields to be enabled");
+ }
+
+ return keyByPartitionAndRecordKey(records)
+ .sortByKey(KEY_COMPARATOR, true, outputSparkPartitions)
+ .values();
+ }
+
+ @Override
+ public boolean arePartitionRecordsSorted() {
+ return true;
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMPartitionPathRepartitionAndSortPartitioner.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMPartitionPathRepartitionAndSortPartitioner.java
new file mode 100644
index 000000000000..1654bd1a62f0
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMPartitionPathRepartitionAndSortPartitioner.java
@@ -0,0 +1,88 @@
+/*
+ * 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.hudi.table.BulkInsertPartitioner;
+
+import org.apache.spark.api.java.JavaRDD;
+
+import scala.Tuple2;
+
+import static
org.apache.hudi.execution.bulkinsert.BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT;
+import static
org.apache.hudi.execution.bulkinsert.LSMBulkInsertRecordSorter.KEY_COMPARATOR;
+import static
org.apache.hudi.execution.bulkinsert.LSMBulkInsertRecordSorter.keyByPartitionAndRecordKey;
+import static
org.apache.hudi.execution.bulkinsert.LSMBulkInsertRecordSorter.sortWithinPartitions;
+
+/**
+ * LSM RDD bulk-insert partitioner for
+ * {@link BulkInsertSortMode#PARTITION_PATH_REPARTITION_AND_SORT}.
+ *
+ * <p>Unlike {@link PartitionPathRepartitionAndSortPartitioner}, which orders
partitioned input
+ * only by partition path and leaves non-partitioned input unsorted, this
implementation orders
+ * every output Spark partition by {@code (partition path, record key)} using
UTF-8 byte ordering.
+ * This stronger ordering is required for records written to LSM base files.
+ *
+ * <p>For a physically partitioned table, {@link PartitionPathRDDPartitioner}
distributes records
+ * using only the partition-path component, keeping all records for the same
table partition
+ * together. {@code repartitionAndSortWithinPartitions} then sorts the
composite key with the LSM
+ * comparator, so records within each table partition are ordered by record
key as well.
+ *
+ * <p>For a physically non-partitioned table, the input is coalesced to the
requested parallelism
+ * and each resulting Spark partition is sorted locally with the same LSM
comparator. Therefore,
+ * both branches produce sorted output and {@link
#arePartitionRecordsSorted()} always returns
+ * {@code true}.
+ */
+public class LSMPartitionPathRepartitionAndSortPartitioner<T>
+ implements BulkInsertPartitioner<JavaRDD<HoodieRecord<T>>> {
+
+ private final boolean isTablePartitioned;
+ private final boolean shouldPopulateMetaFields;
+
+ public LSMPartitionPathRepartitionAndSortPartitioner(boolean
isTablePartitioned,
+ HoodieWriteConfig
config) {
+ this.isTablePartitioned = isTablePartitioned;
+ this.shouldPopulateMetaFields = config.populateMetaFields();
+ }
+
+ @Override
+ public JavaRDD<HoodieRecord<T>> repartitionRecords(JavaRDD<HoodieRecord<T>>
records,
+ int
outputSparkPartitions) {
+ if (!shouldPopulateMetaFields) {
+ throw new HoodieException(
+ PARTITION_PATH_REPARTITION_AND_SORT.name() + " mode requires
meta-fields to be enabled");
+ }
+
+ if (isTablePartitioned) {
+ PartitionPathRDDPartitioner partitioner = new
PartitionPathRDDPartitioner(
+ key -> ((Tuple2<String, String>) key)._1, outputSparkPartitions);
+ return keyByPartitionAndRecordKey(records)
+ .repartitionAndSortWithinPartitions(partitioner, KEY_COMPARATOR)
+ .values();
+ }
+ return sortWithinPartitions(records.coalesce(outputSparkPartitions));
+ }
+
+ @Override
+ public boolean arePartitionRecordsSorted() {
+ return true;
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMPartitionPathRepartitionAndSortPartitionerWithRows.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMPartitionPathRepartitionAndSortPartitionerWithRows.java
new file mode 100644
index 000000000000..84f916f35367
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMPartitionPathRepartitionAndSortPartitionerWithRows.java
@@ -0,0 +1,82 @@
+/*
+ * 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.hudi.table.BulkInsertPartitioner;
+
+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 Dataset Row bulk-insert partitioner for
+ * {@link BulkInsertSortMode#PARTITION_PATH_REPARTITION_AND_SORT}.
+ *
+ * <p>Unlike {@link PartitionPathRepartitionAndSortPartitionerWithRows}, which
orders partitioned
+ * input only by partition path and leaves non-partitioned input unsorted,
this implementation
+ * orders every output Spark partition by the {@link
HoodieRecord#PARTITION_PATH_METADATA_FIELD}
+ * and {@link HoodieRecord#RECORD_KEY_METADATA_FIELD} columns. Spark SQL
stores these columns as
+ * UTF-8 strings, giving LSM base files their required UTF-8 physical ordering
without adding
+ * temporary sort columns or changing the input schema.
+ *
+ * <p>For a physically partitioned table, rows are first repartitioned by the
partition-path
+ * metadata column so that one table partition is not split by the
distribution key, and then
+ * sorted within each resulting Spark partition by partition path and record
key. For a physically
+ * non-partitioned table, rows are coalesced before applying the same local
sort. Both branches
+ * require populated meta fields and produce sorted output, so
+ * {@link #arePartitionRecordsSorted()} always returns {@code true}.
+ */
+public class LSMPartitionPathRepartitionAndSortPartitionerWithRows
+ implements BulkInsertPartitioner<Dataset<Row>> {
+
+ private final boolean isTablePartitioned;
+ private final boolean shouldPopulateMetaFields;
+
+ public LSMPartitionPathRepartitionAndSortPartitionerWithRows(boolean
isTablePartitioned,
+
HoodieWriteConfig config) {
+ this.isTablePartitioned = isTablePartitioned;
+ 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");
+ }
+
+ Dataset<Row> repartitionedRows = isTablePartitioned
+ ? rows.repartition(
+ outputSparkPartitions,
functions.col(HoodieRecord.PARTITION_PATH_METADATA_FIELD))
+ : rows.coalesce(outputSparkPartitions);
+ return repartitionedRows.sortWithinPartitions(
+ functions.col(HoodieRecord.PARTITION_PATH_METADATA_FIELD),
+ functions.col(HoodieRecord.RECORD_KEY_METADATA_FIELD));
+ }
+
+ @Override
+ public boolean arePartitionRecordsSorted() {
+ return true;
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMPartitionSortPartitioner.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMPartitionSortPartitioner.java
new file mode 100644
index 000000000000..8069fc7ba05f
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/LSMPartitionSortPartitioner.java
@@ -0,0 +1,64 @@
+/*
+ * 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.hudi.table.BulkInsertPartitioner;
+
+import org.apache.spark.api.java.JavaRDD;
+
+import static
org.apache.hudi.execution.bulkinsert.BulkInsertSortMode.PARTITION_SORT;
+import static
org.apache.hudi.execution.bulkinsert.LSMBulkInsertRecordSorter.sortWithinPartitions;
+
+/**
+ * LSM RDD bulk-insert partitioner for {@link
BulkInsertSortMode#PARTITION_SORT}.
+ *
+ * <p>Like {@link RDDPartitionSortPartitioner}, this partitioner first
coalesces the input to the
+ * requested parallelism and then materializes and sorts the records within
each Spark partition.
+ * The ordering is intentionally different: {@code
RDDPartitionSortPartitioner} compares its
+ * combined partition-path and record-key string with Java {@link
String#compareTo(String)}, while
+ * an LSM table must compare the partition path and record key separately in
UTF-8 byte order to
+ * preserve the LSM base-file ordering invariant.
+ */
+public class LSMPartitionSortPartitioner<T>
+ implements BulkInsertPartitioner<JavaRDD<HoodieRecord<T>>> {
+
+ private final boolean shouldPopulateMetaFields;
+
+ public LSMPartitionSortPartitioner(HoodieWriteConfig config) {
+ this.shouldPopulateMetaFields = config.populateMetaFields();
+ }
+
+ @Override
+ public JavaRDD<HoodieRecord<T>> repartitionRecords(JavaRDD<HoodieRecord<T>>
records,
+ int
outputSparkPartitions) {
+ if (!shouldPopulateMetaFields) {
+ throw new HoodieException(PARTITION_SORT.name() + " mode requires
meta-fields to be enabled");
+ }
+
+ return sortWithinPartitions(records.coalesce(outputSparkPartitions));
+ }
+
+ @Override
+ public boolean arePartitionRecordsSorted() {
+ return true;
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
index add4d33c4ed7..bc8a3b3c4dc6 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
@@ -328,8 +328,8 @@ public abstract class BaseSparkCommitActionExecutor<T>
extends
if (table.requireSortedRecords()) {
// Partition and sort within each partition as a single step. This is
faster than partitioning first and then
// applying a sort.
- // requireSortedRecords() is true only for HFile base files, which order
keys by UTF-8 bytes,
- // not String (UTF-16) order, so sort with the matching comparator.
+ // HFile base files and LSM tables order keys by UTF-8 bytes, not String
(UTF-16) order,
+ // so sort with the matching comparator.
Comparator<Tuple2<HoodieKey, Option<HoodieRecordLocation>>> comparator =
(Comparator<Tuple2<HoodieKey, Option<HoodieRecordLocation>>> & Serializable)
(t1, t2) -> {
HoodieKey key1 = t1._1;
HoodieKey key2 = t2._1;
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkBulkInsertHelper.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkBulkInsertHelper.java
index b1729baa6b53..c1ba3b43af24 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkBulkInsertHelper.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkBulkInsertHelper.java
@@ -38,6 +38,8 @@ import org.apache.spark.api.java.JavaRDD;
import java.util.List;
+import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
+
/**
* A spark implementation of {@link BaseBulkInsertHelper}.
*
@@ -69,13 +71,18 @@ public class SparkBulkInsertHelper<T, R> extends
BaseBulkInsertHelper<T, HoodieD
final
Option<BulkInsertPartitioner> userDefinedBulkInsertPartitioner) {
HoodieWriteMetadata result = new HoodieWriteMetadata();
+ boolean isLsmTable =
table.getMetaClient().getTableConfig().isLSMTreeStorageLayout();
+ checkArgument(!isLsmTable || userDefinedBulkInsertPartitioner.isEmpty(),
+ "User-defined bulk insert partitioners are not supported for LSM
tables because "
+ + "their record-key ordering cannot be verified");
+ BulkInsertPartitioner partitioner = userDefinedBulkInsertPartitioner
+ .orElseGet(() -> BulkInsertInternalPartitionerFactory.get(table,
config));
+
// Transition bulk_insert state to inflight
table.getActiveTimeline().transitionRequestedToInflight(table.getInstantGenerator().createNewInstant(HoodieInstant.State.REQUESTED,
executor.getCommitActionType(), instantTime), Option.empty(),
config.shouldAllowMultiWriteOnSameInstant());
- BulkInsertPartitioner partitioner =
userDefinedBulkInsertPartitioner.orElseGet(() ->
BulkInsertInternalPartitionerFactory.get(table, config));
-
// Write new files
HoodieData<WriteStatus> writeStatuses =
bulkInsert(inputRecords, instantTime, table, config, performDedupe,
partitioner, false,
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/BaseDatasetBulkInsertCommitActionExecutor.java
b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/BaseDatasetBulkInsertCommitActionExecutor.java
index 8b6466d5117d..e591eb272768 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/BaseDatasetBulkInsertCommitActionExecutor.java
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/BaseDatasetBulkInsertCommitActionExecutor.java
@@ -146,11 +146,21 @@ public abstract class
BaseDatasetBulkInsertCommitActionExecutor implements Seria
return new ConsistentBucketIndexBulkInsertPartitionerWithRows(table,
Collections.emptyMap(), true);
}
} else {
- return DataSourceUtils
- .createUserDefinedBulkInsertPartitionerWithRows(writeConfig)
- .orElseGet(() ->
BulkInsertInternalPartitionerWithRowsFactory.get(writeConfig,
isTablePartitioned));
+ Option<BulkInsertPartitioner<Dataset<Row>>> userDefinedPartitioner =
+
DataSourceUtils.createUserDefinedBulkInsertPartitionerWithRows(writeConfig);
+ if (userDefinedPartitioner.isPresent() &&
table.getMetaClient().getTableConfig().isLSMTreeStorageLayout()) {
+ throw new HoodieException(
+ "User-defined bulk insert partitioners are not supported for LSM
tables because "
+ + "their record-key ordering cannot be verified");
+ }
+ return userDefinedPartitioner.orElseGet(
+ () -> BulkInsertInternalPartitionerWithRowsFactory.get(
+ table.getMetaClient().getTableConfig(), writeConfig,
isTablePartitioned));
}
} else {
+ if (table.getMetaClient().getTableConfig().isLSMTreeStorageLayout()) {
+ throw new HoodieException("The Dataset Row writer requires
hoodie.populate.meta.fields=true for LSM tables.");
+ }
// Sort modes are not yet supported when meta fields are disabled
return new NonSortPartitionerWithRows();
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnLsmStorage.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnLsmStorage.java
new file mode 100644
index 000000000000..5fa1eb002464
--- /dev/null
+++
b/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());
+ }
+
+ private HoodieInstant findCompletedInstant(HoodieTableMetaClient metaClient,
String instantTime) {
+ return
metaClient.reloadActiveTimeline().filterCompletedInstants().getInstants().stream()
+ .filter(instant -> instant.requestedTime().equals(instantTime))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("No completed instant " +
instantTime));
+ }
+
+ private static class LsmTableTestContext {
+ private final String tablePath;
+ private final HoodieTableMetaClient metaClient;
+ private final HoodieWriteConfig writeConfig;
+ private final HoodieTestDataGenerator dataGenerator;
+
+ private LsmTableTestContext(
+ String tablePath,
+ HoodieTableMetaClient metaClient,
+ HoodieWriteConfig writeConfig,
+ HoodieTestDataGenerator dataGenerator) {
+ this.tablePath = tablePath;
+ this.metaClient = metaClient;
+ this.writeConfig = writeConfig;
+ this.dataGenerator = dataGenerator;
+ }
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/execution/bulkinsert/TestLSMBulkInsertPartitioner.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/execution/bulkinsert/TestLSMBulkInsertPartitioner.java
new file mode 100644
index 000000000000..4c56005b4e31
--- /dev/null
+++
b/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)));
+ return locations.iterator();
+ }, true).collect();
+
+ Map<String, Set<Integer>> sparkPartitionsByTablePartition = new
HashMap<>();
+ partitionLocations.forEach(location -> sparkPartitionsByTablePartition
+ .computeIfAbsent(location._1, ignored -> new HashSet<>())
+ .add(location._2));
+ sparkPartitionsByTablePartition.values().forEach(
+ sparkPartitions -> assertEquals(1, sparkPartitions.size()));
+ }
+
+ private <T> void assertSortedSparkPartitions(
+ List<List<T>> sparkPartitions,
+ java.util.function.Function<T, Tuple2<String, String>> keyExtractor) {
+ for (List<T> sparkPartition : sparkPartitions) {
+ Tuple2<String, String> previous = null;
+ for (T record : sparkPartition) {
+ Tuple2<String, String> current = keyExtractor.apply(record);
+ assertTrue(previous == null || KEY_COMPARATOR.compare(previous,
current) <= 0,
+ "Spark partition is not UTF-8 sorted: " + previous + " > " +
current);
+ previous = current;
+ }
+ }
+ }
+
+ private List<HoodieRecord<Object>> createRecords() {
+ List<HoodieRecord<Object>> records = new ArrayList<>();
+ for (Tuple2<String, String> key : createKeys()) {
+ records.add(new HoodieEmptyRecord<>(
+ new HoodieKey(key._2, key._1), HoodieRecord.HoodieRecordType.AVRO));
+ }
+ return records;
+ }
+
+ private List<Row> createRows() {
+ List<Row> rows = new ArrayList<>();
+ int value = 0;
+ for (Tuple2<String, String> key : createKeys()) {
+ rows.add(RowFactory.create(key._1, key._2, value++));
+ }
+ return rows;
+ }
+
+ private List<Tuple2<String, String>> createKeys() {
+ String bmpPrivateUse = new String(Character.toChars(0xE000));
+ String supplementary = new String(Character.toChars(0x20000));
+ return Arrays.asList(
+ new Tuple2<>("p1", supplementary + "-a"),
+ new Tuple2<>("p1", bmpPrivateUse + "-b"),
+ new Tuple2<>("p2", supplementary + "-b"),
+ new Tuple2<>("p2", bmpPrivateUse + "-a"),
+ new Tuple2<>("p2", "ascii"),
+ new Tuple2<>("p3", supplementary + "-c"),
+ new Tuple2<>("p3", bmpPrivateUse + "-c"),
+ new Tuple2<>("p1", "ascii"));
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLSMDataSource.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLSMDataSource.scala
new file mode 100644
index 000000000000..cac27b42c61f
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLSMDataSource.scala
@@ -0,0 +1,441 @@
+/*
+ * 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.{DataSourceUtils, DataSourceWriteOptions}
+import org.apache.hudi.client.SparkRDDWriteClient
+import org.apache.hudi.common.config.HoodieStorageConfig
+import org.apache.hudi.common.model.{HoodieBaseFile, HoodieRecord,
HoodieRecordPayload, HoodieTableType, WriteOperationType}
+import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient}
+import org.apache.hudi.common.testutils.HoodieTestUtils
+import org.apache.hudi.common.util.Option
+import org.apache.hudi.common.util.StringUtils
+import org.apache.hudi.config.{HoodieCompactionConfig, HoodieWriteConfig}
+import org.apache.hudi.exception.HoodieException
+import org.apache.hudi.execution.bulkinsert.{BulkInsertSortMode,
RowCustomColumnsSortPartitioner}
+import org.apache.hudi.testutils.{HoodieClientTestUtils,
SparkClientFunctionalTestHarness}
+import
org.apache.hudi.testutils.SparkClientFunctionalTestHarness.getSparkSqlConf
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{DataFrame, SaveMode}
+import org.junit.jupiter.api.{Tag, Test}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse,
assertThrows, assertTrue}
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.{Arguments, EnumSource, MethodSource}
+
+import scala.collection.JavaConverters._
+
+@Tag("functional")
+class TestLSMDataSource extends SparkClientFunctionalTestHarness {
+
+ private val FirstPartition = "p1"
+ private val SecondPartition = "p2"
+
+ override def conf: SparkConf = conf(getSparkSqlConf)
+
+ @ParameterizedTest
+ @EnumSource(value = classOf[HoodieTableType], names = Array("COPY_ON_WRITE",
"MERGE_ON_READ"))
+ def testStandardWriteOperations(tableType: HoodieTableType): Unit = {
+ val tablePath = s"${basePath}_${tableType.name.toLowerCase}_lsm_dataframe"
+ val options = baseOptions(tableType) +
+ (DataSourceWriteOptions.ENABLE_ROW_WRITER.key -> "false")
+
+ val inserts = rows(Seq(
+ ("😀-insert-p1", "v1", 1L, FirstPartition),
+ ("Ａ-insert-p1", "v1", 1L, FirstPartition),
+ ("middle-insert-p1", "v1", 1L, FirstPartition),
+ ("😀-insert-p2", "v1", 1L, SecondPartition),
+ ("Ａ-insert-p2", "v1", 1L, SecondPartition)))
+ write(inserts, tablePath, options, WriteOperationType.INSERT,
SaveMode.Overwrite)
+ assertLatestBaseFilesSorted(tablePath, WriteOperationType.INSERT)
+ assertSnapshot(tablePath, Map(
+ "😀-insert-p1" -> "v1",
+ "Ａ-insert-p1" -> "v1",
+ "middle-insert-p1" -> "v1",
+ "😀-insert-p2" -> "v1",
+ "Ａ-insert-p2" -> "v1"))
+
+ val updates = rows(Seq(
+ ("😀-insert-p1", "v2", 2L, FirstPartition),
+ ("Ａ-insert-p1", "v2", 2L, FirstPartition)))
+ write(updates, tablePath, options, WriteOperationType.UPSERT)
+ assertChangedFilesSorted(tablePath, tableType, WriteOperationType.UPSERT,
"log")
+ assertSnapshot(tablePath, Map(
+ "😀-insert-p1" -> "v2",
+ "Ａ-insert-p1" -> "v2",
+ "middle-insert-p1" -> "v1",
+ "😀-insert-p2" -> "v1",
+ "Ａ-insert-p2" -> "v1"))
+
+ val deletes = rows(Seq(
+ ("😀-insert-p1", "v2", 3L, FirstPartition),
+ ("Ａ-insert-p1", "v2", 3L, FirstPartition)))
+ write(deletes, tablePath, options, WriteOperationType.DELETE)
+ assertChangedFilesSorted(tablePath, tableType, WriteOperationType.DELETE,
"deletes")
+ assertSnapshot(tablePath, Map(
+ "middle-insert-p1" -> "v1",
+ "😀-insert-p2" -> "v1",
+ "Ａ-insert-p2" -> "v1"))
+
+ val partitionOverwrite = rows(Seq(
+ ("😀-overwrite-p2", "overwrite", 4L, SecondPartition),
+ ("Ａ-overwrite-p2", "overwrite", 4L, SecondPartition)))
+ write(partitionOverwrite, tablePath, options,
WriteOperationType.INSERT_OVERWRITE)
+ assertLatestBaseFilesSorted(tablePath, WriteOperationType.INSERT_OVERWRITE)
+ assertSnapshot(tablePath, Map(
+ "middle-insert-p1" -> "v1",
+ "😀-overwrite-p2" -> "overwrite",
+ "Ａ-overwrite-p2" -> "overwrite"))
+
+ val tableOverwrite = rows(Seq(
+ ("😀-overwrite-table", "table-overwrite", 5L, FirstPartition),
+ ("Ａ-overwrite-table", "table-overwrite", 5L, FirstPartition)))
+ write(tableOverwrite, tablePath, options,
WriteOperationType.INSERT_OVERWRITE_TABLE)
+ assertLatestBaseFilesSorted(tablePath,
WriteOperationType.INSERT_OVERWRITE_TABLE)
+ assertSnapshot(tablePath, Map(
+ "😀-overwrite-table" -> "table-overwrite",
+ "Ａ-overwrite-table" -> "table-overwrite"))
+
+ write(
+ tableOverwrite.limit(0),
+ tablePath,
+ options + (DataSourceWriteOptions.PARTITIONS_TO_DELETE.key ->
FirstPartition),
+ WriteOperationType.DELETE_PARTITION)
+ assertTrue(latestBaseFiles(tablePath).isEmpty)
+ assertSnapshot(tablePath, Map.empty)
+ }
+
+ @ParameterizedTest
+ @MethodSource(Array("bulkInsertWithHoodieRecordPathParams"))
+ def testBulkInsertWithHoodieRecordPath(
+ tableType: HoodieTableType,
+ sortMode: BulkInsertSortMode): Unit = {
+ val tablePath =
s"${basePath}_${tableType.name.toLowerCase}_lsm_bulk_insert_${sortMode.name.toLowerCase}"
+ val options = baseOptions(tableType) ++ Map(
+ DataSourceWriteOptions.ENABLE_ROW_WRITER.key -> "false",
+ HoodieWriteConfig.BULK_INSERT_SORT_MODE.key -> sortMode.name,
+ HoodieWriteConfig.BULKINSERT_PARALLELISM_VALUE.key -> "2")
+ val inserts = rows(Seq(
+ ("😀-bulk-p1", "v1", 1L, FirstPartition),
+ ("Ａ-bulk-p1", "v1", 1L, FirstPartition),
+ ("middle-bulk-p1", "v1", 1L, FirstPartition),
+ ("😀-bulk-p2", "v1", 1L, SecondPartition),
+ ("Ａ-bulk-p2", "v1", 1L, SecondPartition)))
+
+ write(inserts, tablePath, options, WriteOperationType.BULK_INSERT,
SaveMode.Overwrite)
+
+ assertLatestBaseFilesSorted(tablePath, WriteOperationType.BULK_INSERT)
+ assertSnapshot(tablePath, Map(
+ "😀-bulk-p1" -> "v1",
+ "Ａ-bulk-p1" -> "v1",
+ "middle-bulk-p1" -> "v1",
+ "😀-bulk-p2" -> "v1",
+ "Ａ-bulk-p2" -> "v1"))
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = classOf[BulkInsertSortMode], names = Array(
+ "GLOBAL_SORT", "PARTITION_SORT", "PARTITION_PATH_REPARTITION_AND_SORT"))
+ def testBulkInsertWithRowWriter(sortMode: BulkInsertSortMode): Unit = {
+ val tablePath =
s"${basePath}_cow_lsm_row_bulk_insert_${sortMode.name.toLowerCase}"
+ val options = baseOptions(HoodieTableType.COPY_ON_WRITE) ++ Map(
+ DataSourceWriteOptions.ENABLE_ROW_WRITER.key -> "true",
+ HoodieWriteConfig.BULK_INSERT_SORT_MODE.key -> sortMode.name,
+ HoodieWriteConfig.BULKINSERT_PARALLELISM_VALUE.key -> "2")
+ val inserts = rows(Seq(
+ ("😀-row-p1", "v1", 1L, FirstPartition),
+ ("Ａ-row-p1", "v1", 1L, FirstPartition),
+ ("middle-row-p1", "v1", 1L, FirstPartition),
+ ("😀-row-p2", "v1", 1L, SecondPartition),
+ ("Ａ-row-p2", "v1", 1L, SecondPartition)))
+
+ write(inserts, tablePath, options, WriteOperationType.BULK_INSERT,
SaveMode.Overwrite)
+
+ assertLatestBaseFilesSorted(tablePath, WriteOperationType.BULK_INSERT)
+ assertSnapshot(tablePath, Map(
+ "😀-row-p1" -> "v1",
+ "Ａ-row-p1" -> "v1",
+ "middle-row-p1" -> "v1",
+ "😀-row-p2" -> "v1",
+ "Ａ-row-p2" -> "v1"))
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = classOf[BulkInsertSortMode], names = Array(
+ "NONE", "PARTITION_PATH_REPARTITION"))
+ def testBulkInsertRejectsNonSortingModes(sortMode: BulkInsertSortMode): Unit
= {
+ Seq(false, true).foreach { enableRowWriter =>
+ val tablePath =
s"${basePath}_cow_lsm_reject_${sortMode.name.toLowerCase}_$enableRowWriter"
+ val options = baseOptions(HoodieTableType.COPY_ON_WRITE) ++ Map(
+ DataSourceWriteOptions.ENABLE_ROW_WRITER.key ->
enableRowWriter.toString,
+ HoodieWriteConfig.BULK_INSERT_SORT_MODE.key -> sortMode.name)
+ val inserts = rows(Seq(("key-1", "v1", 1L, FirstPartition)))
+
+ val exception = assertThrows(classOf[HoodieException], () =>
+ write(inserts, tablePath, options, WriteOperationType.BULK_INSERT,
SaveMode.Overwrite))
+
+ assertExceptionChainContains(exception,
+ s"""The bulk insert sort mode "${sortMode.name}" does not guarantee
record ordering""")
+ }
+ }
+
+ private def assertExceptionChainContains(exception: Throwable,
expectedMessage: String): Unit = {
+ var current = exception
+ var found = false
+ while (current != null && !found) {
+ found = current.getMessage != null &&
current.getMessage.contains(expectedMessage)
+ current = current.getCause
+ }
+ assertTrue(found,
+ s"Expected exception chain to contain '$expectedMessage', but caught:
$exception")
+ }
+
+ @Test
+ def testLsmRowWriterRejectsDisabledMetaFields(): Unit = {
+ val tablePath = s"${basePath}_cow_lsm_row_without_meta_fields"
+ val options = baseOptions(HoodieTableType.COPY_ON_WRITE) ++ Map(
+ DataSourceWriteOptions.ENABLE_ROW_WRITER.key -> "true",
+ HoodieTableConfig.POPULATE_META_FIELDS.key -> "false")
+ val inserts = rows(Seq(
+ ("key-2", "v2", 1L, FirstPartition),
+ ("key-1", "v1", 1L, FirstPartition)))
+
+ val exception = assertThrows(classOf[HoodieException], () =>
+ write(inserts, tablePath, options, WriteOperationType.BULK_INSERT,
SaveMode.Overwrite))
+
+ assertTrue(exception.getMessage.contains(
+ "The Dataset Row writer requires hoodie.populate.meta.fields=true for
LSM tables"))
+ val instants =
createMetaClient(tablePath).reloadActiveTimeline().getInstants.asScala
+ assertEquals(1, instants.size)
+
assertEquals(org.apache.hudi.common.table.timeline.HoodieInstant.State.REQUESTED,
instants.head.getState)
+ }
+
+ @Test
+ def testLsmRowWriterRejectsCustomPartitionerBeforeInflight(): Unit = {
+ val tablePath = s"${basePath}_cow_lsm_row_custom_partitioner"
+ val options = baseOptions(HoodieTableType.COPY_ON_WRITE) ++ Map(
+ DataSourceWriteOptions.ENABLE_ROW_WRITER.key -> "true",
+ HoodieWriteConfig.BULKINSERT_USER_DEFINED_PARTITIONER_CLASS_NAME.key ->
+ classOf[RowCustomColumnsSortPartitioner].getName,
+ HoodieWriteConfig.BULKINSERT_USER_DEFINED_PARTITIONER_SORT_COLUMNS.key
-> "value")
+ val inserts = rows(Seq(("key-1", "v1", 1L, FirstPartition)))
+
+ assertThrows(classOf[HoodieException], () =>
+ write(inserts, tablePath, options, WriteOperationType.BULK_INSERT,
SaveMode.Overwrite))
+
+ val instants =
createMetaClient(tablePath).reloadActiveTimeline().getInstants.asScala
+ assertEquals(1, instants.size)
+
assertEquals(org.apache.hudi.common.table.timeline.HoodieInstant.State.REQUESTED,
instants.head.getState)
+ }
+
+ @Test
+ def testCompactionProducesSortedBaseFile(): Unit = {
+ val tablePath = s"${basePath}_mor_lsm_compaction"
+ val options = baseOptions(HoodieTableType.MERGE_ON_READ) ++ Map(
+ HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key -> "1")
+
+ write(rows(Seq(
+ ("😀-compact", "v1", 1L, FirstPartition),
+ ("Ａ-compact", "v1", 1L, FirstPartition))),
+ tablePath, options, WriteOperationType.INSERT, SaveMode.Overwrite)
+ write(rows(Seq(
+ ("Ａ-compact", "v2", 2L, FirstPartition))),
+ tablePath, options, WriteOperationType.UPSERT)
+
+ val compactionInstant = withWriteClient(tablePath, options) { client =>
+ val instant = client.scheduleCompaction(Option.empty()).get()
+ val statuses = client.compact(instant, true).getWriteStatuses.collect()
+ assertFalse(statuses.isEmpty)
+ assertTrue(statuses.asScala.forall(status => !status.hasErrors))
+ instant
+ }
+
+ val metaClient = createMetaClient(tablePath)
+
assertTrue(metaClient.reloadActiveTimeline().filterCompletedInstants.containsInstant(compactionInstant))
+ val compactedBaseFiles = latestBaseFiles(tablePath)
+ assertFalse(compactedBaseFiles.isEmpty)
+ assertTrue(compactedBaseFiles.forall(_.getCommitTime == compactionInstant))
+ compactedBaseFiles.foreach(baseFile =>
assertParquetFileSorted(baseFile.getPath, WriteOperationType.COMPACT))
+ assertSnapshot(tablePath, Map(
+ "😀-compact" -> "v1",
+ "Ａ-compact" -> "v2"))
+ }
+
+ @Test
+ def testLogCompactionProducesSortedNativeRun(): Unit = {
+ val tablePath = s"${basePath}_mor_lsm_log_compaction"
+ val options = baseOptions(HoodieTableType.MERGE_ON_READ) ++ Map(
+ HoodieCompactionConfig.ENABLE_LOG_COMPACTION.key -> "true",
+ HoodieCompactionConfig.LOG_COMPACTION_BLOCKS_THRESHOLD.key -> "1")
+
+ write(rows(Seq(
+ ("😀-log-compact", "v1", 1L, FirstPartition),
+ ("Ａ-log-compact", "v1", 1L, FirstPartition))),
+ tablePath, options, WriteOperationType.INSERT, SaveMode.Overwrite)
+ write(rows(Seq(
+ ("Ａ-log-compact", "v2", 2L, FirstPartition))),
+ tablePath, options, WriteOperationType.UPSERT)
+ write(rows(Seq(
+ ("😀-log-compact", "v2", 3L, FirstPartition))),
+ tablePath, options, WriteOperationType.UPSERT)
+
+ val expectedSnapshot = Map(
+ "😀-log-compact" -> "v2",
+ "Ａ-log-compact" -> "v2")
+ assertSnapshot(tablePath, expectedSnapshot)
+
+ val logCompactionInstant = withWriteClient(tablePath, options) { client =>
+ val instant = client.scheduleLogCompaction(Option.empty()).get()
+ client.logCompact(instant, true)
+ instant
+ }
+
+ val metaClient = createMetaClient(tablePath)
+
assertTrue(metaClient.reloadActiveTimeline().filterCompletedInstants.containsInstant(logCompactionInstant))
+ val compactedLogFiles =
HoodieTestUtils.listNativeLogFiles(metaClient.getStorage, tablePath).asScala
+ .filter(path => path.getName.contains(logCompactionInstant) &&
path.getName.endsWith(".log.parquet"))
+ assertEquals(1, compactedLogFiles.size)
+ assertParquetFileSorted(compactedLogFiles.head.toString,
WriteOperationType.LOG_COMPACT)
+ assertSnapshot(tablePath, expectedSnapshot)
+ }
+
+ private def baseOptions(tableType: HoodieTableType): Map[String, String] =
Map(
+ DataSourceWriteOptions.TABLE_TYPE.key -> tableType.name,
+ DataSourceWriteOptions.RECORDKEY_FIELD.key -> "id",
+ DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "partition",
+ DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME.key ->
"org.apache.hudi.keygen.SimpleKeyGenerator",
+ HoodieTableConfig.ORDERING_FIELDS.key -> "ts",
+ HoodieTableConfig.TABLE_STORAGE_LAYOUT.key ->
HoodieTableConfig.TableStorageLayout.LSM_TREE.configValue,
+ HoodieWriteConfig.TBL_NAME.key ->
s"hoodie_lsm_${tableType.name.toLowerCase}",
+ HoodieCompactionConfig.INLINE_COMPACT.key -> "false",
+ HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT.key -> "parquet",
+ "hoodie.insert.shuffle.parallelism" -> "1",
+ "hoodie.upsert.shuffle.parallelism" -> "1",
+ "hoodie.delete.shuffle.parallelism" -> "1")
+
+ private def rows(values: Seq[(String, String, Long, String)]): DataFrame = {
+ val _spark = spark
+ import _spark.implicits._
+ values.toDF("id", "value", "ts", "partition").repartition(1)
+ }
+
+ private def write(
+ input: DataFrame,
+ tablePath: String,
+ options: Map[String, String],
+ operationType: WriteOperationType,
+ saveMode: SaveMode = SaveMode.Append): Unit = {
+ input.write.format("hudi")
+ .options(options)
+ .option(DataSourceWriteOptions.OPERATION.key, operationType.value)
+ .mode(saveMode)
+ .save(tablePath)
+ }
+
+ private def withWriteClient[T](
+ tablePath: String,
+ options: Map[String, String])(
+ operation: SparkRDDWriteClient[HoodieRecordPayload[Nothing]] => T): T = {
+ val client = DataSourceUtils.createHoodieClient(
+ spark.sparkContext,
+ "",
+ tablePath,
+ options(HoodieWriteConfig.TBL_NAME.key),
+ options.asJava)
+ .asInstanceOf[SparkRDDWriteClient[HoodieRecordPayload[Nothing]]]
+ try {
+ operation(client)
+ } finally {
+ client.close()
+ }
+ }
+
+ private def assertChangedFilesSorted(
+ tablePath: String,
+ tableType: HoodieTableType,
+ operationType: WriteOperationType,
+ nativeLogExtension: String): Unit = {
+ if (tableType == HoodieTableType.COPY_ON_WRITE) {
+ assertLatestBaseFilesSorted(tablePath, operationType)
+ } else {
+ val metaClient = createMetaClient(tablePath)
+ val instantTime =
metaClient.getActiveTimeline.filterCompletedInstants.lastInstant.get.requestedTime
+ val nativeLogSuffix = s".$nativeLogExtension.parquet"
+ val logFiles = HoodieTestUtils.listNativeLogFiles(metaClient.getStorage,
tablePath).asScala
+ .filter(path => path.getName.contains(instantTime) &&
path.getName.endsWith(nativeLogSuffix))
+ assertFalse(logFiles.isEmpty, s"$operationType should produce an LSM
native log run")
+ val runSizes = logFiles.map(path =>
assertParquetFileSorted(path.toString, operationType))
+ assertTrue(runSizes.exists(_ > 1), s"$operationType should produce a
non-trivial sorted native run")
+ }
+ }
+
+ private def assertLatestBaseFilesSorted(tablePath: String, operationType:
WriteOperationType): Unit = {
+ val baseFiles = latestBaseFiles(tablePath)
+ assertFalse(baseFiles.isEmpty, s"$operationType should produce an LSM
base-file run")
+ val runSizes = baseFiles.map(baseFile =>
assertParquetFileSorted(baseFile.getPath, operationType))
+ assertTrue(runSizes.exists(_ > 1), s"$operationType should produce a
non-trivial sorted base-file run")
+ }
+
+ private def assertParquetFileSorted(path: String, operationType:
WriteOperationType): Int = {
+ val actualRecordKeys = spark.read.parquet(path)
+ .select(HoodieRecord.RECORD_KEY_METADATA_FIELD)
+ .collect()
+ .map(_.getString(0))
+ .toSeq
+ val expectedRecordKeys = actualRecordKeys.sortWith((left, right) =>
StringUtils.compareUtf8Bytes(left, right) < 0)
+ assertEquals(expectedRecordKeys, actualRecordKeys, s"$operationType output
is not sorted: $path")
+ actualRecordKeys.size
+ }
+
+ private def latestBaseFiles(tablePath: String): Seq[HoodieBaseFile] = {
+ val metaClient = createMetaClient(tablePath)
+ Seq(FirstPartition, SecondPartition).flatMap { partitionPath =>
+ HoodieClientTestUtils.getLatestBaseFiles(
+ tablePath,
+ metaClient.getStorage,
+ s"$tablePath/$partitionPath/*").asScala
+ }
+ }
+
+ private def assertSnapshot(tablePath: String, expected: Map[String,
String]): Unit = {
+ val actual = spark.read.format("hudi").load(tablePath)
+ .select("id", "value")
+ .collect()
+ .map(row => row.getString(0) -> row.getString(1))
+ .toMap
+ assertEquals(expected, actual)
+ }
+
+ private def createMetaClient(tablePath: String): HoodieTableMetaClient =
+ HoodieTableMetaClient.builder()
+ .setBasePath(tablePath)
+ .setConf(storageConf.newInstance())
+ .build()
+}
+
+object TestLSMDataSource {
+
+ def bulkInsertWithHoodieRecordPathParams():
java.util.stream.Stream[Arguments] =
+ java.util.stream.Stream.of(
+ Arguments.of(HoodieTableType.COPY_ON_WRITE,
BulkInsertSortMode.GLOBAL_SORT),
+ Arguments.of(HoodieTableType.COPY_ON_WRITE,
BulkInsertSortMode.PARTITION_SORT),
+ Arguments.of(HoodieTableType.COPY_ON_WRITE,
BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT),
+ Arguments.of(HoodieTableType.MERGE_ON_READ,
BulkInsertSortMode.GLOBAL_SORT),
+ Arguments.of(HoodieTableType.MERGE_ON_READ,
BulkInsertSortMode.PARTITION_SORT),
+ Arguments.of(HoodieTableType.MERGE_ON_READ,
BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT))
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSource.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSource.scala
index 0316423c10e8..c422693cde09 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSource.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSource.scala
@@ -38,7 +38,7 @@ import
org.apache.hudi.metadata.HoodieTableMetadataUtil.{metadataPartitionExists
import org.apache.hudi.storage.{StoragePath, StoragePathInfo}
import org.apache.hudi.table.action.compact.CompactionTriggerStrategy
import org.apache.hudi.table.upgrade.TestUpgradeDowngrade.getFixtureName
-import org.apache.hudi.testutils.{DataSourceTestUtils, HoodieClientTestUtils,
HoodieSparkClientTestBase}
+import org.apache.hudi.testutils.{DataSourceTestUtils,
HoodieSparkClientTestBase}
import org.apache.hudi.util.JFunction
import org.apache.commons.io.FileUtils
@@ -191,6 +191,7 @@ class TestMORDataSource extends HoodieSparkClientTestBase
with SparkDatasetMixin
DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "",
DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME.key ->
"org.apache.hudi.keygen.NonpartitionedKeyGenerator",
HoodieTableConfig.ORDERING_FIELDS.key -> "ts",
+ HoodieTableConfig.TABLE_STORAGE_LAYOUT.key ->
HoodieTableConfig.TableStorageLayout.LSM_TREE.configValue,
HoodieWriteConfig.COMBINE_BEFORE_INSERT.key -> "false",
HoodieWriteConfig.TBL_NAME.key ->
"hoodie_mor_lsm_base_file_only_duplicates",
HoodieMetadataConfig.ENABLE.key -> "false",
@@ -219,19 +220,10 @@ class TestMORDataSource extends HoodieSparkClientTestBase
with SparkDatasetMixin
assertEquals(1,
dataFiles.count(_.getPath.getName.endsWith(HoodieFileFormat.PARQUET.getFileExtension)))
assertFalse(dataFiles.exists(pathInfo =>
org.apache.hudi.common.fs.FSUtils.isLogFile(pathInfo.getPath)))
- // Build the duplicate-bearing base file with the supported default-layout
insert path, then
- // switch only the test fixture to LSM so this test does not imply LSM
insert support.
val metaClient = HoodieTableMetaClient.builder()
.setBasePath(tablePath)
.setConf(storageConf.newInstance())
.build()
- assertFalse(metaClient.getTableConfig.isLSMTreeStorageLayout)
- val tableProps = metaClient.getTableConfig.getProps
- tableProps.setProperty(
- HoodieTableConfig.TABLE_STORAGE_LAYOUT.key,
- HoodieTableConfig.TableStorageLayout.LSM_TREE.configValue)
- HoodieTableConfig.update(metaClient.getStorage, metaClient.getMetaPath,
tableProps)
- metaClient.reloadTableConfig()
assertTrue(metaClient.getTableConfig.isLSMTreeStorageLayout)
// With no log records to merge, the LSM reader delegates directly to the
base-file iterator and
@@ -244,6 +236,24 @@ class TestMORDataSource extends HoodieSparkClientTestBase
with SparkDatasetMixin
assertEquals(2, snapshotRows.length)
assertTrue(snapshotRows.forall(_.getString(0) == duplicateKey))
assertEquals(Set("first", "second"),
snapshotRows.map(_.getString(1)).toSet)
+
+ Seq((duplicateKey, "latest", 3L))
+ .toDF("id", "value", "ts")
+ .repartition(1)
+ .write.format("hudi")
+ .options(writeOpts + (DataSourceWriteOptions.OPERATION.key ->
UPSERT_OPERATION_OPT_VAL))
+ .mode(SaveMode.Append)
+ .save(tablePath)
+
+ // Once an update adds a sorted run, the LSM merge treats equal physical
keys as one logical key.
+ val mergedRows = spark.read.format("hudi")
+ .options(readOpts)
+ .load(tablePath)
+ .select("id", "value")
+ .collect()
+ assertEquals(1, mergedRows.length)
+ assertEquals(duplicateKey, mergedRows.head.getString(0))
+ assertEquals("latest", mergedRows.head.getString(1))
}
@Test
@@ -336,98 +346,6 @@ class TestMORDataSource extends HoodieSparkClientTestBase
with SparkDatasetMixin
assertEquals(Set("emoji-v1"), skipMergeVersions(emojiFaceKey))
}
- @Test
- def testLsmCompactionUsesUtf8Ordering(): Unit = {
- val fullWidthAKey = "Ａ-key"
- val emojiFaceKey = "😀-key"
- val tableName = "hoodie_mor_lsm_compaction"
- val tablePath = s"${basePath}_mor_lsm_compaction"
- val _spark = spark
- import _spark.implicits._
-
- val options = Map[String, String](
- DataSourceWriteOptions.TABLE_TYPE.key ->
HoodieTableType.MERGE_ON_READ.name,
- DataSourceWriteOptions.OPERATION.key -> UPSERT_OPERATION_OPT_VAL,
- DataSourceWriteOptions.RECORDKEY_FIELD.key -> "id",
- DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "",
- DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME.key ->
"org.apache.hudi.keygen.NonpartitionedKeyGenerator",
- HoodieTableConfig.ORDERING_FIELDS.key -> "ts",
- HoodieTableConfig.TABLE_STORAGE_LAYOUT.key ->
HoodieTableConfig.TableStorageLayout.LSM_TREE.configValue,
- HoodieWriteConfig.TBL_NAME.key -> tableName,
- HoodieMetadataConfig.ENABLE.key -> "false",
- HoodieCompactionConfig.INLINE_COMPACT.key -> "false",
- HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key -> "1",
- HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT.key -> "parquet",
- "hoodie.insert.shuffle.parallelism" -> "1",
- "hoodie.upsert.shuffle.parallelism" -> "1")
- val (writeOpts, readOpts) = getWriterReaderOpts(HoodieRecordType.AVRO,
options)
-
- Seq(
- (fullWidthAKey, "full-width-a-v1", 1L),
- (emojiFaceKey, "emoji-v1", 1L))
- .toDF("id", "value", "ts")
- .repartition(1)
- .write.format("hudi")
- .options(writeOpts)
- .mode(SaveMode.Append)
- .save(tablePath)
-
- // Update only the full-width A key so the base and log sorted runs have
different key sets. This
- // exposes an ordering mismatch during the LSM merge instead of merging
two identical runs.
- Seq((fullWidthAKey, "full-width-a-v2", 2L))
- .toDF("id", "value", "ts")
- .repartition(1)
- .write.format("hudi")
- .options(writeOpts)
- .mode(SaveMode.Append)
- .save(tablePath)
-
- val metaClient = HoodieTableMetaClient.builder()
- .setBasePath(tablePath)
- .setConf(storageConf.newInstance())
- .build()
- assertTrue(metaClient.getTableConfig.isLSMTreeStorageLayout)
-
- val client = DataSourceUtils.createHoodieClient(
- spark.sparkContext, "", tablePath, tableName, writeOpts.asJava)
- .asInstanceOf[SparkRDDWriteClient[HoodieRecordPayload[Nothing]]]
- val compactionInstant = try {
- val instant = client.scheduleCompaction(Option.empty()).get()
- val statuses = client.compact(instant, true).getWriteStatuses.collect()
- assertFalse(statuses.isEmpty)
- assertTrue(statuses.asScala.forall(status => !status.hasErrors))
- instant
- } finally {
- client.close()
- }
-
-
assertTrue(metaClient.reloadActiveTimeline().filterCompletedInstants.containsInstant(compactionInstant))
-
- val latestBaseFiles = HoodieClientTestUtils.getLatestBaseFiles(
- tablePath, metaClient.getStorage, s"$tablePath/*")
- assertEquals(1, latestBaseFiles.size())
- assertEquals(compactionInstant, latestBaseFiles.get(0).getCommitTime)
-
- // Compacted LSM base files retain the table-level UTF-8 record-key
ordering contract.
- val physicalBaseFileKeys =
spark.read.parquet(latestBaseFiles.get(0).getPath)
- .select(HoodieRecord.RECORD_KEY_METADATA_FIELD)
- .collect()
- .map(_.getString(0))
- .toList
- assertEquals(Seq(fullWidthAKey, emojiFaceKey), physicalBaseFileKeys)
-
- val actual = spark.read.format("hudi")
- .options(readOpts)
- .load(tablePath)
- .select("id", "value")
- .collect()
- .map(row => row.getString(0) -> row.getString(1))
- .toMap
- assertEquals(Map(
- fullWidthAKey -> "full-width-a-v2",
- emojiFaceKey -> "emoji-v1"), actual)
- }
-
@ParameterizedTest
@CsvSource(Array(
// Inferred as COMMIT_TIME_ORDERING
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestInsertWithLSMLayout.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestInsertWithLSMLayout.scala
new file mode 100644
index 000000000000..c3fa35ddec8e
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestInsertWithLSMLayout.scala
@@ -0,0 +1,238 @@
+/*
+ * 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.spark.sql.hudi.dml.insert
+
+import org.apache.hudi.DataSourceWriteOptions.{ENABLE_ROW_WRITER,
SPARK_SQL_INSERT_INTO_OPERATION}
+import org.apache.hudi.common.config.HoodieStorageConfig
+import org.apache.hudi.common.model.{HoodieRecord, WriteOperationType}
+import org.apache.hudi.common.table.HoodieTableConfig
+import org.apache.hudi.common.util.StringUtils
+import org.apache.hudi.config.HoodieWriteConfig
+import org.apache.hudi.execution.bulkinsert.BulkInsertSortMode
+import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
+
+import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
+import
org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase.getLastCommitMetadata
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse}
+
+import java.nio.file.{Files, Paths}
+
+import scala.collection.JavaConverters._
+
+class TestInsertWithLSMLayout extends HoodieSparkSqlTestBase {
+
+ test("Test INSERT INTO with LSM layout") {
+ Seq("cow", "mor").foreach { tableType =>
+ Seq(WriteOperationType.INSERT, WriteOperationType.UPSERT).foreach {
operation =>
+ withSQLConf(SPARK_SQL_INSERT_INTO_OPERATION.key -> operation.value()) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ |location '$tablePath'
+ |tblproperties (
+ | type = '$tableType',
+ | primaryKey = 'id',
+ | preCombineField = 'ts',
+ | '${HoodieTableConfig.TABLE_STORAGE_LAYOUT.key}' =
'${HoodieTableConfig.TableStorageLayout.LSM_TREE.configValue}',
+ | '${HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT.key}' =
'parquet'
+ |)
+ |""".stripMargin)
+
+ spark.sql(
+ s"""
+ |insert into $tableName values
+ | (3, 'name-3', 30.0, 1000),
+ | (1, 'name-1', 10.0, 1000),
+ | (2, 'name-2', 20.0, 1000)
+ |""".stripMargin)
+
+ checkAnswer(s"select id, name, price, ts from $tableName order by
id")(
+ Seq(1, "name-1", 10.0, 1000),
+ Seq(2, "name-2", 20.0, 1000),
+ Seq(3, "name-3", 30.0, 1000))
+ assertResult(operation) {
+ getLastCommitMetadata(spark, tablePath).getOperationType
+ }
+ assertEquals(
+ HoodieTableConfig.TableStorageLayout.LSM_TREE,
+ createMetaClient(spark,
tablePath).getTableConfig.getTableStorageLayout)
+ }
+ }
+ }
+ }
+ }
+
+ test("Test bulk insert overwrite with LSM row writer") {
+ withSQLConf(SPARK_SQL_INSERT_INTO_OPERATION.key ->
WriteOperationType.BULK_INSERT.value()) {
+ Seq("cow", "mor").foreach { tableType =>
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id string,
+ | name string,
+ | ts long,
+ | dt string
+ |) using hudi
+ |location '$tablePath'
+ |partitioned by (dt)
+ |tblproperties (
+ | type = '$tableType',
+ | primaryKey = 'id',
+ | preCombineField = 'ts',
+ | '${ENABLE_ROW_WRITER.key}' = 'true',
+ | '${HoodieTableConfig.TABLE_STORAGE_LAYOUT.key}' =
'${HoodieTableConfig.TableStorageLayout.LSM_TREE.configValue}',
+ | '${HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT.key}' =
'parquet'
+ |)
+ |""".stripMargin)
+
+ spark.sql(
+ s"""
+ |insert into $tableName values
+ | ('😀-old-p1', 'old', 1, 'p1'),
+ | ('Ａ-old-p1', 'old', 1, 'p1'),
+ | ('middle-old-p2', 'keep', 1, 'p2')
+ |""".stripMargin)
+
+ spark.sql(
+ s"""
+ |insert overwrite table $tableName partition(dt='p1') values
+ | ('😀-partition-overwrite', 'new', 2),
+ | ('Ａ-partition-overwrite', 'new', 2),
+ | ('middle-partition-overwrite', 'new', 2)
+ |""".stripMargin)
+ checkAnswer(s"select id, name, ts, dt from $tableName order by id")(
+ Seq("middle-old-p2", "keep", 1, "p2"),
+ Seq("middle-partition-overwrite", "new", 2, "p1"),
+ Seq("Ａ-partition-overwrite", "new", 2, "p1"),
+ Seq("😀-partition-overwrite", "new", 2, "p1"))
+ assertResult(WriteOperationType.INSERT_OVERWRITE) {
+ getLastCommitMetadata(spark, tablePath).getOperationType
+ }
+ assertBaseFilesSorted(tablePath, WriteOperationType.INSERT_OVERWRITE)
+
+ spark.sql(
+ s"""
+ |insert overwrite table $tableName values
+ | ('😀-table-overwrite', 'table', 3, 'p3'),
+ | ('Ａ-table-overwrite', 'table', 3, 'p3'),
+ | ('middle-table-overwrite', 'table', 3, 'p3')
+ |""".stripMargin)
+ checkAnswer(s"select id, name, ts, dt from $tableName order by id")(
+ Seq("middle-table-overwrite", "table", 3, "p3"),
+ Seq("Ａ-table-overwrite", "table", 3, "p3"),
+ Seq("😀-table-overwrite", "table", 3, "p3"))
+ assertResult(WriteOperationType.INSERT_OVERWRITE_TABLE) {
+ getLastCommitMetadata(spark, tablePath).getOperationType
+ }
+ assertBaseFilesSorted(tablePath,
WriteOperationType.INSERT_OVERWRITE_TABLE)
+ }
+ }
+ }
+ }
+
+ test("Test bulk insert with LSM HoodieRecord writer") {
+ withSQLConf(SPARK_SQL_INSERT_INTO_OPERATION.key ->
WriteOperationType.BULK_INSERT.value()) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id string,
+ | name string,
+ | ts long,
+ | dt string
+ |) using hudi
+ |location '$tablePath'
+ |partitioned by (dt)
+ |tblproperties (
+ | type = 'cow',
+ | primaryKey = 'id',
+ | preCombineField = 'ts',
+ | '${ENABLE_ROW_WRITER.key}' = 'false',
+ | '${HoodieWriteConfig.BULK_INSERT_SORT_MODE.key}' =
+ |
'${BulkInsertSortMode.PARTITION_PATH_REPARTITION_AND_SORT.name}',
+ | '${HoodieWriteConfig.BULKINSERT_PARALLELISM_VALUE.key}' = '2',
+ | '${HoodieTableConfig.TABLE_STORAGE_LAYOUT.key}' =
+ |
'${HoodieTableConfig.TableStorageLayout.LSM_TREE.configValue}'
+ |)
+ |""".stripMargin)
+
+ spark.sql(
+ s"""
+ |insert into $tableName values
+ | ('😀-p1', 'emoji', 1, 'p1'),
+ | ('Ａ-p1', 'full-width', 1, 'p1'),
+ | ('middle-p1', 'ascii', 1, 'p1'),
+ | ('😀-p2', 'emoji', 1, 'p2'),
+ | ('Ａ-p2', 'full-width', 1, 'p2')
+ |""".stripMargin)
+
+ checkAnswer(s"select id, name, ts, dt from $tableName order by id")(
+ Seq("middle-p1", "ascii", 1, "p1"),
+ Seq("Ａ-p1", "full-width", 1, "p1"),
+ Seq("Ａ-p2", "full-width", 1, "p2"),
+ Seq("😀-p1", "emoji", 1, "p1"),
+ Seq("😀-p2", "emoji", 1, "p2"))
+ assertResult(WriteOperationType.BULK_INSERT) {
+ getLastCommitMetadata(spark, tablePath).getOperationType
+ }
+ assertEquals(
+ HoodieTableConfig.TableStorageLayout.LSM_TREE,
+ createMetaClient(spark,
tablePath).getTableConfig.getTableStorageLayout)
+ assertBaseFilesSorted(tablePath, WriteOperationType.BULK_INSERT)
+ }
+ }
+ }
+
+ private def assertBaseFilesSorted(tablePath: String, operationType:
WriteOperationType): Unit = {
+ val pathStream = Files.walk(Paths.get(tablePath))
+ val baseFiles = try {
+ pathStream.iterator().asScala
+ .filter(Files.isRegularFile(_))
+ .filter(path => path.getFileName.toString.endsWith(".parquet"))
+ .filterNot(path => path.toString.contains("/.hoodie/"))
+ .map(_.toString)
+ .toList
+ } finally {
+ pathStream.close()
+ }
+ assertFalse(baseFiles.isEmpty)
+ baseFiles.foreach { baseFile =>
+ val actualKeys = spark.read.parquet(baseFile)
+ .select(HoodieRecord.RECORD_KEY_METADATA_FIELD)
+ .collect()
+ .map(_.getString(0))
+ .toSeq
+ val expectedKeys = actualKeys.sortWith(
+ (left, right) => StringUtils.compareUtf8Bytes(left, right) < 0)
+ assertEquals(expectedKeys, actualKeys, s"$operationType output is not
sorted: $baseFile")
+ }
+ }
+}
