mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-18 09:42:27 +00:00
fix(python): align BlobFile seek and read behavior (#4290)
This commit is contained in:
@@ -34,14 +34,17 @@ FetchBlobsAsync = Callable[[str, pa.Table], Awaitable[pa.Array | pa.ChunkedArray
|
||||
class BlobFile(io.RawIOBase):
|
||||
"""Seekable lazy handle from :meth:`~lancedb.table.Table.fetch_blob_files`.
|
||||
|
||||
Bytes load on ``read`` or ``read_range``, not when the handle is opened.
|
||||
Use :meth:`aread` from async code.
|
||||
Bytes load on ``read``, ``read1``, ``read_range``, or ``read_ranges``, not
|
||||
when the handle is opened. Range reads leave the cursor unchanged. Use
|
||||
:meth:`aread` from async code.
|
||||
"""
|
||||
|
||||
def __init__(self, inner) -> None:
|
||||
self._inner = inner
|
||||
|
||||
async def aread(self) -> bytes:
|
||||
if self.closed:
|
||||
raise ValueError("read of closed file")
|
||||
return await self._inner.read()
|
||||
|
||||
def close(self) -> None:
|
||||
@@ -59,13 +62,16 @@ class BlobFile(io.RawIOBase):
|
||||
|
||||
def seek(self, offset: int, whence: int = io.SEEK_SET) -> int:
|
||||
if whence == io.SEEK_SET:
|
||||
self._inner.seek(offset)
|
||||
position = offset
|
||||
elif whence == io.SEEK_CUR:
|
||||
self._inner.seek(self._inner.tell() + offset)
|
||||
position = self._inner.tell() + offset
|
||||
elif whence == io.SEEK_END:
|
||||
self._inner.seek(self._inner.size() + offset)
|
||||
position = self._inner.size() + offset
|
||||
else:
|
||||
raise ValueError(f"invalid whence: {whence}")
|
||||
if position < 0:
|
||||
raise ValueError(f"negative seek value {position}")
|
||||
self._inner.seek(position)
|
||||
return self._inner.tell()
|
||||
|
||||
def tell(self) -> int:
|
||||
@@ -75,17 +81,39 @@ class BlobFile(io.RawIOBase):
|
||||
return self._inner.size()
|
||||
|
||||
def readall(self) -> bytes:
|
||||
if self.closed:
|
||||
raise ValueError("read of closed file")
|
||||
return self._inner.read_bytes()
|
||||
|
||||
def read(self, size: int = -1) -> bytes:
|
||||
if size == -1:
|
||||
def read(self, size: Optional[int] = -1) -> bytes:
|
||||
if self.closed:
|
||||
raise ValueError("read of closed file")
|
||||
if size is None or size < 0:
|
||||
return self._inner.read_bytes()
|
||||
return super().read(size)
|
||||
|
||||
def read1(self, size: Optional[int] = -1) -> bytes:
|
||||
"""Read up to ``size`` bytes with one underlying read."""
|
||||
if self.closed:
|
||||
raise ValueError("read of closed file")
|
||||
if size is None or size < 0:
|
||||
size = io.DEFAULT_BUFFER_SIZE
|
||||
return self._inner.read_up_to(size)
|
||||
|
||||
def read_range(self, offset: int, length: int) -> bytes:
|
||||
if self.closed:
|
||||
raise ValueError("read of closed file")
|
||||
return self._inner.read_range(offset, length)
|
||||
|
||||
def read_ranges(self, ranges: list[tuple[int, int]]) -> list[bytes]:
|
||||
"""Read ``(offset, length)`` pairs without moving the cursor."""
|
||||
if self.closed:
|
||||
raise ValueError("read of closed file")
|
||||
return self._inner.read_ranges(ranges)
|
||||
|
||||
def readinto(self, b: WriteableBuffer) -> int:
|
||||
if self.closed:
|
||||
raise ValueError("readinto of closed file")
|
||||
view = memoryview(b).cast("B")
|
||||
chunk = self._inner.read_up_to(len(view))
|
||||
view[: len(chunk)] = chunk
|
||||
|
||||
@@ -311,6 +311,7 @@ class BlobFile:
|
||||
def tell(self) -> int: ...
|
||||
def size(self) -> int: ...
|
||||
def read_range(self, offset: int, length: int) -> bytes: ...
|
||||
def read_ranges(self, ranges: list[tuple[int, int]]) -> list[bytes]: ...
|
||||
def read_up_to(self, length: int) -> bytes: ...
|
||||
|
||||
class Job:
|
||||
|
||||
@@ -1335,6 +1335,9 @@ async def test_async_fetch_blob_files_lazy_read():
|
||||
handles = await table.fetch_blob_files("image", hits)
|
||||
assert len(handles) == 1
|
||||
assert await handles[0].aread() == payload
|
||||
handles[0].close()
|
||||
with pytest.raises(ValueError, match="read of closed file"):
|
||||
await handles[0].aread()
|
||||
|
||||
|
||||
def test_fetch_blobs_from_query_result_without_row_id_raises():
|
||||
@@ -1470,6 +1473,49 @@ def test_blob_file_seek_read_and_read_range():
|
||||
handle.seek(0, 99)
|
||||
|
||||
|
||||
def test_blob_file_matches_file_error_and_range_contract():
|
||||
payload = b"abcdef"
|
||||
table = _blob_table("file_contract", [{"id": 1, "image": payload}])
|
||||
row_id = _row_ids_by_id(table)[1]
|
||||
handle = table.fetch_blob_files("image", [row_id])[0]
|
||||
|
||||
handle.seek(2)
|
||||
for offset, whence in [(-1, io.SEEK_SET), (-3, io.SEEK_CUR), (-7, io.SEEK_END)]:
|
||||
with pytest.raises(ValueError, match="negative seek value -1"):
|
||||
handle.seek(offset, whence)
|
||||
assert handle.tell() == 2
|
||||
|
||||
assert handle.read(None) == payload[2:]
|
||||
handle.seek(2)
|
||||
assert handle.read1(2) == payload[2:4]
|
||||
assert handle.read1() == payload[4:]
|
||||
handle.seek(2)
|
||||
assert handle.read_ranges([(0, 2), (4, 2), (2, 0)]) == [b"ab", b"ef", b""]
|
||||
assert handle.tell() == 2
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
handle.read_range(5, 2)
|
||||
with pytest.raises(ValueError):
|
||||
handle.read_ranges([(0, 1), (5, 2)])
|
||||
assert handle.tell() == 2
|
||||
|
||||
handle.close()
|
||||
with pytest.raises(ValueError, match="read of closed file"):
|
||||
handle.read(1)
|
||||
with pytest.raises(ValueError, match="read of closed file"):
|
||||
handle.read(None)
|
||||
with pytest.raises(ValueError, match="read of closed file"):
|
||||
handle.read1(1)
|
||||
with pytest.raises(ValueError, match="read of closed file"):
|
||||
handle.readall()
|
||||
with pytest.raises(ValueError, match="readinto of closed file"):
|
||||
handle.readinto(bytearray(1))
|
||||
with pytest.raises(ValueError):
|
||||
handle.read_range(0, 1)
|
||||
with pytest.raises(ValueError):
|
||||
handle.read_ranges([])
|
||||
|
||||
|
||||
def test_fetch_blob_files_from_query_partial_read():
|
||||
payload = _identifiable_payload(65536)
|
||||
table = _blob_table("query_partial", [{"id": 1, "image": payload}])
|
||||
|
||||
@@ -4,6 +4,7 @@ from concurrent.futures import ThreadPoolExecutor
|
||||
import contextlib
|
||||
from datetime import timedelta
|
||||
import http.server
|
||||
import io
|
||||
import json
|
||||
import multiprocessing as mp
|
||||
import pickle
|
||||
@@ -3503,6 +3504,28 @@ def test_remote_blob_files_are_lazy_seekable_handles():
|
||||
alpha.read_range(0, 1)
|
||||
|
||||
|
||||
def test_remote_blob_file_matches_file_error_and_range_contract():
|
||||
with blob_remote_table() as table:
|
||||
handle = table.fetch_blob_files("image", [10])[0]
|
||||
handle.seek(2)
|
||||
|
||||
with pytest.raises(ValueError, match="negative seek value -1"):
|
||||
handle.seek(-3, io.SEEK_CUR)
|
||||
assert handle.tell() == 2
|
||||
assert handle.read(None) == b"pha"
|
||||
|
||||
handle.seek(2)
|
||||
assert handle.read1(2) == b"ph"
|
||||
assert handle.read_ranges([(0, 2), (3, 2)]) == [b"al", b"ha"]
|
||||
assert handle.tell() == 4
|
||||
with pytest.raises(ValueError):
|
||||
handle.read_range(4, 2)
|
||||
|
||||
handle.close()
|
||||
with pytest.raises(ValueError, match="read of closed file"):
|
||||
handle.read(1)
|
||||
|
||||
|
||||
def test_remote_blob_fetch_accepts_query_table():
|
||||
hits = pa.table({"_rowid": pa.array([10, 20, 30], type=pa.uint64())})
|
||||
|
||||
|
||||
Reference in New Issue
Block a user