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 5f3c5387d2 [format] Reuse one source stream for consecutive BlobRef
copies (#8725)
5f3c5387d2 is described below
commit 5f3c5387d242449a60bbbddb66604126ac59b31e
Author: XiaoHongbo <[email protected]>
AuthorDate: Thu Jul 30 19:46:29 2026 +0800
[format] Reuse one source stream for consecutive BlobRef copies (#8725)
---
.../main/java/org/apache/paimon/data/BlobRef.java | 5 +
.../paimon/data/ReusingBlobRefStreamProvider.java | 220 ++++++++
.../paimon/fs/OffsetSeekableInputStream.java | 10 +-
.../data/ReusingBlobRefStreamProviderTest.java | 347 ++++++++++++
.../paimon/fs/OffsetSeekableInputStreamTest.java | 9 +
.../format/blob/AbstractBlobElementReader.java | 9 +-
.../format/blob/AbstractBlobElementWriter.java | 125 ++++-
.../format/blob/ArrayBlobElementSerializer.java | 6 +-
.../paimon/format/blob/BlobElementSerializer.java | 2 +-
.../paimon/format/blob/BlobFormatWriter.java | 25 +-
.../format/blob/MapBlobElementSerializer.java | 6 +-
.../format/blob/RawBlobElementSerializer.java | 6 +-
.../paimon/format/blob/BlobFormatWriterTest.java | 581 ++++++++++++++++++++-
13 files changed, 1320 insertions(+), 31 deletions(-)
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 7e8a1ddc68..26721936c6 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
@@ -82,6 +82,11 @@ public class BlobRef implements Blob {
descriptor.length());
}
+ /** Package-private, see {@link ReusingBlobRefStreamProvider}. */
+ UriReader uriReader() {
+ return uriReader;
+ }
+
@Override
public boolean equals(Object o) {
if (o == null || getClass() != o.getClass()) {
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/ReusingBlobRefStreamProvider.java
b/paimon-common/src/main/java/org/apache/paimon/data/ReusingBlobRefStreamProvider.java
new file mode 100644
index 0000000000..34c5d4a3b3
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/data/ReusingBlobRefStreamProvider.java
@@ -0,0 +1,220 @@
+/*
+ * 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;
+
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.utils.UriReader;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ * Reuses one underlying source stream across consecutive {@link BlobRef}s
that read from the same
+ * file, so writing N such references opens the source once instead of once
per reference. Callers
+ * only ever receive a stream bounded to a reference's descriptor window; the
underlying {@link
+ * UriReader} never leaves this class, so a caller cannot read past the blob
bounds or open another
+ * URI. Not a public API.
+ */
+public final class ReusingBlobRefStreamProvider implements Closeable {
+
+ @Nullable private UriReader reader;
+ @Nullable private String uri;
+ @Nullable private SeekableInputStream underlying;
+ private long position;
+ // Bumped on every open/close so a previously handed-out view can't read
the new source.
+ private long epoch;
+
+ /**
+ * Positions the open source for {@code ref}, or closes it (so {@link
#openBounded} reopens a
+ * fresh one) when it reads a different file or can't rewind. Its close
error surfaces here
+ * rather than inside {@code openBounded}, so cleanup of a previous,
already-copied source is
+ * never mistaken for {@code ref}'s fetch failure and turned into a NULL
write. Call before
+ * {@code openBounded}.
+ */
+ public void prepareFor(BlobRef ref) throws IOException {
+ epoch++; // invalidate any view handed out for the previous reference
+ if (underlying == null) {
+ return;
+ }
+ long offset = ref.toDescriptor().offset();
+ boolean sameSource = reader == ref.uriReader() &&
ref.toDescriptor().uri().equals(uri);
+ // Different file, or same file that can't rewind: drop it so
openBounded reopens fresh.
+ if (!sameSource || (position != offset && !trySeek(offset))) {
+ close();
+ }
+ }
+
+ /**
+ * Returns a stream bounded to {@code ref}'s descriptor, reusing the
source left open by {@link
+ * #prepareFor} or opening a fresh one, seeking only when not already
positioned. Open/seek
+ * errors propagate so the caller can apply its write-null policy; on such
an error the source
+ * is dropped.
+ */
+ public SeekableInputStream openBounded(BlobRef ref) throws IOException {
+ BlobDescriptor descriptor = ref.toDescriptor();
+ // Fail closed: a mismatched ref would silently read the wrong or
unbounded bytes.
+ if (ref.getClass() != BlobRef.class) {
+ throw new IllegalArgumentException(
+ "ReusingBlobRefStreamProvider does not support BlobRef
subclasses.");
+ }
+ if (descriptor.length() < 0) {
+ throw new IllegalArgumentException(
+ "ReusingBlobRefStreamProvider requires a non-negative
length.");
+ }
+ if (underlying != null && (reader != ref.uriReader() ||
!descriptor.uri().equals(uri))) {
+ throw new IllegalStateException(
+ "openBounded ref differs from prepareFor; call prepareFor
first.");
+ }
+ epoch++; // invalidate any view handed out for the previous reference
+ long offset = descriptor.offset();
+ try {
+ if (underlying == null) {
+ underlying = ref.uriReader().newInputStream(descriptor.uri());
+ reader = ref.uriReader();
+ uri = descriptor.uri();
+ position = 0;
+ }
+ if (position != offset) {
+ underlying.seek(offset);
+ position = offset;
+ }
+ } catch (IOException | RuntimeException | Error e) {
+ discardQuietly();
+ throw e;
+ }
+ return new BoundedSource(epoch, descriptor.length());
+ }
+
+ /** Seeks the open source, returning false instead of throwing when it
can't reposition. */
+ private boolean trySeek(long offset) {
+ try {
+ underlying.seek(offset);
+ position = offset;
+ return true;
+ } catch (IOException | RuntimeException e) {
+ return false;
+ }
+ }
+
+ /** Drops the underlying source, swallowing any close error (used after a
copy failure). */
+ public void discardQuietly() {
+ try {
+ close();
+ } catch (RuntimeException | Error | IOException ignored) {
+ // Swallow so the triggering open/seek/read error isn't masked.
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ epoch++; // invalidate any outstanding view
+ closeUnderlying();
+ }
+
+ private void closeUnderlying() throws IOException {
+ SeekableInputStream toClose = underlying;
+ underlying = null;
+ reader = null;
+ uri = null;
+ position = 0;
+ if (toClose != null) {
+ toClose.close();
+ }
+ }
+
+ /**
+ * A read-only view over the shared underlying stream, capped at the
descriptor length. Reads
+ * advance the shared position; {@link #close()} keeps the underlying open
for the next blob. It
+ * is only valid until the next {@code openBounded}/{@code close}; using
it afterwards throws
+ * rather than reading the new source.
+ */
+ private final class BoundedSource extends SeekableInputStream {
+
+ private final long viewEpoch;
+ private final long length;
+ private long remaining;
+
+ private BoundedSource(long viewEpoch, long length) {
+ this.viewEpoch = viewEpoch;
+ this.length = length;
+ this.remaining = length;
+ }
+
+ private void checkValid() {
+ if (viewEpoch != epoch) {
+ throw new IllegalStateException(
+ "Stale ReusingBlobRefStreamProvider stream: a newer
reference was opened or it was closed.");
+ }
+ }
+
+ @Override
+ public int read() throws IOException {
+ checkValid();
+ if (remaining <= 0) {
+ return -1;
+ }
+ int b = underlying.read();
+ if (b >= 0) {
+ remaining--;
+ position++;
+ }
+ return b;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ checkValid();
+ if (b == null) {
+ throw new NullPointerException();
+ } else if (off < 0 || len < 0 || len > b.length - off) {
+ throw new IndexOutOfBoundsException();
+ } else if (len == 0) {
+ return 0;
+ }
+ if (remaining <= 0) {
+ return -1;
+ }
+ int n = underlying.read(b, off, (int) Math.min(len, remaining));
+ if (n > 0) {
+ remaining -= n;
+ position += n;
+ }
+ return n;
+ }
+
+ @Override
+ public void seek(long desired) {
+ throw new UnsupportedOperationException(
+ "ReusingBlobRefStreamProvider stream is read-only.");
+ }
+
+ @Override
+ public long getPos() {
+ checkValid();
+ return length - remaining;
+ }
+
+ @Override
+ public void close() {
+ remaining = 0;
+ }
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/fs/OffsetSeekableInputStream.java
b/paimon-common/src/main/java/org/apache/paimon/fs/OffsetSeekableInputStream.java
index 66f7f08c0d..d32d7f48b5 100644
---
a/paimon-common/src/main/java/org/apache/paimon/fs/OffsetSeekableInputStream.java
+++
b/paimon-common/src/main/java/org/apache/paimon/fs/OffsetSeekableInputStream.java
@@ -18,6 +18,8 @@
package org.apache.paimon.fs;
+import org.apache.paimon.utils.IOUtils;
+
import java.io.IOException;
/**
@@ -36,7 +38,13 @@ public class OffsetSeekableInputStream extends
SeekableInputStream {
this.offset = offset;
this.length = length;
if (offset != 0) {
- wrapped.seek(offset);
+ try {
+ wrapped.seek(offset);
+ } catch (IOException | RuntimeException | Error e) {
+ // Constructor failed: close the wrapped stream so it doesn't
leak.
+ IOUtils.closeQuietly(wrapped);
+ throw e;
+ }
}
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/data/ReusingBlobRefStreamProviderTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/ReusingBlobRefStreamProviderTest.java
new file mode 100644
index 0000000000..c5deaaaaa7
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/data/ReusingBlobRefStreamProviderTest.java
@@ -0,0 +1,347 @@
+/*
+ * 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;
+
+import org.apache.paimon.fs.ByteArraySeekableStream;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.UriReader;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link ReusingBlobRefStreamProvider}. */
+public class ReusingBlobRefStreamProviderTest {
+
+ /** A UriReader over in-memory files that counts how many streams it
opened. */
+ private static final class CountingUriReader implements UriReader {
+ private final Map<String, byte[]> files;
+ int openCount;
+
+ CountingUriReader(Map<String, byte[]> files) {
+ this.files = files;
+ }
+
+ @Override
+ public SeekableInputStream newInputStream(String uri) {
+ openCount++;
+ return new ByteArraySeekableStream(files.get(uri));
+ }
+ }
+
+ private static byte[] range(int from, int to) {
+ byte[] b = new byte[to - from];
+ for (int i = from; i < to; i++) {
+ b[i - from] = (byte) i;
+ }
+ return b;
+ }
+
+ /** A forward-only stream: reads and getPos work, but seek is unsupported.
*/
+ private static class ForwardOnlyStream extends SeekableInputStream {
+ private final byte[] data;
+ private int pos;
+
+ ForwardOnlyStream(byte[] data) {
+ this.data = data;
+ }
+
+ @Override
+ public void seek(long desired) {
+ throw new UnsupportedOperationException("forward-only");
+ }
+
+ @Override
+ public long getPos() {
+ return pos;
+ }
+
+ @Override
+ public int read() {
+ return pos < data.length ? data[pos++] & 0xff : -1;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) {
+ if (pos >= data.length) {
+ return -1;
+ }
+ int n = Math.min(len, data.length - pos);
+ System.arraycopy(data, pos, b, off, n);
+ pos += n;
+ return n;
+ }
+
+ @Override
+ public void close() {}
+ }
+
+ @Test
+ public void testBoundedToDescriptorWindow() throws IOException {
+ Map<String, byte[]> files = new HashMap<>();
+ files.put("mem://f", range(0, 20));
+ CountingUriReader reader = new CountingUriReader(files);
+ ReusingBlobRefStreamProvider reuse = new
ReusingBlobRefStreamProvider();
+
+ // A window [5, 5) exposes only bytes 5..9 and then EOF -- not the
rest of the file.
+ BlobRef ref = new BlobRef(reader, new BlobDescriptor("mem://f", 5, 5));
+ try (SeekableInputStream in = reuse.openBounded(ref)) {
+ assertThat(IOUtils.readFully(in, false)).containsExactly(range(5,
10));
+ assertThat(in.read()).isEqualTo(-1);
+ }
+ reuse.close();
+ }
+
+ @Test
+ public void testReusesOneOpenAcrossWindowsAndSeeksWithin() throws
IOException {
+ Map<String, byte[]> files = new HashMap<>();
+ files.put("mem://f", range(0, 20));
+ CountingUriReader reader = new CountingUriReader(files);
+ ReusingBlobRefStreamProvider reuse = new
ReusingBlobRefStreamProvider();
+
+ byte[] first = new byte[5];
+ IOUtils.readFully(
+ reuse.openBounded(new BlobRef(reader, new
BlobDescriptor("mem://f", 0, 5))), first);
+ byte[] second = new byte[3];
+ IOUtils.readFully(
+ reuse.openBounded(new BlobRef(reader, new
BlobDescriptor("mem://f", 12, 3))),
+ second);
+ reuse.close();
+
+ assertThat(first).containsExactly(range(0, 5));
+ assertThat(second).containsExactly(range(12, 15));
+ // Same reader + uri across the two windows: opened once, not per
reference.
+ assertThat(reader.openCount).isEqualTo(1);
+ }
+
+ @Test
+ public void testReopensWhenSourceUriChanges() throws IOException {
+ Map<String, byte[]> files = new HashMap<>();
+ files.put("mem://a", range(0, 10));
+ files.put("mem://b", range(100, 110));
+ CountingUriReader reader = new CountingUriReader(files);
+ ReusingBlobRefStreamProvider reuse = new
ReusingBlobRefStreamProvider();
+
+ BlobRef a = new BlobRef(reader, new BlobDescriptor("mem://a", 0, 4));
+ BlobRef b = new BlobRef(reader, new BlobDescriptor("mem://b", 0, 4));
+ reuse.prepareFor(a);
+ reuse.openBounded(a);
+ reuse.prepareFor(b);
+ reuse.openBounded(b);
+ reuse.close();
+
+ assertThat(reader.openCount).isEqualTo(2);
+ }
+
+ @Test
+ public void testPrepareForDifferentSourcePropagatesCloseError() throws
IOException {
+ UriReader reader =
+ u ->
+ u.equals("mem://a")
+ ? new ByteArraySeekableStream(range(0, 4)) {
+ @Override
+ public void close() {
+ throw new RuntimeException("close
failure");
+ }
+ }
+ : new ByteArraySeekableStream(range(0, 4));
+ ReusingBlobRefStreamProvider reuse = new
ReusingBlobRefStreamProvider();
+ reuse.openBounded(new BlobRef(reader, new BlobDescriptor("mem://a", 0,
4)));
+ // Switching away from a surfaces its close error rather than
swallowing it.
+ assertThatThrownBy(
+ () ->
+ reuse.prepareFor(
+ new BlobRef(reader, new
BlobDescriptor("mem://b", 0, 4))))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("close failure");
+ }
+
+ @Test
+ public void testForwardOnlyRewindCloseErrorSurfaces() throws IOException {
+ // Forward-only source can't rewind and also fails to close: the close
error must surface,
+ // not be swallowed by the reopen.
+ UriReader reader =
+ u ->
+ new ForwardOnlyStream(range(0, 5)) {
+ @Override
+ public void close() {
+ throw new RuntimeException("close failure");
+ }
+ };
+ ReusingBlobRefStreamProvider reuse = new
ReusingBlobRefStreamProvider();
+ BlobRef ref = new BlobRef(reader, new BlobDescriptor("mem://f", 0, 5));
+ reuse.prepareFor(ref);
+ IOUtils.readFully(reuse.openBounded(ref), new byte[5]);
+ // Second same-uri ref needs a rewind the source can't do; its failing
close surfaces.
+ assertThatThrownBy(() -> reuse.prepareFor(ref))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("close failure");
+ }
+
+ @Test
+ public void testStaleViewCannotReadNewSource() throws IOException {
+ Map<String, byte[]> files = new HashMap<>();
+ files.put("mem://a", range(0, 20));
+ files.put("mem://b", range(100, 110));
+ CountingUriReader reader = new CountingUriReader(files);
+ ReusingBlobRefStreamProvider reuse = new
ReusingBlobRefStreamProvider();
+
+ // A view over a wide window of file a, then open a second ref into
file b.
+ SeekableInputStream stale =
+ reuse.openBounded(new BlobRef(reader, new
BlobDescriptor("mem://a", 0, 20)));
+ BlobRef b = new BlobRef(reader, new BlobDescriptor("mem://b", 0, 4));
+ reuse.prepareFor(b);
+ reuse.openBounded(b);
+
+ // The stale view must not read file b's bytes with the old (larger)
length.
+ assertThatThrownBy(() -> stale.read(new byte[20], 0, 20))
+ .isInstanceOf(IllegalStateException.class);
+ reuse.close();
+ }
+
+ @Test
+ public void testViewIsStaleAfterClose() throws IOException {
+ Map<String, byte[]> files = new HashMap<>();
+ files.put("mem://f", range(0, 20));
+ ReusingBlobRefStreamProvider reuse = new
ReusingBlobRefStreamProvider();
+ SeekableInputStream view =
+ reuse.openBounded(
+ new BlobRef(
+ new CountingUriReader(files), new
BlobDescriptor("mem://f", 0, 5)));
+ reuse.close();
+
assertThatThrownBy(view::read).isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ public void testSeekErrorNotMaskedByCloseError() {
+ // seek fails (IOException); dropping the source then fails to close
(RuntimeException).
+ UriReader reader =
+ u ->
+ new ByteArraySeekableStream(range(0, 20)) {
+ @Override
+ public void seek(long desired) throws IOException {
+ throw new IOException("seek failure");
+ }
+
+ @Override
+ public void close() {
+ throw new RuntimeException("close failure");
+ }
+ };
+ ReusingBlobRefStreamProvider reuse = new
ReusingBlobRefStreamProvider();
+ // The original seek error must surface, not the follow-on close error.
+ assertThatThrownBy(
+ () ->
+ reuse.openBounded(
+ new BlobRef(reader, new
BlobDescriptor("mem://f", 2, 3))))
+ .isInstanceOf(IOException.class)
+ .hasMessage("seek failure");
+ }
+
+ @Test
+ public void testForwardOnlySourceReopensToRewind() throws IOException {
+ // Two same-uri offset-0 refs: the second needs a rewind the
forward-only stream can't do,
+ // so the source is reopened rather than failing.
+ int[] opens = {0};
+ UriReader reader =
+ u -> {
+ opens[0]++;
+ return new ForwardOnlyStream(range(0, 5));
+ };
+ ReusingBlobRefStreamProvider reuse = new
ReusingBlobRefStreamProvider();
+
+ BlobRef ref = new BlobRef(reader, new BlobDescriptor("mem://f", 0, 5));
+ byte[] first = new byte[5];
+ reuse.prepareFor(ref);
+ IOUtils.readFully(reuse.openBounded(ref), first);
+ byte[] second = new byte[5];
+ reuse.prepareFor(ref);
+ IOUtils.readFully(reuse.openBounded(ref), second);
+ reuse.close();
+
+ assertThat(first).containsExactly(range(0, 5));
+ assertThat(second).containsExactly(range(0, 5));
+ assertThat(opens[0]).isEqualTo(2); // reopened to rewind
+ }
+
+ @Test
+ public void testReadArrayHonorsInputStreamContract() throws IOException {
+ Map<String, byte[]> files = new HashMap<>();
+ files.put("mem://f", range(0, 20));
+ ReusingBlobRefStreamProvider reuse = new
ReusingBlobRefStreamProvider();
+ SeekableInputStream in =
+ reuse.openBounded(
+ new BlobRef(
+ new CountingUriReader(files), new
BlobDescriptor("mem://f", 0, 5)));
+ assertThat(in.read(new byte[4], 0, 0)).isZero(); // len 0 -> 0, not -1
+ assertThatThrownBy(() -> in.read(null, 0,
1)).isInstanceOf(NullPointerException.class);
+ assertThatThrownBy(() -> in.read(new byte[4], -1, 1))
+ .isInstanceOf(IndexOutOfBoundsException.class);
+ assertThatThrownBy(() -> in.read(new byte[4], 0, 5))
+ .isInstanceOf(IndexOutOfBoundsException.class);
+ reuse.close();
+ }
+
+ @Test
+ public void testOpenBoundedRejectsMisuse() throws IOException {
+ Map<String, byte[]> files = new HashMap<>();
+ files.put("mem://a", range(0, 10));
+ files.put("mem://b", range(0, 10));
+ CountingUriReader reader = new CountingUriReader(files);
+ ReusingBlobRefStreamProvider reuse = new
ReusingBlobRefStreamProvider();
+
+ // Subclass (may override newInputStream()) and unknown length are
rejected.
+ BlobRef subclass = new BlobRef(reader, new BlobDescriptor("mem://a",
0, 4)) {};
+ assertThatThrownBy(() -> reuse.openBounded(subclass))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(
+ () ->
+ reuse.openBounded(
+ new BlobRef(reader, new
BlobDescriptor("mem://a", 0, -1))))
+ .isInstanceOf(IllegalArgumentException.class);
+
+ // A ref not matching the open source (prepareFor skipped) is
rejected, not silently reused.
+ reuse.openBounded(new BlobRef(reader, new BlobDescriptor("mem://a", 0,
4)));
+ assertThatThrownBy(
+ () ->
+ reuse.openBounded(
+ new BlobRef(reader, new
BlobDescriptor("mem://b", 0, 4))))
+ .isInstanceOf(IllegalStateException.class);
+ reuse.close();
+ }
+
+ @Test
+ public void testReturnedStreamIsReadOnly() throws IOException {
+ Map<String, byte[]> files = new HashMap<>();
+ files.put("mem://f", range(0, 20));
+ ReusingBlobRefStreamProvider reuse = new
ReusingBlobRefStreamProvider();
+ SeekableInputStream in =
+ reuse.openBounded(
+ new BlobRef(
+ new CountingUriReader(files), new
BlobDescriptor("mem://f", 5, 5)));
+ assertThatThrownBy(() ->
in.seek(0)).isInstanceOf(UnsupportedOperationException.class);
+ reuse.close();
+ }
+}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/fs/OffsetSeekableInputStreamTest.java
b/paimon-common/src/test/java/org/apache/paimon/fs/OffsetSeekableInputStreamTest.java
index 16a5e3924c..8a62fff520 100644
---
a/paimon-common/src/test/java/org/apache/paimon/fs/OffsetSeekableInputStreamTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/fs/OffsetSeekableInputStreamTest.java
@@ -209,6 +209,15 @@ public class OffsetSeekableInputStreamTest {
verify(mockStream, times(1)).close();
}
+ @Test
+ public void testConstructorSeekFailureClosesWrapped() throws IOException {
+ SeekableInputStream mockStream = mock(SeekableInputStream.class);
+ org.mockito.Mockito.doThrow(new IOException("seek
failed")).when(mockStream).seek(5);
+ assertThatThrownBy(() -> new OffsetSeekableInputStream(mockStream, 5,
10))
+ .isInstanceOf(IOException.class);
+ verify(mockStream, times(1)).close();
+ }
+
@Test
public void testReadWithUnlimitedLength() throws IOException {
long offset = 5;
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/blob/AbstractBlobElementReader.java
b/paimon-format/src/main/java/org/apache/paimon/format/blob/AbstractBlobElementReader.java
index 51af6d0dc4..e183be8936 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/blob/AbstractBlobElementReader.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/blob/AbstractBlobElementReader.java
@@ -19,11 +19,14 @@
package org.apache.paimon.format.blob;
import org.apache.paimon.data.Blob;
+import org.apache.paimon.data.BlobDescriptor;
+import org.apache.paimon.data.BlobRef;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.utils.UriReader;
import javax.annotation.Nullable;
@@ -32,7 +35,7 @@ import java.io.IOException;
/** Common reader functionality shared by blob element formats. */
abstract class AbstractBlobElementReader implements
BlobElementSerializer.Reader {
- private final FileIO fileIO;
+ private final UriReader uriReader;
private final String filePath;
private final @Nullable SeekableInputStream in;
private final boolean blobAsDescriptor;
@@ -42,7 +45,7 @@ abstract class AbstractBlobElementReader implements
BlobElementSerializer.Reader
Path filePath,
@Nullable SeekableInputStream in,
boolean blobAsDescriptor) {
- this.fileIO = fileIO;
+ this.uriReader = UriReader.fromFile(fileIO);
this.filePath = filePath.toString();
this.in = in;
this.blobAsDescriptor = blobAsDescriptor;
@@ -58,7 +61,7 @@ abstract class AbstractBlobElementReader implements
BlobElementSerializer.Reader
protected final Blob readBlob(long position, long length) {
if (blobAsDescriptor) {
- return Blob.fromFile(fileIO, filePath, position, length);
+ return new BlobRef(uriReader, new BlobDescriptor(filePath,
position, length));
}
return Blob.fromData(readInlineBlob(position, length));
}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/blob/AbstractBlobElementWriter.java
b/paimon-format/src/main/java/org/apache/paimon/format/blob/AbstractBlobElementWriter.java
index 3a2293cbc9..a79114e8b1 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/blob/AbstractBlobElementWriter.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/blob/AbstractBlobElementWriter.java
@@ -24,6 +24,7 @@ import org.apache.paimon.data.BlobDescriptor;
import org.apache.paimon.data.BlobFetchMetricReporter;
import org.apache.paimon.data.BlobRef;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.ReusingBlobRefStreamProvider;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.fs.SeekableInputStream;
@@ -35,6 +36,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.io.EOFException;
import java.io.IOException;
import java.util.zip.CRC32;
@@ -55,6 +57,7 @@ abstract class AbstractBlobElementWriter implements
BlobElementSerializer.Writer
private final BlobFetchMetricReporter blobFetchMetricReporter;
private final CRC32 crc32;
private final byte[] copyBuffer;
+ private final ReusingBlobRefStreamProvider reuseSource;
private String pathString;
@@ -78,6 +81,7 @@ abstract class AbstractBlobElementWriter implements
BlobElementSerializer.Writer
this.blobFetchMetricReporter = blobFetchMetricReporter;
this.crc32 = new CRC32();
this.copyBuffer = new byte[copyBufferSize];
+ this.reuseSource = new ReusingBlobRefStreamProvider();
}
@Override
@@ -124,10 +128,39 @@ abstract class AbstractBlobElementWriter implements
BlobElementSerializer.Writer
}
}
- protected final @Nullable SeekableInputStream openBlobInputStream(Blob
blob)
- throws IOException {
+ /**
+ * Prepares the byte source for a blob, or returns {@code null} to write
it as NULL (missing
+ * file / fetch failure). An exact {@link BlobRef} with known length
reuses one source stream
+ * (bounded to the descriptor); other blobs open their own stream and read
until EOF.
+ */
+ protected final @Nullable BlobCopySource prepareBlobSource(Blob blob)
throws IOException {
+ // Exact class only: subclasses may override newInputStream() and must
not be bypassed.
+ if (blob != null && blob.getClass() == BlobRef.class) {
+ BlobRef ref = (BlobRef) blob;
+ long length = ref.toDescriptor().length();
+ if (length >= 0) {
+ // Position/release the previous source first; its cleanup
error is not this blob's
+ // fetch failure and must not be turned into a NULL write.
+ reuseSource.prepareFor(ref);
+ SeekableInputStream source = openStream(ref, () ->
reuseSource.openBounded(ref));
+ if (source == null) {
+ return null;
+ }
+ return new BlobCopySource(source, length);
+ }
+ }
+
+ SeekableInputStream source = openStream(blob, blob::newInputStream);
+ if (source == null) {
+ return null;
+ }
+ return new BlobCopySource(source, -1L);
+ }
+
+ @Nullable
+ private SeekableInputStream openStream(Blob blob, StreamOpener opener)
throws IOException {
try {
- return blob.newInputStream();
+ return opener.open();
} catch (IOException | RuntimeException e) {
if (writeNullOnMissingFile && HttpClientUtils.isNotFoundError(e)) {
LOG.warn(
@@ -148,21 +181,55 @@ abstract class AbstractBlobElementWriter implements
BlobElementSerializer.Writer
}
}
- protected final BlobDescriptor writeBlobData(SeekableInputStream in)
throws IOException {
+ protected final BlobDescriptor writeBlobData(BlobCopySource source) throws
IOException {
long blobPosition = out.getPos();
- try (SeekableInputStream stream = in) {
- int bytesRead = stream.read(copyBuffer);
- while (bytesRead >= 0) {
- write(copyBuffer, bytesRead);
- bytesRead = stream.read(copyBuffer);
+ if (source.reused()) {
+ try {
+ copyExactly(source.stream(), source.length());
+ } catch (IOException | RuntimeException e) {
+ blobFetchMetricReporter.recordFetchFailure(e);
+ // Source is at an unknown position now; drop it so the next
blob reopens.
+ reuseSource.discardQuietly();
+ throw e;
+ }
+ } else {
+ try (SeekableInputStream stream = source.stream()) {
+ int bytesRead = stream.read(copyBuffer);
+ while (bytesRead >= 0) {
+ write(copyBuffer, bytesRead);
+ bytesRead = stream.read(copyBuffer);
+ }
+ } catch (IOException | RuntimeException e) {
+ blobFetchMetricReporter.recordFetchFailure(e);
+ throw e;
}
- } catch (IOException | RuntimeException e) {
- blobFetchMetricReporter.recordFetchFailure(e);
- throw e;
}
return new BlobDescriptor(pathString, blobPosition, out.getPos() -
blobPosition);
}
+ /** Copies exactly {@code length} bytes from {@code stream}, throwing on
premature EOF. */
+ private void copyExactly(SeekableInputStream stream, long length) throws
IOException {
+ long remaining = length;
+ while (remaining > 0) {
+ int toRead = (int) Math.min(copyBuffer.length, remaining);
+ int bytesRead = stream.read(copyBuffer, 0, toRead);
+ if (bytesRead < 0) {
+ throw new EOFException(
+ String.format(
+ "Unexpected EOF while copying BLOB payload for
field %s: expected %d "
+ + "bytes but source ended %d bytes
early.",
+ blobFieldName, length, remaining));
+ }
+ if (bytesRead == 0) {
+ throw new IOException(
+ "Source returned 0 bytes while copying BLOB payload
for field "
+ + blobFieldName);
+ }
+ write(copyBuffer, bytesRead);
+ remaining -= bytesRead;
+ }
+ }
+
protected final boolean accept(BlobDescriptor descriptor) throws
IOException {
return writeConsumer != null && writeConsumer.accept(blobFieldName,
descriptor);
}
@@ -236,6 +303,40 @@ abstract class AbstractBlobElementWriter implements
BlobElementSerializer.Writer
|| uri.regionMatches(true, 0, "https://";, 0,
"https://".length());
}
+ @Override
+ public final void close() throws IOException {
+ reuseSource.close();
+ }
+
+ @FunctionalInterface
+ private interface StreamOpener {
+ SeekableInputStream open() throws IOException;
+ }
+
+ /** The byte source of a single blob payload to be copied into the blob
file. */
+ protected static final class BlobCopySource {
+
+ private final SeekableInputStream stream;
+ private final long length;
+
+ private BlobCopySource(SeekableInputStream stream, long length) {
+ this.stream = stream;
+ this.length = length;
+ }
+
+ private boolean reused() {
+ return length >= 0;
+ }
+
+ private SeekableInputStream stream() {
+ return stream;
+ }
+
+ private long length() {
+ return length;
+ }
+ }
+
/** Lazily gets a Blob from a row or array. */
interface BlobGetter {
Blob get();
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/blob/ArrayBlobElementSerializer.java
b/paimon-format/src/main/java/org/apache/paimon/format/blob/ArrayBlobElementSerializer.java
index d6e673ea45..265f0883f2 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/blob/ArrayBlobElementSerializer.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/blob/ArrayBlobElementSerializer.java
@@ -138,12 +138,12 @@ final class ArrayBlobElementSerializer implements
BlobElementSerializer {
elementLengths[i] = NULL_ELEMENT_LENGTH;
continue;
}
- SeekableInputStream in = openBlobInputStream(blob);
- if (in == null) {
+ BlobCopySource source = prepareBlobSource(blob);
+ if (source == null) {
elementLengths[i] = NULL_ELEMENT_LENGTH;
continue;
}
- BlobDescriptor descriptor = writeBlobData(in);
+ BlobDescriptor descriptor = writeBlobData(source);
elementLengths[i] = descriptor.length();
flush |= accept(descriptor);
recordSuccess(descriptor.length());
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobElementSerializer.java
b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobElementSerializer.java
index ca98461053..e1499a89f5 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobElementSerializer.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobElementSerializer.java
@@ -71,7 +71,7 @@ interface BlobElementSerializer {
}
/** Writer used for the lifetime of one output file. */
- interface Writer {
+ interface Writer extends Closeable {
long write(InternalRow row) throws IOException;
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 018b75421c..e95722c2f5 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
@@ -98,10 +98,27 @@ public class BlobFormatWriter implements
FileAwareFormatWriter {
@Override
public void close() throws IOException {
- byte[] indexBytes = DeltaVarintCompressor.compress(lengths.toArray());
- out.write(indexBytes);
- out.write(intToLittleEndian(indexBytes.length));
- out.write(VERSION);
+ Throwable primary = null;
+ try {
+ byte[] indexBytes =
DeltaVarintCompressor.compress(lengths.toArray());
+ out.write(indexBytes);
+ out.write(intToLittleEndian(indexBytes.length));
+ out.write(VERSION);
+ } catch (RuntimeException | Error | IOException e) {
+ primary = e;
+ throw e;
+ } finally {
+ // Surface the footer error as primary and attach a source-close
error as suppressed.
+ if (primary == null) {
+ elementWriter.close();
+ } else {
+ try {
+ elementWriter.close();
+ } catch (RuntimeException | Error | IOException suppressed) {
+ primary.addSuppressed(suppressed);
+ }
+ }
+ }
}
private static BlobElementSerializer.Writer createElementWriter(
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/blob/MapBlobElementSerializer.java
b/paimon-format/src/main/java/org/apache/paimon/format/blob/MapBlobElementSerializer.java
index 8c96edca27..1a4b922552 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/blob/MapBlobElementSerializer.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/blob/MapBlobElementSerializer.java
@@ -208,12 +208,12 @@ final class MapBlobElementSerializer implements
BlobElementSerializer {
valueLengths[i] = NULL_VALUE_LENGTH;
continue;
}
- SeekableInputStream in = openBlobInputStream(blob);
- if (in == null) {
+ BlobCopySource source = prepareBlobSource(blob);
+ if (source == null) {
valueLengths[i] = NULL_VALUE_LENGTH;
continue;
}
- BlobDescriptor descriptor = writeBlobData(in);
+ BlobDescriptor descriptor = writeBlobData(source);
valueLengths[i] = descriptor.length();
flush |= accept(descriptor);
recordSuccess(descriptor.length());
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/blob/RawBlobElementSerializer.java
b/paimon-format/src/main/java/org/apache/paimon/format/blob/RawBlobElementSerializer.java
index 2a73b0947a..d649c61e0b 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/blob/RawBlobElementSerializer.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/blob/RawBlobElementSerializer.java
@@ -106,13 +106,13 @@ final class RawBlobElementSerializer implements
BlobElementSerializer {
return writePlaceholderElement();
}
- SeekableInputStream in = openBlobInputStream(blob);
- if (in == null) {
+ BlobCopySource source = prepareBlobSource(blob);
+ if (source == null) {
return writeNullElement();
}
long recordPosition = startRecord();
- BlobDescriptor descriptor = writeBlobData(in);
+ BlobDescriptor descriptor = writeBlobData(source);
long recordLength = finishRecord(recordPosition);
if (accept(descriptor)) {
flush();
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java
index cf96d40e06..0537c70951 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java
@@ -49,8 +49,13 @@ import org.apache.paimon.utils.UriReaderFactory;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.io.ByteArrayInputStream;
+import java.io.EOFException;
+import java.io.IOException;
import java.nio.file.Files;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -656,6 +661,7 @@ public class BlobFormatWriterTest {
private final Map<String, byte[]> files;
private final List<CountingSeekableInputStream> opened = new
ArrayList<>();
+ private int openCount;
private RecordingUriReader(Map<String, byte[]> files) {
this.files = files;
@@ -667,6 +673,7 @@ public class BlobFormatWriterTest {
if (data == null) {
throw new IllegalArgumentException("Unknown uri: " + uri);
}
+ openCount++;
CountingSeekableInputStream stream = new
CountingSeekableInputStream(data);
opened.add(stream);
return stream;
@@ -679,6 +686,7 @@ public class BlobFormatWriterTest {
private final byte[] data;
private int pos;
private int maxReadRequest;
+ private int closeCount;
private CountingSeekableInputStream(byte[] data) {
this.data = data;
@@ -719,7 +727,9 @@ public class BlobFormatWriterTest {
}
@Override
- public void close() {}
+ public void close() {
+ closeCount++;
+ }
}
@Test
@@ -1019,4 +1029,573 @@ public class BlobFormatWriterTest {
return Blob.fromBytes(descriptorBytes, uriReaderFactory, null);
}
}
+
+ @Test
+ public void testArrayBlobRefsReuseSource(@TempDir java.nio.file.Path
tempDir) throws Exception {
+ String uri = "mem://file";
+ byte[] source = sequentialBytes(30);
+ RecordingUriReader reader = new RecordingUriReader(singleFile(uri,
source));
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+
+ BlobFormatWriter writer =
+ newWriter(outputFile,
RowType.of(DataTypes.ARRAY(DataTypes.BLOB())));
+ writer.addElement(
+ GenericRow.of(
+ new GenericArray(
+ new Object[] {
+ new BlobRef(reader, new
BlobDescriptor(uri, 0, 10)),
+ new BlobRef(reader, new
BlobDescriptor(uri, 10, 10)),
+ new BlobRef(reader, new
BlobDescriptor(uri, 20, 10))
+ })));
+ writer.close();
+
+ assertThat(reader.openCount).isEqualTo(1);
+
+ LocalFileIO fileIO = new LocalFileIO();
+ Path filePath = new Path(outputFile.toUri());
+ long fileSize = Files.size(outputFile);
+ try (SeekableInputStream in = fileIO.newInputStream(filePath)) {
+ BlobFileMeta fileMeta = new BlobFileMeta(in, fileSize, null);
+ assertThat(fileMeta.recordNumber()).isEqualTo(1);
+ BlobFormatReader arrayReader =
+ new BlobFormatReader(
+ fileIO,
+ filePath,
+ fileMeta,
+ in,
+ 1,
+ 0,
+ DataTypes.ARRAY(DataTypes.BLOB()),
+ false);
+ InternalArray array = arrayReader.readBatch().next().getArray(0);
+ assertThat(array.size()).isEqualTo(3);
+
assertThat(readAll(array.getBlob(0))).isEqualTo(Arrays.copyOfRange(source, 0,
10));
+
assertThat(readAll(array.getBlob(1))).isEqualTo(Arrays.copyOfRange(source, 10,
20));
+
assertThat(readAll(array.getBlob(2))).isEqualTo(Arrays.copyOfRange(source, 20,
30));
+ }
+ }
+
+ @Test
+ public void testMapBlobRefsReuseSource(@TempDir java.nio.file.Path
tempDir) throws Exception {
+ String uri = "mem://file";
+ byte[] source = sequentialBytes(30);
+ RecordingUriReader reader = new RecordingUriReader(singleFile(uri,
source));
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+
+ Map<Object, Object> entries = new LinkedHashMap<>();
+ entries.put(
+ BinaryString.fromString("a"), new BlobRef(reader, new
BlobDescriptor(uri, 0, 10)));
+ entries.put(
+ BinaryString.fromString("b"), new BlobRef(reader, new
BlobDescriptor(uri, 10, 10)));
+ entries.put(
+ BinaryString.fromString("c"), new BlobRef(reader, new
BlobDescriptor(uri, 20, 10)));
+
+ BlobFormatWriter writer =
+ newWriter(
+ outputFile,
+ RowType.of(DataTypes.MAP(DataTypes.STRING(),
DataTypes.BLOB())));
+ writer.addElement(GenericRow.of(new GenericMap(entries)));
+ writer.close();
+
+ assertThat(reader.openCount).isEqualTo(1);
+ }
+
+ @Test
+ public void testBlobRefEqualityContractUnchanged() {
+ String uri = "mem://file";
+ RecordingUriReader reader = new RecordingUriReader(singleFile(uri, new
byte[] {1, 2, 3}));
+ BlobDescriptor descriptor = new BlobDescriptor(uri, 0, 3);
+ BlobRef plain = new BlobRef(reader, descriptor);
+ BlobRef sameDescriptor = new BlobRef(reader, descriptor);
+ BlobRef subclass = new BlobRef(reader, descriptor) {};
+
+ // Exact-class equality: plain refs equal, subclasses unequal both
ways.
+ assertThat(plain).isEqualTo(sameDescriptor);
+ assertThat(plain.hashCode()).isEqualTo(sameDescriptor.hashCode());
+ assertThat(plain).isNotEqualTo(subclass);
+ assertThat(subclass).isNotEqualTo(plain);
+ assertThat(new java.util.HashSet<>(Arrays.asList(plain,
subclass))).hasSize(2);
+ assertThat(new java.util.HashSet<>(Arrays.asList(subclass,
plain))).hasSize(2);
+ }
+
+ @Test
+ public void testBlobRefSubclassIsNotBypassed(@TempDir java.nio.file.Path
tempDir)
+ throws Exception {
+ String uri = "mem://file";
+ // Raw source holds [1,2,3]; the subclass overrides newInputStream()
to yield [9,9,9].
+ RecordingUriReader reader = new RecordingUriReader(singleFile(uri, new
byte[] {1, 2, 3}));
+ BlobRef transforming =
+ new BlobRef(reader, new BlobDescriptor(uri, 0, 3)) {
+ @Override
+ public SeekableInputStream newInputStream() {
+ return new CountingSeekableInputStream(new byte[] {9,
9, 9});
+ }
+ };
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+
+ BlobFormatWriter writer = newWriter(outputFile,
RowType.of(DataTypes.BLOB()));
+ writer.addElement(GenericRow.of(transforming));
+ writer.close();
+
+ // The override is honored (slow path); the raw-source fast path did
not bypass it.
+ assertThat(readBackBlobs(outputFile, 1)).containsExactly(new byte[]
{9, 9, 9});
+ assertThat(reader.openCount).isEqualTo(0);
+ }
+
+ @Test
+ public void testConsecutiveBlobRefsShareSingleOpenSource(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ String uri = "mem://file";
+ byte[] source = sequentialBytes(1000);
+ RecordingUriReader reader = new RecordingUriReader(singleFile(uri,
source));
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+
+ BlobFormatWriter writer = newWriter(outputFile,
RowType.of(DataTypes.BLOB()));
+ List<byte[]> expected = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ int offset = i * 10;
+ writer.addElement(
+ GenericRow.of(new BlobRef(reader, new BlobDescriptor(uri,
offset, 10))));
+ expected.add(Arrays.copyOfRange(source, offset, offset + 10));
+ }
+ writer.close();
+
+ // 100 references into the same source only opened it once.
+ assertThat(reader.openCount).isEqualTo(1);
+ assertThat(reader.opened.get(0).closeCount).isEqualTo(1);
+ assertThat(readBackBlobs(outputFile,
100)).containsExactlyElementsOf(expected);
+ }
+
+ @Test
+ public void testEmptyBlobRefsReuseWithoutFailing(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ String uri = "mem://empty";
+ RecordingUriReader reader = new RecordingUriReader(singleFile(uri, new
byte[0]));
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+
+ BlobFormatWriter writer = newWriter(outputFile,
RowType.of(DataTypes.BLOB()));
+ writer.addElement(GenericRow.of(new BlobRef(reader, new
BlobDescriptor(uri, 0, 0))));
+ writer.addElement(GenericRow.of(new BlobRef(reader, new
BlobDescriptor(uri, 0, 0))));
+ writer.close();
+
+ // Two consecutive empty refs reuse one open source and write empty
(non-NULL) blobs.
+ assertThat(reader.openCount).isEqualTo(1);
+ assertThat(readBackBlobs(outputFile, 2)).containsExactly(new byte[0],
new byte[0]);
+ }
+
+ @Test
+ public void testForwardOnlyRewindCloseErrorNotWrittenNull(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ // First source is forward-only and fails to close; the second
same-uri ref needs a rewind.
+ // Even with writeNullOnFetchFailure, the close error must abort, not
become a NULL.
+ boolean[] first = {true};
+ UriReader reader =
+ u -> {
+ if (first[0]) {
+ first[0] = false;
+ return new ForwardOnlyCloseFailingStream(new byte[]
{1, 2, 3});
+ }
+ return new CountingSeekableInputStream(new byte[] {1, 2,
3});
+ };
+ BlobFormatWriter writer =
+ newWriter(tempDir.resolve("blob.out"),
RowType.of(DataTypes.BLOB()), false, true);
+ BlobDescriptor descriptor = new BlobDescriptor("mem://f", 0, 3);
+ writer.addElement(GenericRow.of(new BlobRef(reader, descriptor)));
+ assertThatThrownBy(() -> writer.addElement(GenericRow.of(new
BlobRef(reader, descriptor))))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("close failed");
+ }
+
+ @Test
+ public void testKnownLengthNonSeekableSourceStillWrites(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ byte[] payload = {1, 2, 3};
+ // A non-seekable source (like a wrapped HTTP stream) with known
length and offset 0.
+ UriReader nonSeekable = u -> SeekableInputStream.wrap(new
ByteArrayInputStream(payload));
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+
+ BlobFormatWriter writer = newWriter(outputFile,
RowType.of(DataTypes.BLOB()));
+ writer.addElement(
+ GenericRow.of(new BlobRef(nonSeekable, new
BlobDescriptor("mem://x", 0, 3))));
+ writer.close();
+
+ // Fast-path seek fails, so it falls back to the per-blob stream and
still writes the blob.
+ assertThat(readBackBlobs(outputFile, 1)).containsExactly(payload);
+ }
+
+ @Test
+ public void testNonBlobRefDoesNotDisturbReusableSource(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ String uri = "mem://file";
+ byte[] source = sequentialBytes(30);
+ RecordingUriReader reader = new RecordingUriReader(singleFile(uri,
source));
+ byte[] inlinePayload = "inline-blob".getBytes();
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+
+ BlobFormatWriter writer = newWriter(outputFile,
RowType.of(DataTypes.BLOB()));
+ writer.addElement(GenericRow.of(new BlobRef(reader, new
BlobDescriptor(uri, 0, 5))));
+ // a non-BlobRef blob in between must not close/reopen the reusable
source.
+ writer.addElement(GenericRow.of(Blob.fromData(inlinePayload)));
+ writer.addElement(GenericRow.of(new BlobRef(reader, new
BlobDescriptor(uri, 5, 5))));
+ writer.close();
+
+ assertThat(reader.openCount).isEqualTo(1);
+ assertThat(readBackBlobs(outputFile, 3))
+ .containsExactly(
+ Arrays.copyOfRange(source, 0, 5),
+ inlinePayload,
+ Arrays.copyOfRange(source, 5, 10));
+ }
+
+ @Test
+ public void testOutOfOrderDescriptorsReadViaSeek(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ String uri = "mem://file";
+ byte[] source = sequentialBytes(30);
+ RecordingUriReader reader = new RecordingUriReader(singleFile(uri,
source));
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+
+ BlobFormatWriter writer = newWriter(outputFile,
RowType.of(DataTypes.BLOB()));
+ writer.addElement(GenericRow.of(new BlobRef(reader, new
BlobDescriptor(uri, 20, 10))));
+ writer.addElement(GenericRow.of(new BlobRef(reader, new
BlobDescriptor(uri, 0, 10))));
+ writer.addElement(GenericRow.of(new BlobRef(reader, new
BlobDescriptor(uri, 10, 10))));
+ writer.close();
+
+ assertThat(reader.openCount).isEqualTo(1);
+ assertThat(readBackBlobs(outputFile, 3))
+ .containsExactly(
+ Arrays.copyOfRange(source, 20, 30),
+ Arrays.copyOfRange(source, 0, 10),
+ Arrays.copyOfRange(source, 10, 20));
+ }
+
+ @Test
+ public void testReaderProducedBlobRefsReuseSingleOpen(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ // Write a source blob file with several payloads.
+ byte[][] payloads = {
+ "blob-0".getBytes(),
+ "blob-1".getBytes(),
+ "blob-2-longer-payload".getBytes(),
+ "blob-3".getBytes(),
+ "blob-4".getBytes()
+ };
+ java.nio.file.Path sourceFile = tempDir.resolve("source.blob");
+ BlobFormatWriter sourceWriter = newWriter(sourceFile,
RowType.of(DataTypes.BLOB()));
+ for (byte[] payload : payloads) {
+ sourceWriter.addElement(GenericRow.of(Blob.fromData(payload)));
+ }
+ sourceWriter.close();
+
+ // Read it back as descriptors through the real BlobFormatReader path.
+ CountingLocalFileIO fileIO = new CountingLocalFileIO();
+ Path sourcePath = new Path(sourceFile.toUri());
+ long sourceSize = Files.size(sourceFile);
+ List<InternalRow> rows = new ArrayList<>();
+ try (SeekableInputStream in = fileIO.newInputStream(sourcePath)) {
+ BlobFileMeta fileMeta = new BlobFileMeta(in, sourceSize, null);
+ BlobFormatReader reader =
+ new BlobFormatReader(
+ fileIO, sourcePath, fileMeta, in, 1, 0,
DataTypes.BLOB(), true);
+ FileRecordIterator<InternalRow> iterator = reader.readBatch();
+ for (int i = 0; i < payloads.length; i++) {
+ InternalRow row = iterator.next();
+ // Plain BlobRef: reader refs keep the public equality
contract.
+ assertThat(row.getBlob(0).getClass()).isEqualTo(BlobRef.class);
+ rows.add(GenericRow.of(row.getBlob(0)));
+ }
+ }
+ // Reading only opened the source once (for the reader's own stream).
+ assertThat(fileIO.openCount(sourcePath)).isEqualTo(1);
+
+ // Rewrite the descriptor BlobRefs; the writer must reuse one source
stream, not one per
+ // blob.
+ java.nio.file.Path outputFile = tempDir.resolve("out.blob");
+ BlobFormatWriter writer = newWriter(outputFile,
RowType.of(DataTypes.BLOB()));
+ for (InternalRow row : rows) {
+ writer.addElement(row);
+ }
+ writer.close();
+
+ // Exactly one additional open for all rewritten blobs.
+ assertThat(fileIO.openCount(sourcePath)).isEqualTo(2);
+ assertThat(readBackBlobs(outputFile,
payloads.length)).containsExactly(payloads);
+ }
+
+ @Test
+ public void testSameUriDifferentOffsetAndLength(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ String uri = "mem://file";
+ byte[] source = sequentialBytes(30);
+ RecordingUriReader reader = new RecordingUriReader(singleFile(uri,
source));
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+
+ BlobFormatWriter writer = newWriter(outputFile,
RowType.of(DataTypes.BLOB()));
+ writer.addElement(GenericRow.of(new BlobRef(reader, new
BlobDescriptor(uri, 0, 4))));
+ writer.addElement(GenericRow.of(new BlobRef(reader, new
BlobDescriptor(uri, 4, 11))));
+ writer.addElement(GenericRow.of(new BlobRef(reader, new
BlobDescriptor(uri, 15, 15))));
+ writer.close();
+
+ assertThat(reader.openCount).isEqualTo(1);
+ assertThat(readBackBlobs(outputFile, 3))
+ .containsExactly(
+ Arrays.copyOfRange(source, 0, 4),
+ Arrays.copyOfRange(source, 4, 15),
+ Arrays.copyOfRange(source, 15, 30));
+ }
+
+ @Test
+ public void testSeekFailureRespectsWriteNullConfig(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ RowType rowType = RowType.of(DataTypes.BLOB());
+ UriReader seekFailing = u -> new SeekFailingStream();
+ BlobDescriptor descriptor = new BlobDescriptor("mem://x", 2, 3);
+
+ // writeNullOnFetchFailure=true: a seek failure writes NULL rather
than aborting.
+ java.nio.file.Path nullFile = tempDir.resolve("null.blob");
+ BlobFormatWriter nullWriter = newWriter(nullFile, rowType, false,
true);
+ nullWriter.addElement(GenericRow.of(new BlobRef(seekFailing,
descriptor)));
+ nullWriter.close();
+ LocalFileIO fileIO = new LocalFileIO();
+ try (SeekableInputStream in = fileIO.newInputStream(new
Path(nullFile.toUri()))) {
+ BlobFileMeta meta = new BlobFileMeta(in, Files.size(nullFile),
null);
+ assertThat(meta.recordNumber()).isEqualTo(1);
+ assertThat(meta.isNull(0)).isTrue();
+ }
+
+ // writeNullOnFetchFailure=false: the seek failure propagates.
+ BlobFormatWriter failWriter =
+ newWriter(tempDir.resolve("fail.blob"), rowType, false, false);
+ assertThatThrownBy(
+ () ->
+ failWriter.addElement(
+ GenericRow.of(new BlobRef(seekFailing,
descriptor))))
+ .isInstanceOf(IOException.class);
+ }
+
+ @Test
+ public void testShortSourceThrowsEof(@TempDir java.nio.file.Path tempDir)
throws Exception {
+ String uri = "mem://file";
+ byte[] source = sequentialBytes(5);
+ RecordingUriReader reader = new RecordingUriReader(singleFile(uri,
source));
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+
+ BlobFormatWriter writer = newWriter(outputFile,
RowType.of(DataTypes.BLOB()));
+ // descriptor claims 100 bytes but source only has 5.
+ assertThatThrownBy(
+ () ->
+ writer.addElement(
+ GenericRow.of(
+ new BlobRef(
+ reader, new
BlobDescriptor(uri, 0, 100)))))
+ .isInstanceOf(EOFException.class);
+ // the broken source stream is closed on failure.
+ assertThat(reader.opened.get(0).closeCount).isEqualTo(1);
+ }
+
+ @Test
+ public void testSourceSwitchCloseErrorDoesNotWriteNull(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ // First source fails to close; switching to a second source must not
turn that into a NULL.
+ UriReader reader =
+ u ->
+ u.equals("mem://a")
+ ? new CloseFailingStream(new byte[] {1, 2, 3})
+ : new CountingSeekableInputStream(new byte[]
{4, 5, 6});
+ BlobFormatWriter writer =
+ newWriter(tempDir.resolve("blob.out"),
RowType.of(DataTypes.BLOB()), false, true);
+ writer.addElement(GenericRow.of(new BlobRef(reader, new
BlobDescriptor("mem://a", 0, 3))));
+ // Even with writeNullOnFetchFailure, the switch-close error aborts
rather than losing data.
+ assertThatThrownBy(
+ () ->
+ writer.addElement(
+ GenericRow.of(
+ new BlobRef(
+ reader,
+ new
BlobDescriptor("mem://b", 0, 3)))))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("close failed");
+ }
+
+ @Test
+ public void testSourceSwitchClosesPreviousStream(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ String uriA = "mem://a";
+ String uriB = "mem://b";
+ byte[] sourceA = sequentialBytes(5);
+ byte[] sourceB = sequentialBytes(7);
+ Map<String, byte[]> files = new LinkedHashMap<>();
+ files.put(uriA, sourceA);
+ files.put(uriB, sourceB);
+ RecordingUriReader reader = new RecordingUriReader(files);
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+
+ BlobFormatWriter writer = newWriter(outputFile,
RowType.of(DataTypes.BLOB()));
+ writer.addElement(GenericRow.of(new BlobRef(reader, new
BlobDescriptor(uriA, 0, 5))));
+ writer.addElement(GenericRow.of(new BlobRef(reader, new
BlobDescriptor(uriB, 0, 7))));
+
+ // Switching sources opened a new stream and closed the previous one
immediately.
+ assertThat(reader.openCount).isEqualTo(2);
+ assertThat(reader.opened.get(0).closeCount).isEqualTo(1);
+ assertThat(reader.opened.get(1).closeCount).isEqualTo(0);
+
+ writer.close();
+ assertThat(reader.opened.get(0).closeCount).isEqualTo(1);
+ assertThat(reader.opened.get(1).closeCount).isEqualTo(1);
+ assertThat(readBackBlobs(outputFile, 2)).containsExactly(sourceA,
sourceB);
+ }
+
+ @Test
+ public void testWriterCloseClosesSourceOnce(@TempDir java.nio.file.Path
tempDir)
+ throws Exception {
+ String uri = "mem://file";
+ byte[] source = sequentialBytes(30);
+ RecordingUriReader reader = new RecordingUriReader(singleFile(uri,
source));
+ java.nio.file.Path outputFile = tempDir.resolve("blob.out");
+
+ BlobFormatWriter writer = newWriter(outputFile,
RowType.of(DataTypes.BLOB()));
+ writer.addElement(GenericRow.of(new BlobRef(reader, new
BlobDescriptor(uri, 0, 10))));
+ writer.addElement(GenericRow.of(new BlobRef(reader, new
BlobDescriptor(uri, 10, 10))));
+ assertThat(reader.opened.get(0).closeCount).isEqualTo(0);
+
+ writer.close();
+ assertThat(reader.opened.get(0).closeCount).isEqualTo(1);
+ }
+
+ @Test
+ public void testWriterCloseSurfacesSourceCloseError(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ UriReader closeFailing = u -> new CloseFailingStream(new byte[] {1, 2,
3});
+ BlobFormatWriter writer =
+ newWriter(tempDir.resolve("blob.out"),
RowType.of(DataTypes.BLOB()));
+ writer.addElement(
+ GenericRow.of(new BlobRef(closeFailing, new
BlobDescriptor("mem://x", 0, 3))));
+ // The reusable source's close error propagates from writer.close().
+ assertThatThrownBy(writer::close)
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("close failed");
+ }
+
+ private static final class CloseFailingStream extends SeekableInputStream {
+
+ private final byte[] data;
+ private int pos;
+
+ private CloseFailingStream(byte[] data) {
+ this.data = data;
+ }
+
+ @Override
+ public void seek(long desired) {
+ this.pos = (int) desired;
+ }
+
+ @Override
+ public long getPos() {
+ return pos;
+ }
+
+ @Override
+ public int read() {
+ return pos < data.length ? data[pos++] & 0xFF : -1;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) {
+ if (pos >= data.length) {
+ return -1;
+ }
+ int n = Math.min(len, data.length - pos);
+ System.arraycopy(data, pos, b, off, n);
+ pos += n;
+ return n;
+ }
+
+ @Override
+ public void close() throws IOException {
+ throw new IOException("close failed");
+ }
+ }
+
+ private static final class CountingLocalFileIO extends LocalFileIO {
+
+ private final Map<String, Integer> opens = new HashMap<>();
+
+ @Override
+ public SeekableInputStream newInputStream(Path path) throws
IOException {
+ opens.merge(path.toString(), 1, Integer::sum);
+ return super.newInputStream(path);
+ }
+
+ private int openCount(Path path) {
+ return opens.getOrDefault(path.toString(), 0);
+ }
+ }
+
+ private static final class ForwardOnlyCloseFailingStream extends
SeekableInputStream {
+
+ private final byte[] data;
+ private int pos;
+
+ private ForwardOnlyCloseFailingStream(byte[] data) {
+ this.data = data;
+ }
+
+ @Override
+ public void seek(long desired) {
+ throw new UnsupportedOperationException("forward-only");
+ }
+
+ @Override
+ public long getPos() {
+ return pos;
+ }
+
+ @Override
+ public int read() {
+ return pos < data.length ? data[pos++] & 0xFF : -1;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) {
+ if (pos >= data.length) {
+ return -1;
+ }
+ int n = Math.min(len, data.length - pos);
+ System.arraycopy(data, pos, b, off, n);
+ pos += n;
+ return n;
+ }
+
+ @Override
+ public void close() throws IOException {
+ throw new IOException("close failed");
+ }
+ }
+
+ private static final class SeekFailingStream extends SeekableInputStream {
+
+ @Override
+ public void seek(long desired) throws IOException {
+ throw new IOException("seek failed");
+ }
+
+ @Override
+ public long getPos() {
+ return 0;
+ }
+
+ @Override
+ public int read() {
+ return -1;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) {
+ return -1;
+ }
+
+ @Override
+ public void close() {}
+ }
}
