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 b38855ffb8c7 feat(flink): Sort bulk insert records by record key for
LSM layout (#19390)
b38855ffb8c7 is described below
commit b38855ffb8c733389438751406edf1de055247c0
Author: Shuo Cheng <[email protected]>
AuthorDate: Fri Jul 31 15:47:33 2026 +0800
feat(flink): Sort bulk insert records by record key for LSM layout (#19390)
---
.../apache/hudi/configuration/OptionsResolver.java | 6 +-
.../sink/bucket/BucketBulkInsertWriterHelper.java | 47 ++--
.../bucket/LsmBucketBulkInsertWriterHelper.java | 119 +++++++++
.../hudi/sink/bulk/BulkInsertWriterHelper.java | 25 +-
.../hudi/sink/bulk/LsmBulkInsertWriterHelper.java | 115 +++++++++
.../org/apache/hudi/sink/bulk/WriterHelpers.java | 17 +-
.../java/org/apache/hudi/sink/utils/Pipelines.java | 265 +++++++++++++++------
.../org/apache/hudi/table/HoodieTableFactory.java | 11 +-
.../hudi/configuration/TestOptionsResolver.java | 2 +-
.../sink/bulk/TestLsmBulkInsertWriterHelper.java | 141 +++++++++++
.../hudi/sink/utils/BulkInsertFunctionWrapper.java | 28 ++-
.../apache/hudi/table/ITTestHoodieDataSource.java | 116 +++++++++
.../apache/hudi/table/TestHoodieTableFactory.java | 60 +++--
.../org/apache/hudi/utils/TestStreamerUtil.java | 2 +-
14 files changed, 832 insertions(+), 122 deletions(-)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
index eb891b0a51fa..07f31daa8616 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
@@ -166,13 +166,13 @@ public class OptionsResolver {
/**
* Returns the configured table storage layout.
*
- * <p>Insert and bulk insert operations preserve duplicate record keys and
therefore default to
- * the regular storage layout. Other operations use Flink's LSM tree default.
+ * <p>Insert operations preserve duplicate record keys and therefore default
to the regular
+ * storage layout. Other operations use Flink's LSM tree default.
*/
public static HoodieTableConfig.TableStorageLayout
getTableStorageLayout(Configuration conf) {
return HoodieTableConfig.TableStorageLayout.fromConfigValue(conf.getString(
HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
- (isInsertOperation(conf) || isBulkInsertOperation(conf))
+ isInsertOperation(conf)
? HoodieTableConfig.TableStorageLayout.DEFAULT.configValue()
: HoodieTableConfig.TableStorageLayout.LSM_TREE.configValue()));
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketBulkInsertWriterHelper.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketBulkInsertWriterHelper.java
index 358686f9f6ff..4dc233bb0ed1 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketBulkInsertWriterHelper.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketBulkInsertWriterHelper.java
@@ -47,7 +47,7 @@ import java.util.Map;
public class BucketBulkInsertWriterHelper extends BulkInsertWriterHelper {
public static final String FILE_GROUP_META_FIELD = "_fg";
- private final int recordArity;
+ protected final int recordArity;
private String lastFileId; // for efficient code path
@@ -63,12 +63,7 @@ public class BucketBulkInsertWriterHelper extends
BulkInsertWriterHelper {
String recordKey = keyGen.getRecordKey(record);
String partitionPath = keyGen.getPartitionPath(record);
String fileId = tuple.getString(0).toString();
- if ((lastFileId == null) || !lastFileId.equals(fileId)) {
- log.info("Creating new file for partition path {}", partitionPath);
- handle = getRowCreateHandle(partitionPath, fileId);
- lastFileId = fileId;
- }
- handle.write(recordKey, partitionPath, record);
+ writeRecord(recordKey, partitionPath, fileId, record);
} catch (Throwable throwable) {
IOException ioException = new IOException("Exception happened when bulk
insert.", throwable);
log.error("Global error thrown while trying to write records in
HoodieRowDataCreateHandle", ioException);
@@ -76,6 +71,19 @@ public class BucketBulkInsertWriterHelper extends
BulkInsertWriterHelper {
}
}
+ protected void writeRecord(
+ String recordKey,
+ String partitionPath,
+ String fileId,
+ RowData record) throws IOException {
+ if ((lastFileId == null) || !lastFileId.equals(fileId)) {
+ log.info("Creating new file for partition path {}", partitionPath);
+ handle = getRowCreateHandle(partitionPath, fileId);
+ lastFileId = fileId;
+ }
+ handle.write(recordKey, partitionPath, record);
+ }
+
private HoodieRowDataCreateHandle getRowCreateHandle(String partitionPath,
String fileId) throws IOException {
if (!handles.containsKey(fileId)) { // if there is no handle corresponding
to the fileId
if (this.isInputSorted) {
@@ -93,19 +101,30 @@ public class BucketBulkInsertWriterHelper extends
BulkInsertWriterHelper {
return new SortOperatorGen(rowType, new String[] {FILE_GROUP_META_FIELD});
}
- private static String getFileId(Map<String, String> bucketIdToFileId,
RowDataKeyGen keyGen, RowData record, List<String> indexKeyFields,
- NumBucketsFunction numBucketsFunction,
boolean needFixedFileIdSuffix) {
- String recordKey = keyGen.getRecordKey(record);
- String partition = keyGen.getPartitionPath(record);
- final int numBuckets = numBucketsFunction.getNumBuckets(partition);
+ static String getFileId(
+ Map<String, String> bucketIdToFileId,
+ String recordKey,
+ String partitionPath,
+ List<String> indexKeyFields,
+ NumBucketsFunction numBucketsFunction,
+ boolean needFixedFileIdSuffix) {
+ final int numBuckets = numBucketsFunction.getNumBuckets(partitionPath);
final int bucketNum = BucketIdentifier.getBucketId(recordKey,
indexKeyFields, numBuckets);
- String bucketId = partition + bucketNum;
+ String bucketId = partitionPath + bucketNum;
return bucketIdToFileId.computeIfAbsent(bucketId, k ->
needFixedFileIdSuffix ? BucketIdentifier.newBucketFileIdForNBCC(bucketNum) :
BucketIdentifier.newBucketFileIdPrefix(bucketNum));
}
public static RowData rowWithFileId(Map<String, String> bucketIdToFileId,
RowDataKeyGen keyGen, RowData record, List<String> indexKeyFields,
NumBucketsFunction numBucketsFunction,
boolean needFixedFileIdSuffix) {
- final String fileId = getFileId(bucketIdToFileId, keyGen, record,
indexKeyFields, numBucketsFunction, needFixedFileIdSuffix);
+ String recordKey = keyGen.getRecordKey(record);
+ String partitionPath = keyGen.getPartitionPath(record);
+ final String fileId = getFileId(
+ bucketIdToFileId,
+ recordKey,
+ partitionPath,
+ indexKeyFields,
+ numBucketsFunction,
+ needFixedFileIdSuffix);
return GenericRowData.of(StringData.fromString(fileId), record);
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/LsmBucketBulkInsertWriterHelper.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/LsmBucketBulkInsertWriterHelper.java
new file mode 100644
index 000000000000..23e0b60137ef
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/LsmBucketBulkInsertWriterHelper.java
@@ -0,0 +1,119 @@
+/*
+ * 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.sink.bucket;
+
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.bucket.partition.NumBucketsFunction;
+import org.apache.hudi.sink.bulk.RowDataKeyGen;
+import org.apache.hudi.sink.bulk.sort.SortOperatorGen;
+import org.apache.hudi.table.HoodieTable;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Bucket-index bulk-insert writer helper for LSM input rows.
+ *
+ * <p>The input row contains file ID, encoded record key, and the original
table row.
+ */
+public class LsmBucketBulkInsertWriterHelper extends
BucketBulkInsertWriterHelper {
+
+ private static final String RECORD_KEY_FIELD = "_record_key";
+ private static final String RECORD_FIELD = "record";
+
+ public LsmBucketBulkInsertWriterHelper(
+ Configuration conf,
+ HoodieTable<?, ?, ?, ?> hoodieTable,
+ HoodieWriteConfig writeConfig,
+ String instantTime,
+ int taskPartitionId,
+ long taskId,
+ long taskEpochId,
+ RowType rowType) {
+ super(conf, hoodieTable, writeConfig, instantTime, taskPartitionId,
taskId, taskEpochId, rowType);
+ }
+
+ @Override
+ public void write(RowData sortRow) throws IOException {
+ String fileId = sortRow.getString(0).toString();
+ String recordKey = sortRow.getString(1).toString();
+ RowData record = sortRow.getRow(2, recordArity);
+ String partitionPath = keyGen.getPartitionPath(record);
+ writeRecord(recordKey, partitionPath, fileId, record);
+ }
+
+ public static RowData rowWithFileIdAndKey(
+ Map<String, String> bucketIdToFileId,
+ RowDataKeyGen keyGen,
+ RowData record,
+ List<String> indexKeyFields,
+ NumBucketsFunction numBucketsFunction,
+ boolean needFixedFileIdSuffix) {
+ String recordKey = keyGen.getRecordKey(record);
+ String partitionPath = keyGen.getPartitionPath(record);
+ String fileId = getFileId(
+ bucketIdToFileId,
+ recordKey,
+ partitionPath,
+ indexKeyFields,
+ numBucketsFunction,
+ needFixedFileIdSuffix);
+ return GenericRowData.of(
+ StringData.fromString(fileId),
+ StringData.fromString(recordKey),
+ record);
+ }
+
+ /**
+ * Returns the internal row type used to sort LSM bucket bulk-insert records
by file ID and
+ * record key.
+ *
+ * <p>The fields are ordered as file ID, encoded record key, and original
table row.
+ */
+ public static RowType rowTypeWithFileIdAndKey(RowType rowType) {
+ LogicalType[] types = new LogicalType[] {
+ DataTypes.STRING().getLogicalType(),
+ DataTypes.STRING().getLogicalType(),
+ rowType
+ };
+ String[] names =
+ new String[] {FILE_GROUP_META_FIELD, RECORD_KEY_FIELD, RECORD_FIELD};
+ return RowType.of(types, names);
+ }
+
+ /**
+ * Creates an external sorter ordered by file ID and encoded record key.
+ *
+ * <p>The nested payload is deliberately excluded from the sort keys, so
duplicate record keys
+ * are retained without comparing or aggregating their payloads.
+ */
+ public static SortOperatorGen getFileIdAndKeySorterGen(RowType rowType) {
+ return new SortOperatorGen(
+ rowType, new String[] {FILE_GROUP_META_FIELD, RECORD_KEY_FIELD});
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriterHelper.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriterHelper.java
index 212910bb612a..367f86085613 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriterHelper.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriterHelper.java
@@ -115,7 +115,9 @@ public class BulkInsertWriterHelper implements
AutoCloseable {
? schema
: HoodieSchemaUtils.addMetadataFields(schema,
writeConfig.allowOperationMetadataField());
this.preserveHoodieMetadata = preserveHoodieMetadata;
- this.isInputSorted = OptionsResolver.isBulkInsertOperation(conf) &&
conf.get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT);
+ this.isInputSorted = OptionsResolver.isBulkInsertOperation(conf)
+ && (conf.get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT)
+ || OptionsResolver.isLsmTreeStorageLayout(conf));
this.fileIdPrefix = UUID.randomUUID().toString();
this.keyGen = preserveHoodieMetadata ? null :
RowDataKeyGens.instance(conf, rowType, taskPartitionId, instantTime);
this.writeMetrics = writeMetrics;
@@ -129,14 +131,7 @@ public class BulkInsertWriterHelper implements
AutoCloseable {
String partitionPath = preserveHoodieMetadata
?
record.getString(HoodieRecord.PARTITION_PATH_META_FIELD_ORD).toString()
: keyGen.getPartitionPath(record);
-
- if ((lastKnownPartitionPath == null) ||
!lastKnownPartitionPath.equals(partitionPath) || !handle.canWrite()) {
- handle = getRowCreateHandle(partitionPath);
- lastKnownPartitionPath = partitionPath;
- writeMetrics.ifPresent(FlinkStreamWriteMetrics::markHandleSwitch);
- }
- handle.write(recordKey, partitionPath, record);
- writeMetrics.ifPresent(FlinkStreamWriteMetrics::markRecordIn);
+ writeRecord(recordKey, partitionPath, record);
} catch (Throwable t) {
IOException ioException = new IOException("Exception happened when bulk
insert.", t);
log.error("Global error thrown while trying to write records in
HoodieRowCreateHandle ", ioException);
@@ -144,6 +139,18 @@ public class BulkInsertWriterHelper implements
AutoCloseable {
}
}
+ protected void writeRecord(String recordKey, String partitionPath, RowData
record) throws IOException {
+ if ((lastKnownPartitionPath == null)
+ || !lastKnownPartitionPath.equals(partitionPath)
+ || !handle.canWrite()) {
+ handle = getRowCreateHandle(partitionPath);
+ lastKnownPartitionPath = partitionPath;
+ writeMetrics.ifPresent(FlinkStreamWriteMetrics::markHandleSwitch);
+ }
+ handle.write(recordKey, partitionPath, record);
+ writeMetrics.ifPresent(FlinkStreamWriteMetrics::markRecordIn);
+ }
+
private HoodieRowDataCreateHandle getRowCreateHandle(String partitionPath)
throws IOException {
if (!handles.containsKey(partitionPath)) { // if there is no handle
corresponding to the partition path
// if records are sorted, we can close all existing handles
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/LsmBulkInsertWriterHelper.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/LsmBulkInsertWriterHelper.java
new file mode 100644
index 000000000000..e20aa9688c28
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/LsmBulkInsertWriterHelper.java
@@ -0,0 +1,115 @@
+/*
+ * 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.sink.bulk;
+
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.sink.bulk.sort.SortOperatorGen;
+import org.apache.hudi.table.HoodieTable;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.io.IOException;
+
+/**
+ * Bulk-insert writer helper for LSM input rows.
+ *
+ * <p>The input row contains partition path, encoded record key, and the
original table row. The
+ * routing and record keys are retained after sorting so the writer does not
generate them again.
+ */
+public class LsmBulkInsertWriterHelper extends BulkInsertWriterHelper {
+
+ private static final String PARTITION_META_FIELD = "_partition";
+ private static final String RECORD_KEY_FIELD = "_record_key";
+ private static final String RECORD_FIELD = "record";
+
+ private final int recordArity;
+
+ public LsmBulkInsertWriterHelper(
+ Configuration conf,
+ HoodieTable<?, ?, ?, ?> hoodieTable,
+ HoodieWriteConfig writeConfig,
+ String instantTime,
+ int taskPartitionId,
+ long taskId,
+ long taskEpochId,
+ RowType rowType) {
+ super(conf, hoodieTable, writeConfig, instantTime, taskPartitionId,
taskId, taskEpochId, rowType);
+ this.recordArity = rowType.getFieldCount();
+ }
+
+ @Override
+ public void write(RowData sortRow) throws IOException {
+ String partitionPath = sortRow.getString(0).toString();
+ String recordKey = sortRow.getString(1).toString();
+ RowData record = sortRow.getRow(2, recordArity);
+ writeRecord(recordKey, partitionPath, record);
+ }
+
+ /**
+ * Decorates a table row with the partition path and encoded record key used
by the LSM
+ * sorter.
+ *
+ * @param partitionPath partition path
+ * @param record original table row used to generate the record key
+ * @param keyGen RowData key generator
+ * @return internal LSM sort row containing partition path, record key, and
original table row
+ */
+ public static RowData rowWithPartitionAndKey(
+ String partitionPath,
+ RowData record,
+ RowDataKeyGen keyGen) {
+ return GenericRowData.of(
+ StringData.fromString(partitionPath),
+ StringData.fromString(keyGen.getRecordKey(record)),
+ record);
+ }
+
+ /**
+ * Returns the internal row type used to sort LSM bulk-insert records by
partition and record key.
+ *
+ * <p>The fields are ordered as partition path, encoded record key, and
original table row.
+ */
+ public static RowType rowTypeWithPartitionAndKey(RowType rowType) {
+ LogicalType[] types = new LogicalType[] {
+ DataTypes.STRING().getLogicalType(),
+ DataTypes.STRING().getLogicalType(),
+ rowType
+ };
+ String[] names =
+ new String[] {PARTITION_META_FIELD, RECORD_KEY_FIELD, RECORD_FIELD};
+ return RowType.of(types, names);
+ }
+
+ /**
+ * Creates an external sorter ordered by partition path and encoded record
key.
+ *
+ * <p>The nested payload is deliberately excluded from the sort keys, so
duplicate record keys
+ * are retained without comparing or aggregating their payloads.
+ */
+ public static SortOperatorGen getPartitionAndKeySorterGen(RowType rowType) {
+ return new SortOperatorGen(
+ rowType, new String[] {PARTITION_META_FIELD, RECORD_KEY_FIELD});
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/WriterHelpers.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/WriterHelpers.java
index 99a9ae114cd8..0595830be8c7 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/WriterHelpers.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/WriterHelpers.java
@@ -21,6 +21,7 @@ package org.apache.hudi.sink.bulk;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.configuration.OptionsResolver;
import org.apache.hudi.sink.bucket.BucketBulkInsertWriterHelper;
+import org.apache.hudi.sink.bucket.LsmBucketBulkInsertWriterHelper;
import org.apache.hudi.table.HoodieTable;
import org.apache.flink.configuration.Configuration;
@@ -32,8 +33,18 @@ import org.apache.flink.table.types.logical.RowType;
public class WriterHelpers {
public static BulkInsertWriterHelper getWriterHelper(Configuration conf,
HoodieTable<?, ?, ?, ?> hoodieTable, HoodieWriteConfig writeConfig,
String instantTime, int
taskPartitionId, long taskId, long taskEpochId, RowType rowType) {
- return OptionsResolver.isBucketIndexType(conf)
- ? new BucketBulkInsertWriterHelper(conf, hoodieTable, writeConfig,
instantTime, taskPartitionId, taskId, taskEpochId, rowType)
- : new BulkInsertWriterHelper(conf, hoodieTable, writeConfig,
instantTime, taskPartitionId, taskId, taskEpochId, rowType);
+ if (OptionsResolver.isLsmTreeStorageLayout(conf)) {
+ return OptionsResolver.isBucketIndexType(conf)
+ ? new LsmBucketBulkInsertWriterHelper(
+ conf, hoodieTable, writeConfig, instantTime, taskPartitionId,
taskId, taskEpochId, rowType)
+ : new LsmBulkInsertWriterHelper(
+ conf, hoodieTable, writeConfig, instantTime, taskPartitionId,
taskId, taskEpochId, rowType);
+ } else {
+ return OptionsResolver.isBucketIndexType(conf)
+ ? new BucketBulkInsertWriterHelper(
+ conf, hoodieTable, writeConfig, instantTime, taskPartitionId,
taskId, taskEpochId, rowType)
+ : new BulkInsertWriterHelper(
+ conf, hoodieTable, writeConfig, instantTime, taskPartitionId,
taskId, taskEpochId, rowType);
+ }
}
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java
index 59612323b7db..ea2bf868c19d 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java
@@ -39,8 +39,10 @@ import
org.apache.hudi.sink.bootstrap.batch.BatchBootstrapOperator;
import org.apache.hudi.sink.bucket.BucketBulkInsertWriterHelper;
import org.apache.hudi.sink.bucket.BucketStreamWriteOperator;
import org.apache.hudi.sink.bucket.ConsistentBucketAssignFunction;
+import org.apache.hudi.sink.bucket.LsmBucketBulkInsertWriterHelper;
import org.apache.hudi.sink.buffer.BufferType;
import org.apache.hudi.sink.bulk.BulkInsertWriteOperator;
+import org.apache.hudi.sink.bulk.LsmBulkInsertWriterHelper;
import org.apache.hudi.sink.bulk.RowDataKeyGen;
import org.apache.hudi.sink.bulk.RowDataKeyGens;
import org.apache.hudi.sink.bulk.sort.SortOperatorGen;
@@ -125,82 +127,209 @@ public class Pipelines {
* @return the bulk insert data stream sink
*/
public static DataStream<RowData> bulkInsert(Configuration conf, RowType
rowType, DataStream<RowData> dataStream) {
+ if (OptionsResolver.isRecordLevelIndex(conf)) {
+ throw new HoodieException(
+ "Record level index does not work with bulk insert using FLINK
engine.");
+ }
+ // TODO support bulk insert for consistent bucket index
+ if (OptionsResolver.isConsistentHashingBucketIndexType(conf)) {
+ throw new HoodieException(
+ "Consistent hashing bucket index does not work with bulk insert
using FLINK engine. Use simple bucket index or Spark engine.");
+ }
+
// we need same parallelism for all operators,
// which is equal to write tasks number, to avoid shuffles
- final int PARALLELISM_VALUE = conf.get(FlinkOptions.WRITE_TASKS);
+ final int writeTasks = conf.get(FlinkOptions.WRITE_TASKS);
final boolean isBucketIndexType = OptionsResolver.isBucketIndexType(conf);
+ final boolean isLsmTreeStorageLayout =
OptionsResolver.isLsmTreeStorageLayout(conf);
+
+ DataStream<RowData> preparedDataStream = isBucketIndexType
+ ? bucketShuffleAndSort(
+ conf, rowType, dataStream, writeTasks, isLsmTreeStorageLayout)
+ : shuffleAndSort(
+ conf, rowType, dataStream, writeTasks, isLsmTreeStorageLayout);
+
+ String operatorName =
+ isBucketIndexType ? "bucket_bulk_insert" : "hoodie_bulk_insert_write";
+ return preparedDataStream
+ .transform(opName(operatorName, conf),
+ TypeInformation.of(RowData.class),
BulkInsertWriteOperator.getFactory(conf, rowType))
+ .uid(opUID(operatorName, conf))
+ .setParallelism(writeTasks);
+ }
- if (OptionsResolver.isRecordLevelIndex(conf)) {
- throw new HoodieException(
- "Record level index does not work with bulk insert using FLINK
engine.");
+ /**
+ * Shuffles and sorts the input stream for a bucket bulk insert writer.
+ *
+ * <p>Records are first routed to the write task that owns the target
bucket. For the default
+ * layout, the file ID is appended and the stream is optionally sorted by
file ID. For the LSM
+ * layout, the file ID and record key are appended in the same transform,
then the stream is
+ * sorted by file ID and record key.
+ */
+ private static DataStream<RowData> bucketShuffleAndSort(
+ Configuration conf,
+ RowType rowType,
+ DataStream<RowData> dataStream,
+ int writeTasks,
+ boolean isLsmTreeStorageLayout) {
+ List<String> indexKeyFieldList = OptionsResolver.getIndexKeyFields(conf);
+ // Built once and captured by the per-record map closure
(NumBucketsFunction is Serializable),
+ // avoiding a per-record rebuild from conf inside
BucketBulkInsertWriterHelper.
+ NumBucketsFunction numBucketsFunction = new NumBucketsFunction(
+ conf.get(FlinkOptions.BUCKET_INDEX_PARTITION_EXPRESSIONS),
+ conf.get(FlinkOptions.BUCKET_INDEX_PARTITION_RULE),
+ conf.get(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS));
+ Partitioner<HoodieKey> partitioner =
+ BucketIndexPartitionerFactory.create(conf, indexKeyFieldList);
+ RowDataKeyGen keyGen = RowDataKeyGens.instance(conf, rowType);
+ boolean needFixedFileIdSuffix =
+ OptionsResolver.isNonBlockingConcurrencyControl(conf);
+
+ Map<String, String> bucketIdToFileId = new HashMap<>();
+ DataStream<RowData> routedDataStream =
+ dataStream.partitionCustom(partitioner, keyGen::getHoodieKey);
+
+ if (isLsmTreeStorageLayout) {
+ RowType sortRowType =
+ LsmBucketBulkInsertWriterHelper.rowTypeWithFileIdAndKey(rowType);
+ InternalTypeInfo<RowData> sortTypeInfo =
InternalTypeInfo.of(sortRowType);
+ DataStream<RowData> sortInput = routedDataStream
+ .map(record -> LsmBucketBulkInsertWriterHelper.rowWithFileIdAndKey(
+ bucketIdToFileId,
+ keyGen,
+ record,
+ indexKeyFieldList,
+ numBucketsFunction,
+ needFixedFileIdSuffix), sortTypeInfo)
+ .name("lsm_bulk_insert_sort_keys")
+ .setParallelism(writeTasks);
+ return addBulkInsertSorter(
+ conf,
+ sortInput,
+ sortTypeInfo,
+
LsmBucketBulkInsertWriterHelper.getFileIdAndKeySorterGen(sortRowType),
+ "lsm_sorter:(file_group, record_key)",
+ writeTasks);
}
- if (isBucketIndexType) {
- // TODO support bulk insert for consistent bucket index
- if (OptionsResolver.isConsistentHashingBucketIndexType(conf)) {
- throw new HoodieException(
- "Consistent hashing bucket index does not work with bulk insert
using FLINK engine. Use simple bucket index or Spark engine.");
- }
- List<String> indexKeyFieldList = OptionsResolver.getIndexKeyFields(conf);
- // built once and captured by the per-record map closure
(NumBucketsFunction is Serializable),
- // avoiding a per-record rebuild from conf inside
BucketBulkInsertWriterHelper
- NumBucketsFunction numBucketsFunction = new
NumBucketsFunction(conf.get(FlinkOptions.BUCKET_INDEX_PARTITION_EXPRESSIONS),
- conf.get(FlinkOptions.BUCKET_INDEX_PARTITION_RULE),
conf.get(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS));
- Partitioner<HoodieKey> partitioner =
BucketIndexPartitionerFactory.create(conf, indexKeyFieldList);
- RowDataKeyGen keyGen = RowDataKeyGens.instance(conf, rowType);
- RowType rowTypeWithFileId =
BucketBulkInsertWriterHelper.rowTypeWithFileId(rowType);
- InternalTypeInfo<RowData> typeInfo =
InternalTypeInfo.of(rowTypeWithFileId);
- boolean needFixedFileIdSuffix =
OptionsResolver.isNonBlockingConcurrencyControl(conf);
-
- Map<String, String> bucketIdToFileId = new HashMap<>();
- dataStream = dataStream.partitionCustom(partitioner,
keyGen::getHoodieKey)
- .map(record ->
BucketBulkInsertWriterHelper.rowWithFileId(bucketIdToFileId, keyGen, record,
indexKeyFieldList, numBucketsFunction, needFixedFileIdSuffix), typeInfo)
- .setParallelism(PARALLELISM_VALUE);
- if (conf.get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT)) {
- SortOperatorGen sortOperatorGen =
BucketBulkInsertWriterHelper.getFileIdSorterGen(rowTypeWithFileId);
- dataStream = dataStream.transform("file_sorter", typeInfo,
sortOperatorGen.createSortOperator(conf))
- .setParallelism(PARALLELISM_VALUE);
-
FlinkTransformationUtils.setManagedMemoryWeight(dataStream.getTransformation(),
- conf.get(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L);
- }
- } else if (!FlinkOptions.isDefaultValueDefined(conf,
FlinkOptions.PARTITION_PATH_FIELD)) {
- // if table is not partitioned then we don't need any shuffles,
- // and could add main write operator only
- if (conf.get(FlinkOptions.WRITE_BULK_INSERT_SHUFFLE_INPUT)) {
- // shuffle by partition keys
- // use #partitionCustom instead of #keyBy to avoid duplicate sort
operations,
- // see BatchExecutionUtils#applyBatchExecutionSettings for details.
- Partitioner<String> partitioner = (key, channels) ->
KeyGroupRangeAssignment.assignKeyToParallelOperator(key,
-
KeyGroupRangeAssignment.computeDefaultMaxParallelism(PARALLELISM_VALUE),
channels);
- RowDataKeyGen rowDataKeyGen = RowDataKeyGens.instance(conf, rowType);
- dataStream = dataStream.partitionCustom(partitioner,
rowDataKeyGen::getPartitionPath);
- }
- if (conf.get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT)) {
- final boolean isNeededSortInput =
conf.get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT_BY_RECORD_KEY);
- final String[] partitionFields =
FilePathUtils.extractPartitionKeys(conf);
- final String[] recordKeyFields = OptionsResolver.getRecordKeys(conf);
-
- // if sort input by record key is needed then add record keys to
partition keys
- String[] sortFields = isNeededSortInput
- ? Stream.concat(Arrays.stream(partitionFields),
Arrays.stream(recordKeyFields)).toArray(String[]::new)
- : partitionFields;
- SortOperatorGen sortOperatorGen = new SortOperatorGen(rowType,
sortFields);
- dataStream = dataStream
- .transform(isNeededSortInput ? "sorter:(partition_key,
record_key)" : "sorter:(partition_key)",
- InternalTypeInfo.of(rowType),
sortOperatorGen.createSortOperator(conf))
- .setParallelism(PARALLELISM_VALUE);
-
FlinkTransformationUtils.setManagedMemoryWeight(dataStream.getTransformation(),
- conf.get(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L);
- }
+ RowType rowTypeWithFileId =
BucketBulkInsertWriterHelper.rowTypeWithFileId(rowType);
+ InternalTypeInfo<RowData> typeInfo =
InternalTypeInfo.of(rowTypeWithFileId);
+ DataStream<RowData> rowsWithFileId = routedDataStream
+ .map(record -> BucketBulkInsertWriterHelper.rowWithFileId(
+ bucketIdToFileId,
+ keyGen,
+ record,
+ indexKeyFieldList,
+ numBucketsFunction,
+ needFixedFileIdSuffix), typeInfo)
+ .setParallelism(writeTasks);
+
+ if (!conf.get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT)) {
+ return rowsWithFileId;
}
- // main write operator with following dummy sink in the end
- String opName = isBucketIndexType ? "bucket_bulk_insert" :
"hoodie_bulk_insert_write";
- return dataStream
- .transform(opName(opName, conf),
- TypeInformation.of(RowData.class),
BulkInsertWriteOperator.getFactory(conf, rowType))
- .uid(opUID(opName, conf))
- .setParallelism(PARALLELISM_VALUE);
+ return addBulkInsertSorter(
+ conf,
+ rowsWithFileId,
+ typeInfo,
+ BucketBulkInsertWriterHelper.getFileIdSorterGen(rowTypeWithFileId),
+ "file_sorter",
+ writeTasks);
+ }
+
+ /**
+ * Shuffles and sorts the input stream for a non-bucket bulk insert writer.
+ *
+ * <p>Partitioned input is optionally shuffled by partition path. The LSM
layout then appends
+ * the partition path and record key and sorts by both fields; for a
non-partitioned table the
+ * partition path is empty, so the effective ordering is by record key. The
default layout keeps
+ * the existing behavior: non-partitioned input is passed through without
shuffle or sort, while
+ * partitioned input is sorted only when bulk-insert input sorting is
enabled.
+ */
+ private static DataStream<RowData> shuffleAndSort(
+ Configuration conf,
+ RowType rowType,
+ DataStream<RowData> dataStream,
+ int writeTasks,
+ boolean isLsmTreeStorageLayout) {
+ final boolean isPartitioned =
+ !FlinkOptions.isDefaultValueDefined(conf,
FlinkOptions.PARTITION_PATH_FIELD);
+ final boolean shouldShuffle =
+ isPartitioned &&
conf.get(FlinkOptions.WRITE_BULK_INSERT_SHUFFLE_INPUT);
+ final RowDataKeyGen rowDataKeyGen = RowDataKeyGens.instance(conf, rowType);
+
+ DataStream<RowData> routedDataStream = dataStream;
+ if (shouldShuffle) {
+ // Use #partitionCustom instead of #keyBy to avoid duplicate sort
operations,
+ // see BatchExecutionUtils#applyBatchExecutionSettings for details.
+ Partitioner<String> partitioner =
+ (key, channels) ->
KeyGroupRangeAssignment.assignKeyToParallelOperator(
+ key,
+ KeyGroupRangeAssignment.computeDefaultMaxParallelism(writeTasks),
+ channels);
+ routedDataStream =
+ dataStream.partitionCustom(partitioner,
rowDataKeyGen::getPartitionPath);
+ }
+
+ if (isLsmTreeStorageLayout) {
+ // LSM sorted runs are ordered by partition path and the encoded record
key strings.
+ RowType sortRowType =
LsmBulkInsertWriterHelper.rowTypeWithPartitionAndKey(rowType);
+ InternalTypeInfo<RowData> sortTypeInfo =
InternalTypeInfo.of(sortRowType);
+ DataStream<RowData> sortInput = routedDataStream
+ .map(record -> LsmBulkInsertWriterHelper.rowWithPartitionAndKey(
+ rowDataKeyGen.getPartitionPath(record), record, rowDataKeyGen),
sortTypeInfo)
+ .name("lsm_bulk_insert_sort_keys")
+ .setParallelism(writeTasks);
+ return addBulkInsertSorter(
+ conf,
+ sortInput,
+ sortTypeInfo,
+ LsmBulkInsertWriterHelper.getPartitionAndKeySorterGen(sortRowType),
+ "lsm_sorter:(partition_path, record_key)",
+ writeTasks);
+ }
+
+ if (!isPartitioned ||
!conf.get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT)) {
+ return routedDataStream;
+ }
+
+ // Unlike the LSM path, the default-layout sorter orders the original
record-key fields by
+ // their Flink logical types. The resulting order can differ from encoded
record-key String
+ // ordering; for example, numeric keys are ordered as 2, 10 here but as
"10", "2" for LSM.
+ final boolean sortByRecordKey =
+ conf.get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT_BY_RECORD_KEY);
+ final String[] partitionFields = FilePathUtils.extractPartitionKeys(conf);
+ final String[] recordKeyFields = OptionsResolver.getRecordKeys(conf);
+ String[] sortFields = sortByRecordKey
+ ? Stream.concat(Arrays.stream(partitionFields),
Arrays.stream(recordKeyFields))
+ .toArray(String[]::new)
+ : partitionFields;
+
+ return addBulkInsertSorter(
+ conf,
+ routedDataStream,
+ InternalTypeInfo.of(rowType),
+ new SortOperatorGen(rowType, sortFields),
+ sortByRecordKey
+ ? "sorter:(partition_key, record_key)"
+ : "sorter:(partition_key)",
+ writeTasks);
+ }
+
+ private static DataStream<RowData> addBulkInsertSorter(
+ Configuration conf,
+ DataStream<RowData> dataStream,
+ TypeInformation<RowData> typeInfo,
+ SortOperatorGen sortOperatorGen,
+ String operatorName,
+ int writeTasks) {
+ DataStream<RowData> sortedDataStream = dataStream
+ .transform(operatorName, typeInfo,
sortOperatorGen.createSortOperator(conf))
+ .setParallelism(writeTasks);
+ FlinkTransformationUtils.setManagedMemoryWeight(
+ sortedDataStream.getTransformation(),
+ conf.get(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L);
+ return sortedDataStream;
}
/**
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java
index 92912fd46a17..0d1437c0216c 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java
@@ -193,9 +193,9 @@ public class HoodieTableFactory implements
DynamicTableSourceFactory, DynamicTab
Configuration conf) {
HoodieTableConfig.TableStorageLayout storageLayout =
OptionsResolver.getTableStorageLayout(conf);
ValidationUtils.checkArgument(
- !(OptionsResolver.isInsertOperation(conf) ||
OptionsResolver.isBulkInsertOperation(conf))
+ !OptionsResolver.isInsertOperation(conf)
|| storageLayout != HoodieTableConfig.TableStorageLayout.LSM_TREE,
- "The LSM tree storage layout does not support insert or bulk insert
operations because they allow duplicate record keys.");
+ "The LSM tree storage layout does not support insert operations
because they allow duplicate record keys.");
}
/**
@@ -506,6 +506,13 @@ public class HoodieTableFactory implements
DynamicTableSourceFactory, DynamicTab
* Sets up the table exec sort options.
*/
private void setupSortOptions(Configuration conf, ReadableConfig
contextConfig) {
+ if (OptionsResolver.isBulkInsertOperation(conf)
+ && OptionsResolver.isLsmTreeStorageLayout(conf)) {
+ // An LSM run must always be ordered by its encoded record key. These
options are mandatory
+ // for LSM bulk insert even when the user explicitly disables the
regular bulk-insert sort.
+ conf.set(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT, true);
+ conf.set(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT_BY_RECORD_KEY, true);
+ }
if
(contextConfig.getOptional(TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLES).isPresent()) {
conf.set(TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLES,
contextConfig.get(TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLES));
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestOptionsResolver.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestOptionsResolver.java
index 1cbeb4fb9033..d0ebf5fba637 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestOptionsResolver.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestOptionsResolver.java
@@ -69,7 +69,7 @@ public class TestOptionsResolver {
assertFalse(OptionsResolver.isLsmTreeStorageLayout(conf));
conf.set(FlinkOptions.OPERATION, WriteOperationType.BULK_INSERT.value());
- assertFalse(OptionsResolver.isLsmTreeStorageLayout(conf));
+ assertTrue(OptionsResolver.isLsmTreeStorageLayout(conf));
conf.setString(HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
HoodieTableConfig.TableStorageLayout.LSM_TREE.configValue());
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/TestLsmBulkInsertWriterHelper.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/TestLsmBulkInsertWriterHelper.java
new file mode 100644
index 000000000000..7e1fe68b1738
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/TestLsmBulkInsertWriterHelper.java
@@ -0,0 +1,141 @@
+/*
+ * 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.sink.bulk;
+
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.sink.bulk.sort.SortOperatorGen;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.runtime.generated.RecordComparator;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Tests for LSM sort-row construction and {@link LsmBulkInsertWriterHelper}.
*/
+class TestLsmBulkInsertWriterHelper {
+
+ @Test
+ void testActualCompositeRecordKeyAndPayloadAreRetained() {
+ RowType rowType = RowType.of(
+ new LogicalType[] {
+ new VarCharType(), new VarCharType(), new VarCharType(), new
VarCharType()
+ },
+ new String[] {"id1", "id2", "name", "partition"});
+ Configuration conf = new Configuration();
+ conf.set(FlinkOptions.RECORD_KEY_FIELD, "id1,id2");
+ conf.set(FlinkOptions.PARTITION_PATH_FIELD, "partition");
+ RowDataKeyGen keyGen = RowDataKeyGens.instance(conf, rowType);
+ List<RowData> rows = Arrays.asList(
+ compositeRow("a", "10"),
+ compositeRow("a", "2"),
+ compositeRow("a,", "2"),
+ compositeRow("a", String.valueOf((char) 0xE000)),
+ compositeRow("a", new String(Character.toChars(0x1F600))));
+
+ for (RowData row : rows) {
+ RowData sortRow = LsmBulkInsertWriterHelper.rowWithPartitionAndKey("p1",
row, keyGen);
+ assertEquals("p1", sortRow.getString(0).toString());
+ assertEquals(keyGen.getRecordKey(row), sortRow.getString(1).toString());
+ assertSame(row, sortRow.getRow(2, rowType.getFieldCount()));
+ }
+ }
+
+ @Test
+ void testDecoratedRowsSortByShuffleAndEncodedRecordKey() {
+ RowType rowType = RowType.of(
+ new LogicalType[] {new BigIntType(), new VarCharType(), new
VarCharType()},
+ new String[] {"id", "name", "partition"});
+ Configuration conf = new Configuration();
+ conf.set(FlinkOptions.RECORD_KEY_FIELD, "id");
+ conf.set(FlinkOptions.PARTITION_PATH_FIELD, "partition");
+ RowDataKeyGen keyGen = RowDataKeyGens.instance(conf, rowType);
+ RowType sortRowType =
LsmBulkInsertWriterHelper.rowTypeWithPartitionAndKey(rowType);
+ assertEquals(
+ Arrays.asList("_partition", "_record_key", "record"),
+ sortRowType.getFieldNames());
+
+ RowData row10 = row(10L, "ten", "p1");
+ RowData row2 = row(2L, "two", "p1");
+ RowData rowOtherPartition = row(1L, "one", "p2");
+ RowData sortRow10 = LsmBulkInsertWriterHelper.rowWithPartitionAndKey("p1",
row10, keyGen);
+ RowData sortRow2 = LsmBulkInsertWriterHelper.rowWithPartitionAndKey("p1",
row2, keyGen);
+ RowData sortRowOtherPartition =
+ LsmBulkInsertWriterHelper.rowWithPartitionAndKey("p2",
rowOtherPartition, keyGen);
+
+ SortOperatorGen sortOperatorGen =
+ LsmBulkInsertWriterHelper.getPartitionAndKeySorterGen(sortRowType);
+ RecordComparator comparator =
sortOperatorGen.generateRecordComparator("TestLsmBulkInsertComparator")
+ .newInstance(Thread.currentThread().getContextClassLoader());
+
+ assertTrue(comparator.compare(sortRow10, sortRow2) < 0);
+ assertTrue(comparator.compare(sortRow2, sortRowOtherPartition) < 0);
+ assertEquals("10", sortRow10.getString(1).toString());
+ assertSame(row10, sortRow10.getRow(2, rowType.getFieldCount()));
+ }
+
+ @Test
+ void testDuplicateKeysRemainDistinctPayloads() {
+ RowType rowType = RowType.of(
+ new LogicalType[] {new BigIntType(), new VarCharType(), new
VarCharType()},
+ new String[] {"id", "name", "partition"});
+ Configuration conf = new Configuration();
+ conf.set(FlinkOptions.RECORD_KEY_FIELD, "id");
+ conf.set(FlinkOptions.PARTITION_PATH_FIELD, "partition");
+ RowDataKeyGen keyGen = RowDataKeyGens.instance(conf, rowType);
+
+ RowData first = row(1L, "first", "p1");
+ RowData second = row(1L, "second", "p1");
+ RowData firstSortRow =
+ LsmBulkInsertWriterHelper.rowWithPartitionAndKey("p1", first, keyGen);
+ RowData secondSortRow =
+ LsmBulkInsertWriterHelper.rowWithPartitionAndKey("p1", second, keyGen);
+
+ assertEquals(firstSortRow.getString(1), secondSortRow.getString(1));
+ assertSame(first, firstSortRow.getRow(2, rowType.getFieldCount()));
+ assertSame(second, secondSortRow.getRow(2, rowType.getFieldCount()));
+ }
+
+ private static RowData row(long id, String name, String partition) {
+ return GenericRowData.of(
+ id,
+ StringData.fromString(name),
+ StringData.fromString(partition));
+ }
+
+ private static RowData compositeRow(String id1, String id2) {
+ return GenericRowData.of(
+ StringData.fromString(id1),
+ StringData.fromString(id2),
+ StringData.fromString("payload"),
+ StringData.fromString("p1"));
+ }
+
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/BulkInsertFunctionWrapper.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/BulkInsertFunctionWrapper.java
index 1d5cc2fc9a97..1ad8ccdeec4f 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/BulkInsertFunctionWrapper.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/BulkInsertFunctionWrapper.java
@@ -27,6 +27,7 @@ import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.index.bucket.partition.NumBucketsFunction;
import org.apache.hudi.sink.StreamWriteOperatorCoordinator;
import org.apache.hudi.sink.bucket.BucketBulkInsertWriterHelper;
+import org.apache.hudi.sink.bucket.LsmBucketBulkInsertWriterHelper;
import org.apache.hudi.sink.bulk.BulkInsertWriteFunction;
import org.apache.hudi.sink.bulk.RowDataKeyGen;
import org.apache.hudi.sink.bulk.RowDataKeyGens;
@@ -73,6 +74,7 @@ public class BulkInsertFunctionWrapper<I> implements
TestFunctionWrapper<I> {
private final Configuration conf;
private final RowType rowType;
private final RowType rowTypeWithFileId;
+ private final RowType sortInputRowType;
private final IOManager ioManager;
private final MockStreamingRuntimeContext runtimeContext;
@@ -82,6 +84,7 @@ public class BulkInsertFunctionWrapper<I> implements
TestFunctionWrapper<I> {
@Getter
private StreamWriteOperatorCoordinator coordinator;
private final boolean needSortInput;
+ private final boolean lsmSortInput;
private BulkInsertWriteFunction<RowData> writeFunction;
private MapFunction<RowData, RowData> mapFunction;
@@ -101,9 +104,14 @@ public class BulkInsertFunctionWrapper<I> implements
TestFunctionWrapper<I> {
this.conf = conf;
this.rowType = (RowType)
HoodieSchemaConverter.convertToDataType(StreamerUtil.getSourceSchema(conf)).getLogicalType();
this.rowTypeWithFileId =
BucketBulkInsertWriterHelper.rowTypeWithFileId(rowType);
+ this.lsmSortInput = OptionsResolver.isLsmTreeStorageLayout(conf);
+ this.sortInputRowType = lsmSortInput
+ ? LsmBucketBulkInsertWriterHelper.rowTypeWithFileIdAndKey(rowType)
+ : rowTypeWithFileId;
this.coordinatorContext = new MockOperatorCoordinatorContext(new
OperatorID(), 1);
this.coordinator = new StreamWriteOperatorCoordinator(conf,
this.coordinatorContext);
- this.needSortInput = conf.get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT);
+ this.needSortInput =
+ lsmSortInput || conf.get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT);
}
public void openFunction() throws Exception {
@@ -117,12 +125,12 @@ public class BulkInsertFunctionWrapper<I> implements
TestFunctionWrapper<I> {
}
public void invoke(I record) throws Exception {
- RowData recordWithFileId = mapFunction.map((RowData) record);
+ RowData routedRecord = mapFunction.map((RowData) record);
if (needSortInput) {
// Sort input first, trigger writeFunction at the #endInput
- sortOperator.processElement(new StreamRecord(recordWithFileId));
+ sortOperator.processElement(new StreamRecord(routedRecord));
} else {
- writeFunction.processElement(recordWithFileId, null, null);
+ writeFunction.processElement(routedRecord, null, null);
}
}
@@ -219,7 +227,11 @@ public class BulkInsertFunctionWrapper<I> implements
TestFunctionWrapper<I> {
conf.get(FlinkOptions.BUCKET_INDEX_PARTITION_RULE),
conf.get(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS));
boolean needFixedFileIdSuffix =
OptionsResolver.isNonBlockingConcurrencyControl(conf);
this.bucketIdToFileId = new HashMap<>();
- this.mapFunction = r ->
BucketBulkInsertWriterHelper.rowWithFileId(bucketIdToFileId, keyGen, r,
indexKeyFieldList, numBucketsFunction, needFixedFileIdSuffix);
+ this.mapFunction = lsmSortInput
+ ? r -> LsmBucketBulkInsertWriterHelper.rowWithFileIdAndKey(
+ bucketIdToFileId, keyGen, r, indexKeyFieldList,
numBucketsFunction, needFixedFileIdSuffix)
+ : r -> BucketBulkInsertWriterHelper.rowWithFileId(
+ bucketIdToFileId, keyGen, r, indexKeyFieldList,
numBucketsFunction, needFixedFileIdSuffix);
}
private void setupSortOperator() throws Exception {
@@ -232,13 +244,15 @@ public class BulkInsertFunctionWrapper<I> implements
TestFunctionWrapper<I> {
.setConfig(new StreamConfig(conf))
.setExecutionConfig(new ExecutionConfig().enableObjectReuse())
.build();
- SortOperatorGen sortOperatorGen =
BucketBulkInsertWriterHelper.getFileIdSorterGen(rowTypeWithFileId);
+ SortOperatorGen sortOperatorGen = lsmSortInput
+ ?
LsmBucketBulkInsertWriterHelper.getFileIdAndKeySorterGen(sortInputRowType)
+ : BucketBulkInsertWriterHelper.getFileIdSorterGen(rowTypeWithFileId);
this.sortOperator = (SortOperator)
sortOperatorGen.createSortOperator(conf);
this.sortOperator.setProcessingTimeService(new
TestProcessingTimeService());
this.output = new CollectOutputAdapter<>();
StreamConfig streamConfig = new StreamConfig(conf);
streamConfig.setOperatorID(new OperatorID());
- RowDataSerializer inputSerializer = new
RowDataSerializer(rowTypeWithFileId);
+ RowDataSerializer inputSerializer = new
RowDataSerializer(sortInputRowType);
TestStreamConfigs.setupNetworkInputs(streamConfig, inputSerializer);
streamConfig.setManagedMemoryFractionOperatorOfUseCase(ManagedMemoryUseCase.OPERATOR,
.99);
this.sortOperator.setup(streamTask, streamConfig, output);
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
index f1990de48820..ef017d7ed088 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
@@ -21,6 +21,7 @@ package org.apache.hudi.table;
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.model.DefaultHoodieRecordPayload;
+import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableConfig;
@@ -55,6 +56,7 @@ import org.apache.hudi.utils.TestUtils;
import org.apache.hudi.utils.factory.CollectSinkTableFactory;
import lombok.extern.slf4j.Slf4j;
+import org.apache.avro.generic.GenericRecord;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.core.execution.JobClient;
@@ -70,6 +72,8 @@ import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.data.RowData;
import org.apache.flink.types.Row;
import org.apache.flink.util.CollectionUtil;
+import org.apache.parquet.avro.AvroParquetReader;
+import org.apache.parquet.hadoop.ParquetReader;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
@@ -83,6 +87,8 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
@@ -2004,6 +2010,10 @@ public class ITTestHoodieDataSource {
expected.add("partition=par4" + "00000000");
assertEquals(expected.stream().sorted().collect(Collectors.toList()),
actual.stream().sorted().collect(Collectors.toList()));
+ if ("bulk_insert".equals(operationType)) {
+ assertEquals(TestData.DATA_SET_SOURCE_INSERT.size(),
+ assertBaseFilesAreSortedAndCountRecords(new File(basePath)));
+ }
}
@Test
@@ -2076,6 +2086,80 @@ public class ITTestHoodieDataSource {
+ "+I[id2, Stephen, 33, 1970-01-01T00:00:02, par1]]", 4);
}
+ @ParameterizedTest
+ @MethodSource("tableTypeAndBooleanTrueFalseParams")
+ void testLsmBulkInsertSortsEncodedRecordKeysAndSupportsUpdates(
+ HoodieTableType tableType, boolean partitioned) throws IOException {
+ TestConfigurations.Sql bulkInsertTable = sql("t1")
+ .field("id INT NOT NULL")
+ .field("name STRING")
+ .field("ts BIGINT")
+ .field("pt STRING")
+ .pkField("id")
+ .partitionField("pt")
+ .option(FlinkOptions.PATH, tempFile.getAbsolutePath())
+ .option(FlinkOptions.TABLE_TYPE, tableType)
+ .option(FlinkOptions.OPERATION, "bulk_insert")
+ .option(FlinkOptions.ORDERING_FIELDS, "ts")
+ .option(FlinkOptions.WRITE_TASKS, 1)
+ // LSM bulk insert must enforce record-key sorting even when the user
explicitly disables
+ // the legacy bulk-insert sorting options.
+ .option(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT, false)
+ .option(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT_BY_RECORD_KEY,
false);
+ batchTableEnv.executeSql(partitioned
+ ? bulkInsertTable.end()
+ : bulkInsertTable.noPartition().end());
+
+ // The encoded record keys use String ordering, where "10" < "2" and "11"
< "3".
+ // Keeping the input in numeric order verifies that sorting uses the
encoded Hudi record key
+ // instead of the INT field ordering.
+ execInsertSql(batchTableEnv, "insert into t1 values "
+ + "(2, 'old-2', 1, 'p1'), "
+ + "(10, 'old-10', 1, 'p1'), "
+ + "(3, 'old-3', 1, 'p2'), "
+ + "(11, 'old-11', 1, 'p2')");
+
+ // Validate both the persisted default layout and the physical record-key
ordering in every
+ // base file, rather than relying on query output that may be reordered
during the read.
+ HoodieTableMetaClient metaClient =
HoodieTestUtils.createMetaClient(tempFile.getAbsolutePath());
+ assertEquals(HoodieTableConfig.TableStorageLayout.LSM_TREE,
+ metaClient.getTableConfig().getTableStorageLayout());
+ assertEquals(4, assertBaseFilesAreSortedAndCountRecords(tempFile));
+
+ // Reopen the LSM table with upsert to verify that files created by bulk
insert remain usable
+ // by the update path for both COW and MOR tables.
+ batchTableEnv.executeSql("drop table t1");
+ TestConfigurations.Sql upsertTable = sql("t1")
+ .field("id INT NOT NULL")
+ .field("name STRING")
+ .field("ts BIGINT")
+ .field("pt STRING")
+ .pkField("id")
+ .partitionField("pt")
+ .option(FlinkOptions.PATH, tempFile.getAbsolutePath())
+ .option(FlinkOptions.TABLE_TYPE, tableType)
+ .option(FlinkOptions.OPERATION, "upsert")
+ .option(FlinkOptions.ORDERING_FIELDS, "ts")
+ .option(FlinkOptions.WRITE_TASKS, 1);
+ batchTableEnv.executeSql(partitioned
+ ? upsertTable.end()
+ : upsertTable.noPartition().end());
+
+ execInsertSql(batchTableEnv, "insert into t1 values "
+ + "(2, 'new-2', 2, 'p1'), "
+ + "(10, 'new-10', 2, 'p1'), "
+ + "(3, 'new-3', 2, 'p2'), "
+ + "(11, 'new-11', 2, 'p2')");
+
+ // The snapshot must expose the newer payload for every key after the
update.
+ List<Row> result = execSelectSql(batchTableEnv, "select * from t1");
+ assertRowsEquals(result, "["
+ + "+I[10, new-10, 2, p1], "
+ + "+I[11, new-11, 2, p2], "
+ + "+I[2, new-2, 2, p1], "
+ + "+I[3, new-3, 2, p2]]");
+ }
+
@Test
void testBulkInsertNonPartitionedTable() {
TableEnvironment tableEnv = batchTableEnv;
@@ -3951,6 +4035,38 @@ public class ITTestHoodieDataSource {
return conf.toMap();
}
+ private int assertBaseFilesAreSortedAndCountRecords(File tableBasePath)
throws IOException {
+ List<Path> baseFiles;
+ try (Stream<Path> paths = Files.walk(tableBasePath.toPath())) {
+ baseFiles = paths
+ .filter(Files::isRegularFile)
+ .filter(path -> path.getFileName().toString().endsWith(".parquet"))
+ .filter(path -> !path.toString().contains(
+ File.separator + HoodieTableMetaClient.METAFOLDER_NAME +
File.separator))
+ .collect(Collectors.toList());
+ }
+ assertFalse(baseFiles.isEmpty());
+
+ int totalRecords = 0;
+ for (Path baseFile : baseFiles) {
+ try (ParquetReader<GenericRecord> reader = AvroParquetReader
+ .<GenericRecord>builder(new
org.apache.hadoop.fs.Path(baseFile.toUri()))
+ .build()) {
+ String previousKey = null;
+ GenericRecord record;
+ while ((record = reader.read()) != null) {
+ String currentKey =
record.get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString();
+ String lastKey = previousKey;
+ assertTrue(previousKey == null || previousKey.compareTo(currentKey)
<= 0,
+ () -> "Base file " + baseFile + " is not sorted: " + lastKey + "
> " + currentKey);
+ previousKey = currentKey;
+ totalRecords++;
+ }
+ }
+ }
+ return totalRecords;
+ }
+
/**
* Return test params => (execution mode, table type).
*/
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java
index c9a5f1aa4ee2..1bbfbb65418f 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java
@@ -27,6 +27,7 @@ import org.apache.hudi.common.schema.HoodieSchemaUtils;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.configuration.OptionsResolver;
import org.apache.hudi.exception.HoodieValidationException;
import org.apache.hudi.hive.MultiPartKeysValueExtractor;
import org.apache.hudi.index.HoodieIndex;
@@ -56,7 +57,6 @@ import org.junit.jupiter.params.provider.EnumSource;
import java.io.File;
import java.io.IOException;
-import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
@@ -898,19 +898,42 @@ public class TestHoodieTableFactory {
}
@Test
- void testInsertOperationsDoNotSupportLsmTreeStorageLayout() throws
IOException {
- for (String operation : Arrays.asList("insert", "bulk_insert")) {
- Configuration newTableConf = new Configuration();
- newTableConf.set(FlinkOptions.PATH, new File(tempFile,
operation).getAbsolutePath());
- newTableConf.set(FlinkOptions.TABLE_NAME, "t_" + operation);
- newTableConf.set(FlinkOptions.RECORD_KEY_FIELD, "uuid");
- newTableConf.set(FlinkOptions.OPERATION, operation);
- newTableConf.setString(HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
- HoodieTableConfig.TableStorageLayout.LSM_TREE.configValue());
-
- assertThrows(IllegalArgumentException.class,
- () -> new
HoodieTableFactory().createDynamicTableSink(MockContext.getInstance(newTableConf)));
- }
+ void testStorageLayoutValidationAndBulkInsertSort() throws IOException {
+ Configuration insertConf = new Configuration();
+ insertConf.set(FlinkOptions.PATH, new File(tempFile,
"insert").getAbsolutePath());
+ insertConf.set(FlinkOptions.TABLE_NAME, "t_insert");
+ insertConf.set(FlinkOptions.RECORD_KEY_FIELD, "uuid");
+ insertConf.set(FlinkOptions.OPERATION, "insert");
+ insertConf.setString(HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
+ HoodieTableConfig.TableStorageLayout.LSM_TREE.configValue());
+ assertThrows(IllegalArgumentException.class,
+ () -> new
HoodieTableFactory().createDynamicTableSink(MockContext.getInstance(insertConf)));
+
+ Configuration bulkInsertConf = new Configuration();
+ bulkInsertConf.set(FlinkOptions.PATH, new File(tempFile,
"bulk_insert").getAbsolutePath());
+ bulkInsertConf.set(FlinkOptions.TABLE_NAME, "t_bulk_insert");
+ bulkInsertConf.set(FlinkOptions.RECORD_KEY_FIELD, "uuid");
+ bulkInsertConf.set(FlinkOptions.OPERATION, "bulk_insert");
+ bulkInsertConf.set(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT, false);
+
bulkInsertConf.set(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT_BY_RECORD_KEY,
false);
+ HoodieTableSink lsmSink = (HoodieTableSink) new HoodieTableFactory()
+ .createDynamicTableSink(MockContext.getInstance(bulkInsertConf));
+ assertThat(OptionsResolver.isLsmTreeStorageLayout(lsmSink.getConf()),
is(true));
+
assertThat(lsmSink.getConf().get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT),
is(true));
+
assertThat(lsmSink.getConf().get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT_BY_RECORD_KEY),
is(true));
+
+ Configuration defaultBulkInsertConf = new Configuration(bulkInsertConf);
+ defaultBulkInsertConf.set(FlinkOptions.PATH,
+ new File(tempFile, "default_bulk_insert").getAbsolutePath());
+
defaultBulkInsertConf.setString(HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
+ HoodieTableConfig.TableStorageLayout.DEFAULT.configValue());
+ defaultBulkInsertConf.set(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT,
false);
+
defaultBulkInsertConf.set(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT_BY_RECORD_KEY,
false);
+ HoodieTableSink defaultSink = (HoodieTableSink) new HoodieTableFactory()
+
.createDynamicTableSink(MockContext.getInstance(defaultBulkInsertConf));
+ assertThat(OptionsResolver.isLsmTreeStorageLayout(defaultSink.getConf()),
is(false));
+
assertThat(defaultSink.getConf().get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT),
is(false));
+
assertThat(defaultSink.getConf().get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT_BY_RECORD_KEY),
is(false));
Configuration tableConf = new Configuration();
tableConf.set(FlinkOptions.PATH, new File(tempFile,
"existing_lsm").getAbsolutePath());
@@ -926,6 +949,15 @@ public class TestHoodieTableFactory {
writeConf.set(FlinkOptions.OPERATION, "insert");
assertThrows(IllegalArgumentException.class,
() -> new
HoodieTableFactory().createDynamicTableSink(MockContext.getInstance(writeConf)));
+
+ writeConf.set(FlinkOptions.OPERATION, "bulk_insert");
+ writeConf.set(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT, false);
+ writeConf.set(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT_BY_RECORD_KEY,
false);
+ HoodieTableSink existingLsmSink = (HoodieTableSink) new
HoodieTableFactory()
+ .createDynamicTableSink(MockContext.getInstance(writeConf));
+
assertThat(OptionsResolver.isLsmTreeStorageLayout(existingLsmSink.getConf()),
is(true));
+
assertThat(existingLsmSink.getConf().get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT),
is(true));
+
assertThat(existingLsmSink.getConf().get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT_BY_RECORD_KEY),
is(true));
}
// -------------------------------------------------------------------------
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
index 831637fb7750..ca9c29ae7ff6 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
@@ -192,7 +192,7 @@ class TestStreamerUtil {
conf.set(FlinkOptions.OPERATION, WriteOperationType.BULK_INSERT.value());
metaClient = StreamerUtil.initTableIfNotExists(conf);
- assertFalse(metaClient.getTableConfig().isLSMTreeStorageLayout());
+ assertTrue(metaClient.getTableConfig().isLSMTreeStorageLayout());
}
@Test
