This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 72bae164b4ff fix(flink): preserve arrival order for equal keys in LSM
writes (#20138)
72bae164b4ff is described below
commit 72bae164b4ff70ec8d81590bc2c19b6e5ade8f4f
Author: Shuo Cheng <[email protected]>
AuthorDate: Thu Oct 1 10:01:01 2026 +0800
fix(flink): preserve arrival order for equal keys in LSM writes (#20138)
* fix(flink): preserve arrival order for equal keys in LSM writes
---
.../io/log/block/HoodieFlinkParquetDataBlock.java | 2 +-
.../org/apache/hudi/sink/StreamWriteFunction.java | 12 +-
.../AppendWriteFunctionWithBIMBufferSort.java | 4 +-
...AppendWriteFunctionWithDisruptorBufferSort.java | 2 +-
.../org/apache/hudi/sink/buffer/RowDataBucket.java | 18 +-
.../apache/hudi/sink/buffer/StableSortBuffer.java | 232 ++++++++++++++++++++
.../exception/MemoryPagesExhaustedException.java | 2 +-
.../sink/partitioner/index/IndexWriteFunction.java | 8 +-
.../org/apache/hudi/sink/utils/BufferUtils.java | 24 ++-
.../hudi/sink/utils/NaturalOrderKeyComputer.java | 2 +-
.../sink/utils/NaturalOrderRecordComparator.java | 2 +-
.../org/apache/hudi/sink/TestWriteCopyOnWrite.java | 2 +-
.../org/apache/hudi/sink/TestWriteMergeOnRead.java | 72 +++++++
.../apache/hudi/sink/buffer/TestRowDataBucket.java | 233 ++++++++++++++++++++-
.../partitioner/index/TestIndexWriteFunction.java | 6 +-
.../sink/utils/TestRecordKeySortKeyComputer.java | 6 +-
16 files changed, 590 insertions(+), 37 deletions(-)
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/log/block/HoodieFlinkParquetDataBlock.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/log/block/HoodieFlinkParquetDataBlock.java
index 08a989879b2c..2d312061d7cf 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/log/block/HoodieFlinkParquetDataBlock.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/log/block/HoodieFlinkParquetDataBlock.java
@@ -50,7 +50,7 @@ import static
org.apache.hudi.common.model.HoodieFileFormat.PARQUET;
/**
* HoodieFlinkParquetDataBlock employs an HoodieRecord iterator rather than a
HoodieRecord list for
- * parquet data block, aiming to better utilize the optimizations of {@code
BinaryInMemorySortBuffer},
+ * parquet data block, aiming to better utilize the optimizations of {@code
StableSortBuffer},
* for example, object reusing to decrease GC costs.
*/
public class HoodieFlinkParquetDataBlock extends HoodieParquetDataBlock
implements ColumnRangeMetadataProvider {
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java
index e34dd0b5c11a..4333b358dc89 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java
@@ -36,6 +36,7 @@ import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.metrics.FlinkStreamWriteMetrics;
import org.apache.hudi.sink.buffer.PreemptiveMemorySegmentPool;
import org.apache.hudi.sink.buffer.RowDataBucket;
+import org.apache.hudi.sink.buffer.StableSortBuffer;
import org.apache.hudi.sink.buffer.TotalSizeTracer;
import org.apache.hudi.sink.bulk.RowDataKeyGen;
import org.apache.hudi.sink.bulk.RowDataKeyGens;
@@ -61,7 +62,6 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.binary.BinaryRowData;
import org.apache.flink.table.runtime.generated.NormalizedKeyComputer;
import org.apache.flink.table.runtime.generated.RecordComparator;
-import org.apache.flink.table.runtime.operators.sort.BinaryInMemorySortBuffer;
import org.apache.flink.table.runtime.util.MemorySegmentPool;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.util.Collector;
@@ -84,7 +84,7 @@ import static
org.apache.hudi.common.util.HoodieRecordUtils.getOrderingFieldName
*
* <p><h2>Work Flow</h2>
*
- * <p>The function firstly buffers the data (RowData) in a binary buffer based
on {@code BinaryInMemorySortBuffer}.
+ * <p>The function firstly buffers the data (RowData) in a {@link
StableSortBuffer}.
* It flushes(write) the records batch when the batch size exceeds the
configured size {@link FlinkOptions#WRITE_BATCH_SIZE}
* or the memory of the binary buffer is exhausted, and could not append any
more data or a Flink checkpoint starts.
* After a batch has been written successfully, the function notifies its
operator coordinator {@link StreamWriteOperatorCoordinator}
@@ -355,7 +355,7 @@ public class StreamWriteFunction extends
AbstractStreamWriteFunction<HoodieFlink
boolean success = doBufferRecord(bucketID, record);
if (!success) {
// 2. reclaim pages. A buffer whose write returned false must never be
reused because
- // BinaryInMemorySortBuffer may already have changed its variable-length
storage state.
+ // StableSortBuffer may already have changed its variable-length storage
state.
reclaimMemoryAfterFailedWrite(bucketID);
// 2.1 retry once with a newly-created buffer
@@ -566,12 +566,12 @@ public class StreamWriteFunction extends
AbstractStreamWriteFunction<HoodieFlink
return statuses;
}
- private BinaryInMemorySortBuffer createDataBuffer() {
+ private StableSortBuffer createDataBuffer() {
if (recordKeyComputer == null) {
- return BufferUtils.createBuffer(rowType, preemptiveMemorySegmentPool);
+ return BufferUtils.createStableSortBuffer(rowType,
preemptiveMemorySegmentPool);
}
try {
- return BufferUtils.createBuffer(
+ return BufferUtils.createStableSortBuffer(
rowType,
preemptiveMemorySegmentPool,
recordKeyComputer,
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunctionWithBIMBufferSort.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunctionWithBIMBufferSort.java
index 11017d9e4fc2..3c6f9ca4c04c 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunctionWithBIMBufferSort.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunctionWithBIMBufferSort.java
@@ -92,11 +92,11 @@ public class AppendWriteFunctionWithBIMBufferSort<T>
extends AppendWriteFunction
GeneratedRecordComparator recordComparator =
sortOperatorGen.generateRecordComparator("SortComparator");
this.memorySegmentPools =
this.memorySegmentPoolFactory.createMemorySegmentPools(config, 2,
OptionsResolver.getWriteBufferSizeInBytes(config));
- this.activeBuffer = BufferUtils.createBuffer(rowType,
+ this.activeBuffer = BufferUtils.createSortBuffer(rowType,
memorySegmentPools[0],
keyComputer.newInstance(Thread.currentThread().getContextClassLoader()),
recordComparator.newInstance(Thread.currentThread().getContextClassLoader()));
- this.backgroundBuffer = BufferUtils.createBuffer(rowType,
+ this.backgroundBuffer = BufferUtils.createSortBuffer(rowType,
memorySegmentPools[1],
keyComputer.newInstance(Thread.currentThread().getContextClassLoader()),
recordComparator.newInstance(Thread.currentThread().getContextClassLoader()));
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunctionWithDisruptorBufferSort.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunctionWithDisruptorBufferSort.java
index 347ad81da22d..4653651e254c 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunctionWithDisruptorBufferSort.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunctionWithDisruptorBufferSort.java
@@ -106,7 +106,7 @@ public class AppendWriteFunctionWithDisruptorBufferSort<T>
extends AppendWriteFu
private void initDisruptorBuffer() throws Exception {
if (sortBuffer == null) {
- this.sortBuffer = BufferUtils.createBuffer(rowType,
+ this.sortBuffer = BufferUtils.createSortBuffer(rowType,
memorySegmentPool,
keyComputer.newInstance(Thread.currentThread().getContextClassLoader()),
recordComparator.newInstance(Thread.currentThread().getContextClassLoader()));
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/buffer/RowDataBucket.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/buffer/RowDataBucket.java
index 84ca929100c4..8a6f124af317 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/buffer/RowDataBucket.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/buffer/RowDataBucket.java
@@ -25,17 +25,16 @@ import lombok.Getter;
import org.apache.flink.runtime.operators.sort.QuickSort;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.binary.BinaryRowData;
-import org.apache.flink.table.runtime.operators.sort.BinaryInMemorySortBuffer;
import org.apache.flink.util.MutableObjectIterator;
import java.io.IOException;
/**
- * RowData buffer for a specific data bucket, and the buffer is based on
{@code BinaryInMemorySortBuffer}
- * from Flink, which is backed by a managed {@code MemorySegmentPool} to
minimize GC costs.
+ * RowData buffer for a specific data bucket, backed by a managed {@code
MemorySegmentPool}
+ * to minimize GC costs.
*/
public class RowDataBucket {
- private final BinaryInMemorySortBuffer dataBuffer;
+ private final StableSortBuffer dataBuffer;
@Getter
private final BucketInfo bucketInfo;
private final BufferSizeDetector detector;
@@ -46,7 +45,7 @@ public class RowDataBucket {
public RowDataBucket(
String bucketId,
- BinaryInMemorySortBuffer dataBuffer,
+ StableSortBuffer dataBuffer,
BucketInfo bucketInfo,
Double batchSize) {
this.bucketId = bucketId;
@@ -59,8 +58,13 @@ public class RowDataBucket {
return dataBuffer.getIterator();
}
+ /**
+ * Sorts by record key, retaining arrival order for equal keys.
+ */
public void sort() throws IOException {
- new QuickSort().sort(dataBuffer);
+ if (dataBuffer.size() > 1) {
+ new QuickSort().sort(dataBuffer);
+ }
}
public boolean writeRow(RowData rowData) throws IOException {
@@ -71,7 +75,7 @@ public class RowDataBucket {
if (success) {
detector.detect(rowData);
} else {
- // BinaryInMemorySortBuffer may have partially appended variable-length
data before
+ // The buffer may have partially appended variable-length data before
// returning false. Its internal pointers can no longer be trusted for
another write.
diverged = true;
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/buffer/StableSortBuffer.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/buffer/StableSortBuffer.java
new file mode 100644
index 000000000000..8cddc08b9114
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/buffer/StableSortBuffer.java
@@ -0,0 +1,232 @@
+/*
+ * 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.buffer;
+
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.SimpleCollectingOutputView;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.runtime.generated.NormalizedKeyComputer;
+import org.apache.flink.table.runtime.generated.RecordComparator;
+import org.apache.flink.table.runtime.operators.sort.BinaryIndexedSortable;
+import org.apache.flink.table.runtime.typeutils.AbstractRowDataSerializer;
+import org.apache.flink.table.runtime.typeutils.BinaryRowDataSerializer;
+import org.apache.flink.table.runtime.util.MemorySegmentPool;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Copied from Flink's {@code BinaryInMemorySortBuffer}.
+ *
+ * <p>Provides stable sorting by comparing record offsets when keys are equal,
+ * preserving the original insertion order of records with equal keys.
+ */
+public final class StableSortBuffer extends BinaryIndexedSortable {
+
+ private static final int MIN_REQUIRED_BUFFERS = 3;
+
+ private final AbstractRowDataSerializer<RowData> inputSerializer;
+ private final ArrayList<MemorySegment> recordBufferSegments;
+ private final SimpleCollectingOutputView recordCollector;
+ private final int totalNumBuffers;
+
+ private long currentDataBufferOffset;
+ private long sortIndexBytes;
+
+ /** Create a memory sorter in `insert` way. */
+ public static StableSortBuffer createBuffer(
+ NormalizedKeyComputer normalizedKeyComputer,
+ AbstractRowDataSerializer<RowData> inputSerializer,
+ BinaryRowDataSerializer serializer,
+ RecordComparator comparator,
+ MemorySegmentPool memoryPool) {
+ checkArgument(memoryPool.freePages() >= MIN_REQUIRED_BUFFERS);
+ int totalNumBuffers = memoryPool.freePages();
+ ArrayList<MemorySegment> recordBufferSegments = new ArrayList<>(16);
+ return new StableSortBuffer(
+ normalizedKeyComputer,
+ inputSerializer,
+ serializer,
+ comparator,
+ recordBufferSegments,
+ new SimpleCollectingOutputView(
+ recordBufferSegments, memoryPool, memoryPool.pageSize()),
+ memoryPool,
+ totalNumBuffers);
+ }
+
+ private StableSortBuffer(
+ NormalizedKeyComputer normalizedKeyComputer,
+ AbstractRowDataSerializer<RowData> inputSerializer,
+ BinaryRowDataSerializer serializer,
+ RecordComparator comparator,
+ ArrayList<MemorySegment> recordBufferSegments,
+ SimpleCollectingOutputView recordCollector,
+ MemorySegmentPool pool,
+ int totalNumBuffers) {
+ super(normalizedKeyComputer, serializer, comparator, recordBufferSegments,
pool);
+ this.inputSerializer = inputSerializer;
+ this.recordBufferSegments = recordBufferSegments;
+ this.recordCollector = recordCollector;
+ this.totalNumBuffers = totalNumBuffers;
+ }
+
+ // Preserve arrival order for equal keys using the existing record offsets.
+ @Override
+ public int compare(int segmentNumberI, int segmentOffsetI, int
segmentNumberJ, int segmentOffsetJ) {
+ int result = super.compare(segmentNumberI, segmentOffsetI, segmentNumberJ,
segmentOffsetJ);
+ // ------------------ hoodie update ---------------------
+ if (result != 0) {
+ return result;
+ }
+ // super.compare establishes full-key equality, including fully
determining normalized keys.
+ // Offset order is always ascending, even when the key comparator sorts in
descending order.
+ return Long.compare(sortIndex.get(segmentNumberI).getLong(segmentOffsetI),
+ sortIndex.get(segmentNumberJ).getLong(segmentOffsetJ));
+ }
+
+ /**
+ * Resets the sort buffer back to the state where it is empty. All contained
data is discarded.
+ */
+ public void reset() {
+ // reset all offsets
+ this.numRecords = 0;
+ this.currentSortIndexOffset = 0;
+ this.currentDataBufferOffset = 0;
+ this.sortIndexBytes = 0;
+ // return all memory
+ returnToSegmentPool();
+ // grab first buffers
+ this.currentSortIndexSegment = nextMemorySegment();
+ this.sortIndex.add(this.currentSortIndexSegment);
+ this.recordCollector.reset();
+ }
+
+ public void returnToSegmentPool() {
+ // return all memory
+ this.memorySegmentPool.returnAll(this.sortIndex);
+ this.memorySegmentPool.returnAll(this.recordBufferSegments);
+ this.sortIndex.clear();
+ this.recordBufferSegments.clear();
+ }
+
+ public boolean isEmpty() {
+ return this.numRecords == 0;
+ }
+
+ public void dispose() {
+ returnToSegmentPool();
+ }
+
+ public long getCapacity() {
+ return ((long) this.totalNumBuffers) * memorySegmentPool.pageSize();
+ }
+
+ public long getOccupancy() {
+ return this.currentDataBufferOffset + this.sortIndexBytes;
+ }
+
+ /**
+ * Writes a given record to this sort buffer. The written record will be
appended and take the
+ * last logical position.
+ *
+ * @param record The record to be written.
+ * @return True, if the record was successfully written, false, if the sort
buffer was full.
+ * @throws IOException Thrown, if an error occurred while serializing the
record into the
+ * buffers.
+ */
+ public boolean write(RowData record) throws IOException {
+ // check whether we need a new memory segment for the sort index
+ if (!checkNextIndexOffset()) {
+ return false;
+ }
+
+ // serialize the record into the data buffers
+ int skip;
+ try {
+ skip = this.inputSerializer.serializeToPages(record,
this.recordCollector);
+ } catch (EOFException e) {
+ return false;
+ }
+
+ final long newOffset = this.recordCollector.getCurrentOffset();
+ long currOffset = currentDataBufferOffset + skip;
+
+ writeIndexAndNormalizedKey(record, currOffset);
+
+ this.currentDataBufferOffset = newOffset;
+
+ return true;
+ }
+
+ private BinaryRowData getRecordFromBuffer(BinaryRowData reuse, long pointer)
+ throws IOException {
+ this.recordBuffer.setReadPosition(pointer);
+ return this.serializer.mapFromPages(reuse, this.recordBuffer);
+ }
+
+ /**
+ * Gets an iterator over all records in this buffer in their logical order.
+ *
+ * @return An iterator returning the records in their logical order.
+ */
+ public MutableObjectIterator<BinaryRowData> getIterator() {
+ return new MutableObjectIterator<BinaryRowData>() {
+ private final int size = size();
+ private int current = 0;
+
+ private int currentSegment = 0;
+ private int currentOffset = 0;
+
+ private MemorySegment currentIndexSegment = sortIndex.get(0);
+
+ @Override
+ public BinaryRowData next(BinaryRowData target) {
+ if (this.current < this.size) {
+ this.current++;
+ if (this.currentOffset > lastIndexEntryOffset) {
+ this.currentOffset = 0;
+ this.currentIndexSegment = sortIndex.get(++this.currentSegment);
+ }
+
+ long pointer = this.currentIndexSegment.getLong(this.currentOffset);
+ this.currentOffset += indexEntrySize;
+
+ try {
+ return getRecordFromBuffer(target, pointer);
+ } catch (IOException ioe) {
+ throw new RuntimeException(ioe);
+ }
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public BinaryRowData next() {
+ throw new RuntimeException("Not support!");
+ }
+ };
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/exception/MemoryPagesExhaustedException.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/exception/MemoryPagesExhaustedException.java
index 470c55fe9c2e..32d8e3c1778b 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/exception/MemoryPagesExhaustedException.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/exception/MemoryPagesExhaustedException.java
@@ -22,7 +22,7 @@ import org.apache.hudi.exception.HoodieException;
/**
* Exception thrown for memory pages in {@code MemorySegmentPool} are not
enough to
- * create a {@code BinaryInMemorySortBuffer}.
+ * create a sort buffer.
*/
public class MemoryPagesExhaustedException extends HoodieException {
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/IndexWriteFunction.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/IndexWriteFunction.java
index de8359ca779a..a6c9b1c9b2a0 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/IndexWriteFunction.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/IndexWriteFunction.java
@@ -28,6 +28,7 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.metadata.HoodieMetadataPayload;
+import org.apache.hudi.sink.buffer.StableSortBuffer;
import org.apache.hudi.sink.common.AbstractStreamWriteFunction;
import org.apache.hudi.sink.event.WriteMetadataEvent;
import org.apache.hudi.sink.utils.BufferUtils;
@@ -39,7 +40,6 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.operators.coordination.OperatorEvent;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.binary.BinaryRowData;
-import org.apache.flink.table.runtime.operators.sort.BinaryInMemorySortBuffer;
import org.apache.flink.table.runtime.util.MemorySegmentPool;
import org.apache.flink.util.Collector;
import org.apache.flink.util.IOUtils;
@@ -78,7 +78,7 @@ public class IndexWriteFunction extends
AbstractStreamWriteFunction<RowData> {
private transient MemorySegmentPool memorySegmentPool;
- private transient BinaryInMemorySortBuffer indexDataBuffer;
+ private transient StableSortBuffer indexDataBuffer;
/**
* Hoodie Flink table.
@@ -98,7 +98,7 @@ public class IndexWriteFunction extends
AbstractStreamWriteFunction<RowData> {
this.memorySegmentPool =
this.memorySegmentPoolFactory.createMemorySegmentPool(
config,
config.get(FlinkOptions.INDEX_RLI_WRITE_BUFFER_SIZE) * 1024 * 1024);
- this.indexDataBuffer =
BufferUtils.createBuffer(IndexRowUtils.INDEX_ROW_TYPE, memorySegmentPool);
+ this.indexDataBuffer =
BufferUtils.createStableSortBuffer(IndexRowUtils.INDEX_ROW_TYPE,
memorySegmentPool);
this.dedupKeyExtractor =
this.writeClient.getConfig().isRecordLevelIndexEnabled()
? IndexWriteFunction::getPartitionedDedupKey :
IndexRowUtils::getRecordKey;
}
@@ -155,7 +155,7 @@ public class IndexWriteFunction extends
AbstractStreamWriteFunction<RowData> {
}
}
- private Pair<List<HoodieRecord>, Set<String>>
prepareIndexRecordsAndPartitions(BinaryInMemorySortBuffer indexDataBuffer) {
+ private Pair<List<HoodieRecord>, Set<String>>
prepareIndexRecordsAndPartitions(StableSortBuffer indexDataBuffer) {
BinaryRowData reusedRow = new
BinaryRowData(IndexRowUtils.INDEX_ROW_TYPE.getFieldCount());
Iterator<BinaryRowData> rowItr = new
MutableIteratorWrapperIterator<>(indexDataBuffer.getIterator(), () ->
reusedRow);
Set<String> dataPartitions = new HashSet<>();
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/BufferUtils.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/BufferUtils.java
index 831db7961639..861335ae2a4c 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/BufferUtils.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/BufferUtils.java
@@ -18,6 +18,7 @@
package org.apache.hudi.sink.utils;
+import org.apache.hudi.sink.buffer.StableSortBuffer;
import org.apache.hudi.sink.exception.MemoryPagesExhaustedException;
import org.apache.flink.table.runtime.generated.NormalizedKeyComputer;
@@ -33,16 +34,31 @@ import org.apache.flink.table.types.logical.RowType;
* Utilities to create binary buffer for writing functions.
*/
public class BufferUtils {
- // minimum pages for a BinaryInMemorySortBuffer
+ // minimum pages for either sort buffer
private static final int MIN_REQUIRED_BUFFERS = 3;
- public static BinaryInMemorySortBuffer createBuffer(RowType rowType,
MemorySegmentPool memorySegmentPool) {
- return createBuffer(rowType, memorySegmentPool, new
NaturalOrderKeyComputer(), new NaturalOrderRecordComparator());
+ public static StableSortBuffer createStableSortBuffer(RowType rowType,
MemorySegmentPool memorySegmentPool) {
+ return createStableSortBuffer(rowType, memorySegmentPool, new
NaturalOrderKeyComputer(), new NaturalOrderRecordComparator());
}
- public static BinaryInMemorySortBuffer createBuffer(RowType rowType,
MemorySegmentPool memorySegmentPool, NormalizedKeyComputer keyComputer,
RecordComparator recordComparator) {
+ public static StableSortBuffer createStableSortBuffer(RowType rowType,
MemorySegmentPool memorySegmentPool, NormalizedKeyComputer keyComputer,
RecordComparator recordComparator) {
if (memorySegmentPool.freePages() < MIN_REQUIRED_BUFFERS) {
// there is no enough free pages to create a binary buffer, may need
flush first.
+ throw new MemoryPagesExhaustedException("Free pages are not enough to
create a StableSortBuffer.");
+ }
+ return StableSortBuffer.createBuffer(
+ keyComputer,
+ new RowDataSerializer(rowType),
+ new BinaryRowDataSerializer(rowType.getFieldCount()),
+ recordComparator,
+ memorySegmentPool);
+ }
+
+ /** Creates a Flink sort buffer, which do not require stable ordering. */
+ public static BinaryInMemorySortBuffer createSortBuffer(
+ RowType rowType, MemorySegmentPool memorySegmentPool,
+ NormalizedKeyComputer keyComputer, RecordComparator recordComparator) {
+ if (memorySegmentPool.freePages() < MIN_REQUIRED_BUFFERS) {
throw new MemoryPagesExhaustedException("Free pages are not enough to
create a BinaryInMemorySortBuffer.");
}
return BinaryInMemorySortBuffer.createBuffer(
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/NaturalOrderKeyComputer.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/NaturalOrderKeyComputer.java
index 32b7af13fa3c..7e1303385164 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/NaturalOrderKeyComputer.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/NaturalOrderKeyComputer.java
@@ -24,7 +24,7 @@ import
org.apache.flink.table.runtime.generated.NormalizedKeyComputer;
import org.apache.flink.table.runtime.operators.sort.SortUtil;
/**
- * An implementation of @{code NormalizedKeyComputer} for {@code
BinaryInMemorySortBuffer} in case sort is not needed,
+ * An implementation of @{code NormalizedKeyComputer} for {@code
StableSortBuffer} in case sort is not needed,
* and the returned iterator follows natural order as inserted.
*/
public class NaturalOrderKeyComputer implements NormalizedKeyComputer {
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/NaturalOrderRecordComparator.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/NaturalOrderRecordComparator.java
index 9cd2807d096f..3f891f09317e 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/NaturalOrderRecordComparator.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/NaturalOrderRecordComparator.java
@@ -22,7 +22,7 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.runtime.generated.RecordComparator;
/**
- * An implementation of @{code RecordComparator} for {@code
BinaryInMemorySortBuffer} in case sort is not needed,
+ * An implementation of @{code RecordComparator} for {@code StableSortBuffer}
in case sort is not needed,
* and the returned iterator follows natural order as inserted.
*/
public class NaturalOrderRecordComparator implements RecordComparator {
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
index 3b3798062e7f..7f53f9fb8efc 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
@@ -673,7 +673,7 @@ public class TestWriteCopyOnWrite extends TestWriteBase {
@Test
public void testInsertWithSmallBufferSize() throws Exception {
// reset the config option
- // In rowdata write mode, BinaryInMemorySortBuffer need at least 2 memory
segments for auxiliary information,
+ // In rowdata write mode, StableSortBuffer need at least 2 memory segments
for auxiliary information,
// the page size is tuned to 64 to make sure 3 records from 5 will trigger
a mini-batch write.
conf.set(FlinkOptions.WRITE_MEMORY_SEGMENT_PAGE_SIZE, 64);
conf.set(FlinkOptions.WRITE_TASK_MAX_SIZE, 200 + BUFFER_SIZE_MB);
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnRead.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnRead.java
index 64cdd0405518..cdc530456e80 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnRead.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnRead.java
@@ -20,12 +20,14 @@ package org.apache.hudi.sink;
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
+import org.apache.hudi.common.config.RecordMergeMode;
import org.apache.hudi.common.model.EventTimeAvroPayload;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.model.MetaFieldsMode;
+import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.config.HoodieClusteringConfig;
@@ -34,22 +36,31 @@ import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.util.StreamerUtil;
+import org.apache.hudi.utils.TestConfigurations;
import org.apache.hudi.utils.TestData;
+import org.apache.avro.generic.GenericRecord;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.TimestampData;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
import static org.apache.hudi.utils.TestData.insertRow;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Test cases for delta stream write.
@@ -61,6 +72,67 @@ public class TestWriteMergeOnRead extends
TestWriteCopyOnWrite {
conf.set(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
+ @ParameterizedTest
+ @CsvSource({
+ "COMMIT_TIME_ORDERING, false, false", "COMMIT_TIME_ORDERING, true,
false",
+ "EVENT_TIME_ORDERING, false, false", "EVENT_TIME_ORDERING, true, false",
+ "EVENT_TIME_ORDERING, false, true", "EVENT_TIME_ORDERING, true, true"
+ })
+ void testLsmDuplicateKeysRespectMergeMode(RecordMergeMode mergeMode, boolean
preCombine, boolean equalOrderingValues) throws Exception {
+ conf.setString(HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(), "LSM_TREE");
+ conf.set(FlinkOptions.RECORD_MERGE_MODE, mergeMode.name());
+ conf.set(FlinkOptions.PRE_COMBINE, preCombine);
+ conf.set(FlinkOptions.INDEX_TYPE, "BUCKET");
+ conf.set(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS, 1);
+ conf.set(FlinkOptions.WRITE_BATCH_SIZE, 256.0);
+ boolean commitTimeOrdering = mergeMode ==
RecordMergeMode.COMMIT_TIME_ORDERING;
+ boolean lastArrivalWins = commitTimeOrdering || equalOrderingValues;
+ List<RowData> records = new ArrayList<>();
+ Map<String, Integer> expectedWinners = new HashMap<>();
+ for (int i = 0; i < 1000; i++) {
+ String key = "id" + i % 17;
+ // Decreasing event times distinguish event-time merging from
commit-time merging.
+ // Equal event times must retain the last arrival, just like commit-time
merging.
+ records.add(insertRow(StringData.fromString(key),
StringData.fromString("name"), i,
+ TimestampData.fromEpochMillis(equalOrderingValues ? 1000 : 1000 -
i), StringData.fromString("par1")));
+ if (lastArrivalWins) {
+ expectedWinners.put(key, i);
+ } else {
+ expectedWinners.putIfAbsent(key, i);
+ }
+ }
+ String expected =
expectedWinners.entrySet().stream().sorted(Map.Entry.comparingByKey())
+ .map(entry -> entry.getKey() + ",par1," + entry.getKey() + ",name," +
entry.getValue()
+ + "," + (equalOrderingValues ? 1000 : 1000 - entry.getValue()) +
",par1")
+ .collect(Collectors.joining(", ", "[", "]"));
+ preparePipeline(conf)
+ .consume(records)
+ .checkpoint(1)
+ .assertNextEvent()
+ .checkpointComplete(1)
+ .checkWrittenData(Collections.singletonMap("par1", expected), 1)
+ .end();
+
+ // With pre-combine off, all versions must reach one native log file in
arrival order
+ // within each key, including event-time ties resolved during subsequent
reads.
+ File[] logFiles = new File(tempFile, "par1").listFiles(file ->
file.getName().endsWith(".log.parquet"));
+ assertNotNull(logFiles);
+ assertEquals(1, logFiles.length, "duplicate versions must be tested within
a single native log file");
+ List<GenericRecord> written = TestData.readAllData(tempFile,
TestConfigurations.ROW_TYPE, 1);
+ assertEquals(preCombine ? expectedWinners.size() : records.size(),
written.size());
+ Map<String, Integer> previous = new HashMap<>();
+ Map<String, Integer> actualWinners = new HashMap<>();
+ for (GenericRecord record : written) {
+ String key = record.get("uuid").toString();
+ int arrival = (Integer) record.get("age");
+ Integer oldArrival = previous.put(key, arrival);
+ assertTrue(oldArrival == null || arrival > oldArrival, "file order must
preserve arrivals for " + key);
+ actualWinners.merge(key, arrival, (oldValue, newValue) ->
+ lastArrivalWins ? Math.max(oldValue, newValue) : Math.min(oldValue,
newValue));
+ }
+ assertEquals(expectedWinners, actualWinners);
+ }
+
@Test
public void testIndexStateBootstrapWithMultiFilesInOneSlice() throws
Exception {
// open the function and ingest data
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/buffer/TestRowDataBucket.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/buffer/TestRowDataBucket.java
index e98e26bbd7dc..dc8af04bd423 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/buffer/TestRowDataBucket.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/buffer/TestRowDataBucket.java
@@ -18,30 +18,259 @@
package org.apache.hudi.sink.buffer;
+import org.apache.hudi.configuration.FlinkOptions;
+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.hudi.sink.exception.MemoryPagesExhaustedException;
import org.apache.hudi.sink.utils.BufferUtils;
+import org.apache.hudi.sink.utils.RecordKeySortComparator;
+import org.apache.hudi.sink.utils.RecordKeySortKeyComputer;
import org.apache.hudi.table.action.commit.BucketInfo;
import org.apache.hudi.table.action.commit.BucketType;
import org.apache.hudi.utils.TestData;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.operators.sort.HeapSort;
+import org.apache.flink.runtime.operators.sort.QuickSort;
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.data.binary.BinaryRowData;
+import org.apache.flink.table.runtime.generated.NormalizedKeyComputer;
+import org.apache.flink.table.runtime.generated.RecordComparator;
import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.RowKind;
import org.apache.flink.util.MutableObjectIterator;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import java.util.ArrayList;
+import java.util.Comparator;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/** Tests for {@link RowDataBucket}. */
class TestRowDataBucket {
+ @ParameterizedTest
+ @CsvSource({
+ "true, false, 0", "true, false, 1", "true, false, 12",
+ "true, false, 10000", "false, false, 10000",
+ "true, true, 10000", "false, true, 10000"
+ })
+ void testSortPreservesArrivalOrder(boolean codegen, boolean heapSort, int
count) throws Exception {
+ RowType rowType = (RowType) DataTypes.ROW(
+ DataTypes.FIELD("uuid", DataTypes.STRING()),
+ DataTypes.FIELD("arrival", DataTypes.INT()),
+ DataTypes.FIELD("payload", DataTypes.STRING())).getLogicalType();
+ HeapMemorySegmentPool pool = new HeapMemorySegmentPool(32 * 1024, 4 * 1024
* 1024);
+ int initialFreePages = pool.freePages();
+ NormalizedKeyComputer keyComputer;
+ RecordComparator comparator;
+ if (codegen) {
+ SortOperatorGen generator = new SortOperatorGen(rowType, new String[]
{"uuid"});
+ ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+ keyComputer =
generator.generateNormalizedKeyComputer("StableKeyComputer").newInstance(classLoader);
+ comparator =
generator.generateRecordComparator("StableComparator").newInstance(classLoader);
+ } else {
+ Configuration conf = new Configuration();
+ conf.set(FlinkOptions.RECORD_KEY_FIELD, "uuid");
+ conf.set(FlinkOptions.PARTITION_PATH_FIELD, "");
+ RowDataKeyGen keyGen = RowDataKeyGens.instance(conf, rowType);
+ keyComputer = new RecordKeySortKeyComputer(keyGen, 1);
+ comparator = new RecordKeySortComparator(keyGen);
+ }
+ StableSortBuffer buffer = BufferUtils.createStableSortBuffer(rowType,
pool, keyComputer, comparator);
+ RowDataBucket bucket = new RowDataBucket(
+ "bucket-0", buffer, new BucketInfo(BucketType.INSERT, "file-0",
"partition-0"), 256.0);
+ List<RowData> expected = new ArrayList<>();
+ try {
+ for (int i = 0; i < count; i++) {
+ // Interleaved keys sharing a normalized prefix exercise full-key
comparisons as well as
+ // equal keys. The large input spans several sort-index pages and
exceeds insertion sort.
+ // Variable lengths and records larger than a page exercise byte
offsets and page padding.
+ String payload = repeat('x', i % 997 == 0 ? 40 * 1024 : i % 257);
+ RowData row = GenericRowData.of(StringData.fromString("shared-prefix-"
+ i % 17), i,
+ StringData.fromString(payload));
+ row.setRowKind(i % 3 == 0 ? RowKind.DELETE : RowKind.UPDATE_AFTER);
+ assertTrue(bucket.writeRow(row));
+ expected.add(row);
+ }
+ expected.sort(Comparator.comparing(row -> row.getString(0).toString()));
+ if (heapSort) {
+ // QuickSort's fallback uses the index-addressed compare/swap
overloads.
+ new HeapSort().sort(buffer);
+ } else {
+ bucket.sort();
+ }
+ MutableObjectIterator<BinaryRowData> iterator = bucket.getDataIterator();
+ BinaryRowData reuse = new BinaryRowData(3);
+ boolean[] seen = new boolean[count];
+ for (RowData row : expected) {
+ BinaryRowData actual = iterator.next(reuse);
+ assertNotNull(actual);
+ assertEquals(row.getString(0), actual.getString(0));
+ int arrival = actual.getInt(1);
+ assertFalse(seen[arrival], "each input record must be emitted exactly
once");
+ seen[arrival] = true;
+ assertEquals(StringData.fromString("shared-prefix-" + arrival % 17),
actual.getString(0));
+ assertEquals(row.getInt(1), arrival, "equal keys must retain arrival
order");
+ assertEquals(row.getString(2), actual.getString(2));
+ assertEquals(arrival % 3 == 0 ? RowKind.DELETE : RowKind.UPDATE_AFTER,
actual.getRowKind());
+ }
+ assertNull(iterator.next(reuse));
+ } finally {
+ bucket.dispose();
+ }
+ assertEquals(initialFreePages, pool.freePages());
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "false, false, 1", "false, true, 1", "true, false, 1", "true, true, 1",
+ "false, false, 17", "false, true, 17", "true, false, 17", "true, true,
17"
+ })
+ void testFullyDeterminingKeysPreserveOffsetOrder(boolean heapSort, boolean
descending, int distinctKeys) throws Exception {
+ RowType rowType = (RowType) DataTypes.ROW(
+ DataTypes.FIELD("key", DataTypes.INT()),
+ DataTypes.FIELD("arrival", DataTypes.INT())).getLogicalType();
+ HeapMemorySegmentPool pool = new HeapMemorySegmentPool(128, 1024 * 1024);
+ int initialFreePages = pool.freePages();
+ NormalizedKeyComputer keyComputer = new NormalizedKeyComputer() {
+ @Override
+ public void putKey(RowData record, MemorySegment target, int offset) {
+ target.putInt(offset, record.getInt(0));
+ }
+
+ @Override
+ public int compareKey(MemorySegment left, int leftOffset, MemorySegment
right, int rightOffset) {
+ return Integer.compare(left.getInt(leftOffset),
right.getInt(rightOffset));
+ }
+
+ @Override
+ public void swapKey(MemorySegment left, int leftOffset, MemorySegment
right, int rightOffset) {
+ int key = left.getInt(leftOffset);
+ left.putInt(leftOffset, right.getInt(rightOffset));
+ right.putInt(rightOffset, key);
+ }
+
+ @Override
+ public int getNumKeyBytes() {
+ return Integer.BYTES;
+ }
+
+ @Override
+ public boolean isKeyFullyDetermines() {
+ return true;
+ }
+
+ @Override
+ public boolean invertKey() {
+ return descending;
+ }
+ };
+ StableSortBuffer buffer = BufferUtils.createStableSortBuffer(rowType,
pool, keyComputer, (left, right) -> {
+ throw new AssertionError("Fully determining normalized keys must not
deserialize records for comparison");
+ });
+ List<RowData> expected = new ArrayList<>();
+ try {
+ for (int i = 0; i < 2000; i++) {
+ RowData row = GenericRowData.of(i % distinctKeys - 8, i);
+ assertTrue(buffer.write(row));
+ expected.add(row);
+ }
+ Comparator<RowData> comparator = Comparator.comparingInt(row ->
row.getInt(0));
+ expected.sort(descending ? comparator.reversed() : comparator);
+
+ // Scramble index entries using both swap APIs: byte offsets must
continue to represent
+ // original append order, rather than the positions at which sorting
starts.
+ buffer.swap(0, buffer.size() - 1);
+ buffer.swap(0, buffer.recordSize(), 1, 0);
+ if (heapSort) {
+ new HeapSort().sort(buffer);
+ } else {
+ new QuickSort().sort(buffer);
+ }
+ MutableObjectIterator<BinaryRowData> iterator = buffer.getIterator();
+ BinaryRowData reuse = new BinaryRowData(2);
+ for (RowData row : expected) {
+ BinaryRowData actual = iterator.next(reuse);
+ assertNotNull(actual);
+ assertEquals(row.getInt(0), actual.getInt(0));
+ assertEquals(row.getInt(1), actual.getInt(1));
+ }
+ assertNull(iterator.next(reuse));
+ } finally {
+ buffer.dispose();
+ }
+ assertEquals(initialFreePages, pool.freePages());
+ }
+
+ @Test
+ void testResetPreservesArrivalOrderAcrossBatches() throws Exception {
+ RowType rowType = (RowType) DataTypes.ROW(
+ DataTypes.FIELD("key", DataTypes.STRING()),
+ DataTypes.FIELD("arrival", DataTypes.INT())).getLogicalType();
+ HeapMemorySegmentPool pool = new HeapMemorySegmentPool(32 * 1024, 1024 *
1024);
+ int initialFreePages = pool.freePages();
+ SortOperatorGen generator = new SortOperatorGen(rowType, new String[]
{"key"});
+ ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+ StableSortBuffer buffer = BufferUtils.createStableSortBuffer(rowType, pool,
+
generator.generateNormalizedKeyComputer("ResetKeyComputer").newInstance(classLoader),
+
generator.generateRecordComparator("ResetComparator").newInstance(classLoader));
+ int emptyBufferFreePages = pool.freePages();
+ try {
+ assertEquals((long) initialFreePages * pool.pageSize(),
buffer.getCapacity());
+ for (int batch = 0; batch < 3; batch++) {
+ buffer.reset();
+ assertTrue(buffer.isEmpty());
+ assertEquals(0, buffer.getOccupancy());
+ assertEquals(emptyBufferFreePages, pool.freePages(), "reset must
release pages from the previous batch");
+ List<RowData> expected = new ArrayList<>();
+ for (int i = 0; i < 2000; i++) {
+ RowData row =
GenericRowData.of(StringData.fromString("shared-prefix-" + i % 17), batch *
2000 + i);
+ assertTrue(buffer.write(row));
+ expected.add(row);
+ }
+ assertTrue(buffer.getOccupancy() > 0);
+ assertTrue(pool.freePages() < emptyBufferFreePages);
+ expected.sort(Comparator.comparing(row ->
row.getString(0).toString()));
+ new QuickSort().sort(buffer);
+ MutableObjectIterator<BinaryRowData> iterator = buffer.getIterator();
+ BinaryRowData reuse = new BinaryRowData(2);
+ for (RowData row : expected) {
+ BinaryRowData actual = iterator.next(reuse);
+ assertNotNull(actual);
+ assertEquals(row.getString(0), actual.getString(0));
+ assertEquals(row.getInt(1), actual.getInt(1));
+ }
+ assertNull(iterator.next(reuse));
+ }
+ } finally {
+ buffer.dispose();
+ }
+ assertEquals(initialFreePages, pool.freePages());
+ }
+
+ @Test
+ void testInsufficientPagesDoNotAllocateBuffer() {
+ RowType rowType = (RowType) DataTypes.ROW(DataTypes.FIELD("key",
DataTypes.INT())).getLogicalType();
+ HeapMemorySegmentPool pool = new HeapMemorySegmentPool(32 * 1024, 64 *
1024);
+ int initialFreePages = pool.freePages();
+ assertThrows(MemoryPagesExhaustedException.class, () ->
BufferUtils.createStableSortBuffer(rowType, pool));
+ assertEquals(initialFreePages, pool.freePages());
+ }
+
@Test
void testDivergedBucketCannotBeReusedAndReleasesAllPages() throws Exception {
RowType rowType = (RowType) DataTypes.ROW(
@@ -53,7 +282,7 @@ class TestRowDataBucket {
int initialFreePages = pool.freePages();
RowDataBucket bucket = new RowDataBucket(
"bucket-0",
- BufferUtils.createBuffer(rowType, pool),
+ BufferUtils.createStableSortBuffer(rowType, pool),
new BucketInfo(BucketType.INSERT, "file-0", "partition-0"),
256.0);
@@ -72,7 +301,7 @@ class TestRowDataBucket {
expectedIds.add(id);
}
- assertTrue(writeFailed, "the tiny pool should cause
BinaryInMemorySortBuffer.write to return false");
+ assertTrue(writeFailed, "the tiny pool should cause
StableSortBuffer.write to return false");
assertTrue(bucket.isDiverged());
assertFalse(bucket.isEmpty(), "successful rows written before exhaustion
should remain readable");
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestIndexWriteFunction.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestIndexWriteFunction.java
index b0f17658dc29..2ed12c443d09 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestIndexWriteFunction.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/index/TestIndexWriteFunction.java
@@ -19,11 +19,11 @@
package org.apache.hudi.sink.partitioner.index;
import org.apache.hudi.client.HoodieFlinkWriteClient;
+import org.apache.hudi.sink.buffer.StableSortBuffer;
import org.apache.hudi.sink.utils.BufferUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.memory.MemoryManager;
-import org.apache.flink.table.runtime.operators.sort.BinaryInMemorySortBuffer;
import org.apache.flink.table.runtime.util.LazyMemorySegmentPool;
import org.apache.flink.table.runtime.util.MemorySegmentPool;
import org.junit.jupiter.api.Test;
@@ -69,7 +69,7 @@ class TestIndexWriteFunction {
void testCloseContinuesAfterCleanupFailure(boolean bufferFails) throws
Exception {
HoodieFlinkWriteClient writeClient = mock(HoodieFlinkWriteClient.class);
IndexWriteFunction function = new TestFunction(writeClient);
- BinaryInMemorySortBuffer buffer = mock(BinaryInMemorySortBuffer.class);
+ StableSortBuffer buffer = mock(StableSortBuffer.class);
CloseableMemorySegmentPool pool = mock(CloseableMemorySegmentPool.class);
setField(function, "indexDataBuffer", buffer);
setField(function, "memorySegmentPool", pool);
@@ -99,7 +99,7 @@ class TestIndexWriteFunction {
MemoryManager memoryManager = MemoryManager.create(8L * pageSize,
pageSize);
try {
LazyMemorySegmentPool pool = new LazyMemorySegmentPool(new Object(),
memoryManager, 8);
- BinaryInMemorySortBuffer buffer =
BufferUtils.createBuffer(IndexRowUtils.INDEX_ROW_TYPE, pool);
+ StableSortBuffer buffer =
BufferUtils.createStableSortBuffer(IndexRowUtils.INDEX_ROW_TYPE, pool);
HoodieFlinkWriteClient writeClient = mock(HoodieFlinkWriteClient.class);
IndexWriteFunction function = new TestFunction(writeClient);
setField(function, "indexDataBuffer", buffer);
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestRecordKeySortKeyComputer.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestRecordKeySortKeyComputer.java
index fb0d998448f5..9306295eba9e 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestRecordKeySortKeyComputer.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestRecordKeySortKeyComputer.java
@@ -21,6 +21,7 @@ package org.apache.hudi.sink.utils;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.sink.buffer.HeapMemorySegmentPool;
+import org.apache.hudi.sink.buffer.StableSortBuffer;
import org.apache.hudi.sink.bulk.RowDataKeyGen;
import org.apache.hudi.sink.bulk.RowDataKeyGens;
@@ -33,7 +34,6 @@ 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.data.binary.BinaryRowData;
-import org.apache.flink.table.runtime.operators.sort.BinaryInMemorySortBuffer;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
@@ -212,7 +212,7 @@ class TestRecordKeySortKeyComputer {
}
@Test
- void testBinaryInMemorySortBufferOrdering() throws Exception {
+ void testStableSortBufferOrdering() throws Exception {
RowType longRowType = rowType(new String[] {"key"}, new LogicalType[] {new
BigIntType()});
assertBufferSort(longRowType, "key", Arrays.asList(
GenericRowData.of(2L), GenericRowData.of(10L), GenericRowData.of(-1L),
GenericRowData.of(100L)));
@@ -308,7 +308,7 @@ class TestRecordKeySortKeyComputer {
throws Exception {
RowDataKeyGen keyGen = keyGen(rowType, recordKeyFields);
int recordKeyFieldCount = recordKeyFields.split(",").length;
- BinaryInMemorySortBuffer buffer = BufferUtils.createBuffer(
+ StableSortBuffer buffer = BufferUtils.createStableSortBuffer(
rowType,
new HeapMemorySegmentPool(MemoryManager.DEFAULT_PAGE_SIZE,
BUFFER_SIZE_BYTES),
new RecordKeySortKeyComputer(keyGen, recordKeyFieldCount),
