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 0939062ab2 [common] Optimize BlobRef data materialization (#8663)
0939062ab2 is described below
commit 0939062ab226b55a46b60574823578f5bb90d807
Author: umi <[email protected]>
AuthorDate: Thu Jul 16 13:18:38 2026 +0800
[common] Optimize BlobRef data materialization (#8663)
---
.../main/java/org/apache/paimon/data/BlobRef.java | 19 ++++++-
.../test/java/org/apache/paimon/data/BlobTest.java | 63 ++++++++++++++++++++++
2 files changed, 81 insertions(+), 1 deletion(-)
diff --git a/paimon-common/src/main/java/org/apache/paimon/data/BlobRef.java
b/paimon-common/src/main/java/org/apache/paimon/data/BlobRef.java
index 0248454ee9..7e8a1ddc68 100644
--- a/paimon-common/src/main/java/org/apache/paimon/data/BlobRef.java
+++ b/paimon-common/src/main/java/org/apache/paimon/data/BlobRef.java
@@ -35,6 +35,9 @@ import java.util.Objects;
@Public
public class BlobRef implements Blob {
+ /** The maximum safe size of a byte array. Some VMs reserve header words
in an array. */
+ private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
+
private final UriReader uriReader;
private final BlobDescriptor descriptor;
@@ -46,7 +49,21 @@ public class BlobRef implements Blob {
@Override
public byte[] toData() {
try {
- return IOUtils.readFully(newInputStream(), true);
+ long length = descriptor.length();
+ if (length > MAX_ARRAY_SIZE) {
+ throw new IOException(
+ String.format(
+ "Blob is too large to materialize as byte[]:
%d, maximum is %d",
+ length, MAX_ARRAY_SIZE));
+ }
+ try (SeekableInputStream inputStream = newInputStream()) {
+ if (length >= 0) {
+ byte[] data = new byte[(int) length];
+ IOUtils.readFully(inputStream, data);
+ return data;
+ }
+ return IOUtils.readFully(inputStream, false);
+ }
} catch (IOException e) {
throw new RuntimeException(e);
}
diff --git a/paimon-common/src/test/java/org/apache/paimon/data/BlobTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/BlobTest.java
index 27703662da..50e34b2939 100644
--- a/paimon-common/src/test/java/org/apache/paimon/data/BlobTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/data/BlobTest.java
@@ -18,7 +18,9 @@
package org.apache.paimon.data;
+import org.apache.paimon.fs.ByteArraySeekableStream;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.utils.UriReader;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -28,8 +30,11 @@ import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicBoolean;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link Blob}. */
public class BlobTest {
@@ -79,4 +84,62 @@ public class BlobTest {
Blob blob = Blob.fromHttp(uri);
assertThat(blob).isInstanceOf(BlobRef.class);
}
+
+ @Test
+ public void testBlobRefReadsKnownLengthDirectly() {
+ byte[] data = new byte[8194];
+ for (int i = 0; i < data.length; i++) {
+ data[i] = (byte) i;
+ }
+ TrackingSeekableInputStream inputStream = new
TrackingSeekableInputStream(data);
+ UriReader uriReader = uri -> inputStream;
+ Blob blob = Blob.fromDescriptor(uriReader, new BlobDescriptor("test",
2, data.length - 2));
+
+ assertThat(blob.toData()).isEqualTo(Arrays.copyOfRange(data, 2,
data.length));
+ assertThat(inputStream.readCalls).isEqualTo(1);
+ assertThat(inputStream.maxRequestedBytes).isEqualTo(data.length - 2);
+ assertThat(inputStream.closed).isTrue();
+ }
+
+ @Test
+ public void testBlobRefRejectsTooLargeLengthBeforeOpeningStream() {
+ AtomicBoolean streamOpened = new AtomicBoolean();
+ UriReader uriReader =
+ uri -> {
+ streamOpened.set(true);
+ return new ByteArraySeekableStream(new byte[0]);
+ };
+ long length = (long) Integer.MAX_VALUE - 7;
+ Blob blob = Blob.fromDescriptor(uriReader, new BlobDescriptor("test",
0, length));
+
+ assertThatThrownBy(blob::toData)
+ .isInstanceOf(RuntimeException.class)
+ .hasCauseInstanceOf(IOException.class)
+ .hasMessageContaining("Blob is too large to materialize as
byte[]");
+ assertThat(streamOpened).isFalse();
+ }
+
+ private static class TrackingSeekableInputStream extends
ByteArraySeekableStream {
+
+ private int readCalls;
+ private int maxRequestedBytes;
+ private boolean closed;
+
+ private TrackingSeekableInputStream(byte[] data) {
+ super(data);
+ }
+
+ @Override
+ public int read(byte[] bytes, int offset, int length) throws
IOException {
+ readCalls++;
+ maxRequestedBytes = Math.max(maxRequestedBytes, length);
+ return super.read(bytes, offset, length);
+ }
+
+ @Override
+ public void close() throws IOException {
+ closed = true;
+ super.close();
+ }
+ }
}
