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 19b7ee7073 [core] Prevent integer overflow in sort buffer size with
isEmpty check (#8209)
19b7ee7073 is described below
commit 19b7ee70738a0fcb9d89bd228e54d2fc7784ac65
Author: Liurnly <[email protected]>
AuthorDate: Fri Jun 12 15:46:18 2026 +0800
[core] Prevent integer overflow in sort buffer size with isEmpty check
(#8209)
---
.../org/apache/paimon/append/cluster/Sorter.java | 2 +-
.../paimon/crosspartition/GlobalIndexAssigner.java | 2 +-
.../apache/paimon/mergetree/MergeTreeWriter.java | 2 +-
.../paimon/mergetree/SortBufferWriteBuffer.java | 5 +
.../org/apache/paimon/mergetree/WriteBuffer.java | 2 +
.../mergetree/localmerge/HashMapLocalMerger.java | 5 +
.../paimon/mergetree/localmerge/LocalMerger.java | 2 +
.../localmerge/SortBufferLocalMerger.java | 5 +
.../paimon/sort/BinaryExternalSortBuffer.java | 15 ++-
.../paimon/sort/BinaryInMemorySortBuffer.java | 3 +-
.../java/org/apache/paimon/sort/SortBuffer.java | 2 +
.../SortBufferWriteBufferOverflowTest.java | 131 +++++++++++++++++++++
.../paimon/sort/BinaryExternalSortBufferTest.java | 29 +++++
.../paimon/flink/sink/LocalMergeOperator.java | 2 +-
.../apache/paimon/flink/sorter/SortOperator.java | 2 +-
15 files changed, 201 insertions(+), 8 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/Sorter.java
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/Sorter.java
index 41786992de..e1cfe9fbe3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/cluster/Sorter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/cluster/Sorter.java
@@ -104,7 +104,7 @@ public abstract class Sorter {
buffer.write(rowWithKey);
}
- if (buffer.size() > 0) {
+ if (!buffer.isEmpty()) {
return buffer.sortedIterator();
} else {
throw new IllegalStateException("numRecords after sorting is 0.");
diff --git
a/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
b/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
index 12f90cd447..cb7d45a648 100644
---
a/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
@@ -207,7 +207,7 @@ public class GlobalIndexAssigner implements Serializable,
Closeable {
throws Exception {
bootstrap = false;
boolean isEmpty = true;
- if (bootstrapKeys.size() > 0) {
+ if (!bootstrapKeys.isEmpty()) {
RocksDBBulkLoader bulkLoader = keyIndex.createBulkLoader();
MutableObjectIterator<BinaryRow> keyIterator =
bootstrapKeys.sortedIterator();
BinaryRow row = new BinaryRow(2);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
index 488b3992dd..beb2651f1f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
@@ -208,7 +208,7 @@ public class MergeTreeWriter implements
RecordWriter<KeyValue>, MemoryOwner {
private void flushWriteBuffer(boolean waitForLatestCompaction, boolean
forcedFullCompaction)
throws Exception {
- if (writeBuffer.size() > 0) {
+ if (!writeBuffer.isEmpty()) {
if (compactManager.shouldWaitForLatestCompaction()) {
waitForLatestCompaction = true;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java
index 17fc5af625..6f6d84c935 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java
@@ -150,6 +150,11 @@ public class SortBufferWriteBuffer implements WriteBuffer {
return buffer.size();
}
+ @Override
+ public boolean isEmpty() {
+ return buffer.isEmpty();
+ }
+
@Override
public long memoryOccupancy() {
return buffer.getOccupancy();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/WriteBuffer.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/WriteBuffer.java
index 9b2749fb7f..ddeb54286f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/WriteBuffer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/WriteBuffer.java
@@ -45,6 +45,8 @@ public interface WriteBuffer {
/** Record size of this table. */
int size();
+ boolean isEmpty();
+
/** Memory occupancy size of this table. */
long memoryOccupancy();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/localmerge/HashMapLocalMerger.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/localmerge/HashMapLocalMerger.java
index 1a395fb36c..2db4576506 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/localmerge/HashMapLocalMerger.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/localmerge/HashMapLocalMerger.java
@@ -118,6 +118,11 @@ public class HashMapLocalMerger implements LocalMerger {
return buffer.getNumElements();
}
+ @Override
+ public boolean isEmpty() {
+ return buffer.getNumElements() == 0;
+ }
+
@Override
public void forEach(Consumer<InternalRow> consumer) throws IOException {
KeyValueIterator<BinaryRow, BinaryRow> iterator =
buffer.getEntryIterator(false);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/localmerge/LocalMerger.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/localmerge/LocalMerger.java
index bec71808a7..8e0e47da80 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/localmerge/LocalMerger.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/localmerge/LocalMerger.java
@@ -32,6 +32,8 @@ public interface LocalMerger {
int size();
+ boolean isEmpty();
+
void forEach(Consumer<InternalRow> consumer) throws IOException;
void clear();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/localmerge/SortBufferLocalMerger.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/localmerge/SortBufferLocalMerger.java
index 198e6c67d3..d35f499c81 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/localmerge/SortBufferLocalMerger.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/localmerge/SortBufferLocalMerger.java
@@ -58,6 +58,11 @@ public class SortBufferLocalMerger implements LocalMerger {
return sortBuffer.size();
}
+ @Override
+ public boolean isEmpty() {
+ return sortBuffer.isEmpty();
+ }
+
@Override
public void forEach(Consumer<InternalRow> consumer) throws IOException {
sortBuffer.forEach(
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 5521295ea3..960965c901 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
@@ -59,7 +59,7 @@ public class BinaryExternalSortBuffer implements SortBuffer {
private final List<ChannelWithMeta> spillChannelIDs;
private final MemorySize maxDiskSize;
- private int numRecords = 0;
+ private long numRecords = 0;
public BinaryExternalSortBuffer(
BinaryRowSerializer serializer,
@@ -122,7 +122,18 @@ public class BinaryExternalSortBuffer implements
SortBuffer {
@Override
public int size() {
- return numRecords;
+ if (numRecords > Integer.MAX_VALUE) {
+ throw new RuntimeException(
+ "numRecords "
+ + numRecords
+ + " exceeds Integer.MAX_VALUE, use isEmpty()
instead of size().");
+ }
+ return (int) numRecords;
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return numRecords == 0;
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/sort/BinaryInMemorySortBuffer.java
b/paimon-core/src/main/java/org/apache/paimon/sort/BinaryInMemorySortBuffer.java
index 2fe3ebdbae..6f0af79a8d 100644
---
a/paimon-core/src/main/java/org/apache/paimon/sort/BinaryInMemorySortBuffer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/sort/BinaryInMemorySortBuffer.java
@@ -148,7 +148,8 @@ public class BinaryInMemorySortBuffer extends
BinaryIndexedSortable implements S
return false;
}
- boolean isEmpty() {
+ @Override
+ public boolean isEmpty() {
return this.numRecords == 0;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/sort/SortBuffer.java
b/paimon-core/src/main/java/org/apache/paimon/sort/SortBuffer.java
index 723ec21c01..127165c173 100644
--- a/paimon-core/src/main/java/org/apache/paimon/sort/SortBuffer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/sort/SortBuffer.java
@@ -29,6 +29,8 @@ public interface SortBuffer {
int size();
+ boolean isEmpty();
+
void clear();
long getOccupancy();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/SortBufferWriteBufferOverflowTest.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/SortBufferWriteBufferOverflowTest.java
new file mode 100644
index 0000000000..f357b9a28c
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/SortBufferWriteBufferOverflowTest.java
@@ -0,0 +1,131 @@
+/*
+ * 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.paimon.mergetree;
+
+import org.apache.paimon.compression.CompressOptions;
+import org.apache.paimon.data.serializer.AbstractRowDataSerializer;
+import org.apache.paimon.data.serializer.BinaryRowSerializer;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.memory.HeapMemorySegmentPool;
+import org.apache.paimon.memory.MemorySegmentPool;
+import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.sort.BinaryExternalSortBuffer;
+import org.apache.paimon.sort.BinaryInMemorySortBuffer;
+import org.apache.paimon.sort.IntNormalizedKeyComputer;
+import org.apache.paimon.sort.IntRecordComparator;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import sun.misc.Unsafe;
+
+import java.lang.reflect.Field;
+import java.nio.file.Path;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Test that {@link SortBufferWriteBuffer} (used by {@link MergeTreeWriter})
handles int overflow in
+ * numRecords correctly by using isEmpty() instead of size().
+ */
+public class SortBufferWriteBufferOverflowTest {
+
+ @TempDir Path tempDir;
+
+ private IOManager ioManager;
+ private MemorySegmentPool memorySegmentPool;
+
+ @BeforeEach
+ public void setUp() {
+ ioManager = IOManager.create(tempDir.toString());
+ memorySegmentPool = new HeapMemorySegmentPool(32 * 1024 * 3L, 32 *
1024);
+ }
+
+ @AfterEach
+ public void tearDown() throws Exception {
+ ioManager.close();
+ }
+
+ private BinaryExternalSortBuffer createSortBuffer() {
+ BinaryRowSerializer serializer = new BinaryRowSerializer(1);
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ BinaryInMemorySortBuffer inMemorySortBuffer =
+ BinaryInMemorySortBuffer.createBuffer(
+ IntNormalizedKeyComputer.INSTANCE,
+ (AbstractRowDataSerializer) serializer,
+ IntRecordComparator.INSTANCE,
+ memorySegmentPool);
+ return new BinaryExternalSortBuffer(
+ serializer,
+ IntRecordComparator.INSTANCE,
+ memorySegmentPool.pageSize(),
+ inMemorySortBuffer,
+ ioManager,
+ 128,
+ CompressOptions.defaultOptions(),
+ MemorySize.MAX_VALUE);
+ }
+
+ private static SortBufferWriteBuffer
createWriteBuffer(BinaryExternalSortBuffer buffer)
+ throws Exception {
+ Field theUnsafeField = Unsafe.class.getDeclaredField("theUnsafe");
+ theUnsafeField.setAccessible(true);
+ Unsafe unsafe = (Unsafe) theUnsafeField.get(null);
+ SortBufferWriteBuffer writeBuffer =
+ (SortBufferWriteBuffer)
unsafe.allocateInstance(SortBufferWriteBuffer.class);
+
+ Field bufferField =
SortBufferWriteBuffer.class.getDeclaredField("buffer");
+ bufferField.setAccessible(true);
+ bufferField.set(writeBuffer, buffer);
+ return writeBuffer;
+ }
+
+ private static void setNumRecords(BinaryExternalSortBuffer buffer, long
numRecords)
+ throws Exception {
+ Field numRecordsField =
BinaryExternalSortBuffer.class.getDeclaredField("numRecords");
+ numRecordsField.setAccessible(true);
+ numRecordsField.setLong(buffer, numRecords);
+ }
+
+ @Test
+ public void testIsEmptyWorksWhenNumRecordsExceedsIntMax() throws Exception
{
+ BinaryExternalSortBuffer buffer = createSortBuffer();
+ SortBufferWriteBuffer writeBuffer = createWriteBuffer(buffer);
+
+ assertThat(writeBuffer.size()).isEqualTo(0);
+ assertThat(writeBuffer.isEmpty()).isTrue();
+
+ setNumRecords(buffer, Integer.MAX_VALUE);
+ assertThat(writeBuffer.size()).isEqualTo(Integer.MAX_VALUE);
+ assertThat(writeBuffer.isEmpty()).isFalse();
+
+ setNumRecords(buffer, (long) Integer.MAX_VALUE + 1);
+
+ assertThat(writeBuffer.isEmpty()).isFalse();
+ assertThatThrownBy(writeBuffer::size)
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("exceeds Integer.MAX_VALUE");
+
+ writeBuffer.clear();
+ assertThat(writeBuffer.isEmpty()).isTrue();
+ assertThat(writeBuffer.size()).isEqualTo(0);
+ }
+}
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 0ff35a0606..8aea0a76ce 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
@@ -37,6 +37,7 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.io.IOException;
+import java.lang.reflect.Field;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
@@ -46,6 +47,7 @@ import java.util.List;
import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link BinaryExternalSortBuffer}. */
public class BinaryExternalSortBufferTest {
@@ -96,6 +98,33 @@ public class BinaryExternalSortBufferTest {
assertThat(files).isEmpty();
}
+ private static void setNumRecords(BinaryExternalSortBuffer sorter, long
numRecords)
+ throws Exception {
+ Field numRecordsField =
BinaryExternalSortBuffer.class.getDeclaredField("numRecords");
+ numRecordsField.setAccessible(true);
+ numRecordsField.setLong(sorter, numRecords);
+ }
+
+ @Test
+ public void testSizeBoundary() throws Exception {
+ BinaryExternalSortBuffer sorter = createBuffer();
+
+ assertThat(sorter.size()).isEqualTo(0);
+ assertThat(sorter.isEmpty()).isTrue();
+
+ setNumRecords(sorter, Integer.MAX_VALUE);
+ assertThat(sorter.size()).isEqualTo(Integer.MAX_VALUE);
+ assertThat(sorter.isEmpty()).isFalse();
+
+ setNumRecords(sorter, (long) Integer.MAX_VALUE + 1);
+ assertThat(sorter.isEmpty()).isFalse();
+ assertThatThrownBy(sorter::size)
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("exceeds Integer.MAX_VALUE");
+
+ sorter.clear();
+ }
+
@Test
public void testSortNoSpill() throws Exception {
int size = 1_000_000;
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java
index 401ca43682..4852171a64 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java
@@ -201,7 +201,7 @@ public class LocalMergeOperator extends
AbstractStreamOperator<InternalRow>
}
private void flushBuffer() throws Exception {
- if (merger.size() == 0) {
+ if (merger.isEmpty()) {
return;
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java
index dfa1e432a1..01c349ac83 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java
@@ -107,7 +107,7 @@ public class SortOperator extends
TableStreamOperator<InternalRow>
@Override
public void endInput() throws Exception {
- if (buffer.size() > 0) {
+ if (!buffer.isEmpty()) {
MutableObjectIterator<BinaryRow> iterator =
buffer.sortedIterator();
BinaryRow binaryRow = new BinaryRow(arity);
while ((binaryRow = iterator.next(binaryRow)) != null) {
