diff --git a/docs/docs/concepts/spec/rowformat.md b/docs/docs/concepts/spec/rowformat.md
index 04315cc66ce0..c28943256635 100644
--- a/docs/docs/concepts/spec/rowformat.md
+++ b/docs/docs/concepts/spec/rowformat.md
@@ -182,11 +182,12 @@ To read a row by its zero-based row number within the file:
 
 1. **Read Footer**: Seek to file end - 32 bytes, read the 32-byte footer. Validate magic number.
 2. **Read Block Index**: Seek to `indexOffset`, read `indexLength` bytes, decode the three arrays. Compute block offsets by prefix sum of `blockCompressedSizes[]`.
-3. **Select Block**: Find block `b` where `blockRowStarts[b] <= rowNum < blockEnd`. For the last block, `blockEnd` is `totalRowCount`; otherwise it is `blockRowStarts[b + 1]`.
-4. **Read Block**: Seek to `blockOffset(b)`, read `blockCompressedSizes[b]` bytes.
-5. **Decompress**: ZSTD decompress into a buffer of size `blockUncompressedSizes[b]`.
-6. **Locate Row**: Compute `localIdx = rowNum - blockRowStarts[b]`. Read `offsets[localIdx]` from the offset array at the end of the decompressed block.
-7. **Deserialize**: Read the row starting at the computed offset using the row serialization format.
+3. **Check Consistency**: The three arrays must have the same length, that length must equal `blockCount`, and `blockCompressedSizes[]` must sum to `indexOffset`, because the blocks are written contiguously from position 0 and the index follows the last one. A reader that bounds its block loop by one of the two — the footer's `blockCount` or the index array length — must reject a file where they disagree rather than silently reading fewer blocks.
+4. **Select Block**: Find block `b` where `blockRowStarts[b] <= rowNum < blockEnd`. For the last block, `blockEnd` is `totalRowCount`; otherwise it is `blockRowStarts[b + 1]`.
+5. **Read Block**: Seek to `blockOffset(b)`, read `blockCompressedSizes[b]` bytes.
+6. **Decompress**: ZSTD decompress into a buffer of size `blockUncompressedSizes[b]`.
+7. **Locate Row**: Compute `localIdx = rowNum - blockRowStarts[b]`. Read `offsets[localIdx]` from the offset array at the end of the decompressed block.
+8. **Deserialize**: Read the row starting at the computed offset using the row serialization format.
 
 ## Projection
 
diff --git a/paimon-format/src/main/java/org/apache/paimon/format/row/RowBlockIndex.java b/paimon-format/src/main/java/org/apache/paimon/format/row/RowBlockIndex.java
index 2987d28bf9c5..76c5b0048498 100644
--- a/paimon-format/src/main/java/org/apache/paimon/format/row/RowBlockIndex.java
+++ b/paimon-format/src/main/java/org/apache/paimon/format/row/RowBlockIndex.java
@@ -26,6 +26,8 @@
 
 import java.io.IOException;
 
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
 /** Block index that maps row numbers to block locations. */
 class RowBlockIndex {
 
@@ -36,12 +38,87 @@ class RowBlockIndex {
 
     RowBlockIndex(
             long[] blockCompressedSizes, long[] blockUncompressedSizes, long[] blockRowStarts) {
+        checkArgument(
+                blockCompressedSizes.length == blockUncompressedSizes.length
+                        && blockCompressedSizes.length == blockRowStarts.length,
+                "Row file block index arrays disagree on the block count: %s compressed sizes, %s uncompressed sizes, %s row starts.",
+                blockCompressedSizes.length,
+                blockUncompressedSizes.length,
+                blockRowStarts.length);
         this.blockCompressedSizes = blockCompressedSizes;
         this.blockUncompressedSizes = blockUncompressedSizes;
         this.blockRowStarts = blockRowStarts;
         this.blockOffsets = computeOffsets(blockCompressedSizes);
     }
 
+    /**
+     * Checks the index against the footer, which is the only place both are in hand. Blocks are
+     * written contiguously from position 0 and the index follows the last one, so the compressed
+     * sizes must sum to exactly {@code indexOffset} — see the row format spec. Row starts must
+     * cover every row exactly once, because {@code RowFormatReader} turns consecutive starts into
+     * the row range of a block and skips a block whose range a selection does not intersect: a
+     * first start past 0, a repeated start, or a last start at the row count would drop rows
+     * silently.
+     */
+    void validate(RowFileFooter footer) throws IOException {
+        if (blockCount() != footer.blockCount) {
+            throw new IOException(
+                    String.format(
+                            "Row file block index holds %d blocks, but the footer declares %d.",
+                            blockCount(), footer.blockCount));
+        }
+
+        long blocksEnd = 0;
+        for (int i = 0; i < blockCount(); i++) {
+            if (blockCompressedSizes[i] < 0) {
+                throw new IOException(
+                        String.format(
+                                "Row file block %d has a negative compressed size %d.",
+                                i, blockCompressedSizes[i]));
+            }
+            blocksEnd += blockCompressedSizes[i];
+        }
+        if (blocksEnd != footer.indexOffset) {
+            throw new IOException(
+                    String.format(
+                            "Row file blocks end at %d, but the footer puts the block index at %d.",
+                            blocksEnd, footer.indexOffset));
+        }
+
+        if (blockCount() == 0) {
+            if (footer.totalRowCount != 0) {
+                throw new IOException(
+                        String.format(
+                                "Row file block index is empty, but the footer declares %d rows.",
+                                footer.totalRowCount));
+            }
+            return;
+        }
+
+        if (blockRowStarts[0] != 0) {
+            throw new IOException(
+                    String.format(
+                            "Row file block 0 starts at row %d, so rows before it are unreachable.",
+                            blockRowStarts[0]));
+        }
+        for (int i = 1; i < blockCount(); i++) {
+            if (blockRowStarts[i] <= blockRowStarts[i - 1]) {
+                throw new IOException(
+                        String.format(
+                                "Row file block %d starts at row %d, not after block %d at row %d.",
+                                i, blockRowStarts[i], i - 1, blockRowStarts[i - 1]));
+            }
+        }
+        if (blockRowStarts[blockCount() - 1] >= footer.totalRowCount) {
+            throw new IOException(
+                    String.format(
+                            "Row file block %d starts at row %d, which the declared row count %d does not reach.",
+                            blockCount() - 1,
+                            blockRowStarts[blockCount() - 1],
+                            footer.totalRowCount));
+        }
+    }
+
     int blockCount() {
         return blockCompressedSizes.length;
     }
diff --git a/paimon-format/src/main/java/org/apache/paimon/format/row/RowFileFooter.java b/paimon-format/src/main/java/org/apache/paimon/format/row/RowFileFooter.java
index c6d0026d11f0..0f1426de7b8c 100644
--- a/paimon-format/src/main/java/org/apache/paimon/format/row/RowFileFooter.java
+++ b/paimon-format/src/main/java/org/apache/paimon/format/row/RowFileFooter.java
@@ -19,7 +19,6 @@
 package org.apache.paimon.format.row;
 
 import org.apache.paimon.fs.PositionOutputStream;
-import org.apache.paimon.fs.SeekableInputStream;
 
 import java.io.IOException;
 
@@ -54,13 +53,6 @@ void writeTo(PositionOutputStream out) throws IOException {
         out.write(buf);
     }
 
-    static RowFileFooter readFrom(SeekableInputStream in, long fileSize) throws IOException {
-        in.seek(fileSize - FOOTER_SIZE);
-        byte[] buf = new byte[FOOTER_SIZE];
-        readFully(in, buf);
-        return readFrom(buf, 0);
-    }
-
     static RowFileFooter readFrom(byte[] buf, int offset) throws IOException {
         int magic = readIntLE(buf, offset + 28);
         if (magic != MAGIC) {
@@ -82,14 +74,25 @@ static RowFileFooter readFrom(byte[] buf, int offset) throws IOException {
         return new RowFileFooter(totalRowCount, blockCount, indexOffset, indexLength);
     }
 
-    private static void readFully(SeekableInputStream in, byte[] buf) throws IOException {
-        int off = 0;
-        while (off < buf.length) {
-            int read = in.read(buf, off, buf.length - off);
-            if (read < 0) {
-                throw new IOException("Unexpected end of file");
-            }
-            off += read;
+    /**
+     * Checks that the block index lies inside the file and ahead of the footer. The offsets come
+     * from the file itself, and they size the buffer the index is read into.
+     */
+    void validate(long fileSize) throws IOException {
+        // written this way so that a huge indexOffset cannot overflow the comparison
+        if (indexOffset < 0
+                || indexLength < 0
+                || indexOffset > fileSize - FOOTER_SIZE - indexLength) {
+            throw new IOException(
+                    String.format(
+                            "Invalid row file block index location: offset %d, length %d, in a file of %d bytes.",
+                            indexOffset, indexLength, fileSize));
+        }
+        if (blockCount < 0) {
+            throw new IOException("Invalid row file block count: " + blockCount);
+        }
+        if (totalRowCount < 0) {
+            throw new IOException("Invalid row file row count: " + totalRowCount);
         }
     }
 
diff --git a/paimon-format/src/main/java/org/apache/paimon/format/row/RowFormatReaderFactory.java b/paimon-format/src/main/java/org/apache/paimon/format/row/RowFormatReaderFactory.java
index 476dffb9a41b..316499f3e6c1 100644
--- a/paimon-format/src/main/java/org/apache/paimon/format/row/RowFormatReaderFactory.java
+++ b/paimon-format/src/main/java/org/apache/paimon/format/row/RowFormatReaderFactory.java
@@ -55,6 +55,13 @@ public FileRecordReader<InternalRow> createReader(Context context) throws IOExce
         // before it parses lengths and offsets taken from the file itself, so a truncated or
         // corrupt file can throw anywhere in between and would otherwise leak the stream.
         try {
+            if (fileSize < RowFileFooter.FOOTER_SIZE) {
+                throw new IOException(
+                        String.format(
+                                "Row file %s holds %d bytes, too few for a %d-byte footer.",
+                                path, fileSize, RowFileFooter.FOOTER_SIZE));
+            }
+
             int tailSize = (int) Math.min(TAIL_PREFETCH_SIZE, fileSize);
             long tailOffset = fileSize - tailSize;
             in.seek(tailOffset);
@@ -63,6 +70,7 @@ public FileRecordReader<InternalRow> createReader(Context context) throws IOExce
 
             RowFileFooter footer =
                     RowFileFooter.readFrom(tailBuf, tailSize - RowFileFooter.FOOTER_SIZE);
+            footer.validate(fileSize);
 
             RowBlockIndex blockIndex;
             if (footer.indexOffset >= tailOffset) {
@@ -73,6 +81,7 @@ public FileRecordReader<InternalRow> createReader(Context context) throws IOExce
             } else {
                 blockIndex = RowBlockIndex.readFrom(in, footer.indexOffset, footer.indexLength);
             }
+            blockIndex.validate(footer);
 
             return new RowFormatReader(
                     in,
diff --git a/paimon-format/src/test/java/org/apache/paimon/format/row/RowFileIndexConsistencyTest.java b/paimon-format/src/test/java/org/apache/paimon/format/row/RowFileIndexConsistencyTest.java
new file mode 100644
index 000000000000..815394f6b7db
--- /dev/null
+++ b/paimon-format/src/test/java/org/apache/paimon/format/row/RowFileIndexConsistencyTest.java
@@ -0,0 +1,207 @@
+/*
+ * 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.row;
+
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.FormatReaderContext;
+import org.apache.paimon.format.FormatReaderFactory;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * The footer and the block index describe the same blocks twice, and until they were cross-checked
+ * only the index was consulted: {@code blockCount} was written into every row file and never read,
+ * so a Java reader bounded its block loop by the index while the Python reader bounded it by the
+ * footer. These tests pin the agreement the spec implies — the compressed sizes sum to {@code
+ * indexOffset}, and the block count matches.
+ */
+class RowFileIndexConsistencyTest {
+
+    private static final RowType ROW_TYPE = RowType.of(DataTypes.INT());
+
+    @TempDir java.nio.file.Path tempDir;
+
+    @Test
+    void testBlockIndexArraysMustAgreeOnTheBlockCount() {
+        assertThatThrownBy(
+                        () ->
+                                new RowBlockIndex(
+                                        new long[] {10, 20}, new long[] {100, 200}, new long[] {0}))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("2 compressed sizes")
+                .hasMessageContaining("1 row starts");
+    }
+
+    @Test
+    void testFooterBlockCountMustMatchTheIndex() throws Exception {
+        Path path = writeRowFile("block-count.row");
+        // footer blockCount is a little-endian int at footer offset 8
+        patch(path, footerOffset(path) + 8, intLE(99));
+
+        assertThatThrownBy(() -> openReader(path))
+                .isInstanceOf(IOException.class)
+                .hasMessageContaining("the footer declares 99");
+    }
+
+    @Test
+    void testCompressedSizesMustSumToTheIndexOffset() {
+        // two blocks of 10 and 20 compressed bytes occupy [0, 30), so the index starts at 30
+        RowBlockIndex index =
+                new RowBlockIndex(new long[] {10, 20}, new long[] {100, 200}, new long[] {0, 5});
+        assertThatCode(() -> index.validate(new RowFileFooter(9, 2, 30, 7)))
+                .doesNotThrowAnyException();
+        assertThatThrownBy(() -> index.validate(new RowFileFooter(9, 2, 31, 7)))
+                .isInstanceOf(IOException.class)
+                .hasMessageContaining("blocks end at 30")
+                .hasMessageContaining("block index at 31");
+    }
+
+    @Test
+    void testRowStartsMustCoverEveryRowExactlyOnce() {
+        // RowFormatReader turns consecutive starts into a block's row range and skips a block whose
+        // range the selection does not intersect, so each of these drops rows without an error
+        assertThatThrownBy(() -> validateRowStarts(new long[] {10, 20}, 30))
+                .isInstanceOf(IOException.class)
+                .hasMessageContaining("block 0 starts at row 10");
+        assertThatThrownBy(() -> validateRowStarts(new long[] {0, 0}, 30))
+                .isInstanceOf(IOException.class)
+                .hasMessageContaining("not after block 0 at row 0");
+        assertThatThrownBy(() -> validateRowStarts(new long[] {0, 5}, 5))
+                .isInstanceOf(IOException.class)
+                .hasMessageContaining("the declared row count 5 does not reach");
+        assertThatThrownBy(() -> validateRowStarts(new long[] {0}, 0))
+                .isInstanceOf(IOException.class)
+                .hasMessageContaining("the declared row count 0 does not reach");
+    }
+
+    @Test
+    void testAnEmptyIndexNeedsAnEmptyFile() {
+        RowBlockIndex empty = new RowBlockIndex(new long[0], new long[0], new long[0]);
+        assertThatCode(() -> empty.validate(new RowFileFooter(0, 0, 0, 7)))
+                .doesNotThrowAnyException();
+        assertThatThrownBy(() -> empty.validate(new RowFileFooter(7, 0, 0, 7)))
+                .isInstanceOf(IOException.class)
+                .hasMessageContaining("empty, but the footer declares 7 rows");
+    }
+
+    @Test
+    void testNegativeCompressedSizeIsRejected() {
+        // the sizes sum to the declared indexOffset only because the second cancels the first
+        RowBlockIndex index =
+                new RowBlockIndex(new long[] {200, -100}, new long[] {100, 200}, new long[] {0, 5});
+        assertThatThrownBy(() -> index.validate(new RowFileFooter(9, 2, 100, 7)))
+                .isInstanceOf(IOException.class)
+                .hasMessageContaining("block 1 has a negative compressed size -100");
+    }
+
+    private static void validateRowStarts(long[] rowStarts, long totalRowCount) throws IOException {
+        long[] sizes = new long[rowStarts.length];
+        Arrays.fill(sizes, 10);
+        new RowBlockIndex(sizes, sizes.clone(), rowStarts)
+                .validate(
+                        new RowFileFooter(totalRowCount, rowStarts.length, 10L * sizes.length, 7));
+    }
+
+    @Test
+    void testIndexOutsideTheFileIsRejected() throws Exception {
+        Path path = writeRowFile("index-outside.row");
+        patch(path, footerOffset(path) + 12, longLE(1L << 40));
+
+        assertThatThrownBy(() -> openReader(path))
+                .isInstanceOf(IOException.class)
+                .hasMessageContaining("Invalid row file block index location");
+    }
+
+    @Test
+    void testFileTooShortForAFooterIsRejected() throws Exception {
+        Path path = new Path(new Path(tempDir.toString()), "short.row");
+        try (PositionOutputStream out = new LocalFileIO().newOutputStream(path, false)) {
+            out.write(new byte[RowFileFooter.FOOTER_SIZE - 1]);
+        }
+
+        assertThatThrownBy(() -> openReader(path))
+                .isInstanceOf(IOException.class)
+                .hasMessageContaining("too few for a 32-byte footer");
+    }
+
+    private Path writeRowFile(String name) throws IOException {
+        Path path = new Path(new Path(tempDir.toString()), name);
+        LocalFileIO fileIO = new LocalFileIO();
+        FileFormat format = FileFormat.fromIdentifier("row", new Options());
+        try (PositionOutputStream out = fileIO.newOutputStream(path, false)) {
+            FormatWriter writer = format.createWriterFactory(ROW_TYPE).create(out, "zstd");
+            for (int i = 0; i < 1000; i++) {
+                writer.addElement(GenericRow.of(i));
+            }
+            writer.close();
+        }
+        assertThat(fileIO.getFileSize(path)).isGreaterThan(RowFileFooter.FOOTER_SIZE);
+        return path;
+    }
+
+    private void openReader(Path path) throws IOException {
+        LocalFileIO fileIO = new LocalFileIO();
+        FormatReaderFactory readerFactory =
+                FileFormat.fromIdentifier("row", new Options())
+                        .createReaderFactory(ROW_TYPE, ROW_TYPE, new ArrayList<>());
+        readerFactory.createReader(
+                new FormatReaderContext(fileIO, path, fileIO.getFileSize(path), null, null));
+    }
+
+    private long footerOffset(Path path) throws IOException {
+        return new LocalFileIO().getFileSize(path) - RowFileFooter.FOOTER_SIZE;
+    }
+
+    private void patch(Path path, long offset, byte[] bytes) throws IOException {
+        java.nio.file.Path file = java.nio.file.Paths.get(path.toUri().getPath());
+        byte[] all = Files.readAllBytes(file);
+        System.arraycopy(bytes, 0, all, (int) offset, bytes.length);
+        Files.write(file, all);
+    }
+
+    private static byte[] intLE(int value) {
+        byte[] buf = new byte[4];
+        RowFileFooter.writeIntLE(buf, 0, value);
+        return buf;
+    }
+
+    private static byte[] longLE(long value) {
+        byte[] buf = new byte[8];
+        RowFileFooter.writeLongLE(buf, 0, value);
+        return buf;
+    }
+}
diff --git a/paimon-python/pypaimon/read/reader/format_row_reader.py b/paimon-python/pypaimon/read/reader/format_row_reader.py
index 5b0bed51f49d..cca6369e3cd0 100644
--- a/paimon-python/pypaimon/read/reader/format_row_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_row_reader.py
@@ -158,6 +158,9 @@ def close(self):
         pass
 
     def _read_metadata(self):
+        if self._file_size < FOOTER_SIZE:
+            raise IOError(f"Invalid row file: {self._file_size} bytes hold no {FOOTER_SIZE}-byte footer")
+
         with self._file_io.new_input_stream(self._file_path) as f:
             f.seek(self._file_size - FOOTER_SIZE)
             footer_bytes = f.read(FOOTER_SIZE)
@@ -175,9 +178,21 @@ def _read_metadata(self):
 
         self._total_row_count = struct.unpack_from('<q', footer_bytes, 0)[0]
         self._block_count = struct.unpack_from('<i', footer_bytes, 8)[0]
-        index_offset = struct.unpack_from('<q', footer_bytes, 12)[0]
+        self._index_offset = struct.unpack_from('<q', footer_bytes, 12)[0]
+        index_offset = self._index_offset
         index_length = struct.unpack_from('<i', footer_bytes, 20)[0]
 
+        # the block index has to lie inside the file and ahead of the footer, and these are the
+        # numbers the seek and the read below are given
+        if (index_offset < 0 or index_length < 0
+                or index_offset > self._file_size - FOOTER_SIZE - index_length):
+            raise IOError(f"Invalid row file block index location: offset {index_offset}, "
+                          f"length {index_length}, in a file of {self._file_size} bytes")
+        if self._block_count < 0:
+            raise IOError(f"Invalid row file block count: {self._block_count}")
+        if self._total_row_count < 0:
+            raise IOError(f"Invalid row file row count: {self._total_row_count}")
+
         with self._file_io.new_input_stream(self._file_path) as f:
             f.seek(index_offset)
             index_bytes = f.read(index_length)
@@ -192,17 +207,22 @@ def _parse_block_index(self, index_data: bytes):
 
         len1, consumed = _decode_var_int(index_data, pos)
         pos += consumed
-        self._block_compressed_sizes = DeltaVarintCompressor.decompress(index_data[pos:pos + len1])
+        self._block_compressed_sizes = DeltaVarintCompressor.decompress(
+            self._chunk(index_data, pos, len1))
         pos += len1
 
         len2, consumed = _decode_var_int(index_data, pos)
         pos += consumed
-        self._block_uncompressed_sizes = DeltaVarintCompressor.decompress(index_data[pos:pos + len2])
+        self._block_uncompressed_sizes = DeltaVarintCompressor.decompress(
+            self._chunk(index_data, pos, len2))
         pos += len2
 
         len3, consumed = _decode_var_int(index_data, pos)
         pos += consumed
-        self._block_row_starts = DeltaVarintCompressor.decompress(index_data[pos:pos + len3])
+        self._block_row_starts = DeltaVarintCompressor.decompress(
+            self._chunk(index_data, pos, len3))
+
+        self._validate_block_index()
 
         offset = 0
         self._block_offsets = []
@@ -210,6 +230,60 @@ def _parse_block_index(self, index_data: bytes):
             self._block_offsets.append(offset)
             offset += size
 
+    @staticmethod
+    def _chunk(index_data: bytes, pos: int, length: int) -> bytes:
+        # slicing clamps, and DeltaVarintCompressor.decompress stops at a truncated varint without
+        # raising, so an out-of-range length would silently shorten one array
+        if length < 0 or pos + length > len(index_data):
+            raise IOError(f"Invalid row file block index: a {length}-byte array at offset {pos} "
+                          f"does not fit in {len(index_data)} bytes")
+        return index_data[pos:pos + length]
+
+    def _validate_block_index(self):
+        """Cross-check the index against the footer, as the row format spec requires.
+
+        Blocks are written contiguously from position 0 and the index follows the last one, so the
+        compressed sizes sum to exactly index_offset. Row starts become the row range of a block,
+        and a block whose range a selection does not intersect is skipped, so a first start past 0,
+        a repeated start or a last start at the row count would drop rows silently.
+        """
+        counts = (len(self._block_compressed_sizes), len(self._block_uncompressed_sizes),
+                  len(self._block_row_starts))
+        if len(set(counts)) != 1:
+            raise IOError(f"Row file block index arrays disagree on the block count: "
+                          f"{counts[0]} compressed sizes, {counts[1]} uncompressed sizes, "
+                          f"{counts[2]} row starts")
+        if counts[0] != self._block_count:
+            raise IOError(f"Row file block index holds {counts[0]} blocks, but the footer "
+                          f"declares {self._block_count}")
+
+        blocks_end = 0
+        for i, size in enumerate(self._block_compressed_sizes):
+            if size < 0:
+                raise IOError(f"Row file block {i} has a negative compressed size {size}")
+            blocks_end += size
+        if blocks_end != self._index_offset:
+            raise IOError(f"Row file blocks end at {blocks_end}, but the footer puts the "
+                          f"block index at {self._index_offset}")
+
+        if self._block_count == 0:
+            if self._total_row_count != 0:
+                raise IOError(f"Row file block index is empty, but the footer declares "
+                              f"{self._total_row_count} rows")
+            return
+
+        if self._block_row_starts[0] != 0:
+            raise IOError(f"Row file block 0 starts at row {self._block_row_starts[0]}, "
+                          f"so rows before it are unreachable")
+        for i in range(1, self._block_count):
+            if self._block_row_starts[i] <= self._block_row_starts[i - 1]:
+                raise IOError(f"Row file block {i} starts at row {self._block_row_starts[i]}, "
+                              f"not after block {i - 1} at row {self._block_row_starts[i - 1]}")
+        if self._block_row_starts[-1] >= self._total_row_count:
+            raise IOError(f"Row file block {self._block_count - 1} starts at row "
+                          f"{self._block_row_starts[-1]}, which the declared row count "
+                          f"{self._total_row_count} does not reach")
+
     def _read_and_decompress_block(self, block_idx: int) -> bytes:
         import zstandard as zstd
 
diff --git a/paimon-python/pypaimon/tests/test_format_row_reader_writer.py b/paimon-python/pypaimon/tests/test_format_row_reader_writer.py
index cff237b5be0d..b14f7a2b538a 100644
--- a/paimon-python/pypaimon/tests/test_format_row_reader_writer.py
+++ b/paimon-python/pypaimon/tests/test_format_row_reader_writer.py
@@ -16,13 +16,14 @@
 # under the License.
 
 import os
+import struct
 import tempfile
 from decimal import Decimal
 
 import pyarrow as pa
 import pytest
 
-from pypaimon.read.reader.format_row_reader import FormatRowReader
+from pypaimon.read.reader.format_row_reader import FOOTER_SIZE, FormatRowReader
 from pypaimon.schema.data_types import (
     ArrayType, AtomicType, DataField, MapType, RowType
 )
@@ -532,3 +533,118 @@ def test_data_evolution_row_id_read(self):
             ]
         finally:
             shutil.rmtree(tempdir, ignore_errors=True)
+
+
+class TestRowFileIndexConsistency:
+    """The footer and the block index describe the same blocks twice.
+
+    Until they were cross-checked, this reader bounded its block loop by the footer's block_count
+    while the Java reader bounded it by the index array length, so a file the two disagreed about
+    was read differently by each. These pin the agreement the row format spec requires.
+    """
+
+    FIELDS = [DataField(0, "id", AtomicType("INT"))]
+
+    def _write(self, path, rows=1000):
+        data = pa.table({"id": pa.array(list(range(rows)), type=pa.int32())})
+        _write_row_file(path, self.FIELDS, data)
+
+    def _patch(self, path, offset_from_footer, packed):
+        size = os.path.getsize(path)
+        with open(path, 'r+b') as f:
+            f.seek(size - FOOTER_SIZE + offset_from_footer)
+            f.write(packed)
+
+    def test_footer_block_count_must_match_the_index(self):
+        with tempfile.NamedTemporaryFile(suffix=".row", delete=False) as tmp:
+            path = tmp.name
+        try:
+            self._write(path)
+            # block_count is a little-endian int at footer offset 8; zero used to make a full scan
+            # return nothing while a row-id read still returned rows
+            self._patch(path, 8, struct.pack('<i', 0))
+            with pytest.raises(IOError, match="the footer declares 0"):
+                _read_row_file(path, self.FIELDS)
+        finally:
+            os.unlink(path)
+
+    def test_index_outside_the_file_is_rejected(self):
+        with tempfile.NamedTemporaryFile(suffix=".row", delete=False) as tmp:
+            path = tmp.name
+        try:
+            self._write(path)
+            self._patch(path, 12, struct.pack('<q', 1 << 40))
+            with pytest.raises(IOError, match="Invalid row file block index location"):
+                _read_row_file(path, self.FIELDS)
+        finally:
+            os.unlink(path)
+
+    def test_file_too_short_for_a_footer_is_rejected(self):
+        with tempfile.NamedTemporaryFile(suffix=".row", delete=False) as tmp:
+            path = tmp.name
+        try:
+            with open(path, 'wb') as f:
+                f.write(b'\0' * (FOOTER_SIZE - 1))
+            with pytest.raises(IOError, match="hold no 32-byte footer"):
+                _read_row_file(path, self.FIELDS)
+        finally:
+            os.unlink(path)
+
+    def test_row_starts_must_cover_every_row_exactly_once(self):
+        reader = self._reader_with(compressed=[10, 20], row_starts=[10, 20], total_rows=30)
+        with pytest.raises(IOError, match="block 0 starts at row 10"):
+            reader._validate_block_index()
+
+        reader = self._reader_with(compressed=[10, 20], row_starts=[0, 0], total_rows=30)
+        with pytest.raises(IOError, match="not after block 0 at row 0"):
+            reader._validate_block_index()
+
+        reader = self._reader_with(compressed=[10, 20], row_starts=[0, 5], total_rows=5)
+        with pytest.raises(IOError, match="row count 5 does not reach"):
+            reader._validate_block_index()
+
+    def test_compressed_sizes_must_sum_to_the_index_offset(self):
+        # two blocks of 10 and 20 compressed bytes occupy [0, 30), so the index starts at 30
+        reader = self._reader_with(compressed=[10, 20], row_starts=[0, 5], total_rows=30,
+                                   index_offset=30)
+        reader._validate_block_index()
+
+        reader = self._reader_with(compressed=[10, 20], row_starts=[0, 5], total_rows=30,
+                                   index_offset=31)
+        with pytest.raises(IOError, match="blocks end at 30"):
+            reader._validate_block_index()
+
+        # the sum has to be checked for an empty index too, where it is the only thing left
+        reader = self._reader_with(compressed=[], row_starts=[], total_rows=0, index_offset=7)
+        with pytest.raises(IOError, match="blocks end at 0"):
+            reader._validate_block_index()
+
+    def test_negative_compressed_size_is_rejected(self):
+        # the sizes sum to the declared index_offset only because the second cancels the first
+        reader = self._reader_with(compressed=[200, -100], row_starts=[0, 5], total_rows=30,
+                                   index_offset=100)
+        with pytest.raises(IOError, match="block 1 has a negative compressed size -100"):
+            reader._validate_block_index()
+
+    def test_arrays_must_agree_on_the_block_count(self):
+        reader = self._reader_with(compressed=[10, 20], row_starts=[0], total_rows=30)
+        with pytest.raises(IOError, match="2 compressed sizes"):
+            reader._validate_block_index()
+
+    def test_an_empty_index_needs_an_empty_file(self):
+        reader = self._reader_with(compressed=[], row_starts=[], total_rows=0)
+        reader._validate_block_index()
+
+        reader = self._reader_with(compressed=[], row_starts=[], total_rows=7)
+        with pytest.raises(IOError, match="empty, but the footer declares 7 rows"):
+            reader._validate_block_index()
+
+    def _reader_with(self, compressed, row_starts, total_rows, index_offset=None):
+        reader = FormatRowReader.__new__(FormatRowReader)
+        reader._block_compressed_sizes = compressed
+        reader._block_uncompressed_sizes = [100] * len(compressed)
+        reader._block_row_starts = row_starts
+        reader._block_count = len(compressed)
+        reader._total_row_count = total_rows
+        reader._index_offset = sum(compressed) if index_offset is None else index_offset
+        return reader