[format] Cross-check the row-file block index against its footer

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
 

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,62 @@ 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 index the
+     * arrays of every later lookup and size the per-block selection array.
+     */
+    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 =
+                blockCount() == 0
+                        ? 0
+                        : blockOffset(blockCount() - 1) + blockCompressedSize(blockCount() - 1);
+        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));
+        }
+
+        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, before block %d at row %d.",
+                                i, blockRowStarts[i], i - 1, blockRowStarts[i - 1]));
+            }
+        }
+        if (blockCount() > 0 && blockRowStarts[blockCount() - 1] > footer.totalRowCount) {
+            throw new IOException(
+                    String.format(
+                            "Row file block %d starts at row %d, past the declared row count %d.",
+                            blockCount() - 1,
+                            blockRowStarts[blockCount() - 1],
+                            footer.totalRowCount));
+        }
+    }
+
     int blockCount() {
         return blockCompressedSizes.length;
     }

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,24 @@ 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 {
+        if (indexOffset < 0
+                || indexLength < 0
+                || indexOffset + indexLength > fileSize - FOOTER_SIZE) {
+            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);
         }
     }
 

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,

paimon-format/src/test/java/org/apache/paimon/format/row/RowFileIndexConsistencyTest.java

@@ -0,0 +1,169 @@
+/*
+ * 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 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 testRowStartsMustNotGoBackwards() {
+        RowBlockIndex index =
+                new RowBlockIndex(new long[] {10, 20}, new long[] {100, 200}, new long[] {5, 0});
+        assertThatThrownBy(() -> index.validate(new RowFileFooter(9, 2, 30, 7)))
+                .isInstanceOf(IOException.class)
+                .hasMessageContaining("before block 0 at row 5");
+    }
+
+    @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;
+    }
+}