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 eb891b0a51fa1..07f31daa86162 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 static boolean isCowTable(Configuration conf) {
/**
* Returns the configured table storage layout.
*
- * 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.
+ *
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 358686f9f6ff8..4dc233bb0ed17 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 @@
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 void write(RowData tuple) throws IOException {
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 void write(RowData tuple) throws IOException {
}
}
+ 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 static SortOperatorGen getFileIdSorterGen(RowType rowType) {
return new SortOperatorGen(rowType, new String[] {FILE_GROUP_META_FIELD});
}
- private static String getFileId(Map bucketIdToFileId, RowDataKeyGen keyGen, RowData record, List 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 bucketIdToFileId,
+ String recordKey,
+ String partitionPath,
+ List 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 bucketIdToFileId, RowDataKeyGen keyGen, RowData record, List 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 0000000000000..23e0b60137ef4
--- /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.
+ *
+ * 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 bucketIdToFileId,
+ RowDataKeyGen keyGen,
+ RowData record,
+ List 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.
+ *
+ * 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.
+ *
+ *
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 30a19e1423aa6..123dd191a4c43 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 BulkInsertWriterHelper(Configuration conf, HoodieTable hoodieTable, Hoodi
? 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 void write(RowData record) throws IOException {
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 void write(RowData record) throws IOException {
}
}
+ 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
@@ -228,4 +235,3 @@ private WriteStatus closeWriteHandle(HoodieRowDataCreateHandle rowCreateHandle)
}
}
-
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 0000000000000..e20aa9688c28c
--- /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.
+ *
+ *
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.
+ *
+ *
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.
+ *
+ *
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 99a9ae114cd8e..0595830be8c77 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 @@
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 @@
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 59612323b7db2..ea2bf868c19df 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.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 bulkInsert(Configuration conf, RowType rowType, DataStream 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 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.
+ *
+ * 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 bucketShuffleAndSort(
+ Configuration conf,
+ RowType rowType,
+ DataStream dataStream,
+ int writeTasks,
+ boolean isLsmTreeStorageLayout) {
+ List 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 partitioner =
+ BucketIndexPartitionerFactory.create(conf, indexKeyFieldList);
+ RowDataKeyGen keyGen = RowDataKeyGens.instance(conf, rowType);
+ boolean needFixedFileIdSuffix =
+ OptionsResolver.isNonBlockingConcurrencyControl(conf);
+
+ Map bucketIdToFileId = new HashMap<>();
+ DataStream routedDataStream =
+ dataStream.partitionCustom(partitioner, keyGen::getHoodieKey);
+
+ if (isLsmTreeStorageLayout) {
+ RowType sortRowType =
+ LsmBucketBulkInsertWriterHelper.rowTypeWithFileIdAndKey(rowType);
+ InternalTypeInfo sortTypeInfo = InternalTypeInfo.of(sortRowType);
+ DataStream 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 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 partitioner = BucketIndexPartitionerFactory.create(conf, indexKeyFieldList);
- RowDataKeyGen keyGen = RowDataKeyGens.instance(conf, rowType);
- RowType rowTypeWithFileId = BucketBulkInsertWriterHelper.rowTypeWithFileId(rowType);
- InternalTypeInfo typeInfo = InternalTypeInfo.of(rowTypeWithFileId);
- boolean needFixedFileIdSuffix = OptionsResolver.isNonBlockingConcurrencyControl(conf);
-
- Map 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 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 typeInfo = InternalTypeInfo.of(rowTypeWithFileId);
+ DataStream 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.
+ *
+ * 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 shuffleAndSort(
+ Configuration conf,
+ RowType rowType,
+ DataStream 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 routedDataStream = dataStream;
+ if (shouldShuffle) {
+ // Use #partitionCustom instead of #keyBy to avoid duplicate sort operations,
+ // see BatchExecutionUtils#applyBatchExecutionSettings for details.
+ Partitioner 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 sortTypeInfo = InternalTypeInfo.of(sortRowType);
+ DataStream 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 addBulkInsertSorter(
+ Configuration conf,
+ DataStream dataStream,
+ TypeInformation typeInfo,
+ SortOperatorGen sortOperatorGen,
+ String operatorName,
+ int writeTasks) {
+ DataStream 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 92912fd46a17f..0d1437c0216c5 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 @@ private void checkStorageLayout(
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 @@ private static void setupWriteOptions(Configuration conf) {
* 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 bd22f66415843..1ce45c4e8d6d0 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
@@ -56,7 +56,7 @@ void testTableStorageLayoutDefaultsByOperation() {
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 0000000000000..7e1fe68b1738e
--- /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 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 1d5cc2fc9a97e..1ad8ccdeec4f2 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.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 implements TestFunctionWrapper {
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 implements TestFunctionWrapper {
@Getter
private StreamWriteOperatorCoordinator coordinator;
private final boolean needSortInput;
+ private final boolean lsmSortInput;
private BulkInsertWriteFunction writeFunction;
private MapFunction mapFunction;
@@ -101,9 +104,14 @@ public BulkInsertFunctionWrapper(String tablePath, Configuration conf) throws Ex
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 void openFunction() throws Exception {
}
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 @@ private void setupMapFunction() {
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 @@ private void setupSortOperator() throws Exception {
.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 f1990de488205..ef017d7ed0880 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 @@
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.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.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 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 @@ void testBulkInsertWithPartitionBucketIndex(String operationType, String tableTy
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 @@ void testBulkInsertWithSortByRecordKey() {
+ "+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 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 static Map getDefaultKeys() {
return conf.toMap();
}
+ private int assertBaseFilesAreSortedAndCountRecords(File tableBasePath) throws IOException {
+ List baseFiles;
+ try (Stream 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 reader = AvroParquetReader
+ .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 c9a5f1aa4ee2a..1bbfbb65418fe 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.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 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 @@ void testLanceFormatSupportedForFlinkTables() {
}
@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 @@ void testInsertOperationsDoNotSupportLsmTreeStorageLayout() throws IOException {
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 831637fb7750d..ca9c29ae7ff6d 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 @@ void testInitInsertTableStorageLayout() throws IOException {
conf.set(FlinkOptions.OPERATION, WriteOperationType.BULK_INSERT.value());
metaClient = StreamerUtil.initTableIfNotExists(conf);
- assertFalse(metaClient.getTableConfig().isLSMTreeStorageLayout());
+ assertTrue(metaClient.getTableConfig().isLSMTreeStorageLayout());
}
@Test