You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/04/20 01:44:53 UTC
(paimon) branch master updated: [python] Fix blob read OOM by wrapping lazy stream in BlobRef (#7672)
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 05c2d3df0a [python] Fix blob read OOM by wrapping lazy stream in BlobRef (#7672)
05c2d3df0a is described below
commit 05c2d3df0a23e3751b3c892a84a679d226a78c92
Author: XiaoHongbo <xi...@alibaba-inc.com>
AuthorDate: Mon Apr 20 09:44:48 2026 +0800
[python] Fix blob read OOM by wrapping lazy stream in BlobRef (#7672)
---
paimon-python/pypaimon/table/row/blob.py | 88 ++++++++++++++++---
paimon-python/pypaimon/tests/blob_test.py | 140 ++++++++++++++++++++++++++++++
2 files changed, 216 insertions(+), 12 deletions(-)
diff --git a/paimon-python/pypaimon/table/row/blob.py b/paimon-python/pypaimon/table/row/blob.py
index 738198537b..484c7c32e9 100644
--- a/paimon-python/pypaimon/table/row/blob.py
+++ b/paimon-python/pypaimon/table/row/blob.py
@@ -19,7 +19,7 @@
import io
import struct
from abc import ABC, abstractmethod
-from typing import Optional, Union
+from typing import BinaryIO, Optional, Union
from urllib.parse import urlparse
from pypaimon.common.uri_reader import UriReader, FileUriReader
@@ -163,6 +163,72 @@ class BlobDescriptor:
return self.__str__()
+class OffsetInputStream(io.RawIOBase):
+
+ def __init__(self, wrapped, offset: int, length: int):
+ self._wrapped = wrapped
+ self._offset = offset
+ self._length = length
+ if offset != 0:
+ wrapped.seek(offset)
+
+ def readable(self) -> bool:
+ return True
+
+ def seekable(self) -> bool:
+ return True
+
+ def readinto(self, b):
+ if self._length != -1:
+ remaining = self._length - self.tell()
+ if remaining <= 0:
+ return 0
+ if len(b) > remaining:
+ b = memoryview(b)[:remaining]
+ n = self._wrapped.readinto(b)
+ return n if n is not None else 0
+
+ def read(self, size=-1):
+ if size is None:
+ size = -1
+ if self._length != -1:
+ remaining = self._length - self.tell()
+ if remaining <= 0:
+ return b''
+ if size < 0 or size > remaining:
+ size = remaining
+ if size < 0:
+ return self._wrapped.read()
+ return self._wrapped.read(size)
+
+ def seek(self, pos, whence=io.SEEK_SET):
+ if whence == io.SEEK_SET:
+ if pos < 0:
+ raise ValueError(f"Negative seek position: {pos}")
+ target = self._offset + pos
+ elif whence == io.SEEK_CUR:
+ target = self._wrapped.tell() + pos
+ target = max(target, self._offset)
+ elif whence == io.SEEK_END:
+ if self._length != -1:
+ target = self._offset + self._length + pos
+ else:
+ end = self._wrapped.seek(0, io.SEEK_END)
+ target = max(end + pos, self._offset)
+ target = max(target, self._offset)
+ else:
+ raise ValueError(f"Invalid whence: {whence}")
+ return self._wrapped.seek(target) - self._offset
+
+ def tell(self) -> int:
+ return self._wrapped.tell() - self._offset
+
+ def close(self):
+ if not self.closed:
+ self._wrapped.close()
+ super().close()
+
+
class Blob(ABC):
@abstractmethod
@@ -174,7 +240,7 @@ class Blob(ABC):
pass
@abstractmethod
- def new_input_stream(self) -> io.BytesIO:
+ def new_input_stream(self) -> BinaryIO:
pass
@staticmethod
@@ -236,7 +302,7 @@ class BlobData(Blob):
def to_descriptor(self) -> 'BlobDescriptor':
raise RuntimeError("Blob data can not convert to descriptor.")
- def new_input_stream(self) -> io.BytesIO:
+ def new_input_stream(self) -> BinaryIO:
return io.BytesIO(self._data)
def __eq__(self, other) -> bool:
@@ -264,18 +330,16 @@ class BlobRef(Blob):
def to_descriptor(self) -> BlobDescriptor:
return self._descriptor
- def new_input_stream(self) -> io.BytesIO:
+ def new_input_stream(self) -> BinaryIO:
uri = self._descriptor.uri
offset = self._descriptor.offset
length = self._descriptor.length
- with self._uri_reader.new_input_stream(uri) as input_stream:
- if offset > 0:
- input_stream.seek(offset)
- if length == -1:
- data = input_stream.read()
- else:
- data = input_stream.read(length)
- return io.BytesIO(data)
+ stream = self._uri_reader.new_input_stream(uri)
+ try:
+ return OffsetInputStream(stream, offset, length)
+ except Exception:
+ stream.close()
+ raise
def __eq__(self, other) -> bool:
if not isinstance(other, BlobRef):
diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py
index 0d2f4f4c89..cd951eedef 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -15,6 +15,7 @@ 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 io
import os
import shutil
import struct
@@ -1158,5 +1159,144 @@ class BlobEndToEndTest(unittest.TestCase):
reader_content.close()
+class OffsetInputStreamTest(unittest.TestCase):
+
+ def setUp(self):
+ self.test_data = bytes(range(20))
+ self.wrapped = io.BytesIO(self.test_data)
+
+ def test_constructor(self):
+ from pypaimon.table.row.blob import OffsetInputStream
+ stream = OffsetInputStream(io.BytesIO(self.test_data), 5, 10)
+ self.assertEqual(stream.tell(), 0)
+ stream.close()
+
+ def test_get_pos_and_seek(self):
+ from pypaimon.table.row.blob import OffsetInputStream
+ stream = OffsetInputStream(io.BytesIO(self.test_data), 5, 10)
+ stream.seek(3)
+ self.assertEqual(stream.tell(), 3)
+ stream.seek(10)
+ self.assertEqual(stream.tell(), 10)
+ stream.seek(0)
+ self.assertEqual(stream.tell(), 0)
+ stream.close()
+
+ def test_read_single_byte(self):
+ from pypaimon.table.row.blob import OffsetInputStream
+ stream = OffsetInputStream(io.BytesIO(self.test_data), 5, 10)
+ data = stream.read(1)
+ self.assertEqual(data[0], self.test_data[5])
+ self.assertEqual(stream.tell(), 1)
+ stream.seek(9)
+ data = stream.read(1)
+ self.assertEqual(data[0], self.test_data[14])
+ self.assertEqual(stream.tell(), 10)
+ stream.close()
+
+ def test_read_single_byte_at_end(self):
+ from pypaimon.table.row.blob import OffsetInputStream
+ stream = OffsetInputStream(io.BytesIO(self.test_data), 5, 10)
+ stream.seek(10)
+ data = stream.read(1)
+ self.assertEqual(data, b'')
+ stream.close()
+
+ def test_read_byte_array(self):
+ from pypaimon.table.row.blob import OffsetInputStream
+ stream = OffsetInputStream(io.BytesIO(self.test_data), 5, 10)
+ data = stream.read(5)
+ self.assertEqual(data, self.test_data[5:10])
+ self.assertEqual(stream.tell(), 5)
+ stream.close()
+
+ def test_read_byte_array_hitting_end(self):
+ from pypaimon.table.row.blob import OffsetInputStream
+ stream = OffsetInputStream(io.BytesIO(self.test_data), 5, 10)
+ stream.seek(7)
+ data = stream.read(5)
+ self.assertEqual(len(data), 3)
+ self.assertEqual(data, self.test_data[12:15])
+ self.assertEqual(stream.tell(), 10)
+ stream.close()
+
+ def test_read_byte_array_at_end(self):
+ from pypaimon.table.row.blob import OffsetInputStream
+ stream = OffsetInputStream(io.BytesIO(self.test_data), 5, 10)
+ stream.seek(10)
+ data = stream.read(5)
+ self.assertEqual(data, b'')
+ stream.close()
+
+ def test_read_with_unlimited_length(self):
+ from pypaimon.table.row.blob import OffsetInputStream
+ stream = OffsetInputStream(io.BytesIO(self.test_data), 5, -1)
+ data = stream.read(10)
+ self.assertEqual(data, self.test_data[5:15])
+ self.assertEqual(stream.tell(), 10)
+ stream.close()
+
+ def test_seek_set_negative_raises(self):
+ from pypaimon.table.row.blob import OffsetInputStream
+ stream = OffsetInputStream(io.BytesIO(self.test_data), 5, 10)
+ with self.assertRaises(ValueError):
+ stream.seek(-1, io.SEEK_SET)
+ stream.close()
+
+ def test_seek_cur_underflow_clamps_to_zero(self):
+ from pypaimon.table.row.blob import OffsetInputStream
+ stream = OffsetInputStream(io.BytesIO(self.test_data), 5, 10)
+ stream.seek(2)
+ stream.seek(-5, io.SEEK_CUR)
+ self.assertEqual(stream.tell(), 0)
+ data = stream.read(1)
+ self.assertEqual(data[0], self.test_data[5])
+ stream.close()
+
+ def test_seek_end_underflow_clamps_to_zero(self):
+ from pypaimon.table.row.blob import OffsetInputStream
+ stream = OffsetInputStream(io.BytesIO(self.test_data), 5, 10)
+ stream.seek(-20, io.SEEK_END)
+ self.assertEqual(stream.tell(), 0)
+ data = stream.read(1)
+ self.assertEqual(data[0], self.test_data[5])
+ stream.close()
+
+ def test_seek_end_underflow_unlimited_length(self):
+ from pypaimon.table.row.blob import OffsetInputStream
+ stream = OffsetInputStream(io.BytesIO(self.test_data), 5, -1)
+ stream.seek(-30, io.SEEK_END)
+ self.assertEqual(stream.tell(), 0)
+ data = stream.read(2)
+ self.assertEqual(data, self.test_data[5:7])
+ stream.close()
+
+ def test_seek_end_with_real_file(self):
+ from pypaimon.table.row.blob import OffsetInputStream
+ with tempfile.NamedTemporaryFile(delete=False) as tmp:
+ tmp.write(self.test_data)
+ tmp_path = tmp.name
+ try:
+ f = open(tmp_path, 'rb')
+ stream = OffsetInputStream(f, 5, 10)
+ stream.seek(0, io.SEEK_END)
+ self.assertEqual(stream.tell(), 10)
+ stream.seek(-3, io.SEEK_END)
+ self.assertEqual(stream.tell(), 7)
+ data = stream.read(3)
+ self.assertEqual(data, self.test_data[12:15])
+ stream.close()
+
+ f = open(tmp_path, 'rb')
+ stream = OffsetInputStream(f, 5, -1)
+ stream.seek(-30, io.SEEK_END)
+ self.assertEqual(stream.tell(), 0)
+ data = stream.read(2)
+ self.assertEqual(data, self.test_data[5:7])
+ stream.close()
+ finally:
+ os.remove(tmp_path)
+
+
if __name__ == '__main__':
unittest.main()