This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 4bb54123ac [format] Cross-check the row-file block index against its
footer (#10001)
4bb54123ac is described below
commit 4bb54123ac51455c2aed61538248149cae3a392a
Author: jackylee <[email protected]>
AuthorDate: Mon Sep 21 11:27:06 2026 +0800
[format] Cross-check the row-file block index against its footer (#10001)
---
docs/docs/concepts/spec/rowformat.md | 11 +-
.../apache/paimon/format/row/RowBlockIndex.java | 77 ++++++++
.../apache/paimon/format/row/RowFileFooter.java | 35 ++--
.../paimon/format/row/RowFormatReaderFactory.java | 9 +
.../format/row/RowFileIndexConsistencyTest.java | 207 +++++++++++++++++++++
.../pypaimon/read/reader/format_row_reader.py | 82 +++++++-
.../tests/test_format_row_reader_writer.py | 118 +++++++++++-
7 files changed, 513 insertions(+), 26 deletions(-)
diff --git a/docs/docs/concepts/spec/rowformat.md
b/docs/docs/concepts/spec/rowformat.md
index 04315cc66c..c289432566 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 2987d28bf9..76c5b00484 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 org.apache.paimon.utils.VarLengthIntUtils;
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 c6d0026d11..0f1426de7b 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 @@ class RowFileFooter {
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 @@ class RowFileFooter {
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 476dffb9a4..316499f3e6 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 class RowFormatReaderFactory implements
FormatReaderFactory {
// 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 class RowFormatReaderFactory implements
FormatReaderFactory {
RowFileFooter footer =
RowFileFooter.readFrom(tailBuf, tailSize -
RowFileFooter.FOOTER_SIZE);
+ footer.validate(fileSize);
RowBlockIndex blockIndex;
if (footer.indexOffset >= tailOffset) {
@@ -73,6 +81,7 @@ public class RowFormatReaderFactory implements
FormatReaderFactory {
} 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 0000000000..815394f6b7
--- /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 5b0bed51f4..cca6369e3c 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 @@ class FormatRowReader(RecordBatchReader):
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 @@ class FormatRowReader(RecordBatchReader):
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 @@ class FormatRowReader(RecordBatchReader):
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 @@ class FormatRowReader(RecordBatchReader):
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 cff237b5be..b14f7a2b53 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 @@ class TestFormatRowReaderWriter:
]
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
