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 c64ca86c6e [format][python] State every block index invariant in the
row format spec (#10069)
c64ca86c6e is described below
commit c64ca86c6ebcb62c22812bd9a4aba829afe591e3
Author: jackylee <[email protected]>
AuthorDate: Tue Sep 22 10:43:07 2026 +0800
[format][python] State every block index invariant in the row format spec
(#10069)
---
docs/docs/concepts/spec/rowformat.md | 16 +++++++++++++++-
.../java/org/apache/paimon/format/row/RowBlockIndex.java | 10 ++++++++++
.../paimon/format/row/RowFileIndexConsistencyTest.java | 10 ++++++++++
paimon-python/pypaimon/read/reader/format_row_reader.py | 13 ++++++++++---
.../pypaimon/tests/test_format_row_reader_writer.py | 12 ++++++++++++
5 files changed, 57 insertions(+), 4 deletions(-)
diff --git a/docs/docs/concepts/spec/rowformat.md
b/docs/docs/concepts/spec/rowformat.md
index c289432566..698e5deee2 100644
--- a/docs/docs/concepts/spec/rowformat.md
+++ b/docs/docs/concepts/spec/rowformat.md
@@ -159,6 +159,20 @@ The arrays are:
- **blockUncompressedSizes**: Uncompressed size of each block (needed to
allocate decompression buffer)
- **blockRowStarts**: Cumulative row count at the start of each block (for
binary search)
+A well-formed index satisfies all of the following, and a reader rejects a
file that does not:
+
+- The three arrays have the same length, and that length equals the footer's
`blockCount`.
+- `blockCompressedSizes[]` sums to exactly `indexOffset`. Blocks are written
contiguously from
+ position 0 and the index follows the last one, so any other sum means the
two disagree about
+ where the blocks end.
+- No size is negative. The sum alone does not imply this: two sizes can cancel.
+- `blockRowStarts[0]` is 0, and each later start is strictly greater than the
one before it. A
+ reader turns consecutive starts into a block's row range, so a first start
past 0 leaves the rows
+ before it unreachable and a repeated start gives a block an empty range.
+- `blockRowStarts[last]` is less than the footer's `totalRowCount`, so the
last block holds at
+ least one row.
+- An empty index has `blockCount` 0 and `totalRowCount` 0.
+
## Footer
The footer occupies the last 32 bytes of the file. Offsets below are relative
to the start of
@@ -182,7 +196,7 @@ 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. **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.
+3. **Check Consistency**: Verify the block index against the footer, as
described under Block Index. 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]`.
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 76c5b00484..4e24a37d0d 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
@@ -85,6 +85,16 @@ class RowBlockIndex {
blocksEnd, footer.indexOffset));
}
+ for (int i = 0; i < blockCount(); i++) {
+ // nothing in the footer bounds this one, and it sizes the
decompression buffer
+ if (blockUncompressedSizes[i] < 0) {
+ throw new IOException(
+ String.format(
+ "Row file block %d has a negative uncompressed
size %d.",
+ i, blockUncompressedSizes[i]));
+ }
+ }
+
if (blockCount() == 0) {
if (footer.totalRowCount != 0) {
throw new IOException(
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
index 815394f6b7..ffc54f0cdc 100644
---
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
@@ -128,6 +128,16 @@ class RowFileIndexConsistencyTest {
.hasMessageContaining("block 1 has a negative compressed size
-100");
}
+ @Test
+ void testNegativeUncompressedSizeIsRejected() {
+ // the footer bounds the compressed sizes through the sum, but nothing
bounds these
+ RowBlockIndex index =
+ new RowBlockIndex(new long[] {10, 20}, new long[] {100, -1},
new long[] {0, 5});
+ assertThatThrownBy(() -> index.validate(new RowFileFooter(9, 2, 30,
7)))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("block 1 has a negative uncompressed
size -1");
+ }
+
private static void validateRowStarts(long[] rowStarts, long
totalRowCount) throws IOException {
long[] sizes = new long[rowStarts.length];
Arrays.fill(sizes, 10);
diff --git a/paimon-python/pypaimon/read/reader/format_row_reader.py
b/paimon-python/pypaimon/read/reader/format_row_reader.py
index cca6369e3c..56ddd15fc0 100644
--- a/paimon-python/pypaimon/read/reader/format_row_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_row_reader.py
@@ -243,9 +243,11 @@ class FormatRowReader(RecordBatchReader):
"""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.
+ compressed sizes sum to exactly index_offset. Row starts have to start
at 0 and increase
+ strictly because _compute_blocks_for_indices bisects them: a first
start past 0 makes
+ bisect_right return 0 for the rows before it, so block_idx is -1,
row_starts[-1] is the last
+ start, and the local row goes negative — this reader would decode a
row from the wrong place
+ in the block rather than skip it.
"""
counts = (len(self._block_compressed_sizes),
len(self._block_uncompressed_sizes),
len(self._block_row_starts))
@@ -266,6 +268,11 @@ class FormatRowReader(RecordBatchReader):
raise IOError(f"Row file blocks end at {blocks_end}, but the
footer puts the "
f"block index at {self._index_offset}")
+ for i, size in enumerate(self._block_uncompressed_sizes):
+ # nothing in the footer bounds this one, and it sizes the
decompression buffer
+ if size < 0:
+ raise IOError(f"Row file block {i} has a negative uncompressed
size {size}")
+
if self._block_count == 0:
if self._total_row_count != 0:
raise IOError(f"Row file block index is empty, but the footer
declares "
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 b14f7a2b53..2023dd3fdd 100644
--- a/paimon-python/pypaimon/tests/test_format_row_reader_writer.py
+++ b/paimon-python/pypaimon/tests/test_format_row_reader_writer.py
@@ -603,6 +603,18 @@ class TestRowFileIndexConsistency:
with pytest.raises(IOError, match="row count 5 does not reach"):
reader._validate_block_index()
+ # one block and no declared rows: the block would hold nothing
+ reader = self._reader_with(compressed=[10], row_starts=[0],
total_rows=0)
+ with pytest.raises(IOError, match="row count 0 does not reach"):
+ reader._validate_block_index()
+
+ def test_negative_uncompressed_size_is_rejected(self):
+ # the footer bounds the compressed sizes through the sum, but nothing
bounds these
+ reader = self._reader_with(compressed=[10, 20], row_starts=[0, 5],
total_rows=30)
+ reader._block_uncompressed_sizes = [100, -1]
+ with pytest.raises(IOError, match="block 1 has a negative uncompressed
size -1"):
+ 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,
