From 138733cc68f322296ab0e84c2e865f9d5730d1fc Mon Sep 17 00:00:00 2001 From: huangxiaoping <1754789345@qq.com> Date: Wed, 15 Jul 2026 19:10:10 +0800 Subject: [PATCH 1/4] [core] Improve vector capacity growth for huge batches --- .../writable/AbstractWritableVector.java | 25 +++- .../columnar/heap/HeapVectorCapacityTest.java | 42 ++++++ .../writable/AbstractWritableVectorTest.java | 124 ++++++++++++++++++ 3 files changed, 187 insertions(+), 4 deletions(-) create mode 100644 paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java create mode 100644 paimon-common/src/test/java/org/apache/paimon/data/columnar/writable/AbstractWritableVectorTest.java diff --git a/paimon-common/src/main/java/org/apache/paimon/data/columnar/writable/AbstractWritableVector.java b/paimon-common/src/main/java/org/apache/paimon/data/columnar/writable/AbstractWritableVector.java index fd4981b403d0..810c95509b26 100644 --- a/paimon-common/src/main/java/org/apache/paimon/data/columnar/writable/AbstractWritableVector.java +++ b/paimon-common/src/main/java/org/apache/paimon/data/columnar/writable/AbstractWritableVector.java @@ -31,6 +31,10 @@ public abstract class AbstractWritableVector implements WritableColumnVector, Se private static final long serialVersionUID = 1L; + static final int MAX_ROUNDED_ARRAY_LENGTH = Integer.MAX_VALUE - 15; + static final int DEFAULT_HUGE_VECTOR_THRESHOLD = 1 << 20; + static final double DEFAULT_HUGE_VECTOR_RESERVE_RATIO = 1.2; + // If the whole column vector has no nulls, this is true, otherwise false. protected boolean noNulls = true; @@ -41,6 +45,8 @@ public abstract class AbstractWritableVector implements WritableColumnVector, Se protected int capacity; + private final int initialCapacity; + /** * The Dictionary for this column. If it's not null, will be used to decode the value in get(). */ @@ -48,6 +54,7 @@ public abstract class AbstractWritableVector implements WritableColumnVector, Se public AbstractWritableVector(int capacity) { this.capacity = capacity; + this.initialCapacity = capacity; } /** Update the dictionary. */ @@ -91,8 +98,10 @@ public int getCapacity() { @Override public void reset() { - // To reduce copy, Ww don't result the capacity to initial capacity here. Which means the - // capacity will be the same as expand. + // Keep normal expanded capacity for reuse, but do not hold very large arrays forever. + if (capacity > DEFAULT_HUGE_VECTOR_THRESHOLD) { + capacity = initialCapacity; + } noNulls = true; isAllNull = false; elementsAppended = 0; @@ -103,7 +112,7 @@ public void reserve(int requiredCapacity) { if (requiredCapacity < 0) { throw new IllegalArgumentException("Invalid capacity: " + requiredCapacity); } else if (requiredCapacity > capacity) { - int newCapacity = (int) Math.min(Integer.MAX_VALUE, requiredCapacity * 2L); + int newCapacity = calculateNewCapacity(requiredCapacity); if (requiredCapacity <= newCapacity) { try { reserveInternal(newCapacity); @@ -113,11 +122,19 @@ public void reserve(int requiredCapacity) { } } else { throw new UnsupportedOperationException( - "Cannot allocate :" + newCapacity + " elements"); + "Cannot allocate " + requiredCapacity + " elements"); } capacity = newCapacity; } } + static int calculateNewCapacity(int requiredCapacity) { + long newCapacity = + requiredCapacity < DEFAULT_HUGE_VECTOR_THRESHOLD + ? requiredCapacity * 2L + : (long) (requiredCapacity * DEFAULT_HUGE_VECTOR_RESERVE_RATIO); + return (int) Math.min(MAX_ROUNDED_ARRAY_LENGTH, newCapacity); + } + protected abstract void reserveInternal(int newCapacity); } diff --git a/paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java b/paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java new file mode 100644 index 000000000000..57667415f976 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java @@ -0,0 +1,42 @@ +/* + * 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.data.columnar.heap; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for heap vector capacity reuse. */ +class HeapVectorCapacityTest { + + @Test + void testHugeVectorCapacityIsReleasedOnReset() { + int threshold = 1 << 20; + HeapIntVector vector = new HeapIntVector(4); + + vector.reserve(threshold); + assertThat(vector.getCapacity()).isGreaterThan(threshold); + assertThat(vector.vector.length).isEqualTo(vector.getCapacity()); + + vector.reset(); + + assertThat(vector.getCapacity()).isEqualTo(4); + assertThat(vector.vector).hasSize(4); + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/data/columnar/writable/AbstractWritableVectorTest.java b/paimon-common/src/test/java/org/apache/paimon/data/columnar/writable/AbstractWritableVectorTest.java new file mode 100644 index 000000000000..94b62426a65e --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/data/columnar/writable/AbstractWritableVectorTest.java @@ -0,0 +1,124 @@ +/* + * 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.data.columnar.writable; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link AbstractWritableVector} capacity growth. */ +class AbstractWritableVectorTest { + + @Test + void testSmallVectorDoublesRequiredCapacity() { + TestVector vector = new TestVector(4); + + vector.reserve(5); + + assertThat(vector.getCapacity()).isEqualTo(10); + assertThat(vector.reservedCapacity).isEqualTo(10); + } + + @Test + void testHugeVectorUsesConservativeGrowth() { + int threshold = AbstractWritableVector.DEFAULT_HUGE_VECTOR_THRESHOLD; + TestVector vector = new TestVector(threshold - 1); + + vector.reserve(threshold); + + assertThat(vector.getCapacity()) + .isEqualTo( + (int) + (threshold + * AbstractWritableVector + .DEFAULT_HUGE_VECTOR_RESERVE_RATIO)); + } + + @Test + void testHugeVectorCapacityResetsToInitialCapacity() { + int threshold = AbstractWritableVector.DEFAULT_HUGE_VECTOR_THRESHOLD; + TestVector vector = new TestVector(4); + + vector.reserve(threshold); + vector.addElementsAppended(1); + vector.reset(); + + assertThat(vector.getCapacity()).isEqualTo(4); + assertThat(vector.getElementsAppended()).isZero(); + } + + @Test + void testNormalExpandedCapacityIsRetainedAfterReset() { + TestVector vector = new TestVector(4); + + vector.reserve(5); + vector.reset(); + + assertThat(vector.getCapacity()).isEqualTo(10); + } + + @Test + void testReserveRejectsCapacityBeyondMaxRoundedArrayLength() { + TestVector vector = new TestVector(4); + + assertThatThrownBy(() -> vector.reserve(Integer.MAX_VALUE)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Cannot allocate " + Integer.MAX_VALUE + " elements"); + } + + private static class TestVector extends AbstractWritableVector { + + private int reservedCapacity; + + private TestVector(int capacity) { + super(capacity); + this.reservedCapacity = capacity; + } + + @Override + public void setNullAt(int rowId) {} + + @Override + public void setNulls(int rowId, int count) {} + + @Override + public void fillWithNulls() {} + + @Override + public WritableIntVector reserveDictionaryIds(int capacity) { + return null; + } + + @Override + public WritableIntVector getDictionaryIds() { + return null; + } + + @Override + public boolean isNullAt(int i) { + return false; + } + + @Override + protected void reserveInternal(int newCapacity) { + this.reservedCapacity = newCapacity; + } + } +} From c635588122c756ee6e0bcb83d4a28287686130da Mon Sep 17 00:00:00 2001 From: huangxiaoping <1754789345@qq.com> Date: Tue, 21 Jul 2026 23:59:12 +0800 Subject: [PATCH 2/4] Add shrink ratio hysteresis to avoid allocation thrashing --- .../writable/AbstractWritableVector.java | 10 ++++++++-- .../columnar/heap/HeapVectorCapacityTest.java | 17 +++++++++++++++++ .../writable/AbstractWritableVectorTest.java | 18 +++++++++++++++++- 3 files changed, 42 insertions(+), 3 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/data/columnar/writable/AbstractWritableVector.java b/paimon-common/src/main/java/org/apache/paimon/data/columnar/writable/AbstractWritableVector.java index 810c95509b26..59e5ce91d1fb 100644 --- a/paimon-common/src/main/java/org/apache/paimon/data/columnar/writable/AbstractWritableVector.java +++ b/paimon-common/src/main/java/org/apache/paimon/data/columnar/writable/AbstractWritableVector.java @@ -34,6 +34,7 @@ public abstract class AbstractWritableVector implements WritableColumnVector, Se static final int MAX_ROUNDED_ARRAY_LENGTH = Integer.MAX_VALUE - 15; static final int DEFAULT_HUGE_VECTOR_THRESHOLD = 1 << 20; static final double DEFAULT_HUGE_VECTOR_RESERVE_RATIO = 1.2; + static final int DEFAULT_HUGE_VECTOR_SHRINK_RATIO = 4; // If the whole column vector has no nulls, this is true, otherwise false. protected boolean noNulls = true; @@ -98,8 +99,8 @@ public int getCapacity() { @Override public void reset() { - // Keep normal expanded capacity for reuse, but do not hold very large arrays forever. - if (capacity > DEFAULT_HUGE_VECTOR_THRESHOLD) { + // Keep normal expanded capacity for reuse, but release buffers inflated by one-off spikes. + if (shouldShrinkCapacity()) { capacity = initialCapacity; } noNulls = true; @@ -128,6 +129,11 @@ public void reserve(int requiredCapacity) { } } + private boolean shouldShrinkCapacity() { + return capacity > DEFAULT_HUGE_VECTOR_THRESHOLD + && capacity > (long) elementsAppended * DEFAULT_HUGE_VECTOR_SHRINK_RATIO; + } + static int calculateNewCapacity(int requiredCapacity) { long newCapacity = requiredCapacity < DEFAULT_HUGE_VECTOR_THRESHOLD diff --git a/paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java b/paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java index 57667415f976..08c737174ad4 100644 --- a/paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java @@ -18,6 +18,8 @@ package org.apache.paimon.data.columnar.heap; +import org.apache.paimon.data.RowHelper; + import org.junit.jupiter.api.Test; import static org.assertj.core.api.Assertions.assertThat; @@ -34,9 +36,24 @@ void testHugeVectorCapacityIsReleasedOnReset() { assertThat(vector.getCapacity()).isGreaterThan(threshold); assertThat(vector.vector.length).isEqualTo(vector.getCapacity()); + vector.addElementsAppended(1); vector.reset(); assertThat(vector.getCapacity()).isEqualTo(4); assertThat(vector.vector).hasSize(4); } + + @Test + void testHugeVectorCapacityIsRetainedWhenUsageIsNotTooSmall() { + int threshold = 1 << 20; + HeapIntVector vector = new HeapIntVector(4); + + vector.reserve(threshold); + int expandedCapacity = vector.getCapacity(); + vector.addElementsAppended(expandedCapacity / RowHelper.SHRINK_RATIO); + vector.reset(); + + assertThat(vector.getCapacity()).isEqualTo(expandedCapacity); + assertThat(vector.vector).hasSize(expandedCapacity); + } } diff --git a/paimon-common/src/test/java/org/apache/paimon/data/columnar/writable/AbstractWritableVectorTest.java b/paimon-common/src/test/java/org/apache/paimon/data/columnar/writable/AbstractWritableVectorTest.java index 94b62426a65e..722eae38a5ca 100644 --- a/paimon-common/src/test/java/org/apache/paimon/data/columnar/writable/AbstractWritableVectorTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/data/columnar/writable/AbstractWritableVectorTest.java @@ -52,7 +52,7 @@ void testHugeVectorUsesConservativeGrowth() { } @Test - void testHugeVectorCapacityResetsToInitialCapacity() { + void testHugeVectorCapacityResetsToInitialCapacityWhenUsageIsSmall() { int threshold = AbstractWritableVector.DEFAULT_HUGE_VECTOR_THRESHOLD; TestVector vector = new TestVector(4); @@ -64,6 +64,22 @@ void testHugeVectorCapacityResetsToInitialCapacity() { assertThat(vector.getElementsAppended()).isZero(); } + @Test + void testHugeVectorCapacityIsRetainedWhenUsageIsNotTooSmall() { + int threshold = AbstractWritableVector.DEFAULT_HUGE_VECTOR_THRESHOLD; + TestVector vector = new TestVector(4); + + vector.reserve(threshold); + int expandedCapacity = vector.getCapacity(); + int retainedUsage = + expandedCapacity / AbstractWritableVector.DEFAULT_HUGE_VECTOR_SHRINK_RATIO; + vector.addElementsAppended(retainedUsage); + vector.reset(); + + assertThat(vector.getCapacity()).isEqualTo(expandedCapacity); + assertThat(vector.getElementsAppended()).isZero(); + } + @Test void testNormalExpandedCapacityIsRetainedAfterReset() { TestVector vector = new TestVector(4); From e6d5b6e8f3da53eb42ee226aa999abcb595a1a1d Mon Sep 17 00:00:00 2001 From: huangxiaoping <1754789345@qq.com> Date: Wed, 22 Jul 2026 00:07:28 +0800 Subject: [PATCH 3/4] Fix integer truncation in shrink ratio tests --- .../paimon/data/columnar/heap/HeapVectorCapacityTest.java | 5 ++--- .../data/columnar/writable/AbstractWritableVectorTest.java | 4 +++- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java b/paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java index 08c737174ad4..aafac3113506 100644 --- a/paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java @@ -18,8 +18,6 @@ package org.apache.paimon.data.columnar.heap; -import org.apache.paimon.data.RowHelper; - import org.junit.jupiter.api.Test; import static org.assertj.core.api.Assertions.assertThat; @@ -50,7 +48,8 @@ void testHugeVectorCapacityIsRetainedWhenUsageIsNotTooSmall() { vector.reserve(threshold); int expandedCapacity = vector.getCapacity(); - vector.addElementsAppended(expandedCapacity / RowHelper.SHRINK_RATIO); + // Use ceiling division (shrink ratio = 4) so capacity is not strictly greater than usage * 4. + vector.addElementsAppended((expandedCapacity + 3) / 4); vector.reset(); assertThat(vector.getCapacity()).isEqualTo(expandedCapacity); diff --git a/paimon-common/src/test/java/org/apache/paimon/data/columnar/writable/AbstractWritableVectorTest.java b/paimon-common/src/test/java/org/apache/paimon/data/columnar/writable/AbstractWritableVectorTest.java index 722eae38a5ca..9e63d4933f84 100644 --- a/paimon-common/src/test/java/org/apache/paimon/data/columnar/writable/AbstractWritableVectorTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/data/columnar/writable/AbstractWritableVectorTest.java @@ -71,8 +71,10 @@ void testHugeVectorCapacityIsRetainedWhenUsageIsNotTooSmall() { vector.reserve(threshold); int expandedCapacity = vector.getCapacity(); + // Use ceiling division so capacity is not strictly greater than usage * 4. int retainedUsage = - expandedCapacity / AbstractWritableVector.DEFAULT_HUGE_VECTOR_SHRINK_RATIO; + (expandedCapacity + AbstractWritableVector.DEFAULT_HUGE_VECTOR_SHRINK_RATIO - 1) + / AbstractWritableVector.DEFAULT_HUGE_VECTOR_SHRINK_RATIO; vector.addElementsAppended(retainedUsage); vector.reset(); From 09dd2bfeca2682b97bb59c1769e405b788932be7 Mon Sep 17 00:00:00 2001 From: huangxiaoping <1754789345@qq.com> Date: Wed, 22 Jul 2026 13:02:30 +0800 Subject: [PATCH 4/4] [spark] Fix test case --- .../paimon/data/columnar/heap/HeapVectorCapacityTest.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java b/paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java index aafac3113506..907601281b16 100644 --- a/paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/data/columnar/heap/HeapVectorCapacityTest.java @@ -48,7 +48,8 @@ void testHugeVectorCapacityIsRetainedWhenUsageIsNotTooSmall() { vector.reserve(threshold); int expandedCapacity = vector.getCapacity(); - // Use ceiling division (shrink ratio = 4) so capacity is not strictly greater than usage * 4. + // Use ceiling division (shrink ratio = 4) + // so capacity is not strictly greater than usage * 4. vector.addElementsAppended((expandedCapacity + 3) / 4); vector.reset();