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 909424f6d0 [common] Avoid temporary long[] allocations when
compressing blob footer index (#9307)
909424f6d0 is described below
commit 909424f6d07f2e896708b20a3b9dc301f17a15c5
Author: Wenchao Wu <[email protected]>
AuthorDate: Thu Aug 20 09:17:39 2026 +0800
[common] Avoid temporary long[] allocations when compressing blob footer
index (#9307)
---
.../apache/paimon/utils/DeltaVarintCompressor.java | 91 ++++++++++----
.../paimon/utils/DeltaVarintCompressorTest.java | 139 +++++++++++++++++++++
.../paimon/format/blob/BlobFormatWriter.java | 2 +-
3 files changed, 210 insertions(+), 22 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/utils/DeltaVarintCompressor.java
b/paimon-common/src/main/java/org/apache/paimon/utils/DeltaVarintCompressor.java
index 77606cdaf1..62f6ec9f0f 100644
---
a/paimon-common/src/main/java/org/apache/paimon/utils/DeltaVarintCompressor.java
+++
b/paimon-common/src/main/java/org/apache/paimon/utils/DeltaVarintCompressor.java
@@ -19,7 +19,6 @@
package org.apache.paimon.utils;
import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
/**
* Combining Delta Encoding and Varints Encoding, suitable for integer
sequences that are increasing
@@ -33,21 +32,57 @@ public class DeltaVarintCompressor {
return new byte[0];
}
- LongArrayList deltas = new LongArrayList(data.length);
- // Store the first element
- deltas.add(data[0]);
+ // First pass: compute the exact encoded size without allocating.
+ long size = varintSize(data[0]);
+ long previous = data[0];
for (int i = 1; i < data.length; i++) {
- // Compute delta
- deltas.add(data[i] - data[i - 1]);
+ long current = data[i];
+ size += varintSize(current - previous);
+ previous = current;
}
+ checkIndexSize(size);
- // Pre-allocate space
- ByteArrayOutputStream out = new ByteArrayOutputStream(data.length *
10);
- for (int i = 0; i < deltas.size(); i++) {
- // Apply ZigZag and Varints
- encodeVarint(deltas.get(i), out);
+ // Second pass: encode into a precisely sized buffer.
+ byte[] out = new byte[(int) size];
+ int pos = 0;
+ previous = data[0];
+ pos = encodeVarint(previous, out, pos);
+ for (int i = 1; i < data.length; i++) {
+ long current = data[i];
+ pos = encodeVarint(current - previous, out, pos);
+ previous = current;
+ }
+ return out;
+ }
+
+ // Compresses a LongArrayList directly, avoiding a copy into a temporary
long[].
+ public static byte[] compressLongArrayList(LongArrayList data) {
+ if (data == null || data.size() == 0) {
+ return new byte[0];
+ }
+
+ int count = data.size();
+ // First pass: compute the exact encoded size without allocating.
+ long size = varintSize(data.get(0));
+ long previous = data.get(0);
+ for (int i = 1; i < count; i++) {
+ long current = data.get(i);
+ size += varintSize(current - previous);
+ previous = current;
+ }
+ checkIndexSize(size);
+
+ // Second pass: encode into a precisely sized buffer.
+ byte[] out = new byte[(int) size];
+ int pos = 0;
+ previous = data.get(0);
+ pos = encodeVarint(previous, out, pos);
+ for (int i = 1; i < count; i++) {
+ long current = data.get(i);
+ pos = encodeVarint(current - previous, out, pos);
+ previous = current;
}
- return out.toByteArray();
+ return out;
}
// Decompresses a byte array back to the original long array
@@ -74,19 +109,33 @@ public class DeltaVarintCompressor {
return result;
}
- // Encodes a long value using ZigZag and Varints
- private static void encodeVarint(long value, ByteArrayOutputStream out) {
- // ZigZag transformation for long
+ // Number of bytes a value occupies after ZigZag and Varints encoding.
+ private static int varintSize(long value) {
long tmp = (value << 1) ^ (value >> 63);
- // Check if multiple bytes are needed
+ int size = 1;
while ((tmp & ~0x7FL) != 0) {
- // Set MSB to 1 (continuation)
- out.write(((int) tmp & 0x7F) | 0x80);
- // Unsigned right shift
+ size++;
tmp >>>= 7;
}
- // Final byte with MSB set to 0
- out.write((byte) tmp);
+ return size;
+ }
+
+ // Encodes a value using ZigZag and Varints into out starting at pos,
returning the new pos.
+ private static int encodeVarint(long value, byte[] out, int pos) {
+ long tmp = (value << 1) ^ (value >> 63);
+ while ((tmp & ~0x7FL) != 0) {
+ out[pos++] = (byte) (((int) tmp & 0x7F) | 0x80);
+ tmp >>>= 7;
+ }
+ out[pos++] = (byte) tmp;
+ return pos;
+ }
+
+ // The BLOB footer stores the index length as a signed 32-bit
little-endian int.
+ private static void checkIndexSize(long size) {
+ if (size > Integer.MAX_VALUE) {
+ throw new IllegalArgumentException("Compressed index too large: "
+ size + " bytes");
+ }
}
// Decodes a Varints-encoded value and reverses ZigZag transformation
diff --git
a/paimon-common/src/test/java/org/apache/paimon/utils/DeltaVarintCompressorTest.java
b/paimon-common/src/test/java/org/apache/paimon/utils/DeltaVarintCompressorTest.java
index f3b0fe669d..95c5130082 100644
---
a/paimon-common/src/test/java/org/apache/paimon/utils/DeltaVarintCompressorTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/utils/DeltaVarintCompressorTest.java
@@ -136,4 +136,143 @@ class DeltaVarintCompressorTest {
DeltaVarintCompressor.decompress(corrupted);
});
}
+
+ @Test
+ void testCompressLongArrayListNullAndEmpty() {
+ assertArrayEquals(new byte[0],
DeltaVarintCompressor.compressLongArrayList(null));
+ assertArrayEquals(new byte[0],
DeltaVarintCompressor.compressLongArrayList(toList()));
+ }
+
+ @Test
+ void testCompressLongArrayListMatchesArray() {
+ assertByteEquivalence(new long[] {42L});
+ assertByteEquivalence(new long[] {80L, 50L, 90L, 80L, 70L});
+ assertByteEquivalence(new long[] {7L, 7L, 7L, 7L});
+ assertByteEquivalence(new long[] {1L, 2L, 3L, 4L, 5L});
+ assertByteEquivalence(new long[] {100L, 90L, 80L, 70L, 60L});
+ assertByteEquivalence(new long[] {-1L, -2L});
+ assertByteEquivalence(new long[] {-1L, 5L, -2L, 8L});
+ assertByteEquivalence(new long[] {Long.MIN_VALUE, Long.MAX_VALUE});
+ assertByteEquivalence(new long[] {Long.MAX_VALUE, Long.MIN_VALUE});
+ assertByteEquivalence(new long[] {Long.MIN_VALUE, 0L, Long.MAX_VALUE});
+ assertByteEquivalence(new long[] {-3L, Long.MAX_VALUE, Long.MIN_VALUE,
3L});
+ }
+
+ @Test
+ void testCompressLongArrayListMatchesArrayRandom() {
+ long[] original = new long[100];
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+ for (int i = 0; i < original.length; i++) {
+ original[i] = rnd.nextLong();
+ }
+ assertByteEquivalence(original);
+ }
+
+ @Test
+ void testCompressGoldenBytes() {
+ // ZigZag(42) = 84 -> single varint byte.
+ assertArrayEquals(new byte[] {0x54},
DeltaVarintCompressor.compress(new long[] {42L}));
+ assertArrayEquals(
+ new byte[] {0x54},
DeltaVarintCompressor.compressLongArrayList(toList(42L)));
+
+ // {80, 50, 90, 80, 70}: deltas are 80, -30, 40, -10, -10.
+ byte[] expected = {(byte) 0xA0, 0x01, 0x3B, 0x50, 0x13, 0x13};
+ assertArrayEquals(
+ expected, DeltaVarintCompressor.compress(new long[] {80L, 50L,
90L, 80L, 70L}));
+ assertArrayEquals(
+ expected,
+ DeltaVarintCompressor.compressLongArrayList(toList(80L, 50L,
90L, 80L, 70L)));
+ }
+
+ @Test
+ void testCompressGoldenBytesVarintWidths() {
+ // 2-byte varint boundary: ZigZag(64) = 128, ZigZag(-64) = 127.
+ assertGolden(new long[] {64L}, new byte[] {(byte) 0x80, 0x01});
+ assertGolden(new long[] {-64L}, new byte[] {0x7F});
+
+ // 3-byte varint boundary: ZigZag(8192) = 16384, ZigZag(-8192) = 16383.
+ assertGolden(new long[] {8192L}, new byte[] {(byte) 0x80, (byte) 0x80,
0x01});
+ assertGolden(new long[] {-8192L}, new byte[] {(byte) 0xFF, 0x7F});
+
+ // Full-width 10-byte varints.
+ assertGolden(
+ new long[] {Long.MIN_VALUE},
+ new byte[] {
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ 0x01
+ });
+ assertGolden(
+ new long[] {Long.MAX_VALUE},
+ new byte[] {
+ (byte) 0xFE,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ 0x01
+ });
+
+ // Adjacent values whose delta overflows long.
+ assertGolden(
+ new long[] {Long.MAX_VALUE, Long.MIN_VALUE},
+ new byte[] {
+ (byte) 0xFE,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ 0x01,
+ 0x02
+ });
+ assertGolden(
+ new long[] {Long.MIN_VALUE, Long.MAX_VALUE},
+ new byte[] {
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ 0x01,
+ 0x01
+ });
+ }
+
+ private static void assertGolden(long[] values, byte[] expected) {
+ assertArrayEquals(expected, DeltaVarintCompressor.compress(values));
+ assertArrayEquals(expected,
DeltaVarintCompressor.compressLongArrayList(toList(values)));
+ }
+
+ private static void assertByteEquivalence(long[] values) {
+ assertArrayEquals(
+ DeltaVarintCompressor.compress(values),
+ DeltaVarintCompressor.compressLongArrayList(toList(values)));
+ }
+
+ private static LongArrayList toList(long... values) {
+ LongArrayList list = new LongArrayList(values.length);
+ for (long value : values) {
+ list.add(value);
+ }
+ return list;
+ }
}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java
b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java
index e95722c2f5..99e367447f 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java
@@ -100,7 +100,7 @@ public class BlobFormatWriter implements
FileAwareFormatWriter {
public void close() throws IOException {
Throwable primary = null;
try {
- byte[] indexBytes =
DeltaVarintCompressor.compress(lengths.toArray());
+ byte[] indexBytes =
DeltaVarintCompressor.compressLongArrayList(lengths);
out.write(indexBytes);
out.write(intToLittleEndian(indexBytes.length));
out.write(VERSION);
