From ca3c2d5e1ec0b39f13e6d183910d3b7d131fc864 Mon Sep 17 00:00:00 2001 From: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Date: Wed, 23 Sep 2026 18:27:16 +0000 Subject: [PATCH] fix(python): align BlobFile seek and read behavior --- python/python/lancedb/_blob.py | 42 ++++++++++++++++++++---- python/python/lancedb/_lancedb.pyi | 1 + python/python/tests/test_blob.py | 46 ++++++++++++++++++++++++++ python/python/tests/test_remote_db.py | 23 +++++++++++++ python/src/table.rs | 47 +++++++++++++++++++++------ 5 files changed, 142 insertions(+), 17 deletions(-) diff --git a/python/python/lancedb/_blob.py b/python/python/lancedb/_blob.py index dc4ed37df..6884edee1 100644 --- a/python/python/lancedb/_blob.py +++ b/python/python/lancedb/_blob.py @@ -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 diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index ba4e3a0df..1b7edde5b 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -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: diff --git a/python/python/tests/test_blob.py b/python/python/tests/test_blob.py index 0d3a89c21..840dbd14c 100644 --- a/python/python/tests/test_blob.py +++ b/python/python/tests/test_blob.py @@ -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}]) diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py index c03862a49..01d7bcf16 100644 --- a/python/python/tests/test_remote_db.py +++ b/python/python/tests/test_remote_db.py @@ -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())}) diff --git a/python/src/table.rs b/python/src/table.rs index a3dad3d86..19c34758e 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -682,6 +682,14 @@ pub struct PyBlobFile { inner: Arc, } +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> { @@ -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> { 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 { @@ -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>> { + 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::>>()?; + 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> { 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()) } }