From dd5cf33ffa6e19155a7a574a0d2b81b8dac6349e Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Mon, 20 Jul 2026 21:24:28 -0700 Subject: [PATCH] [format] Reuse one source stream for consecutive BlobRef copies BLOB compaction opens a fresh stream (file open / HTTP GET) per BlobRef -- thousands of redundant opens when one compacted file draws from one source. Reuse one stream across references into the same source (same UriReader + URI), seeking to each offset and copying exactly descriptor.length bytes, bounded to the descriptor window so a caller never reads past the blob. Reuse state lives in an internal BlobReuseSource (paimon-common) that only ever hands back a stream bounded to a reference's descriptor; the raw UriReader never leaves the class. A stale view, a mismatched ref, an unknown length, or a BlobRef subclass are rejected; a source that can't rewind is reopened, and its cleanup errors surface instead of becoming a NULL write. Non-BlobRef / unknown-length values keep the open-per-blob path, so output bytes and CRC are unchanged, and BlobRef.equals keeps exact-class semantics. Behavior change: a source shorter than the descriptor throws EOFException instead of writing a truncated blob. --- .../java/org/apache/paimon/data/BlobRef.java | 5 + .../apache/paimon/data/BlobReuseSource.java | 218 +++++++ .../paimon/fs/OffsetSeekableInputStream.java | 10 +- .../paimon/data/BlobReuseSourceTest.java | 347 +++++++++++ .../fs/OffsetSeekableInputStreamTest.java | 9 + .../paimon/format/blob/BlobFormatReader.java | 17 +- .../paimon/format/blob/BlobFormatWriter.java | 203 +++++-- .../format/blob/BlobFormatWriterTest.java | 556 +++++++++++++++++- 8 files changed, 1323 insertions(+), 42 deletions(-) create mode 100644 paimon-common/src/main/java/org/apache/paimon/data/BlobReuseSource.java create mode 100644 paimon-common/src/test/java/org/apache/paimon/data/BlobReuseSourceTest.java 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 7e8a1ddc68a8..89c6c872323d 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 SeekableInputStream newInputStream() throws IOException { descriptor.length()); } + /** Package-private, see {@link BlobReuseSource}. */ + 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/BlobReuseSource.java b/paimon-common/src/main/java/org/apache/paimon/data/BlobReuseSource.java new file mode 100644 index 000000000000..687f6cdcb1c6 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/data/BlobReuseSource.java @@ -0,0 +1,218 @@ +/* + * 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 BlobReuseSource 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( + "BlobReuseSource does not support BlobRef subclasses."); + } + if (descriptor.length() < 0) { + throw new IllegalArgumentException("BlobReuseSource 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 BlobReuseSource 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("BlobReuseSource 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 66f7f08c0dbd..d32d7f48b516 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 OffsetSeekableInputStream(SeekableInputStream wrapped, long offset, long 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/BlobReuseSourceTest.java b/paimon-common/src/test/java/org/apache/paimon/data/BlobReuseSourceTest.java new file mode 100644 index 000000000000..ad6c1f2a8871 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/data/BlobReuseSourceTest.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 BlobReuseSource}. */ +public class BlobReuseSourceTest { + + /** A UriReader over in-memory files that counts how many streams it opened. */ + private static final class CountingUriReader implements UriReader { + private final Map files; + int openCount; + + CountingUriReader(Map 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 files = new HashMap<>(); + files.put("mem://f", range(0, 20)); + CountingUriReader reader = new CountingUriReader(files); + BlobReuseSource reuse = new BlobReuseSource(); + + // 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 files = new HashMap<>(); + files.put("mem://f", range(0, 20)); + CountingUriReader reader = new CountingUriReader(files); + BlobReuseSource reuse = new BlobReuseSource(); + + 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 files = new HashMap<>(); + files.put("mem://a", range(0, 10)); + files.put("mem://b", range(100, 110)); + CountingUriReader reader = new CountingUriReader(files); + BlobReuseSource reuse = new BlobReuseSource(); + + 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)); + BlobReuseSource reuse = new BlobReuseSource(); + 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"); + } + }; + BlobReuseSource reuse = new BlobReuseSource(); + 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 files = new HashMap<>(); + files.put("mem://a", range(0, 20)); + files.put("mem://b", range(100, 110)); + CountingUriReader reader = new CountingUriReader(files); + BlobReuseSource reuse = new BlobReuseSource(); + + // 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 files = new HashMap<>(); + files.put("mem://f", range(0, 20)); + BlobReuseSource reuse = new BlobReuseSource(); + 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"); + } + }; + BlobReuseSource reuse = new BlobReuseSource(); + // 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)); + }; + BlobReuseSource reuse = new BlobReuseSource(); + + 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 files = new HashMap<>(); + files.put("mem://f", range(0, 20)); + BlobReuseSource reuse = new BlobReuseSource(); + 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 files = new HashMap<>(); + files.put("mem://a", range(0, 10)); + files.put("mem://b", range(0, 10)); + CountingUriReader reader = new CountingUriReader(files); + BlobReuseSource reuse = new BlobReuseSource(); + + // 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 files = new HashMap<>(); + files.put("mem://f", range(0, 20)); + BlobReuseSource reuse = new BlobReuseSource(); + 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 16a5e3924cd4..8a62fff5207f 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 void testClose() throws IOException { 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/BlobFormatReader.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatReader.java index 6fd4ccb75699..3284029ca4c9 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatReader.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatReader.java @@ -20,7 +20,9 @@ import org.apache.paimon.data.Blob; import org.apache.paimon.data.BlobArrayPlaceholder; +import org.apache.paimon.data.BlobDescriptor; import org.apache.paimon.data.BlobPlaceholder; +import org.apache.paimon.data.BlobRef; import org.apache.paimon.data.GenericArray; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; @@ -33,6 +35,7 @@ import org.apache.paimon.types.DataTypeRoot; import org.apache.paimon.utils.DeltaVarintCompressor; import org.apache.paimon.utils.IOUtils; +import org.apache.paimon.utils.UriReader; import javax.annotation.Nullable; @@ -48,7 +51,9 @@ public class BlobFormatReader implements FileRecordReader { private static final int MIN_ARRAY_PAYLOAD_LENGTH = ARRAY_HEADER_LENGTH + ARRAY_INDEX_LENGTH_SIZE; - private final FileIO fileIO; + // Shared by all BlobRefs this reader produces, so a BlobFormatWriter can reuse one source + // stream. + private final UriReader uriReader; private final Path filePath; private final String filePathString; private final BlobFileMeta fileMeta; @@ -70,7 +75,7 @@ public BlobFormatReader( int blobIndex, DataType blobFieldType, boolean blobAsDescriptor) { - this.fileIO = fileIO; + this.uriReader = UriReader.fromFile(fileIO); this.filePath = filePath; this.filePathString = filePath.toString(); this.fileMeta = fileMeta; @@ -144,7 +149,7 @@ private Blob readBlob(@Nullable SeekableInputStream in, long position, long leng if (in != null && !blobAsDescriptor) { return Blob.fromData(readInlineBlob(in, position, length)); } - return Blob.fromFile(fileIO, filePathString, position, length); + return new BlobRef(uriReader, new BlobDescriptor(filePathString, position, length)); } @Override @@ -259,7 +264,11 @@ private GenericArray readBlobArray(SeekableInputStream in, long position, long l if (elementLength == BlobFormatWriter.ARRAY_NULL_ELEMENT_LENGTH) { blobs[i] = null; } else if (blobAsDescriptor) { - blobs[i] = Blob.fromFile(fileIO, filePathString, elementOffset, elementLength); + blobs[i] = + new BlobRef( + uriReader, + new BlobDescriptor( + filePathString, elementOffset, elementLength)); elementOffset += elementLength; } else { blobs[i] = Blob.fromData(readInlineBlob(in, elementOffset, elementLength)); 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 5e2bc9f9bb24..547850b89301 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 @@ -25,6 +25,7 @@ import org.apache.paimon.data.BlobFetchMetricReporter; import org.apache.paimon.data.BlobPlaceholder; import org.apache.paimon.data.BlobRef; +import org.apache.paimon.data.BlobReuseSource; import org.apache.paimon.data.InternalArray; import org.apache.paimon.data.InternalRow; import org.apache.paimon.format.FileAwareFormatWriter; @@ -43,6 +44,7 @@ import javax.annotation.Nullable; +import java.io.EOFException; import java.io.IOException; import java.util.zip.CRC32; @@ -79,6 +81,10 @@ public class BlobFormatWriter implements FileAwareFormatWriter { private String pathString; + // Reuses one source stream across consecutive references into the same file, so compaction of + // descriptor/remote blobs opens each source once instead of once per blob. + private final BlobReuseSource reuseSource = new BlobReuseSource(); + public BlobFormatWriter( PositionOutputStream out, @Nullable BlobConsumer writeConsumer, @@ -179,8 +185,8 @@ public void addElement(InternalRow element) throws IOException { } private void addBlob(Blob blob) throws IOException { - SeekableInputStream in = openBlobInputStream(blob); - if (in == null) { + BlobCopySource source = prepareBlobSource(blob); + if (source == null) { writeNullElement(); return; } @@ -188,7 +194,7 @@ private void addBlob(Blob blob) throws IOException { crc32.reset(); write(MAGIC_NUMBER_BYTES); - BlobDescriptor descriptor = writeBlobData(in); + BlobDescriptor descriptor = writeBlobData(source); long binLength = out.getPos() - previousPos + 12; lengths.add(binLength); @@ -229,12 +235,12 @@ private void addBlobArray(InternalArray array) throws IOException { elementLengths[i] = ARRAY_NULL_ELEMENT_LENGTH; continue; } - SeekableInputStream in = openBlobInputStream(blob); - if (in == null) { + BlobCopySource source = prepareBlobSource(blob); + if (source == null) { elementLengths[i] = ARRAY_NULL_ELEMENT_LENGTH; continue; } - BlobDescriptor descriptor = writeBlobData(in); + BlobDescriptor descriptor = writeBlobData(source); elementLengths[i] = descriptor.length(); if (writeConsumer != null) { flush |= writeConsumer.accept(blobFieldName, descriptor); @@ -272,46 +278,154 @@ private Blob getArrayBlob(InternalArray array, int pos) { } } + /** + * 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. + */ @Nullable - private SeekableInputStream openBlobInputStream(Blob blob) throws IOException { + private 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 in = openStream(blob, blob::newInputStream); + if (in == null) { + return null; + } + return new BlobCopySource(in, -1L); + } + + @FunctionalInterface + private interface StreamOpener { + SeekableInputStream open() throws IOException; + } + + /** + * Opens a source stream via {@code opener}, converting missing-file / fetch-failure errors into + * a {@code null} result when the writer is configured to write NULL for such blobs. + */ + @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( - "Failed to open blob from {} (HTTP 404), writing NULL for BLOB field {}.", - blobUri(blob), - blobFieldName, - e); - blobFetchMetricReporter.recordMissingFileNullWritten(true); + if (writeNullOnError(blob, e)) { return null; } - if (shouldWriteNullOnFetchFailure(e)) { - logWriteNullOnFetchFailure(e, blob); - blobFetchMetricReporter.recordFetchFailureNullWritten(e); - return null; - } - blobFetchMetricReporter.recordFetchFailure(e); throw e; } } - private BlobDescriptor writeBlobData(SeekableInputStream in) throws IOException { + /** Applies the write-null-on-missing/fetch-failure policy; returns true to write NULL. */ + private boolean writeNullOnError(Blob blob, Throwable e) { + if (writeNullOnMissingFile && HttpClientUtils.isNotFoundError(e)) { + LOG.warn( + "Failed to open blob from {} (HTTP 404), writing NULL for BLOB field {}.", + blobUri(blob), + blobFieldName, + e); + blobFetchMetricReporter.recordMissingFileNullWritten(true); + return true; + } + if (shouldWriteNullOnFetchFailure(e)) { + logWriteNullOnFetchFailure(e, blob); + blobFetchMetricReporter.recordFetchFailureNullWritten(e); + return true; + } + blobFetchMetricReporter.recordFetchFailure(e); + return false; + } + + private BlobDescriptor writeBlobData(BlobCopySource source) throws IOException { long blobPos = out.getPos(); - try (SeekableInputStream stream = in) { - int bytesRead = stream.read(tmpBuffer); - while (bytesRead >= 0) { - write(tmpBuffer, bytesRead); - bytesRead = stream.read(tmpBuffer); + if (source.reused()) { + // Bounded stream from reuseSource; copy exactly descriptor.length bytes. + 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(tmpBuffer); + while (bytesRead >= 0) { + write(tmpBuffer, bytesRead); + bytesRead = stream.read(tmpBuffer); + } + } catch (IOException | RuntimeException e) { + blobFetchMetricReporter.recordFetchFailure(e); + throw e; } - } catch (IOException | RuntimeException e) { - blobFetchMetricReporter.recordFetchFailure(e); - throw e; } return new BlobDescriptor(pathString, blobPos, out.getPos() - blobPos); } + /** 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(tmpBuffer.length, remaining); + int bytesRead = stream.read(tmpBuffer, 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(tmpBuffer, bytesRead); + remaining -= bytesRead; + } + } + + /** The byte source of a single blob payload to be copied into the blob file. */ + private static final class BlobCopySource { + + private final SeekableInputStream stream; + private final long length; + + private BlobCopySource(SeekableInputStream stream, long length) { + this.stream = stream; + this.length = length; + } + + /** + * Whether the payload is copied from a reused, pre-positioned source (exact-length read). + */ + private boolean reused() { + return length >= 0; + } + + private SeekableInputStream stream() { + return stream; + } + + private long length() { + return length; + } + } + private boolean shouldWriteNullOnFetchFailure(Throwable e) { return writeNullOnFetchFailure && !HttpClientUtils.isNotFoundError(e); } @@ -388,11 +502,28 @@ public boolean reachTargetSize(boolean suggestedCheck, long targetSize) throws I @Override public void close() throws IOException { - // index - byte[] indexBytes = DeltaVarintCompressor.compress(lengths.toArray()); - out.write(indexBytes); - // header - out.write(intToLittleEndian(indexBytes.length)); - out.write(VERSION); + Throwable primary = null; + try { + // index + byte[] indexBytes = DeltaVarintCompressor.compress(lengths.toArray()); + out.write(indexBytes); + // header + 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) { + reuseSource.close(); + } else { + try { + reuseSource.close(); + } catch (RuntimeException | Error | IOException suppressed) { + primary.addSuppressed(suppressed); + } + } + } } } 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 633d32e44126..3e5465b3317a 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 @@ -42,8 +42,13 @@ 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; @@ -541,6 +546,7 @@ private static final class RecordingUriReader implements UriReader { private final Map files; private final List opened = new ArrayList<>(); + private int openCount; private RecordingUriReader(Map files) { this.files = files; @@ -552,6 +558,7 @@ public SeekableInputStream newInputStream(String uri) { if (data == null) { throw new IllegalArgumentException("Unknown uri: " + uri); } + openCount++; CountingSeekableInputStream stream = new CountingSeekableInputStream(data); opened.add(stream); return stream; @@ -564,6 +571,7 @@ private static final class CountingSeekableInputStream extends SeekableInputStre private final byte[] data; private int pos; private int maxReadRequest; + private int closeCount; private CountingSeekableInputStream(byte[] data) { this.data = data; @@ -604,7 +612,9 @@ public int read(byte[] b, int off, int len) { } @Override - public void close() {} + public void close() { + closeCount++; + } } private static void assertBlobPayload(Blob blob, byte[] expected) throws Exception { @@ -802,4 +812,548 @@ public Blob getBlob(int pos) { 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 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 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 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 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 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 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() {} + } }