This is an automated email from the ASF dual-hosted git repository.
ferenc-csaky pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 371ba17ab9f [FLINK-40959][table-runtime] Fix `BinaryExternalSorter`
emitting unsorted rows
371ba17ab9f is described below
commit 371ba17ab9fde660097fafbcec4bae3dbc6a7425
Author: Mate Czagany <[email protected]>
AuthorDate: Wed Oct 14 14:01:03 2026 +0200
[FLINK-40959][table-runtime] Fix `BinaryExternalSorter` emitting unsorted
rows
Generated-by: Claude Code (Opus 5.5)
---
.../operators/join/SortMergeJoinFunction.java | 4 +-
.../operators/sort/BinaryExternalSorter.java | 22 +++++---
.../table/runtime/operators/sort/SortOperator.java | 5 +-
.../operators/sort/BinaryExternalSorterTest.java | 62 ++++++++++++++++------
4 files changed, 66 insertions(+), 27 deletions(-)
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/SortMergeJoinFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/SortMergeJoinFunction.java
index 652939efea7..10cde411a44 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/SortMergeJoinFunction.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/SortMergeJoinFunction.java
@@ -181,7 +181,7 @@ public class SortMergeJoinFunction implements Serializable {
inputSerializer1,
serializer1,
computer1.newInstance(cl),
- comparator1.newInstance(cl),
+ () -> comparator1.newInstance(cl),
maxNumFileHandles,
compressionEnabled,
compressionBlockSize,
@@ -198,7 +198,7 @@ public class SortMergeJoinFunction implements Serializable {
inputSerializer2,
serializer2,
computer2.newInstance(cl),
- comparator2.newInstance(cl),
+ () -> comparator2.newInstance(cl),
maxNumFileHandles,
compressionEnabled,
compressionBlockSize,
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sort/BinaryExternalSorter.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sort/BinaryExternalSorter.java
index c94969a240d..c69518a07ae 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sort/BinaryExternalSorter.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sort/BinaryExternalSorter.java
@@ -54,6 +54,7 @@ import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -176,7 +177,7 @@ public class BinaryExternalSorter implements
Sorter<BinaryRowData> {
AbstractRowDataSerializer<RowData> inputSerializer,
BinaryRowDataSerializer serializer,
NormalizedKeyComputer normalizedKeyComputer,
- RecordComparator comparator,
+ Supplier<RecordComparator> comparatorFactory,
int maxNumFileHandles,
boolean compressionEnabled,
int compressionBlockSize,
@@ -189,7 +190,7 @@ public class BinaryExternalSorter implements
Sorter<BinaryRowData> {
inputSerializer,
serializer,
normalizedKeyComputer,
- comparator,
+ comparatorFactory,
maxNumFileHandles,
compressionEnabled,
compressionBlockSize,
@@ -205,7 +206,7 @@ public class BinaryExternalSorter implements
Sorter<BinaryRowData> {
AbstractRowDataSerializer<RowData> inputSerializer,
BinaryRowDataSerializer serializer,
NormalizedKeyComputer normalizedKeyComputer,
- RecordComparator comparator,
+ Supplier<RecordComparator> comparatorFactory,
int maxNumFileHandles,
boolean compressionEnabled,
int compressionBlockSize,
@@ -224,6 +225,11 @@ public class BinaryExternalSorter implements
Sorter<BinaryRowData> {
checkNotNull(ioManager);
checkNotNull(normalizedKeyComputer);
checkNotNull(memoryManager);
+
+ // Generated comparators may keep per-call state in member fields
(e.g. after code
+ // splitting), so the sorting thread and the merging path must not
share an instance.
+ final RecordComparator sortComparator = comparatorFactory.get();
+ final RecordComparator mergeComparator = comparatorFactory.get();
this.serializer = (BinaryRowDataSerializer) serializer.duplicate();
this.memorySegmentSize = memoryManager.getPageSize();
@@ -275,7 +281,11 @@ public class BinaryExternalSorter implements
Sorter<BinaryRowData> {
this.sortReadMemory.add(pool);
final BinaryInMemorySortBuffer buffer =
BinaryInMemorySortBuffer.createBuffer(
- normalizedKeyComputer, inputSerializer,
serializer, comparator, pool);
+ normalizedKeyComputer,
+ inputSerializer,
+ serializer,
+ sortComparator,
+ pool);
// add to empty queue
CircularElement element = new CircularElement(i, buffer);
@@ -314,7 +324,7 @@ public class BinaryExternalSorter implements
Sorter<BinaryRowData> {
maxNumFileHandles,
channelManager,
(BinaryRowDataSerializer) serializer.duplicate(),
- comparator,
+ mergeComparator,
compressionEnabled,
compressionCodecFactory,
compressionBlockSize);
@@ -329,7 +339,7 @@ public class BinaryExternalSorter implements
Sorter<BinaryRowData> {
circularQueues,
ioManager,
(BinaryRowDataSerializer) serializer.duplicate(),
- comparator);
+ mergeComparator);
// start the thread that handles merging from second storage
this.mergeThread =
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sort/SortOperator.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sort/SortOperator.java
index 927c5d97e15..8e455647d6f 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sort/SortOperator.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sort/SortOperator.java
@@ -28,7 +28,6 @@ import org.apache.flink.table.data.binary.BinaryRowData;
import org.apache.flink.table.runtime.generated.GeneratedNormalizedKeyComputer;
import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
import org.apache.flink.table.runtime.generated.NormalizedKeyComputer;
-import org.apache.flink.table.runtime.generated.RecordComparator;
import org.apache.flink.table.runtime.operators.TableStreamOperator;
import org.apache.flink.table.runtime.typeutils.AbstractRowDataSerializer;
import org.apache.flink.table.runtime.typeutils.BinaryRowDataSerializer;
@@ -84,7 +83,7 @@ public class SortOperator extends TableStreamOperator<RowData>
this.binarySerializer = new
BinaryRowDataSerializer(inputSerializer.getArity());
NormalizedKeyComputer computer = gComputer.newInstance(cl);
- RecordComparator comparator = gComparator.newInstance(cl);
+ GeneratedRecordComparator genComparator = gComparator;
gComputer = null;
gComparator = null;
@@ -98,7 +97,7 @@ public class SortOperator extends TableStreamOperator<RowData>
inputSerializer,
binarySerializer,
computer,
- comparator,
+ () -> genComparator.newInstance(cl),
maxNumFileHandles,
compressionEnabled,
compressionBlockSize,
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/sort/BinaryExternalSorterTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/sort/BinaryExternalSorterTest.java
index 4dff42e164b..3a92f0f0142 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/sort/BinaryExternalSorterTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/sort/BinaryExternalSorterTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.runtime.operators.sort;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
import org.apache.flink.runtime.memory.MemoryManager;
@@ -48,6 +49,8 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import static org.assertj.core.api.Assertions.assertThat;
@@ -68,9 +71,7 @@ class BinaryExternalSorterTest {
if (!spillCompress) {
conf.set(ExecutionConfigOptions.TABLE_EXEC_SPILL_COMPRESSION_ENABLED, false);
}
- if (asyncMerge) {
-
conf.set(ExecutionConfigOptions.TABLE_EXEC_SORT_ASYNC_MERGE_ENABLED, true);
- }
+ conf.set(ExecutionConfigOptions.TABLE_EXEC_SORT_ASYNC_MERGE_ENABLED,
asyncMerge);
}
@Parameters(name = "spillCompress-{0} asyncMerge-{1}")
@@ -134,7 +135,7 @@ class BinaryExternalSorterTest {
(AbstractRowDataSerializer) serializer,
serializer,
IntNormalizedKeyComputer.INSTANCE,
- IntRecordComparator.INSTANCE,
+ () -> IntRecordComparator.INSTANCE,
conf.get(ExecutionConfigOptions.TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLES),
conf.get(ExecutionConfigOptions.TABLE_EXEC_SPILL_COMPRESSION_ENABLED),
(int)
@@ -181,7 +182,7 @@ class BinaryExternalSorterTest {
(AbstractRowDataSerializer) serializer,
serializer,
IntNormalizedKeyComputer.INSTANCE,
- IntRecordComparator.INSTANCE,
+ () -> IntRecordComparator.INSTANCE,
conf.get(ExecutionConfigOptions.TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLES),
conf.get(ExecutionConfigOptions.TABLE_EXEC_SPILL_COMPRESSION_ENABLED),
(int)
@@ -229,7 +230,7 @@ class BinaryExternalSorterTest {
return false;
}
},
- IntRecordComparator.INSTANCE,
+ () -> IntRecordComparator.INSTANCE,
conf.get(ExecutionConfigOptions.TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLES),
conf.get(ExecutionConfigOptions.TABLE_EXEC_SPILL_COMPRESSION_ENABLED),
(int)
@@ -278,7 +279,7 @@ class BinaryExternalSorterTest {
(AbstractRowDataSerializer) serializer,
serializer,
IntNormalizedKeyComputer.INSTANCE,
- IntRecordComparator.INSTANCE,
+ () -> IntRecordComparator.INSTANCE,
conf.get(ExecutionConfigOptions.TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLES),
conf.get(ExecutionConfigOptions.TABLE_EXEC_SPILL_COMPRESSION_ENABLED),
(int)
@@ -328,12 +329,13 @@ class BinaryExternalSorterTest {
return true;
}
},
- new IntRecordComparator() {
- @Override
- public int compare(RowData o1, RowData o2) {
- return -super.compare(o1, o2);
- }
- },
+ () ->
+ new IntRecordComparator() {
+ @Override
+ public int compare(RowData o1, RowData o2)
{
+ return -super.compare(o1, o2);
+ }
+ },
conf.get(ExecutionConfigOptions.TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLES),
conf.get(ExecutionConfigOptions.TABLE_EXEC_SPILL_COMPRESSION_ENABLED),
(int)
@@ -376,6 +378,7 @@ class BinaryExternalSorterTest {
long minMemorySize =
memoryManager.computeNumberOfPages(0.01) *
MemoryManager.DEFAULT_PAGE_SIZE;
conf.set(ExecutionConfigOptions.TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLES,
8);
+ List<Set<String>> comparatorThreads = new ArrayList<>();
BinaryExternalSorter sorter =
new BinaryExternalSorter(
@@ -385,8 +388,30 @@ class BinaryExternalSorterTest {
this.ioManager,
(AbstractRowDataSerializer) serializer,
serializer,
- IntNormalizedKeyComputer.INSTANCE,
- IntRecordComparator.INSTANCE,
+ new IntNormalizedKeyComputer() {
+ @Override
+ public int compareKey(
+ MemorySegment segI,
+ int offsetI,
+ MemorySegment segJ,
+ int offsetJ) {
+ // make the sorting thread call the record
comparator
+ return 0;
+ }
+
+ @Override
+ public boolean isKeyFullyDetermines() {
+ return false;
+ }
+ },
+ () -> {
+ Set<String> threads =
ConcurrentHashMap.newKeySet();
+ comparatorThreads.add(threads);
+ return (o1, o2) -> {
+ threads.add(Thread.currentThread().getName());
+ return
IntRecordComparator.INSTANCE.compare(o1, o2);
+ };
+ },
conf.get(ExecutionConfigOptions.TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLES),
conf.get(ExecutionConfigOptions.TABLE_EXEC_SPILL_COMPRESSION_ENABLED),
(int)
@@ -409,6 +434,11 @@ class BinaryExternalSorterTest {
}
sorter.close();
+
+ // generated comparators can be stateful, so the two threads must not
share one
+ assertThat(comparatorThreads)
+ .anyMatch(threads -> threads.contains("SortMerger sorting
thread"))
+ .anyMatch(threads -> threads.contains("SortMerger merging
thread"));
}
@TestTemplate
@@ -431,7 +461,7 @@ class BinaryExternalSorterTest {
(AbstractRowDataSerializer) serializer,
serializer,
IntNormalizedKeyComputer.INSTANCE,
- IntRecordComparator.INSTANCE,
+ () -> IntRecordComparator.INSTANCE,
conf.get(ExecutionConfigOptions.TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLES),
conf.get(ExecutionConfigOptions.TABLE_EXEC_SPILL_COMPRESSION_ENABLED),
(int)
