fix(python): align BlobFile seek and read behavior

This commit is contained in:
Gatefixer
2026-09-23 18:27:16 +00:00
parent df7bfcdb65
commit ca3c2d5e1e
5 changed files with 142 additions and 17 deletions
+35 -7
View File
@@ -33,14 +33,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:
@@ -58,13 +61,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:
@@ -74,17 +80,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
+1
View File
@@ -281,6 +281,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:
+46
View File
@@ -967,6 +967,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():
@@ -1102,6 +1105,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}])
+23
View File
@@ -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
@@ -2472,6 +2473,28 @@ def test_remote_blob_files_are_lazy_seekable_handles():
assert gamma.read() == b"mma"
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())})
+37 -10
View File
@@ -682,6 +682,14 @@ pub struct PyBlobFile {
inner: Arc<BlobFile>,
}
fn blob_file_error(operation: &str, error: lance_core::Error) -> pyo3::PyErr {
let message = format!("blob {operation} failed: {error}");
match error {
lance_core::Error::InvalidInput { .. } => PyValueError::new_err(message),
_ => PyRuntimeError::new_err(message),
}
}
#[pymethods]
impl PyBlobFile {
fn read_bytes(self_: PyRef<'_, Self>) -> PyResult<Py<PyBytes>> {
@@ -689,17 +697,14 @@ impl PyBlobFile {
let py = self_.py();
let bytes = py
.detach(move || block_on(async move { inner.read().await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob read failed: {e}")))?;
.map_err(|e| blob_file_error("read", e))?;
Ok(PyBytes::new(py, bytes.as_ref()).unbind())
}
pub fn read(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner.clone();
future_into_py(self_.py(), async move {
let bytes = inner
.read()
.await
.map_err(|e| PyRuntimeError::new_err(format!("blob read failed: {e}")))?;
let bytes = inner.read().await.map_err(|e| blob_file_error("read", e))?;
Python::attach(|py| Ok(PyBytes::new(py, bytes.as_ref()).unbind()))
})
}
@@ -709,7 +714,7 @@ impl PyBlobFile {
self_
.py()
.detach(move || block_on(async move { inner.close().await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob close failed: {e}")))
.map_err(|e| blob_file_error("close", e))
}
fn is_closed(self_: PyRef<'_, Self>) -> bool {
@@ -724,7 +729,7 @@ impl PyBlobFile {
self_
.py()
.detach(move || block_on(async move { inner.seek(position).await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob seek failed: {e}")))
.map_err(|e| blob_file_error("seek", e))
}
fn tell(self_: PyRef<'_, Self>) -> PyResult<u64> {
@@ -732,7 +737,7 @@ impl PyBlobFile {
self_
.py()
.detach(move || block_on(async move { inner.tell().await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob tell failed: {e}")))
.map_err(|e| blob_file_error("tell", e))
}
fn size(self_: PyRef<'_, Self>) -> u64 {
@@ -748,16 +753,38 @@ impl PyBlobFile {
let py = self_.py();
let bytes = py
.detach(move || block_on(async move { inner.read_range(offset..end).await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob read_range failed: {e}")))?;
.map_err(|e| blob_file_error("read_range", e))?;
Ok(PyBytes::new(py, bytes.as_ref()).unbind())
}
/// Read blob-local byte ranges without moving the cursor.
fn read_ranges(self_: PyRef<'_, Self>, ranges: Vec<(u64, u64)>) -> PyResult<Vec<Py<PyBytes>>> {
let ranges = ranges
.into_iter()
.map(|(offset, length)| {
offset
.checked_add(length)
.map(|end| offset..end)
.ok_or_else(|| PyValueError::new_err("offset + length overflowed"))
})
.collect::<PyResult<Vec<_>>>()?;
let inner = self_.inner.clone();
let py = self_.py();
let bytes = py
.detach(move || block_on(async move { inner.read_ranges(&ranges).await }))
.map_err(|e| blob_file_error("read_ranges", e))?;
Ok(bytes
.iter()
.map(|bytes| PyBytes::new(py, bytes.as_ref()).unbind())
.collect())
}
fn read_up_to(self_: PyRef<'_, Self>, length: usize) -> PyResult<Py<PyBytes>> {
let inner = self_.inner.clone();
let py = self_.py();
let bytes = py
.detach(move || block_on(async move { inner.read_up_to(length).await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob read_up_to failed: {e}")))?;
.map_err(|e| blob_file_error("read_up_to", e))?;
Ok(PyBytes::new(py, bytes.as_ref()).unbind())
}
}