[core] Expose Avro header and physical block locations

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;

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");
     }
 }

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);
@@ -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();

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,18 +1130,26 @@ 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();
                     rowCount++;
                 }
                 blockCount++;
             }
+            assertThat(nextOffset).isEqualTo(bytes.length);
         }
 
         assertThat(blockCount).isGreaterThan(1);

paimon-format/src/main/java/org/apache/avro/file/RawBlockReader.java

@@ -18,26 +18,77 @@
 
 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.IOException;
-import java.io.InputStream;
+import java.util.NoSuchElementException;
 
 /** Package bridge exposing Avro's compressed blocks without reflection. */
 public final class RawBlockReader extends DataFileStream<Void> {
 
-    public RawBlockReader(InputStream input) throws IOException {
+    private final SeekableInputStream input;
+    private final byte[] headerBytes;
+    private long blockOffset;
+    private long blockLength;
+    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<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);
+    }
+
+    /** Returns a copy of the complete OCF header, including schema, codec and 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.getPos() - 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());

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) {
@@ -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());