This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 99c6fdf00e400ded1af532b010a69b67821b25d8 Author: YeJunHao <[email protected]> AuthorDate: Tue Sep 15 17:48:32 2026 +0800 [core] Support input streams in Avro block readers (#9842) --- .../apache/paimon/manifest/ManifestAvroReader.java | 9 ++- .../apache/paimon/manifest/ManifestFileTest.java | 9 ++- .../java/org/apache/avro/file/RawBlockReader.java | 61 +++++++++++---- .../apache/paimon/format/avro/AvroBlockReader.java | 7 +- .../paimon/format/avro/AvroBlockReaderTest.java | 87 +++++----------------- 5 files changed, 83 insertions(+), 90 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java index 4e84c690ee..95134f36a0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java @@ -26,7 +26,6 @@ import org.apache.paimon.format.avro.AvroRawBlock; import org.apache.paimon.format.avro.AvroRecordDecoder; import org.apache.paimon.format.avro.AvroRecordDecoder.FieldDecoder; import org.apache.paimon.format.avro.AvroRecordDecoder.FieldType; -import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.CloseableIterator; @@ -35,6 +34,7 @@ import org.apache.paimon.utils.IOUtils; import javax.annotation.Nullable; import java.io.IOException; +import java.io.InputStream; import java.io.UncheckedIOException; import java.nio.ByteBuffer; import java.util.Arrays; @@ -56,7 +56,7 @@ public final class ManifestAvroReader implements AutoCloseable { private long blockOrdinal = -1; - ManifestAvroReader(SeekableInputStream input) throws IOException { + ManifestAvroReader(InputStream input) throws IOException { AvroBlockReader blockReader = null; try { blockReader = new AvroBlockReader(input); @@ -75,7 +75,10 @@ public final class ManifestAvroReader implements AutoCloseable { return blockReader.headerBytes(); } - /** Returns the physical block offset; read immediately after {@link #next()}. */ + /** + * Returns the block offset relative to the initial input position; read immediately after + * {@link #next()}. + */ public long blockOffset() { return blockReader.blockOffset(); } diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java index b5ff0fb002..5fb0a7e049 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java @@ -55,7 +55,9 @@ import org.junit.jupiter.params.provider.MethodSource; import javax.annotation.Nullable; +import java.io.ByteArrayInputStream; import java.io.IOException; +import java.io.SequenceInputStream; import java.io.UncheckedIOException; import java.nio.ByteBuffer; import java.nio.file.Files; @@ -1131,8 +1133,13 @@ public class ManifestFileTest { int blockCount = 0; int rowCount = 0; byte[] bytes = Files.readAllBytes(tempDir.resolve("manifest").resolve(manifest.fileName())); + int split = bytes.length / 2; - try (ManifestAvroReader reader = openManifestReader(manifest)) { + try (ManifestAvroReader reader = + new ManifestAvroReader( + new SequenceInputStream( + new ByteArrayInputStream(bytes, 0, split), + new ByteArrayInputStream(bytes, split, bytes.length - split)))) { byte[] header = reader.headerBytes(); assertThat(header).isEqualTo(Arrays.copyOf(bytes, header.length)); long nextOffset = header.length; diff --git a/paimon-format/src/main/java/org/apache/avro/file/RawBlockReader.java b/paimon-format/src/main/java/org/apache/avro/file/RawBlockReader.java index 60ae9b195a..17f39bdebb 100644 --- a/paimon-format/src/main/java/org/apache/avro/file/RawBlockReader.java +++ b/paimon-format/src/main/java/org/apache/avro/file/RawBlockReader.java @@ -18,38 +18,35 @@ package org.apache.avro.file; -import org.apache.paimon.fs.SeekableInputStream; -import org.apache.paimon.utils.IOUtils; - import org.apache.avro.Schema; import org.apache.avro.io.DatumReader; import org.apache.avro.io.Decoder; +import java.io.ByteArrayOutputStream; +import java.io.FilterInputStream; import java.io.IOException; +import java.io.InputStream; +import java.util.Arrays; import java.util.NoSuchElementException; /** Package bridge exposing Avro's compressed blocks without reflection. */ public final class RawBlockReader extends DataFileStream<Void> { - private final SeekableInputStream input; + private final CountingInput input; private final byte[] headerBytes; private long blockOffset; private long blockLength; private boolean pending; - public RawBlockReader(SeekableInputStream input) throws IOException { - this(input, input.getPos()); + public RawBlockReader(InputStream input) throws IOException { + this(new CountingInput(input)); } - private RawBlockReader(SeekableInputStream input, long headerOffset) throws IOException { + private RawBlockReader(CountingInput input) throws IOException { super(input, new NoOpDatumReader<Void>()); this.input = input; - this.headerBytes = new byte[Math.toIntExact(position() - headerOffset)]; - long resumePosition = input.getPos(); - input.seek(headerOffset); - IOUtils.readFully(input, headerBytes); - // Preserve the position past any bytes already buffered by the Avro decoder. - input.seek(resumePosition); + this.headerBytes = Arrays.copyOf(input.prefix.toByteArray(), Math.toIntExact(position())); + input.prefix = null; } /** Returns a copy of the complete OCF header, including schema, codec and sync marker. */ @@ -58,7 +55,8 @@ public final class RawBlockReader extends DataFileStream<Void> { } /** - * Returns the physical block offset; read immediately after {@link #nextRawBlock(RawBlock)}. + * Returns the block offset relative to the initial input position; read immediately after + * {@link #nextRawBlock(RawBlock)}. */ public long blockOffset() { return blockOffset; @@ -71,7 +69,7 @@ public final class RawBlockReader extends DataFileStream<Void> { private long position() throws IOException { // This is the same read-ahead adjustment used by DataFileReader.blockFinished(). - return input.getPos() - vin.inputStream().available(); + return input.position - vin.inputStream().available(); } public boolean hasNextRawBlock() throws IOException { @@ -94,6 +92,39 @@ public final class RawBlockReader extends DataFileStream<Void> { : reuse.replace(raw, resolveCodec(), getSchema()); } + private static final class CountingInput extends FilterInputStream { + private long position; + private ByteArrayOutputStream prefix = new ByteArrayOutputStream(); + + private CountingInput(InputStream input) { + super(input); + } + + @Override + public int read() throws IOException { + int value = in.read(); + if (value >= 0) { + position++; + if (prefix != null) { + prefix.write(value); + } + } + return value; + } + + @Override + public int read(byte[] bytes, int offset, int length) throws IOException { + int n = in.read(bytes, offset, length); + if (n > 0) { + position += n; + if (prefix != null) { + prefix.write(bytes, offset, n); + } + } + return n; + } + } + private static final class NoOpDatumReader<D> implements DatumReader<D> { @Override diff --git a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java index 61e387d580..03b70840b6 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java @@ -18,7 +18,6 @@ package org.apache.paimon.format.avro; -import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.IOUtils; @@ -31,6 +30,7 @@ import javax.annotation.Nullable; import java.io.Closeable; import java.io.IOException; +import java.io.InputStream; import java.util.Collections; /** @@ -45,7 +45,7 @@ public final class AvroBlockReader implements Closeable { private @Nullable AvroRawBlock borrowedRawBlock; - public AvroBlockReader(SeekableInputStream input) throws IOException { + public AvroBlockReader(InputStream input) throws IOException { try { this.reader = new RawBlockReader(input); } catch (IOException | RuntimeException | Error e) { @@ -60,7 +60,8 @@ public final class AvroBlockReader implements Closeable { } /** - * Returns the physical block offset; read immediately after {@link #nextBorrowedRawBlock()}. + * Returns the block offset relative to the initial input position; read immediately after + * {@link #nextBorrowedRawBlock()}. */ public long blockOffset() { return reader.blockOffset(); diff --git a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroBlockReaderTest.java b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroBlockReaderTest.java index c0faacd3eb..1aabaa0b12 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroBlockReaderTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroBlockReaderTest.java @@ -18,12 +18,6 @@ package org.apache.paimon.format.avro; -import org.apache.paimon.fs.ByteArraySeekableStream; -import org.apache.paimon.fs.Path; -import org.apache.paimon.fs.SeekableInputStream; -import org.apache.paimon.fs.SeekableInputStreamWrapper; -import org.apache.paimon.fs.local.LocalFileIO; - import org.apache.avro.Schema; import org.apache.avro.file.CodecFactory; import org.apache.avro.file.DataFileStream; @@ -31,17 +25,15 @@ import org.apache.avro.file.DataFileWriter; import org.apache.avro.generic.GenericDatumReader; import org.apache.avro.generic.GenericDatumWriter; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; +import java.io.FilterInputStream; import java.io.IOException; -import java.nio.file.Files; -import java.util.ArrayList; +import java.io.InputStream; import java.util.Arrays; -import java.util.List; import java.util.NoSuchElementException; import java.util.concurrent.atomic.AtomicBoolean; @@ -53,8 +45,6 @@ class AvroBlockReaderTest { private static final Schema SCHEMA = Schema.create(Schema.Type.LONG); - @TempDir private java.nio.file.Path tempDir; - @ParameterizedTest @ValueSource(strings = {"null", "deflate", "snappy", "zstandard"}) void blockMetadataMatchesWriterBoundaries(String codec) throws Exception { @@ -78,31 +68,19 @@ class AvroBlockReaderTest { assertThat(boundaries[values.length]).isEqualTo(bytes.length); for (int maxRead : new int[] {1, 7, Integer.MAX_VALUE}) { - List<Long> seeks = new ArrayList<>(); - SeekableInputStream input = - new SeekableInputStreamWrapper(open(bytes)) { + InputStream input = + new FilterInputStream(new ByteArrayInputStream(bytes)) { @Override public int read(byte[] data, int offset, int length) throws IOException { return super.read(data, offset, Math.min(length, maxRead)); } - - @Override - public void seek(long position) throws IOException { - seeks.add(position); - super.seek(position); - } }; try (AvroBlockReader reader = new AvroBlockReader(input)) { - long resumePosition = input.getPos(); - assertThat(seeks).containsExactly(0L, resumePosition); byte[] header = reader.headerBytes(); assertThat(header).isEqualTo(Arrays.copyOf(bytes, (int) boundaries[0])); - assertThat(input.getPos()).isEqualTo(resumePosition); - assertThat(seeks).containsExactly(0L, resumePosition); byte[] anotherHeader = reader.headerBytes(); anotherHeader[0] = 0; assertThat(reader.headerBytes()).isEqualTo(header); - assertThat(seeks).hasSize(2); AvroRawBlock previous = null; for (int i = 0; i < values.length; i++) { // Exercise next() both directly and after repeated look-ahead calls. @@ -136,7 +114,7 @@ class AvroBlockReaderTest { writer.create(SCHEMA, output); } byte[] bytes = output.toByteArray(); - try (AvroBlockReader reader = new AvroBlockReader(open(bytes))) { + try (AvroBlockReader reader = new AvroBlockReader(new ByteArrayInputStream(bytes))) { assertThat(reader.headerBytes()).isEqualTo(bytes); assertThat(reader.hasNextBlock()).isFalse(); assertThatThrownBy(reader::nextBorrowedRawBlock) @@ -145,46 +123,28 @@ class AvroBlockReaderTest { } @Test - void failedHeaderReadClosesTheInput() throws IOException { - ByteArrayOutputStream output = new ByteArrayOutputStream(); - try (DataFileWriter<Long> writer = new DataFileWriter<>(new GenericDatumWriter<>(SCHEMA))) { - writer.create(SCHEMA, output); - writer.append(11L); - } - byte[] bytes = output.toByteArray(); + void failedHeaderReadClosesTheInput() { AtomicBoolean closed = new AtomicBoolean(); - SeekableInputStream input = - new SeekableInputStreamWrapper(open(bytes)) { - private boolean readingHeader; - - @Override - public void seek(long position) throws IOException { - super.seek(position); - readingHeader = position == 0; - } - + InputStream input = + new InputStream() { @Override - public int read(byte[] data, int offset, int length) throws IOException { - if (readingHeader) { - throw new IOException("header read failed"); - } - return super.read(data, offset, length); + public int read() throws IOException { + throw new IOException("header read failed"); } @Override - public void close() throws IOException { - super.close(); + public void close() { closed.set(true); } }; assertThatThrownBy(() -> new AvroBlockReader(input)) .isInstanceOf(IOException.class) - .hasMessage("header read failed"); + .hasRootCauseMessage("header read failed"); assertThat(closed.get()).isTrue(); } @Test - void headerFromMemoryCanRestoreEof() throws IOException { + void headerExcludesBufferedBlockData() throws IOException { for (int records : new int[] {0, 1}) { ByteArrayOutputStream output = new ByteArrayOutputStream(); long headerLength; @@ -197,12 +157,12 @@ class AvroBlockReaderTest { } } byte[] bytes = output.toByteArray(); - ByteArraySeekableStream input = new ByteArraySeekableStream(bytes); + ByteArrayInputStream input = new ByteArrayInputStream(bytes); try (AvroBlockReader reader = new AvroBlockReader(input)) { - assertThat(input.getPos()).isEqualTo(bytes.length); + assertThat(input.available()).isZero(); byte[] header = reader.headerBytes(); assertThat(header).isEqualTo(Arrays.copyOf(bytes, (int) headerLength)); - assertThat(input.getPos()).isEqualTo(bytes.length); + assertThat(input.available()).isZero(); if (records > 0) { assertThat(reader.nextBorrowedRawBlock().recordCount()).isEqualTo(records); assertBlockReadable( @@ -230,27 +190,18 @@ class AvroBlockReaderTest { int prefixLength = 13; byte[] bytes = new byte[prefixLength + avro.length]; System.arraycopy(avro, 0, bytes, prefixLength, avro.length); - SeekableInputStream input = open(bytes); - input.seek(prefixLength); + InputStream input = new ByteArrayInputStream(bytes, prefixLength, avro.length); try (AvroBlockReader reader = new AvroBlockReader(input)) { - long resumePosition = input.getPos(); byte[] header = reader.headerBytes(); assertThat(header).isEqualTo(Arrays.copyOf(avro, (int) headerLength)); - assertThat(input.getPos()).isEqualTo(resumePosition); assertThat(reader.nextBorrowedRawBlock().recordCount()).isEqualTo(1); - assertThat(reader.blockOffset()).isEqualTo(prefixLength + headerLength); + assertThat(reader.blockOffset()).isEqualTo(headerLength); assertBlockReadable( - header, bytes, reader.blockOffset(), reader.blockLength(), new long[] {17L}); + header, avro, reader.blockOffset(), reader.blockLength(), new long[] {17L}); assertThat(reader.hasNextBlock()).isFalse(); } } - private SeekableInputStream open(byte[] bytes) throws IOException { - java.nio.file.Path file = Files.createTempFile(tempDir, "blocks-", ".avro"); - Files.write(file, bytes); - return LocalFileIO.create().newInputStream(new Path(file.toUri())); - } - private static void assertBlockReadable( byte[] header, byte[] file, long offset, long length, long[] expected) throws IOException {
