This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 6deb55a2a9 [core] Fix cascading re-merge in BinaryExternalSortBuffer
spill merge (#9303)
6deb55a2a9 is described below
commit 6deb55a2a98e47d35fbe26980432d1929770b6c8
Author: zhoulii <[email protected]>
AuthorDate: Thu Aug 20 09:20:21 2026 +0800
[core] Fix cascading re-merge in BinaryExternalSortBuffer spill merge
(#9303)
---
.../paimon/sort/BinaryExternalSortBuffer.java | 19 ++++++---
.../paimon/sort/BinaryExternalSortBufferTest.java | 45 ++++++++++++++++++++++
2 files changed, 58 insertions(+), 6 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/sort/BinaryExternalSortBuffer.java
b/paimon-core/src/main/java/org/apache/paimon/sort/BinaryExternalSortBuffer.java
index 960965c901..8319f458bf 100644
---
a/paimon-core/src/main/java/org/apache/paimon/sort/BinaryExternalSortBuffer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/sort/BinaryExternalSortBuffer.java
@@ -196,12 +196,6 @@ public class BinaryExternalSortBuffer implements
SortBuffer {
+ "or increase the table option
'write-buffer-size'.");
} else {
spill();
-
- if (spillChannelIDs.size() >= maxNumFileHandles) {
- List<ChannelWithMeta> merged =
merger.mergeChannelList(spillChannelIDs);
- spillChannelIDs.clear();
- spillChannelIDs.addAll(merged);
- }
}
}
}
@@ -217,6 +211,19 @@ public class BinaryExternalSortBuffer implements
SortBuffer {
private MutableObjectIterator<BinaryRow> spilledIterator() throws
IOException {
spill();
+ // Merge the spilled sorted runs until the number of file handles fits
within the
+ // fan-in limit. This is performed once here, after spilling is
finished, instead of
+ // incrementally during write(). Doing it during write() would re-add
the merged output
+ // to the spill list and re-merge it together with subsequently
spilled files, causing
+ // cascading re-merge and degrading the merge IO from O(N*log N) to
O(N^2).
+ // This mirrors Flink's SpillingThread#mergeOnDisk: while (channels >
maxFanIn)
+ // channels = mergeChannelList(channels).
+ while (spillChannelIDs.size() > maxNumFileHandles) {
+ List<ChannelWithMeta> merged =
merger.mergeChannelList(spillChannelIDs);
+ spillChannelIDs.clear();
+ spillChannelIDs.addAll(merged);
+ }
+
List<FileIOChannel> openChannels = new ArrayList<>();
BinaryMergeIterator<BinaryRow> iterator =
merger.getMergingIterator(spillChannelIDs, openChannels);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/sort/BinaryExternalSortBufferTest.java
b/paimon-core/src/test/java/org/apache/paimon/sort/BinaryExternalSortBufferTest.java
index 8aea0a76ce..a23f5104cb 100644
---
a/paimon-core/src/test/java/org/apache/paimon/sort/BinaryExternalSortBufferTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/sort/BinaryExternalSortBufferTest.java
@@ -224,6 +224,51 @@ public class BinaryExternalSortBufferTest {
innerTestSpilling(buffer);
}
+ @Test
+ public void testNoCascadingMergeAtFinalIterator() throws Exception {
+ // Use a tiny memory so that writing a moderate number of records
produces many spill
+ // files, and a very small fan-in (maxNumFileHandles = 2) so that a
correct final merge
+ // must invoke mergeChannelList more than once.
+ final int maxNumFileHandles = 2;
+ initMemorySegmentPool(128 * 1024); // 4 pages of 32 KB
+
+ BinaryExternalSortBuffer sorter = createBuffer(maxNumFileHandles);
+
+ int size = 50_000;
+ MockBinaryRowReader reader = new MockBinaryRowReader(size);
+ sorter.write(reader);
+ assertThat(sorter.size()).isEqualTo(size);
+
+ // No merge should happen during write(): the number of spilled files
must exceed
+ // maxNumFileHandles^2. The previous (cascading) implementation merged
during write()
+ // and re-added the merged output to the spill list, which would leave
at most
+ // maxNumFileHandles files here and cause O(N^2) re-merge.
+ int spilledFilesAfterWrite = spillChannelCount(sorter);
+ assertThat(spilledFilesAfterWrite).isGreaterThan(maxNumFileHandles *
maxNumFileHandles);
+
+ MutableObjectIterator<BinaryRow> iterator = sorter.sortedIterator();
+
+ // The final merge must bring the number of opened files within the
fan-in limit.
+ // When spilledFilesAfterWrite > maxNumFileHandles^2 a single
mergeChannelList call is
+ // not enough, so this also guards against using a one-shot `if`
instead of `while`.
+ int finalFileHandles = spillChannelCount(sorter);
+ assertThat(finalFileHandles).isLessThanOrEqualTo(maxNumFileHandles);
+
+ BinaryRow next = serializer.createInstance();
+ for (int i = 0; i < size; i++) {
+ next = iterator.next(next);
+ assertThat(next.getInt(0)).isEqualTo(i);
+ assertThat(next.getString(1).toString()).isEqualTo(getString(i));
+ }
+ sorter.clear();
+ }
+
+ private static int spillChannelCount(BinaryExternalSortBuffer sorter)
throws Exception {
+ Field field =
BinaryExternalSortBuffer.class.getDeclaredField("spillChannelIDs");
+ field.setAccessible(true);
+ return ((List<?>) field.get(sorter)).size();
+ }
+
private void innerTestSpilling(BinaryExternalSortBuffer sorter) throws
Exception {
int size = 2000_000;
