From ce927bc2d858db1b732d46ffa4880a70ad06886c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Tue, 15 Sep 2026 15:02:27 +0800 Subject: [PATCH 1/4] [core] Expose Avro header and physical block locations --- .../paimon/manifest/ManifestAvroReader.java | 15 ++ .../paimon/manifest/ManifestFileTest.java | 13 +- .../org/apache/avro/file/RawBlockReader.java | 86 ++++++++++- .../paimon/format/avro/AvroBlockReader.java | 17 +++ .../format/avro/AvroBlockReaderTest.java | 138 ++++++++++++++++++ 5 files changed, 265 insertions(+), 4 deletions(-) create mode 100644 paimon-format/src/test/java/org/apache/paimon/format/avro/AvroBlockReaderTest.java 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 995834c37e7e..1a4dbbb47189 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 @@ -70,6 +70,21 @@ public final class ManifestAvroReader implements AutoCloseable { } } + /** Returns a copy of the complete OCF header, including schema, codec and sync marker. */ + public byte[] headerBytes() { + return blockReader.headerBytes(); + } + + /** Returns the physical block offset; read immediately after {@link #next()}. */ + public long blockOffset() { + return blockReader.blockOffset(); + } + + /** Returns the last-read block's encoded length, including its header and sync marker. */ + public long blockLength() { + return blockReader.blockLength(); + } + /** Returns whether another raw Avro block is available. */ public boolean hasNext() throws IOException { return blockReader.hasNextBlock(); 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 d8eab09df774..b5ff0fb00249 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 @@ -58,6 +58,7 @@ import java.io.IOException; import java.io.UncheckedIOException; import java.nio.ByteBuffer; +import java.nio.file.Files; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -1129,11 +1130,18 @@ void testBlockReaderReadsAcrossMultipleBlocks() throws Exception { ProjectedManifestEntry.Projection projection = projection(DataFileMeta.FILE_NAME); int blockCount = 0; int rowCount = 0; + byte[] bytes = Files.readAllBytes(tempDir.resolve("manifest").resolve(manifest.fileName())); try (ManifestAvroReader reader = openManifestReader(manifest)) { + byte[] header = reader.headerBytes(); + assertThat(header).isEqualTo(Arrays.copyOf(bytes, header.length)); + long nextOffset = header.length; while (reader.hasNext()) { - ManifestAvroReader.RowIterator rows = - reader.next().toRows(projection.projectedType()); + ManifestAvroReader.RawBlock block = reader.next(); + assertThat(reader.blockOffset()).isEqualTo(nextOffset); + assertThat(reader.blockLength()).isPositive(); + nextOffset += reader.blockLength(); + ManifestAvroReader.RowIterator rows = block.toRows(projection.projectedType()); assertThat(rows.hasNext()).isTrue(); while (rows.hasNext()) { rows.next(); @@ -1141,6 +1149,7 @@ void testBlockReaderReadsAcrossMultipleBlocks() throws Exception { } blockCount++; } + assertThat(nextOffset).isEqualTo(bytes.length); } assertThat(blockCount).isGreaterThan(1); 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 43a68c54a44d..637a4a925f8f 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 @@ -22,27 +22,109 @@ 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 { + private final CountingInput input; + private final byte[] headerBytes; + private long blockOffset; + private long blockLength; + private boolean pending; + public RawBlockReader(InputStream input) throws IOException { + this(new CountingInput(input)); + } + + private RawBlockReader(CountingInput input) throws IOException { super(input, new NoOpDatumReader()); + this.input = input; + long length = position(); + this.headerBytes = Arrays.copyOf(input.prefix.toByteArray(), (int) length); + input.prefix = null; + } + + /** Returns a copy of the complete OCF header, including its sync marker. */ + public byte[] headerBytes() { + return headerBytes.clone(); + } + + /** + * Returns the physical block offset; read immediately after {@link #nextRawBlock(RawBlock)}. + */ + public long blockOffset() { + return blockOffset; + } + + /** Returns the last-read block's encoded length, including its header and sync marker. */ + public long blockLength() { + return blockLength; } - public boolean hasNextRawBlock() { - return super.hasNextBlock(); + private long position() throws IOException { + // This is the same read-ahead adjustment used by DataFileReader.blockFinished(). + return input.position - vin.inputStream().available(); + } + + public boolean hasNextRawBlock() throws IOException { + if (!pending) { + blockOffset = position(); + pending = super.hasNextBlock(); + } + return pending; } public RawBlock nextRawBlock(RawBlock reuse) throws IOException { + if (!hasNextRawBlock()) { + throw new NoSuchElementException(); + } DataBlock raw = super.nextRawBlock(reuse == null ? null : reuse.dataBlock()); + blockLength = position() - blockOffset; + pending = false; return reuse == null ? new RawBlock(raw, resolveCodec(), getSchema()) : 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 implements DatumReader { @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 c4359afcda43..15260bbea298 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 @@ -54,6 +54,23 @@ public AvroBlockReader(InputStream input) throws IOException { } } + /** Returns a copy of the complete OCF header, including schema, codec and sync marker. */ + public byte[] headerBytes() { + return reader.headerBytes(); + } + + /** + * Returns the physical block offset; read immediately after {@link #nextBorrowedRawBlock()}. + */ + public long blockOffset() { + return reader.blockOffset(); + } + + /** Returns the last-read block's encoded length, including its header and sync marker. */ + public long blockLength() { + return reader.blockLength(); + } + /** Creates a record decoder from the writer schema stored in the Avro file header. */ public AvroRecordDecoder createRecordDecoder() { return new AvroRecordDecoder(reader.getSchema()); 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 new file mode 100644 index 000000000000..c4b9cc3aacbf --- /dev/null +++ b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroBlockReaderTest.java @@ -0,0 +1,138 @@ +/* + * 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.format.avro; + +import org.apache.avro.Schema; +import org.apache.avro.file.CodecFactory; +import org.apache.avro.file.DataFileStream; +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.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.Arrays; +import java.util.NoSuchElementException; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for physical Avro block metadata. */ +class AvroBlockReaderTest { + + private static final Schema SCHEMA = Schema.create(Schema.Type.LONG); + + @ParameterizedTest + @ValueSource(strings = {"null", "deflate", "snappy", "zstandard"}) + void blockMetadataMatchesWriterBoundaries(String codec) throws Exception { + long[][] values = {{0L, 1L, Long.MAX_VALUE}, {100L}, {1000L, 1001L}}; + long[] boundaries = new long[values.length + 1]; + ByteArrayOutputStream output = new ByteArrayOutputStream(); + try (DataFileWriter writer = new DataFileWriter<>(new GenericDatumWriter<>(SCHEMA))) { + writer.setCodec(CodecFactory.fromString(codec)); + // Exercise headers larger than the decoder's read-ahead buffer. + writer.setMeta("test.padding", new byte[20_000]); + writer.create(SCHEMA, output); + boundaries[0] = writer.sync(); + for (int i = 0; i < values.length; i++) { + for (long value : values[i]) { + writer.append(value); + } + boundaries[i + 1] = writer.sync(); + } + } + byte[] bytes = output.toByteArray(); + assertThat(boundaries[values.length]).isEqualTo(bytes.length); + + for (int maxRead : new int[] {1, 7, Integer.MAX_VALUE}) { + ByteArrayInputStream input = + new ByteArrayInputStream(bytes) { + @Override + public synchronized int read(byte[] data, int offset, int length) { + return super.read(data, offset, Math.min(length, maxRead)); + } + }; + try (AvroBlockReader reader = new AvroBlockReader(input)) { + byte[] header = reader.headerBytes(); + assertThat(header).isEqualTo(Arrays.copyOf(bytes, (int) boundaries[0])); + byte[] anotherHeader = reader.headerBytes(); + anotherHeader[0] = 0; + assertThat(reader.headerBytes()).isEqualTo(header); + AvroRawBlock previous = null; + for (int i = 0; i < values.length; i++) { + // Exercise next() both directly and after repeated look-ahead calls. + if (i > 0) { + assertThat(reader.hasNextBlock()).isTrue(); + assertThat(reader.hasNextBlock()).isTrue(); + } + AvroRawBlock block = reader.nextBorrowedRawBlock(); + if (previous != null) { + assertThat(block).isSameAs(previous); + } + previous = block; + assertThat(block.recordCount()).isEqualTo(values[i].length); + assertThat(reader.blockOffset()).isEqualTo(boundaries[i]); + assertThat(reader.blockLength()).isEqualTo(boundaries[i + 1] - boundaries[i]); + assertBlockReadable( + header, bytes, reader.blockOffset(), reader.blockLength(), values[i]); + } + assertThat(reader.hasNextBlock()).isFalse(); + assertThat(reader.hasNextBlock()).isFalse(); + assertThatThrownBy(reader::nextBorrowedRawBlock) + .isInstanceOf(NoSuchElementException.class); + } + } + } + + @Test + void emptyFileContainsOnlyTheHeader() throws IOException { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + try (DataFileWriter writer = new DataFileWriter<>(new GenericDatumWriter<>(SCHEMA))) { + writer.create(SCHEMA, output); + } + byte[] bytes = output.toByteArray(); + try (AvroBlockReader reader = new AvroBlockReader(new ByteArrayInputStream(bytes))) { + assertThat(reader.headerBytes()).isEqualTo(bytes); + assertThat(reader.hasNextBlock()).isFalse(); + assertThatThrownBy(reader::nextBorrowedRawBlock) + .isInstanceOf(NoSuchElementException.class); + } + } + + private static void assertBlockReadable( + byte[] header, byte[] file, long offset, long length, long[] expected) + throws IOException { + ByteArrayOutputStream selected = new ByteArrayOutputStream(); + selected.write(header); + selected.write(file, (int) offset, (int) length); + try (DataFileStream reader = + new DataFileStream<>( + new ByteArrayInputStream(selected.toByteArray()), + new GenericDatumReader<>())) { + for (long value : expected) { + assertThat(reader.next()).isEqualTo(value); + } + assertThat(reader.hasNext()).isFalse(); + } + } +} From dc6031fc8051bae0281864fc0366f2f869f542ec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Tue, 15 Sep 2026 15:18:35 +0800 Subject: [PATCH 2/4] [core] Use seekable streams for Avro block metadata --- .../paimon/manifest/ManifestAvroReader.java | 6 +- .../org/apache/avro/file/RawBlockReader.java | 75 ++++++----------- .../paimon/format/avro/AvroBlockReader.java | 6 +- .../format/avro/AvroBlockReaderTest.java | 83 ++++++++++++++++++- 4 files changed, 110 insertions(+), 60 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 1a4dbbb47189..76b5dd9c51fd 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,6 +26,7 @@ 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; @@ -34,7 +35,6 @@ 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(InputStream input) throws IOException { + ManifestAvroReader(SeekableInputStream input) throws IOException { AvroBlockReader blockReader = null; try { blockReader = new AvroBlockReader(input); @@ -71,7 +71,7 @@ public final class ManifestAvroReader implements AutoCloseable { } /** Returns a copy of the complete OCF header, including schema, codec and sync marker. */ - public byte[] headerBytes() { + public byte[] headerBytes() throws IOException { return blockReader.headerBytes(); } 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 637a4a925f8f..a72c50d97e86 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,40 +18,48 @@ 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 javax.annotation.Nullable; + 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 { - private final CountingInput input; - private final byte[] headerBytes; + private final SeekableInputStream input; + private final long headerLength; + @Nullable private byte[] headerBytes; private long blockOffset; private long blockLength; private boolean pending; - public RawBlockReader(InputStream input) throws IOException { - this(new CountingInput(input)); - } - - private RawBlockReader(CountingInput input) throws IOException { + public RawBlockReader(SeekableInputStream input) throws IOException { super(input, new NoOpDatumReader()); this.input = input; - long length = position(); - this.headerBytes = Arrays.copyOf(input.prefix.toByteArray(), (int) length); - input.prefix = null; + this.headerLength = position(); } - /** Returns a copy of the complete OCF header, including its sync marker. */ - public byte[] headerBytes() { + /** Returns a copy of the complete OCF header, reading and caching it on first access. */ + public byte[] headerBytes() throws IOException { + if (headerBytes == null) { + byte[] bytes = new byte[Math.toIntExact(headerLength)]; + long resumePosition = input.getPos(); + try { + input.seek(0); + IOUtils.readFully(input, bytes); + } finally { + // Preserve the position past any bytes already buffered by the Avro decoder. + input.seek(resumePosition); + } + headerBytes = bytes; + } return headerBytes.clone(); } @@ -69,7 +77,7 @@ public long blockLength() { private long position() throws IOException { // This is the same read-ahead adjustment used by DataFileReader.blockFinished(). - return input.position - vin.inputStream().available(); + return input.getPos() - vin.inputStream().available(); } public boolean hasNextRawBlock() throws IOException { @@ -92,39 +100,6 @@ public RawBlock nextRawBlock(RawBlock reuse) throws IOException { : 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 implements DatumReader { @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 15260bbea298..7c3a79ae5cbd 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,6 +18,7 @@ package org.apache.paimon.format.avro; +import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.IOUtils; @@ -30,7 +31,6 @@ 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(InputStream input) throws IOException { + public AvroBlockReader(SeekableInputStream input) throws IOException { try { this.reader = new RawBlockReader(input); } catch (IOException | RuntimeException | Error e) { @@ -55,7 +55,7 @@ public AvroBlockReader(InputStream input) throws IOException { } /** Returns a copy of the complete OCF header, including schema, codec and sync marker. */ - public byte[] headerBytes() { + public byte[] headerBytes() throws IOException { return reader.headerBytes(); } 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 c4b9cc3aacbf..268cb8ed2fde 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,6 +18,11 @@ package org.apache.paimon.format.avro; +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; @@ -25,14 +30,19 @@ 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.IOException; +import java.nio.file.Files; +import java.util.ArrayList; import java.util.Arrays; +import java.util.List; import java.util.NoSuchElementException; +import java.util.concurrent.atomic.AtomicBoolean; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -42,6 +52,8 @@ 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 { @@ -65,19 +77,31 @@ void blockMetadataMatchesWriterBoundaries(String codec) throws Exception { assertThat(boundaries[values.length]).isEqualTo(bytes.length); for (int maxRead : new int[] {1, 7, Integer.MAX_VALUE}) { - ByteArrayInputStream input = - new ByteArrayInputStream(bytes) { + List seeks = new ArrayList<>(); + SeekableInputStream input = + new SeekableInputStreamWrapper(open(bytes)) { @Override - public synchronized int read(byte[] data, int offset, int length) { + 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)) { + assertThat(seeks).isEmpty(); + long resumePosition = input.getPos(); 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. @@ -111,7 +135,7 @@ void emptyFileContainsOnlyTheHeader() throws IOException { writer.create(SCHEMA, output); } byte[] bytes = output.toByteArray(); - try (AvroBlockReader reader = new AvroBlockReader(new ByteArrayInputStream(bytes))) { + try (AvroBlockReader reader = new AvroBlockReader(open(bytes))) { assertThat(reader.headerBytes()).isEqualTo(bytes); assertThat(reader.hasNextBlock()).isFalse(); assertThatThrownBy(reader::nextBorrowedRawBlock) @@ -119,6 +143,57 @@ void emptyFileContainsOnlyTheHeader() throws IOException { } } + @Test + void failedHeaderReadRestoresThePositionAndCanBeRetried() throws IOException { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + long headerLength; + long secondBlockOffset; + try (DataFileWriter writer = new DataFileWriter<>(new GenericDatumWriter<>(SCHEMA))) { + writer.create(SCHEMA, output); + headerLength = writer.sync(); + writer.append(11L); + secondBlockOffset = writer.sync(); + writer.append(22L); + } + byte[] bytes = output.toByteArray(); + AtomicBoolean failRead = new AtomicBoolean(); + SeekableInputStream input = + new SeekableInputStreamWrapper(open(bytes)) { + @Override + public int read(byte[] data, int offset, int length) throws IOException { + if (failRead.getAndSet(false)) { + throw new IOException("header read failed"); + } + return super.read(data, offset, length); + } + }; + try (AvroBlockReader reader = new AvroBlockReader(input)) { + reader.nextBorrowedRawBlock(); + assertThat(reader.hasNextBlock()).isTrue(); + long resumePosition = input.getPos(); + failRead.set(true); + assertThatThrownBy(reader::headerBytes) + .isInstanceOf(IOException.class) + .hasMessage("header read failed"); + assertThat(input.getPos()).isEqualTo(resumePosition); + + byte[] header = reader.headerBytes(); + assertThat(header).isEqualTo(Arrays.copyOf(bytes, (int) headerLength)); + assertThat(input.getPos()).isEqualTo(resumePosition); + assertThat(reader.nextBorrowedRawBlock().recordCount()).isEqualTo(1); + assertThat(reader.blockOffset()).isEqualTo(secondBlockOffset); + assertBlockReadable( + header, bytes, reader.blockOffset(), reader.blockLength(), new long[] {22L}); + 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 { From 6f0286086278f4b9a582ecf9e637ab72ff180081 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Tue, 15 Sep 2026 15:29:02 +0800 Subject: [PATCH 3/4] [core] Fix Avro header reads at stream boundaries --- .../paimon/fs/ByteArraySeekableStream.java | 2 +- .../fs/ByteArraySeekableStreamTest.java | 17 ++++- .../org/apache/avro/file/RawBlockReader.java | 10 ++- .../format/avro/AvroBlockReaderTest.java | 63 +++++++++++++++++++ 4 files changed, 87 insertions(+), 5 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/ByteArraySeekableStream.java b/paimon-common/src/main/java/org/apache/paimon/fs/ByteArraySeekableStream.java index d6536927b100..2b83cd5b4636 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/ByteArraySeekableStream.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/ByteArraySeekableStream.java @@ -92,7 +92,7 @@ public ByteArrayStream(byte[] buf) { } public void seek(int position) throws IOException { - if (position >= count) { + if (position > count) { throw new EOFException("Can't seek position: " + position + ", length is " + count); } pos = position; diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/ByteArraySeekableStreamTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/ByteArraySeekableStreamTest.java index 2df02e85db49..31d37a6a2f43 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/ByteArraySeekableStreamTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/fs/ByteArraySeekableStreamTest.java @@ -65,12 +65,25 @@ public void testBasic() throws IOException { } } + @Test + public void testSeekToEnd() throws IOException { + for (int length : new int[] {0, 10}) { + try (ByteArraySeekableStream stream = + new ByteArraySeekableStream(randomBytes(length))) { + stream.seek(length); + Assertions.assertThat(stream.getPos()).isEqualTo(length); + Assertions.assertThat(stream.available()).isZero(); + Assertions.assertThat(stream.read()).isEqualTo(-1); + } + } + } + @Test public void testThrow() { int bl = 10; byte[] b = randomBytes(bl); ByteArraySeekableStream byteArraySeekableStream = new ByteArraySeekableStream(b); - Assertions.assertThatCode(() -> byteArraySeekableStream.seek(10)) - .hasMessage("Can't seek position: 10, length is 10"); + Assertions.assertThatCode(() -> byteArraySeekableStream.seek(11)) + .hasMessage("Can't seek position: 11, length is 10"); } } 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 a72c50d97e86..356c894ec87b 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 @@ -34,6 +34,7 @@ public final class RawBlockReader extends DataFileStream { private final SeekableInputStream input; + private final long headerOffset; private final long headerLength; @Nullable private byte[] headerBytes; private long blockOffset; @@ -41,9 +42,14 @@ public final class RawBlockReader extends DataFileStream { private boolean pending; public RawBlockReader(SeekableInputStream input) throws IOException { + this(input, input.getPos()); + } + + private RawBlockReader(SeekableInputStream input, long headerOffset) throws IOException { super(input, new NoOpDatumReader()); this.input = input; - this.headerLength = position(); + this.headerOffset = headerOffset; + this.headerLength = position() - headerOffset; } /** Returns a copy of the complete OCF header, reading and caching it on first access. */ @@ -52,7 +58,7 @@ public byte[] headerBytes() throws IOException { byte[] bytes = new byte[Math.toIntExact(headerLength)]; long resumePosition = input.getPos(); try { - input.seek(0); + input.seek(headerOffset); IOUtils.readFully(input, bytes); } finally { // Preserve the position past any bytes already buffered by the Avro decoder. 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 268cb8ed2fde..98d047b0f8f1 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,6 +18,7 @@ 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; @@ -188,6 +189,68 @@ public int read(byte[] data, int offset, int length) throws IOException { } } + @Test + void headerFromMemoryCanRestoreEof() throws IOException { + for (int records : new int[] {0, 1}) { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + long headerLength; + try (DataFileWriter writer = + new DataFileWriter<>(new GenericDatumWriter<>(SCHEMA))) { + writer.create(SCHEMA, output); + headerLength = writer.sync(); + if (records > 0) { + writer.append(17L); + } + } + byte[] bytes = output.toByteArray(); + ByteArraySeekableStream input = new ByteArraySeekableStream(bytes); + try (AvroBlockReader reader = new AvroBlockReader(input)) { + assertThat(input.getPos()).isEqualTo(bytes.length); + byte[] header = reader.headerBytes(); + assertThat(header).isEqualTo(Arrays.copyOf(bytes, (int) headerLength)); + assertThat(input.getPos()).isEqualTo(bytes.length); + if (records > 0) { + assertThat(reader.nextBorrowedRawBlock().recordCount()).isEqualTo(records); + assertBlockReadable( + header, + bytes, + reader.blockOffset(), + reader.blockLength(), + new long[] {17L}); + } + assertThat(reader.hasNextBlock()).isFalse(); + } + } + } + + @Test + void headerStartsAtInitialStreamPosition() throws IOException { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + long headerLength; + try (DataFileWriter writer = new DataFileWriter<>(new GenericDatumWriter<>(SCHEMA))) { + writer.create(SCHEMA, output); + headerLength = writer.sync(); + writer.append(17L); + } + byte[] avro = output.toByteArray(); + 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); + 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); + assertBlockReadable( + header, bytes, 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); From f9c7b53bce7038d148b4db747acff775e15b7339 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Tue, 15 Sep 2026 15:40:30 +0800 Subject: [PATCH 4/4] [core] Capture Avro header bytes during construction --- .../paimon/manifest/ManifestAvroReader.java | 2 +- .../org/apache/avro/file/RawBlockReader.java | 30 ++++------- .../paimon/format/avro/AvroBlockReader.java | 2 +- .../format/avro/AvroBlockReaderTest.java | 50 ++++++++----------- 4 files changed, 33 insertions(+), 51 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 76b5dd9c51fd..4e84c690ee97 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 @@ -71,7 +71,7 @@ public final class ManifestAvroReader implements AutoCloseable { } /** Returns a copy of the complete OCF header, including schema, codec and sync marker. */ - public byte[] headerBytes() throws IOException { + public byte[] headerBytes() { return blockReader.headerBytes(); } 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 356c894ec87b..60ae9b195abb 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 @@ -25,8 +25,6 @@ import org.apache.avro.io.DatumReader; import org.apache.avro.io.Decoder; -import javax.annotation.Nullable; - import java.io.IOException; import java.util.NoSuchElementException; @@ -34,9 +32,7 @@ public final class RawBlockReader extends DataFileStream { private final SeekableInputStream input; - private final long headerOffset; - private final long headerLength; - @Nullable private byte[] headerBytes; + private final byte[] headerBytes; private long blockOffset; private long blockLength; private boolean pending; @@ -48,24 +44,16 @@ public RawBlockReader(SeekableInputStream input) throws IOException { private RawBlockReader(SeekableInputStream input, long headerOffset) throws IOException { super(input, new NoOpDatumReader()); this.input = input; - this.headerOffset = headerOffset; - this.headerLength = position() - headerOffset; + 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); } - /** Returns a copy of the complete OCF header, reading and caching it on first access. */ - public byte[] headerBytes() throws IOException { - if (headerBytes == null) { - byte[] bytes = new byte[Math.toIntExact(headerLength)]; - long resumePosition = input.getPos(); - try { - input.seek(headerOffset); - IOUtils.readFully(input, bytes); - } finally { - // Preserve the position past any bytes already buffered by the Avro decoder. - input.seek(resumePosition); - } - headerBytes = bytes; - } + /** Returns a copy of the complete OCF header, including schema, codec and sync marker. */ + public byte[] headerBytes() { return headerBytes.clone(); } 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 7c3a79ae5cbd..61e387d580c8 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 @@ -55,7 +55,7 @@ public AvroBlockReader(SeekableInputStream input) throws IOException { } /** Returns a copy of the complete OCF header, including schema, codec and sync marker. */ - public byte[] headerBytes() throws IOException { + public byte[] headerBytes() { return reader.headerBytes(); } 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 98d047b0f8f1..c0faacd3eb8b 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 @@ -93,8 +93,8 @@ public void seek(long position) throws IOException { } }; try (AvroBlockReader reader = new AvroBlockReader(input)) { - assertThat(seeks).isEmpty(); 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); @@ -145,48 +145,42 @@ void emptyFileContainsOnlyTheHeader() throws IOException { } @Test - void failedHeaderReadRestoresThePositionAndCanBeRetried() throws IOException { + void failedHeaderReadClosesTheInput() throws IOException { ByteArrayOutputStream output = new ByteArrayOutputStream(); - long headerLength; - long secondBlockOffset; try (DataFileWriter writer = new DataFileWriter<>(new GenericDatumWriter<>(SCHEMA))) { writer.create(SCHEMA, output); - headerLength = writer.sync(); writer.append(11L); - secondBlockOffset = writer.sync(); - writer.append(22L); } byte[] bytes = output.toByteArray(); - AtomicBoolean failRead = new AtomicBoolean(); + 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; + } + @Override public int read(byte[] data, int offset, int length) throws IOException { - if (failRead.getAndSet(false)) { + if (readingHeader) { throw new IOException("header read failed"); } return super.read(data, offset, length); } - }; - try (AvroBlockReader reader = new AvroBlockReader(input)) { - reader.nextBorrowedRawBlock(); - assertThat(reader.hasNextBlock()).isTrue(); - long resumePosition = input.getPos(); - failRead.set(true); - assertThatThrownBy(reader::headerBytes) - .isInstanceOf(IOException.class) - .hasMessage("header read failed"); - assertThat(input.getPos()).isEqualTo(resumePosition); - byte[] header = reader.headerBytes(); - assertThat(header).isEqualTo(Arrays.copyOf(bytes, (int) headerLength)); - assertThat(input.getPos()).isEqualTo(resumePosition); - assertThat(reader.nextBorrowedRawBlock().recordCount()).isEqualTo(1); - assertThat(reader.blockOffset()).isEqualTo(secondBlockOffset); - assertBlockReadable( - header, bytes, reader.blockOffset(), reader.blockLength(), new long[] {22L}); - assertThat(reader.hasNextBlock()).isFalse(); - } + @Override + public void close() throws IOException { + super.close(); + closed.set(true); + } + }; + assertThatThrownBy(() -> new AvroBlockReader(input)) + .isInstanceOf(IOException.class) + .hasMessage("header read failed"); + assertThat(closed.get()).isTrue(); } @Test