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 8c182bf0db [python] Fix RecursionError when reading a shard or slice
of a large file (#10113)
8c182bf0db is described below
commit 8c182bf0db768541ea987a7684addfa5378cb735
Author: jackylee <[email protected]>
AuthorDate: Thu Sep 24 13:09:28 2026 +0800
[python] Fix RecursionError when reading a shard or slice of a large file
(#10113)
---
.../pypaimon/read/reader/shard_batch_reader.py | 13 ++-
.../pypaimon/tests/shard_batch_reader_test.py | 94 ++++++++++++++++++++++
2 files changed, 103 insertions(+), 4 deletions(-)
diff --git a/paimon-python/pypaimon/read/reader/shard_batch_reader.py
b/paimon-python/pypaimon/read/reader/shard_batch_reader.py
index aa00814b82..445ff5f802 100644
--- a/paimon-python/pypaimon/read/reader/shard_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/shard_batch_reader.py
@@ -38,8 +38,14 @@ class ShardBatchReader(RecordBatchReader):
if isinstance(self.reader.format_reader, FormatBlobReader):
# For blob reader, pass begin_idx and end_idx parameters
return self.reader.read_arrow_batch(start_idx=self.start_pos,
end_idx=self.end_pos)
- else:
- # For non-blob reader (DataFileBatchReader), use standard
read_arrow_batch
+
+ # For non-blob reader (DataFileBatchReader), use standard
read_arrow_batch.
+ # Loop rather than recurse over skipped batches: a slice/shard whose
range
+ # sits deep in a file (default parquet batch_size is 1024 rows) skips
one
+ # batch per step, so recursing here overflows the stack
(RecursionError)
+ # once the skipped count exceeds the interpreter limit. Mirrors the
+ # while-loop skip pattern in ConcatBatchReader /
ApplyDeletionVectorReader.
+ while True:
batch = self.reader.read_arrow_batch()
if batch is None:
@@ -56,8 +62,7 @@ class ShardBatchReader(RecordBatchReader):
return batch.slice(self.start_pos - batch_begin, self.end_pos
- self.start_pos)
elif batch_begin < self.end_pos < self.current_pos: # batch ends
after the desired range
return batch.slice(0, self.end_pos - batch_begin)
- else: # batch is outside the desired range
- return self.read_arrow_batch()
+ # else: batch is outside the desired range -> read the next one
(loop)
def close(self):
self.reader.close()
diff --git a/paimon-python/pypaimon/tests/shard_batch_reader_test.py
b/paimon-python/pypaimon/tests/shard_batch_reader_test.py
new file mode 100644
index 0000000000..7dc4badbc2
--- /dev/null
+++ b/paimon-python/pypaimon/tests/shard_batch_reader_test.py
@@ -0,0 +1,94 @@
+# 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.
+
+import unittest
+
+import pyarrow as pa
+
+from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
+from pypaimon.read.reader.shard_batch_reader import ShardBatchReader
+
+
+class _BatchReader(RecordBatchReader):
+ """A non-blob reader that replays an explicit list of arrow batches."""
+
+ format_reader = None # not a FormatBlobReader -> ShardBatchReader takes
the row-range path
+
+ def __init__(self, batches):
+ self._batches = iter(batches)
+
+ def read_arrow_batch(self):
+ return next(self._batches, None)
+
+ def close(self):
+ pass
+
+
+def _single_row_batches(count):
+ return [pa.record_batch([pa.array([i])], names=["id"]) for i in
range(count)]
+
+
+def _read_all(reader):
+ got = []
+ while True:
+ batch = reader.read_arrow_batch()
+ if batch is None:
+ break
+ got.extend(batch.column("id").to_pylist())
+ return got
+
+
+class ShardBatchReaderTest(unittest.TestCase):
+
+ def test_slice_deep_into_file_does_not_recurse(self):
+ # A slice/shard whose range sits many batches into a file must not
recurse
+ # once per skipped batch. With the default parquet batch_size of 1024
rows a
+ # slice starting ~1M rows in skips >1000 batches; recursing there
overflowed
+ # the stack with RecursionError. 2000 single-row batches reproduce
that.
+ batch_count = 2000
+ reader = ShardBatchReader(
+ _BatchReader(_single_row_batches(batch_count)), batch_count - 1,
batch_count)
+
+ batch = reader.read_arrow_batch()
+
+ self.assertIsNotNone(batch)
+ self.assertEqual(batch.column("id").to_pylist(), [batch_count - 1])
+ self.assertIsNone(reader.read_arrow_batch())
+
+ def test_slice_returns_only_rows_in_range(self):
+ # Semantics guard: with single-row batches, slice [2, 5) yields rows
2, 3, 4
+ # and nothing else, so the loop refactor preserves the range filtering.
+ reader = ShardBatchReader(_BatchReader(_single_row_batches(8)), 2, 5)
+
+ self.assertEqual(_read_all(reader), [2, 3, 4])
+
+ def test_slice_straddling_batch_boundaries(self):
+ # Multi-row batches so the two slice() branches are exercised: the
first
+ # batch straddles start_pos (2 in [0,4)) and the last straddles end_pos
+ # (9 in [8,12)); slice [2, 9) must yield exactly rows 2..8.
+ batches = [
+ pa.record_batch([pa.array([0, 1, 2, 3])], names=["id"]),
+ pa.record_batch([pa.array([4, 5, 6, 7])], names=["id"]),
+ pa.record_batch([pa.array([8, 9, 10, 11])], names=["id"]),
+ ]
+ reader = ShardBatchReader(_BatchReader(batches), 2, 9)
+
+ self.assertEqual(_read_all(reader), [2, 3, 4, 5, 6, 7, 8])
+
+
+if __name__ == "__main__":
+ unittest.main()
