From 50bca1a9460e21c0165da0a6a9ad45577e0e8f84 Mon Sep 17 00:00:00 2001 From: Shuo Cheng Date: Tue, 28 Jul 2026 14:34:26 +0800 Subject: [PATCH 1/6] feat(flink): sort bulk insert records by record key for LSM layout --- .../hudi/configuration/OptionsResolver.java | 6 +- .../bucket/BucketBulkInsertWriterHelper.java | 47 ++++-- .../LsmBucketBulkInsertWriterHelper.java | 85 +++++++++++ .../sink/bulk/BulkInsertWriterHelper.java | 26 ++-- .../sink/bulk/LsmBulkInsertWriterHelper.java | 81 ++++++++++ .../apache/hudi/sink/bulk/WriterHelpers.java | 17 ++- .../bulk/sort/LsmBulkInsertSortUtils.java | 80 ++++++++++ .../org/apache/hudi/sink/utils/Pipelines.java | 77 ++++++++-- .../apache/hudi/table/HoodieTableFactory.java | 11 +- .../configuration/TestOptionsResolver.java | 2 +- .../bulk/sort/TestLsmBulkInsertSortUtils.java | 139 ++++++++++++++++++ .../sink/utils/BulkInsertFunctionWrapper.java | 29 +++- .../hudi/table/ITTestHoodieDataSource.java | 116 +++++++++++++++ .../hudi/table/TestHoodieTableFactory.java | 60 ++++++-- .../apache/hudi/utils/TestStreamerUtil.java | 2 +- 15 files changed, 709 insertions(+), 69 deletions(-) create mode 100644 hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/LsmBucketBulkInsertWriterHelper.java create mode 100644 hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/LsmBulkInsertWriterHelper.java create mode 100644 hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/LsmBulkInsertSortUtils.java create mode 100644 hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/sort/TestLsmBulkInsertSortUtils.java 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..eebbcd61ef464 --- /dev/null +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/LsmBucketBulkInsertWriterHelper.java @@ -0,0 +1,85 @@ +/* + * 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.table.HoodieTable; + +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.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 { + + 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 rowWithFileIdAndRecordKey( + 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); + } +} 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..be17a6a433378 --- /dev/null +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/LsmBulkInsertWriterHelper.java @@ -0,0 +1,81 @@ +/* + * 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.table.HoodieTable; + +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.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 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 Hudi record key used by the LSM + * sorter. + * + * @param partitionPath partition path + * @param record original table row used to generate the Hudi record key + * @param keyGen Hudi 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); + } +} 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..6b60f0bd8fbd4 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/bulk/sort/LsmBulkInsertSortUtils.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/LsmBulkInsertSortUtils.java new file mode 100644 index 0000000000000..7025fac3bc27c --- /dev/null +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/LsmBulkInsertSortUtils.java @@ -0,0 +1,80 @@ +/* + * 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.sort; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; + +/** + * Utilities for defining and sorting the internal rows used by Flink LSM bulk insert. + * + *

Each internal row contains a shuffle key, the actual encoded record key, and the original + * table row. The shuffle key is the partition path for the regular bulk-insert path and the file ID + * for the bucket-index path. The external sorter orders rows by the shuffle key first and the + * record key second, so records for each writer route form a record-key-ordered run. + * + *

The original table row remains nested in the internal row and is not part of the sort key. + * After sorting, the LSM writer helpers consume the internal row directly and reuse the retained + * record key. + */ +public final class LsmBulkInsertSortUtils { + + public static final String SHUFFLE_KEY_FIELD = "_shuffle_key"; + public static final String RECORD_KEY_FIELD = "_record_key"; + public static final String PAYLOAD_FIELD = "_payload"; + + private LsmBulkInsertSortUtils() { + } + + /** + * Returns the internal row type passed from LSM key decoration through the external sorter. + * + *

The fields are ordered as shuffle key, encoded Hudi record key, and original table row. + * + * @param payloadType logical type of the original table row + * @return logical type of the internal LSM sort row + */ + public static RowType sortRowType(RowType payloadType) { + LogicalType[] types = new LogicalType[] { + DataTypes.STRING().getLogicalType(), + DataTypes.STRING().getLogicalType(), + payloadType + }; + String[] names = new String[] {SHUFFLE_KEY_FIELD, RECORD_KEY_FIELD, PAYLOAD_FIELD}; + return RowType.of(types, names); + } + + /** + * Creates an external sorter ordered by shuffle key 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. + * + * @param sortRowType logical type returned by {@link #sortRowType(RowType)} + * @return generator for the LSM bulk-insert sort operator + */ + public static SortOperatorGen getLsmSorterGen(RowType sortRowType) { + return new SortOperatorGen(sortRowType, + new String[] { + SHUFFLE_KEY_FIELD, + RECORD_KEY_FIELD + }); + } +} 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..0b4b52208c93f 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,10 +39,13 @@ 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.LsmBulkInsertSortUtils; import org.apache.hudi.sink.bulk.sort.SortOperatorGen; import org.apache.hudi.sink.clustering.ClusteringCommitEvent; import org.apache.hudi.sink.clustering.ClusteringCommitSink; @@ -129,6 +132,7 @@ public static DataStream bulkInsert(Configuration conf, RowType rowType // which is equal to write tasks number, to avoid shuffles final int PARALLELISM_VALUE = conf.get(FlinkOptions.WRITE_TASKS); final boolean isBucketIndexType = OptionsResolver.isBucketIndexType(conf); + final boolean isLsmTreeStorageLayout = OptionsResolver.isLsmTreeStorageLayout(conf); if (OptionsResolver.isRecordLevelIndex(conf)) { throw new HoodieException( @@ -147,35 +151,80 @@ public static DataStream bulkInsert(Configuration conf, RowType rowType 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)) + dataStream = dataStream.partitionCustom(partitioner, keyGen::getHoodieKey); + if (isLsmTreeStorageLayout) { + RowType sortRowType = LsmBulkInsertSortUtils.sortRowType(rowType); + InternalTypeInfo sortTypeInfo = InternalTypeInfo.of(sortRowType); + dataStream = dataStream + .map(record -> LsmBucketBulkInsertWriterHelper.rowWithFileIdAndRecordKey( + bucketIdToFileId, + keyGen, + record, + indexKeyFieldList, + numBucketsFunction, + needFixedFileIdSuffix), sortTypeInfo) + .name("lsm_bulk_insert_sort_keys") + .setParallelism(PARALLELISM_VALUE); + SortOperatorGen sortOperatorGen = LsmBulkInsertSortUtils.getLsmSorterGen(sortRowType); + dataStream = dataStream + .transform("lsm_sorter:(file_group, record_key)", sortTypeInfo, + sortOperatorGen.createSortOperator(conf)) .setParallelism(PARALLELISM_VALUE); FlinkTransformationUtils.setManagedMemoryWeight(dataStream.getTransformation(), conf.get(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L); + } else { + RowType rowTypeWithFileId = BucketBulkInsertWriterHelper.rowTypeWithFileId(rowType); + InternalTypeInfo typeInfo = InternalTypeInfo.of(rowTypeWithFileId); + dataStream = dataStream + .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)) { + } else { + 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); + + if (shouldShuffle) { // 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)) { + if (isLsmTreeStorageLayout) { + RowType sortRowType = LsmBulkInsertSortUtils.sortRowType(rowType); + InternalTypeInfo sortTypeInfo = InternalTypeInfo.of(sortRowType); + dataStream = dataStream + .map(record -> LsmBulkInsertWriterHelper.rowWithPartitionAndKey( + rowDataKeyGen.getPartitionPath(record), record, rowDataKeyGen), sortTypeInfo) + .name("lsm_bulk_insert_sort_keys") + .setParallelism(PARALLELISM_VALUE); + SortOperatorGen sortOperatorGen = LsmBulkInsertSortUtils.getLsmSorterGen(sortRowType); + dataStream = dataStream + .transform("lsm_sorter:(partition_path, record_key)", sortTypeInfo, + sortOperatorGen.createSortOperator(conf)) + .setParallelism(PARALLELISM_VALUE); + FlinkTransformationUtils.setManagedMemoryWeight(dataStream.getTransformation(), + conf.get(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L); + } else if (isPartitioned && 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); 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/sort/TestLsmBulkInsertSortUtils.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/sort/TestLsmBulkInsertSortUtils.java new file mode 100644 index 0000000000000..1e97a42822b41 --- /dev/null +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/sort/TestLsmBulkInsertSortUtils.java @@ -0,0 +1,139 @@ +/* + * 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.sort; + +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.sink.bulk.LsmBulkInsertWriterHelper; +import org.apache.hudi.sink.bulk.RowDataKeyGen; +import org.apache.hudi.sink.bulk.RowDataKeyGens; + +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 LsmBulkInsertSortUtils}. */ +class TestLsmBulkInsertSortUtils { + + @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 = LsmBulkInsertSortUtils.sortRowType(rowType); + + 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 = LsmBulkInsertSortUtils.getLsmSorterGen(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..be0ed5477fef5 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,9 +27,11 @@ 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; +import org.apache.hudi.sink.bulk.sort.LsmBulkInsertSortUtils; import org.apache.hudi.sink.bulk.sort.SortOperator; import org.apache.hudi.sink.bulk.sort.SortOperatorGen; import org.apache.hudi.sink.common.AbstractWriteFunction; @@ -73,6 +75,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 +85,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 +105,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 + ? LsmBulkInsertSortUtils.sortRowType(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 +126,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 +228,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.rowWithFileIdAndRecordKey( + bucketIdToFileId, keyGen, r, indexKeyFieldList, numBucketsFunction, needFixedFileIdSuffix) + : r -> BucketBulkInsertWriterHelper.rowWithFileId( + bucketIdToFileId, keyGen, r, indexKeyFieldList, numBucketsFunction, needFixedFileIdSuffix); } private void setupSortOperator() throws Exception { @@ -232,13 +245,15 @@ private void setupSortOperator() throws Exception { .setConfig(new StreamConfig(conf)) .setExecutionConfig(new ExecutionConfig().enableObjectReuse()) .build(); - SortOperatorGen sortOperatorGen = BucketBulkInsertWriterHelper.getFileIdSorterGen(rowTypeWithFileId); + SortOperatorGen sortOperatorGen = lsmSortInput + ? LsmBulkInsertSortUtils.getLsmSorterGen(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..332c447ef9d5c 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,70 @@ void testBulkInsertWithSortByRecordKey() { + "+I[id2, Stephen, 33, 1970-01-01T00:00:02, par1]]", 4); } + @ParameterizedTest + @MethodSource("lsmBulkInsertParams") + 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) + .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()); + + 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')"); + + HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(tempFile.getAbsolutePath()); + assertEquals(HoodieTableConfig.TableStorageLayout.LSM_TREE, + metaClient.getTableConfig().getTableStorageLayout()); + assertEquals(4, assertBaseFilesAreSortedAndCountRecords(tempFile)); + + 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')"); + + 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 +4025,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). */ @@ -3979,6 +4085,16 @@ private static Stream executionModeAndTableTypeParams() { return Stream.of(data).map(Arguments::of); } + private static Stream lsmBulkInsertParams() { + Object[][] data = + new Object[][] { + {HoodieTableType.COPY_ON_WRITE, false}, + {HoodieTableType.COPY_ON_WRITE, true}, + {HoodieTableType.MERGE_ON_READ, false}, + {HoodieTableType.MERGE_ON_READ, true}}; + return Stream.of(data).map(Arguments::of); + } + /** * Return test params => (execution mode, hive style partitioning). */ 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 From 51b5e6042ae94f94e17db533e961d205e11cfa6a Mon Sep 17 00:00:00 2001 From: Shuo Cheng Date: Tue, 28 Jul 2026 19:59:05 +0800 Subject: [PATCH 2/6] update and fix test --- .../org/apache/hudi/sink/utils/Pipelines.java | 302 +++++++++++------- .../hudi/table/ITTestHoodieDataSource.java | 22 +- 2 files changed, 200 insertions(+), 124 deletions(-) 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 0b4b52208c93f..fb1878cbb2b20 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 @@ -128,128 +128,204 @@ 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); - if (OptionsResolver.isRecordLevelIndex(conf)) { - throw new HoodieException( - "Record level index does not work with bulk insert using FLINK engine."); + DataStream preparedDataStream = isBucketIndexType + ? prepareBucketBulkInsert( + conf, rowType, dataStream, writeTasks, isLsmTreeStorageLayout) + : prepareNonBucketBulkInsert( + 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); + } + + /** + * Prepares 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 prepareBucketBulkInsert( + 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 = LsmBulkInsertSortUtils.sortRowType(rowType); + InternalTypeInfo sortTypeInfo = InternalTypeInfo.of(sortRowType); + DataStream sortInput = routedDataStream + .map(record -> LsmBucketBulkInsertWriterHelper.rowWithFileIdAndRecordKey( + bucketIdToFileId, + keyGen, + record, + indexKeyFieldList, + numBucketsFunction, + needFixedFileIdSuffix), sortTypeInfo) + .name("lsm_bulk_insert_sort_keys") + .setParallelism(writeTasks); + return addBulkInsertSorter( + conf, + sortInput, + sortTypeInfo, + LsmBulkInsertSortUtils.getLsmSorterGen(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); - boolean needFixedFileIdSuffix = OptionsResolver.isNonBlockingConcurrencyControl(conf); - - Map bucketIdToFileId = new HashMap<>(); - dataStream = dataStream.partitionCustom(partitioner, keyGen::getHoodieKey); - if (isLsmTreeStorageLayout) { - RowType sortRowType = LsmBulkInsertSortUtils.sortRowType(rowType); - InternalTypeInfo sortTypeInfo = InternalTypeInfo.of(sortRowType); - dataStream = dataStream - .map(record -> LsmBucketBulkInsertWriterHelper.rowWithFileIdAndRecordKey( - bucketIdToFileId, - keyGen, - record, - indexKeyFieldList, - numBucketsFunction, - needFixedFileIdSuffix), sortTypeInfo) - .name("lsm_bulk_insert_sort_keys") - .setParallelism(PARALLELISM_VALUE); - SortOperatorGen sortOperatorGen = LsmBulkInsertSortUtils.getLsmSorterGen(sortRowType); - dataStream = dataStream - .transform("lsm_sorter:(file_group, record_key)", sortTypeInfo, - sortOperatorGen.createSortOperator(conf)) - .setParallelism(PARALLELISM_VALUE); - FlinkTransformationUtils.setManagedMemoryWeight(dataStream.getTransformation(), - conf.get(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L); - } else { - RowType rowTypeWithFileId = BucketBulkInsertWriterHelper.rowTypeWithFileId(rowType); - InternalTypeInfo typeInfo = InternalTypeInfo.of(rowTypeWithFileId); - dataStream = dataStream - .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 { - 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); - - if (shouldShuffle) { - // 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); - dataStream = dataStream.partitionCustom(partitioner, rowDataKeyGen::getPartitionPath); - } - if (isLsmTreeStorageLayout) { - RowType sortRowType = LsmBulkInsertSortUtils.sortRowType(rowType); - InternalTypeInfo sortTypeInfo = InternalTypeInfo.of(sortRowType); - dataStream = dataStream - .map(record -> LsmBulkInsertWriterHelper.rowWithPartitionAndKey( - rowDataKeyGen.getPartitionPath(record), record, rowDataKeyGen), sortTypeInfo) - .name("lsm_bulk_insert_sort_keys") - .setParallelism(PARALLELISM_VALUE); - SortOperatorGen sortOperatorGen = LsmBulkInsertSortUtils.getLsmSorterGen(sortRowType); - dataStream = dataStream - .transform("lsm_sorter:(partition_path, record_key)", sortTypeInfo, - sortOperatorGen.createSortOperator(conf)) - .setParallelism(PARALLELISM_VALUE); - FlinkTransformationUtils.setManagedMemoryWeight(dataStream.getTransformation(), - conf.get(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L); - } else if (isPartitioned && 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); + } + + /** + * Prepares 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 prepareNonBucketBulkInsert( + 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) { + RowType sortRowType = LsmBulkInsertSortUtils.sortRowType(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, + LsmBulkInsertSortUtils.getLsmSorterGen(sortRowType), + "lsm_sorter:(partition_path, record_key)", + writeTasks); + } + + if (!isPartitioned || !conf.get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT)) { + return routedDataStream; + } + + 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/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java index 332c447ef9d5c..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 @@ -2087,7 +2087,7 @@ void testBulkInsertWithSortByRecordKey() { } @ParameterizedTest - @MethodSource("lsmBulkInsertParams") + @MethodSource("tableTypeAndBooleanTrueFalseParams") void testLsmBulkInsertSortsEncodedRecordKeysAndSupportsUpdates( HoodieTableType tableType, boolean partitioned) throws IOException { TestConfigurations.Sql bulkInsertTable = sql("t1") @@ -2102,23 +2102,32 @@ void testLsmBulkInsertSortsEncodedRecordKeysAndSupportsUpdates( .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") @@ -2142,6 +2151,7 @@ void testLsmBulkInsertSortsEncodedRecordKeysAndSupportsUpdates( + "(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], " @@ -4085,16 +4095,6 @@ private static Stream executionModeAndTableTypeParams() { return Stream.of(data).map(Arguments::of); } - private static Stream lsmBulkInsertParams() { - Object[][] data = - new Object[][] { - {HoodieTableType.COPY_ON_WRITE, false}, - {HoodieTableType.COPY_ON_WRITE, true}, - {HoodieTableType.MERGE_ON_READ, false}, - {HoodieTableType.MERGE_ON_READ, true}}; - return Stream.of(data).map(Arguments::of); - } - /** * Return test params => (execution mode, hive style partitioning). */ From 544022e0fccf947cf5e5ecc4487ad8ec55e9dc9b Mon Sep 17 00:00:00 2001 From: Shuo Cheng Date: Thu, 30 Jul 2026 09:44:47 +0800 Subject: [PATCH 3/6] fix style --- .../src/main/java/org/apache/hudi/sink/bulk/WriterHelpers.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 6b60f0bd8fbd4..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 @@ -42,7 +42,7 @@ public static BulkInsertWriterHelper getWriterHelper(Configuration conf, HoodieT } else { return OptionsResolver.isBucketIndexType(conf) ? new BucketBulkInsertWriterHelper( - conf, hoodieTable, writeConfig, instantTime, taskPartitionId, taskId, taskEpochId, rowType) + conf, hoodieTable, writeConfig, instantTime, taskPartitionId, taskId, taskEpochId, rowType) : new BulkInsertWriterHelper( conf, hoodieTable, writeConfig, instantTime, taskPartitionId, taskId, taskEpochId, rowType); } From ac1c18598664ea4c514438ac6290909b54207ad6 Mon Sep 17 00:00:00 2001 From: Shuo Cheng Date: Thu, 30 Jul 2026 12:04:46 +0800 Subject: [PATCH 4/6] fix comment --- .../sink/bulk/LsmBulkInsertWriterHelper.java | 6 +++--- .../sink/bulk/sort/LsmBulkInsertSortUtils.java | 12 ++++++------ .../org/apache/hudi/sink/utils/Pipelines.java | 16 ++++++++++------ 3 files changed, 19 insertions(+), 15 deletions(-) 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 index be17a6a433378..ccd5bdd9e7189 100644 --- 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 @@ -61,12 +61,12 @@ public void write(RowData sortRow) throws IOException { } /** - * Decorates a table row with the partition path and encoded Hudi record key used by the LSM + * 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 Hudi record key - * @param keyGen Hudi RowData key generator + * @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( diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/LsmBulkInsertSortUtils.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/LsmBulkInsertSortUtils.java index 7025fac3bc27c..a6d25c2cae940 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/LsmBulkInsertSortUtils.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/LsmBulkInsertSortUtils.java @@ -38,7 +38,7 @@ public final class LsmBulkInsertSortUtils { public static final String SHUFFLE_KEY_FIELD = "_shuffle_key"; public static final String RECORD_KEY_FIELD = "_record_key"; - public static final String PAYLOAD_FIELD = "_payload"; + public static final String RECORD_FIELD = "_record"; private LsmBulkInsertSortUtils() { } @@ -46,18 +46,18 @@ private LsmBulkInsertSortUtils() { /** * Returns the internal row type passed from LSM key decoration through the external sorter. * - *

The fields are ordered as shuffle key, encoded Hudi record key, and original table row. + *

The fields are ordered as shuffle key, encoded record key, and original table row. * - * @param payloadType logical type of the original table row + * @param rowType logical type of the original table row * @return logical type of the internal LSM sort row */ - public static RowType sortRowType(RowType payloadType) { + public static RowType sortRowType(RowType rowType) { LogicalType[] types = new LogicalType[] { DataTypes.STRING().getLogicalType(), DataTypes.STRING().getLogicalType(), - payloadType + rowType }; - String[] names = new String[] {SHUFFLE_KEY_FIELD, RECORD_KEY_FIELD, PAYLOAD_FIELD}; + String[] names = new String[] {SHUFFLE_KEY_FIELD, RECORD_KEY_FIELD, RECORD_FIELD}; return RowType.of(types, names); } 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 fb1878cbb2b20..855e2eaa44052 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 @@ -145,9 +145,9 @@ public static DataStream bulkInsert(Configuration conf, RowType rowType final boolean isLsmTreeStorageLayout = OptionsResolver.isLsmTreeStorageLayout(conf); DataStream preparedDataStream = isBucketIndexType - ? prepareBucketBulkInsert( + ? bucketShuffleAndSort( conf, rowType, dataStream, writeTasks, isLsmTreeStorageLayout) - : prepareNonBucketBulkInsert( + : shuffleAndSort( conf, rowType, dataStream, writeTasks, isLsmTreeStorageLayout); String operatorName = @@ -160,14 +160,14 @@ public static DataStream bulkInsert(Configuration conf, RowType rowType } /** - * Prepares the input stream for a bucket bulk insert writer. + * 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 prepareBucketBulkInsert( + private static DataStream bucketShuffleAndSort( Configuration conf, RowType rowType, DataStream dataStream, @@ -238,7 +238,7 @@ private static DataStream prepareBucketBulkInsert( } /** - * Prepares the input stream for a non-bucket bulk insert writer. + * 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 @@ -246,7 +246,7 @@ private static DataStream prepareBucketBulkInsert( * 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 prepareNonBucketBulkInsert( + private static DataStream shuffleAndSort( Configuration conf, RowType rowType, DataStream dataStream, @@ -272,6 +272,7 @@ private static DataStream prepareNonBucketBulkInsert( } if (isLsmTreeStorageLayout) { + // LSM sorted runs are ordered by partition path and the encoded record key strings. RowType sortRowType = LsmBulkInsertSortUtils.sortRowType(rowType); InternalTypeInfo sortTypeInfo = InternalTypeInfo.of(sortRowType); DataStream sortInput = routedDataStream @@ -292,6 +293,9 @@ private static DataStream prepareNonBucketBulkInsert( 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); From e4e9d4ef5ef66caba070758269a2cf7db3776137 Mon Sep 17 00:00:00 2001 From: Shuo Cheng Date: Thu, 30 Jul 2026 19:32:45 +0800 Subject: [PATCH 5/6] fix comment --- .../sink/bulk/sort/LsmBulkInsertSortUtils.java | 16 +++++++++------- .../org/apache/hudi/sink/utils/Pipelines.java | 6 ++++-- .../bulk/sort/TestLsmBulkInsertSortUtils.java | 6 +++++- .../sink/utils/BulkInsertFunctionWrapper.java | 2 +- 4 files changed, 19 insertions(+), 11 deletions(-) diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/LsmBulkInsertSortUtils.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/LsmBulkInsertSortUtils.java index a6d25c2cae940..8ba2e87080589 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/LsmBulkInsertSortUtils.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/LsmBulkInsertSortUtils.java @@ -28,7 +28,7 @@ *

Each internal row contains a shuffle key, the actual encoded record key, and the original * table row. The shuffle key is the partition path for the regular bulk-insert path and the file ID * for the bucket-index path. The external sorter orders rows by the shuffle key first and the - * record key second, so records for each writer route form a record-key-ordered run. + * record key second, so records for each shuffle key form a record-key-ordered run. * *

The original table row remains nested in the internal row and is not part of the sort key. * After sorting, the LSM writer helpers consume the internal row directly and reuse the retained @@ -36,9 +36,10 @@ */ public final class LsmBulkInsertSortUtils { - public static final String SHUFFLE_KEY_FIELD = "_shuffle_key"; + public static final String FILE_GROUP_META_FIELD = "_fg"; + public static final String PARTITION_META_FIELD = "_partition"; public static final String RECORD_KEY_FIELD = "_record_key"; - public static final String RECORD_FIELD = "_record"; + public static final String RECORD_FIELD = "record"; private LsmBulkInsertSortUtils() { } @@ -49,15 +50,16 @@ private LsmBulkInsertSortUtils() { *

The fields are ordered as shuffle key, encoded record key, and original table row. * * @param rowType logical type of the original table row + * @param shuffleKeyField field name for the partition path or file group * @return logical type of the internal LSM sort row */ - public static RowType sortRowType(RowType rowType) { + public static RowType sortRowType(RowType rowType, String shuffleKeyField) { LogicalType[] types = new LogicalType[] { DataTypes.STRING().getLogicalType(), DataTypes.STRING().getLogicalType(), rowType }; - String[] names = new String[] {SHUFFLE_KEY_FIELD, RECORD_KEY_FIELD, RECORD_FIELD}; + String[] names = new String[] {shuffleKeyField, RECORD_KEY_FIELD, RECORD_FIELD}; return RowType.of(types, names); } @@ -67,13 +69,13 @@ public static RowType sortRowType(RowType rowType) { *

The nested payload is deliberately excluded from the sort keys, so duplicate record keys * are retained without comparing or aggregating their payloads. * - * @param sortRowType logical type returned by {@link #sortRowType(RowType)} + * @param sortRowType logical type returned by {@link #sortRowType(RowType, String)} * @return generator for the LSM bulk-insert sort operator */ public static SortOperatorGen getLsmSorterGen(RowType sortRowType) { return new SortOperatorGen(sortRowType, new String[] { - SHUFFLE_KEY_FIELD, + sortRowType.getFieldNames().get(0), RECORD_KEY_FIELD }); } 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 855e2eaa44052..53c283d60e8d8 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 @@ -191,7 +191,8 @@ private static DataStream bucketShuffleAndSort( dataStream.partitionCustom(partitioner, keyGen::getHoodieKey); if (isLsmTreeStorageLayout) { - RowType sortRowType = LsmBulkInsertSortUtils.sortRowType(rowType); + RowType sortRowType = LsmBulkInsertSortUtils.sortRowType( + rowType, LsmBulkInsertSortUtils.FILE_GROUP_META_FIELD); InternalTypeInfo sortTypeInfo = InternalTypeInfo.of(sortRowType); DataStream sortInput = routedDataStream .map(record -> LsmBucketBulkInsertWriterHelper.rowWithFileIdAndRecordKey( @@ -273,7 +274,8 @@ private static DataStream shuffleAndSort( if (isLsmTreeStorageLayout) { // LSM sorted runs are ordered by partition path and the encoded record key strings. - RowType sortRowType = LsmBulkInsertSortUtils.sortRowType(rowType); + RowType sortRowType = LsmBulkInsertSortUtils.sortRowType( + rowType, LsmBulkInsertSortUtils.PARTITION_META_FIELD); InternalTypeInfo sortTypeInfo = InternalTypeInfo.of(sortRowType); DataStream sortInput = routedDataStream .map(record -> LsmBulkInsertWriterHelper.rowWithPartitionAndKey( diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/sort/TestLsmBulkInsertSortUtils.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/sort/TestLsmBulkInsertSortUtils.java index 1e97a42822b41..f72a37415d988 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/sort/TestLsmBulkInsertSortUtils.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/sort/TestLsmBulkInsertSortUtils.java @@ -79,7 +79,11 @@ void testDecoratedRowsSortByShuffleAndEncodedRecordKey() { conf.set(FlinkOptions.RECORD_KEY_FIELD, "id"); conf.set(FlinkOptions.PARTITION_PATH_FIELD, "partition"); RowDataKeyGen keyGen = RowDataKeyGens.instance(conf, rowType); - RowType sortRowType = LsmBulkInsertSortUtils.sortRowType(rowType); + RowType sortRowType = LsmBulkInsertSortUtils.sortRowType( + rowType, LsmBulkInsertSortUtils.PARTITION_META_FIELD); + assertEquals( + Arrays.asList("_partition", "_record_key", "record"), + sortRowType.getFieldNames()); RowData row10 = row(10L, "ten", "p1"); RowData row2 = row(2L, "two", "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 be0ed5477fef5..879375bf5b320 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 @@ -107,7 +107,7 @@ public BulkInsertFunctionWrapper(String tablePath, Configuration conf) throws Ex this.rowTypeWithFileId = BucketBulkInsertWriterHelper.rowTypeWithFileId(rowType); this.lsmSortInput = OptionsResolver.isLsmTreeStorageLayout(conf); this.sortInputRowType = lsmSortInput - ? LsmBulkInsertSortUtils.sortRowType(rowType) + ? LsmBulkInsertSortUtils.sortRowType(rowType, LsmBulkInsertSortUtils.FILE_GROUP_META_FIELD) : rowTypeWithFileId; this.coordinatorContext = new MockOperatorCoordinatorContext(new OperatorID(), 1); this.coordinator = new StreamWriteOperatorCoordinator(conf, this.coordinatorContext); From badecf916b143bc50a885fff0f6011d17db92525 Mon Sep 17 00:00:00 2001 From: Shuo Cheng Date: Fri, 31 Jul 2026 11:46:43 +0800 Subject: [PATCH 6/6] fix comment --- .../LsmBucketBulkInsertWriterHelper.java | 36 +++++++- .../sink/bulk/LsmBulkInsertWriterHelper.java | 34 ++++++++ .../bulk/sort/LsmBulkInsertSortUtils.java | 82 ------------------- .../org/apache/hudi/sink/utils/Pipelines.java | 14 ++-- ...ava => TestLsmBulkInsertWriterHelper.java} | 16 ++-- .../sink/utils/BulkInsertFunctionWrapper.java | 7 +- 6 files changed, 85 insertions(+), 104 deletions(-) delete mode 100644 hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/LsmBulkInsertSortUtils.java rename hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/{sort/TestLsmBulkInsertSortUtils.java => TestLsmBulkInsertWriterHelper.java} (92%) 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 index eebbcd61ef464..23e0b60137ef4 100644 --- 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 @@ -21,12 +21,15 @@ 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; @@ -40,6 +43,9 @@ */ 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, @@ -61,7 +67,7 @@ public void write(RowData sortRow) throws IOException { writeRecord(recordKey, partitionPath, fileId, record); } - public static RowData rowWithFileIdAndRecordKey( + public static RowData rowWithFileIdAndKey( Map bucketIdToFileId, RowDataKeyGen keyGen, RowData record, @@ -82,4 +88,32 @@ public static RowData rowWithFileIdAndRecordKey( 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/LsmBulkInsertWriterHelper.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/LsmBulkInsertWriterHelper.java index ccd5bdd9e7189..e20aa9688c28c 100644 --- 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 @@ -19,12 +19,15 @@ 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; @@ -37,6 +40,10 @@ */ 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( @@ -78,4 +85,31 @@ public static RowData rowWithPartitionAndKey( 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/sort/LsmBulkInsertSortUtils.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/LsmBulkInsertSortUtils.java deleted file mode 100644 index 8ba2e87080589..0000000000000 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/LsmBulkInsertSortUtils.java +++ /dev/null @@ -1,82 +0,0 @@ -/* - * 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.sort; - -import org.apache.flink.table.api.DataTypes; -import org.apache.flink.table.types.logical.LogicalType; -import org.apache.flink.table.types.logical.RowType; - -/** - * Utilities for defining and sorting the internal rows used by Flink LSM bulk insert. - * - *

Each internal row contains a shuffle key, the actual encoded record key, and the original - * table row. The shuffle key is the partition path for the regular bulk-insert path and the file ID - * for the bucket-index path. The external sorter orders rows by the shuffle key first and the - * record key second, so records for each shuffle key form a record-key-ordered run. - * - *

The original table row remains nested in the internal row and is not part of the sort key. - * After sorting, the LSM writer helpers consume the internal row directly and reuse the retained - * record key. - */ -public final class LsmBulkInsertSortUtils { - - public static final String FILE_GROUP_META_FIELD = "_fg"; - public static final String PARTITION_META_FIELD = "_partition"; - public static final String RECORD_KEY_FIELD = "_record_key"; - public static final String RECORD_FIELD = "record"; - - private LsmBulkInsertSortUtils() { - } - - /** - * Returns the internal row type passed from LSM key decoration through the external sorter. - * - *

The fields are ordered as shuffle key, encoded record key, and original table row. - * - * @param rowType logical type of the original table row - * @param shuffleKeyField field name for the partition path or file group - * @return logical type of the internal LSM sort row - */ - public static RowType sortRowType(RowType rowType, String shuffleKeyField) { - LogicalType[] types = new LogicalType[] { - DataTypes.STRING().getLogicalType(), - DataTypes.STRING().getLogicalType(), - rowType - }; - String[] names = new String[] {shuffleKeyField, RECORD_KEY_FIELD, RECORD_FIELD}; - return RowType.of(types, names); - } - - /** - * Creates an external sorter ordered by shuffle key 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. - * - * @param sortRowType logical type returned by {@link #sortRowType(RowType, String)} - * @return generator for the LSM bulk-insert sort operator - */ - public static SortOperatorGen getLsmSorterGen(RowType sortRowType) { - return new SortOperatorGen(sortRowType, - new String[] { - sortRowType.getFieldNames().get(0), - RECORD_KEY_FIELD - }); - } -} 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 53c283d60e8d8..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 @@ -45,7 +45,6 @@ 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.LsmBulkInsertSortUtils; import org.apache.hudi.sink.bulk.sort.SortOperatorGen; import org.apache.hudi.sink.clustering.ClusteringCommitEvent; import org.apache.hudi.sink.clustering.ClusteringCommitSink; @@ -191,11 +190,11 @@ private static DataStream bucketShuffleAndSort( dataStream.partitionCustom(partitioner, keyGen::getHoodieKey); if (isLsmTreeStorageLayout) { - RowType sortRowType = LsmBulkInsertSortUtils.sortRowType( - rowType, LsmBulkInsertSortUtils.FILE_GROUP_META_FIELD); + RowType sortRowType = + LsmBucketBulkInsertWriterHelper.rowTypeWithFileIdAndKey(rowType); InternalTypeInfo sortTypeInfo = InternalTypeInfo.of(sortRowType); DataStream sortInput = routedDataStream - .map(record -> LsmBucketBulkInsertWriterHelper.rowWithFileIdAndRecordKey( + .map(record -> LsmBucketBulkInsertWriterHelper.rowWithFileIdAndKey( bucketIdToFileId, keyGen, record, @@ -208,7 +207,7 @@ private static DataStream bucketShuffleAndSort( conf, sortInput, sortTypeInfo, - LsmBulkInsertSortUtils.getLsmSorterGen(sortRowType), + LsmBucketBulkInsertWriterHelper.getFileIdAndKeySorterGen(sortRowType), "lsm_sorter:(file_group, record_key)", writeTasks); } @@ -274,8 +273,7 @@ private static DataStream shuffleAndSort( if (isLsmTreeStorageLayout) { // LSM sorted runs are ordered by partition path and the encoded record key strings. - RowType sortRowType = LsmBulkInsertSortUtils.sortRowType( - rowType, LsmBulkInsertSortUtils.PARTITION_META_FIELD); + RowType sortRowType = LsmBulkInsertWriterHelper.rowTypeWithPartitionAndKey(rowType); InternalTypeInfo sortTypeInfo = InternalTypeInfo.of(sortRowType); DataStream sortInput = routedDataStream .map(record -> LsmBulkInsertWriterHelper.rowWithPartitionAndKey( @@ -286,7 +284,7 @@ private static DataStream shuffleAndSort( conf, sortInput, sortTypeInfo, - LsmBulkInsertSortUtils.getLsmSorterGen(sortRowType), + LsmBulkInsertWriterHelper.getPartitionAndKeySorterGen(sortRowType), "lsm_sorter:(partition_path, record_key)", writeTasks); } diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/sort/TestLsmBulkInsertSortUtils.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/TestLsmBulkInsertWriterHelper.java similarity index 92% rename from hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/sort/TestLsmBulkInsertSortUtils.java rename to hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/TestLsmBulkInsertWriterHelper.java index f72a37415d988..7e1fe68b1738e 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/sort/TestLsmBulkInsertSortUtils.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/TestLsmBulkInsertWriterHelper.java @@ -16,12 +16,10 @@ * limitations under the License. */ -package org.apache.hudi.sink.bulk.sort; +package org.apache.hudi.sink.bulk; import org.apache.hudi.configuration.FlinkOptions; -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; import org.apache.flink.configuration.Configuration; import org.apache.flink.table.data.GenericRowData; @@ -41,8 +39,8 @@ import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; -/** Tests for LSM sort-row construction and {@link LsmBulkInsertSortUtils}. */ -class TestLsmBulkInsertSortUtils { +/** Tests for LSM sort-row construction and {@link LsmBulkInsertWriterHelper}. */ +class TestLsmBulkInsertWriterHelper { @Test void testActualCompositeRecordKeyAndPayloadAreRetained() { @@ -79,8 +77,7 @@ void testDecoratedRowsSortByShuffleAndEncodedRecordKey() { conf.set(FlinkOptions.RECORD_KEY_FIELD, "id"); conf.set(FlinkOptions.PARTITION_PATH_FIELD, "partition"); RowDataKeyGen keyGen = RowDataKeyGens.instance(conf, rowType); - RowType sortRowType = LsmBulkInsertSortUtils.sortRowType( - rowType, LsmBulkInsertSortUtils.PARTITION_META_FIELD); + RowType sortRowType = LsmBulkInsertWriterHelper.rowTypeWithPartitionAndKey(rowType); assertEquals( Arrays.asList("_partition", "_record_key", "record"), sortRowType.getFieldNames()); @@ -93,7 +90,8 @@ void testDecoratedRowsSortByShuffleAndEncodedRecordKey() { RowData sortRowOtherPartition = LsmBulkInsertWriterHelper.rowWithPartitionAndKey("p2", rowOtherPartition, keyGen); - SortOperatorGen sortOperatorGen = LsmBulkInsertSortUtils.getLsmSorterGen(sortRowType); + SortOperatorGen sortOperatorGen = + LsmBulkInsertWriterHelper.getPartitionAndKeySorterGen(sortRowType); RecordComparator comparator = sortOperatorGen.generateRecordComparator("TestLsmBulkInsertComparator") .newInstance(Thread.currentThread().getContextClassLoader()); 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 879375bf5b320..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 @@ -31,7 +31,6 @@ import org.apache.hudi.sink.bulk.BulkInsertWriteFunction; import org.apache.hudi.sink.bulk.RowDataKeyGen; import org.apache.hudi.sink.bulk.RowDataKeyGens; -import org.apache.hudi.sink.bulk.sort.LsmBulkInsertSortUtils; import org.apache.hudi.sink.bulk.sort.SortOperator; import org.apache.hudi.sink.bulk.sort.SortOperatorGen; import org.apache.hudi.sink.common.AbstractWriteFunction; @@ -107,7 +106,7 @@ public BulkInsertFunctionWrapper(String tablePath, Configuration conf) throws Ex this.rowTypeWithFileId = BucketBulkInsertWriterHelper.rowTypeWithFileId(rowType); this.lsmSortInput = OptionsResolver.isLsmTreeStorageLayout(conf); this.sortInputRowType = lsmSortInput - ? LsmBulkInsertSortUtils.sortRowType(rowType, LsmBulkInsertSortUtils.FILE_GROUP_META_FIELD) + ? LsmBucketBulkInsertWriterHelper.rowTypeWithFileIdAndKey(rowType) : rowTypeWithFileId; this.coordinatorContext = new MockOperatorCoordinatorContext(new OperatorID(), 1); this.coordinator = new StreamWriteOperatorCoordinator(conf, this.coordinatorContext); @@ -229,7 +228,7 @@ private void setupMapFunction() { boolean needFixedFileIdSuffix = OptionsResolver.isNonBlockingConcurrencyControl(conf); this.bucketIdToFileId = new HashMap<>(); this.mapFunction = lsmSortInput - ? r -> LsmBucketBulkInsertWriterHelper.rowWithFileIdAndRecordKey( + ? r -> LsmBucketBulkInsertWriterHelper.rowWithFileIdAndKey( bucketIdToFileId, keyGen, r, indexKeyFieldList, numBucketsFunction, needFixedFileIdSuffix) : r -> BucketBulkInsertWriterHelper.rowWithFileId( bucketIdToFileId, keyGen, r, indexKeyFieldList, numBucketsFunction, needFixedFileIdSuffix); @@ -246,7 +245,7 @@ private void setupSortOperator() throws Exception { .setExecutionConfig(new ExecutionConfig().enableObjectReuse()) .build(); SortOperatorGen sortOperatorGen = lsmSortInput - ? LsmBulkInsertSortUtils.getLsmSorterGen(sortInputRowType) + ? LsmBucketBulkInsertWriterHelper.getFileIdAndKeySorterGen(sortInputRowType) : BucketBulkInsertWriterHelper.getFileIdSorterGen(rowTypeWithFileId); this.sortOperator = (SortOperator) sortOperatorGen.createSortOperator(conf); this.sortOperator.setProcessingTimeService(new TestProcessingTimeService());