feat(remote): add seekable blob range reads (#3750)

## Summary

- Implements Cloud `fetch_blob_files`: returns real seekable `BlobFile`
handles over HTTP Range instead of `NotSupported`.
- Completes the second Cloud blob read verb after #3684 (`fetch_blobs` =
eager whole bytes; this = lazy / partial / sequential reads).
- Same public handle API as local (`read_range`, `read_up_to`, `seek`,
`tell`, `close`), so one code path works for local and Cloud.

Large blobs (video, audio, PDFs) should not require downloading the
whole object to inspect a header or stream a slice. After search,
callers open a handle and read only what they need:

```python
hits = table.search(vec).select(["id", "video"]).limit(5).to_arrow()

with table.fetch_blob_files("video", hits)[0] as f:
    header = f.read_range(0, 256)
    f.seek(keyframe_offset)
    chunk = f.read_up_to(1 << 20)
```

### Behavior

- Handle creation probes size with `bytes=0-0` (bounded concurrency,
input order preserved).
- `204` → null (`None`); `416` with `bytes */0` → valid empty blob;
other `416` → error.
- `read_range` validates `Content-Range` and body length; OOB ranges
fail with `invalid_input` before the request (aligned with Lance).
- `read_up_to` reuses one open-ended Range response across sequential
reads; `seek` drops it.
- Servers older than 0.5.0 get a clear `NotSupported` (does not suggest
`fetch_blobs`, which they also lack).

## Testing

- `cargo test --features remote -p lancedb remote_blob`
- `cargo test --features remote -p lancedb test_blob`
- `cargo clippy --features remote --tests --examples` (no new warnings
from this change)

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Drew Gallardo
2026-08-03 08:38:08 -07:00
committed by GitHub
parent 9e26bf3fba
commit 3dd9c598e9
7 changed files with 1473 additions and 60 deletions
+2 -2
View File
@@ -1595,8 +1595,8 @@ class Table(ABC):
Prefer this over :meth:`fetch_blobs` for large payloads. ``row_ids`` is
a ``list[int]`` or a query ``pyarrow.Table`` carrying row identity via
``_rowid`` or a ``_lance_row_id`` field on the blob descriptor. Null
rows are ``None``. Unsupported on LanceDB Cloud, where
:meth:`fetch_blobs` returns full bytes instead.
rows are ``None``. Remote tables require LanceDB Cloud server 0.5.0 or
newer.
"""
@abstractmethod
+33 -2
View File
@@ -2055,6 +2055,24 @@ def blob_remote_table(*, server_version=Version("0.5.0")):
request.send_header("phalanx-version", str(server_version))
request.end_headers()
request.wfile.write(json.dumps(BLOB_DESCRIBE_RESPONSE).encode())
elif request.path.startswith("/v1/table/test/blob/image/"):
path = request.path.partition("?")[0]
row_id = int(path.split("/")[-2])
payload = {10: b"alpha", 20: None, 30: b"gamma"}[row_id]
if payload is None:
request.send_response(204)
request.end_headers()
return
byte_range = request.headers["Range"].removeprefix("bytes=")
start_text, end_text = byte_range.split("-", maxsplit=1)
start = int(start_text)
end = int(end_text) if end_text else len(payload) - 1
chunk = payload[start : end + 1]
request.send_response(206)
request.send_header("Content-Range", f"bytes {start}-{end}/{len(payload)}")
request.send_header("Content-Length", str(len(chunk)))
request.end_headers()
request.wfile.write(chunk)
elif request.path == "/v1/table/test/query/":
content_len = int(request.headers.get("Content-Length", 0))
body = json.loads(request.rfile.read(content_len))
@@ -2092,8 +2110,21 @@ def test_remote_blob_columns_and_fetch():
assert table.blob_columns() == ["image"]
blobs = table.fetch_blobs("image", [10, 20, 30])
assert blobs.to_pylist() == [b"alpha", None, b"gamma"]
with pytest.raises(NotImplementedError, match="Use fetch_blobs for full bytes"):
table.fetch_blob_files("image", [10, 20, 30])
def test_remote_blob_files_are_lazy_seekable_handles():
with blob_remote_table() as table:
files = table.fetch_blob_files("image", [10, 20, 30])
assert len(files) == 3
alpha, null_row, gamma = files
assert null_row is None
assert alpha is not None
assert gamma is not None
assert alpha.size() == 5
assert alpha.read_range(1, 3) == b"lph"
gamma.seek(2)
assert gamma.read() == b"mma"
def test_remote_blob_fetch_accepts_query_table():
+25 -11
View File
@@ -426,9 +426,11 @@ pub struct PyBlobFile {
impl PyBlobFile {
fn read_bytes(self_: PyRef<'_, Self>) -> PyResult<Py<PyBytes>> {
let inner = self_.inner.clone();
let bytes = block_on(async move { inner.read().await })
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}")))?;
Ok(PyBytes::new(self_.py(), bytes.as_ref()).unbind())
Ok(PyBytes::new(py, bytes.as_ref()).unbind())
}
pub fn read(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
@@ -444,24 +446,32 @@ impl PyBlobFile {
fn close(self_: PyRef<'_, Self>) -> PyResult<()> {
let inner = self_.inner.clone();
block_on(async move { inner.close().await })
self_
.py()
.detach(move || block_on(async move { inner.close().await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob close failed: {e}")))
}
fn is_closed(self_: PyRef<'_, Self>) -> bool {
let inner = self_.inner.clone();
block_on(async move { inner.is_closed().await })
self_
.py()
.detach(move || block_on(async move { inner.is_closed().await }))
}
fn seek(self_: PyRef<'_, Self>, position: u64) -> PyResult<()> {
let inner = self_.inner.clone();
block_on(async move { inner.seek(position).await })
self_
.py()
.detach(move || block_on(async move { inner.seek(position).await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob seek failed: {e}")))
}
fn tell(self_: PyRef<'_, Self>) -> PyResult<u64> {
let inner = self_.inner.clone();
block_on(async move { inner.tell().await })
self_
.py()
.detach(move || block_on(async move { inner.tell().await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob tell failed: {e}")))
}
@@ -475,16 +485,20 @@ impl PyBlobFile {
.checked_add(length as u64)
.ok_or_else(|| PyValueError::new_err("offset + length overflowed"))?;
let inner = self_.inner.clone();
let bytes = block_on(async move { inner.read_range(offset..end).await })
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}")))?;
Ok(PyBytes::new(self_.py(), bytes.as_ref()).unbind())
Ok(PyBytes::new(py, bytes.as_ref()).unbind())
}
fn read_up_to(self_: PyRef<'_, Self>, length: usize) -> PyResult<Py<PyBytes>> {
let inner = self_.inner.clone();
let bytes = block_on(async move { inner.read_up_to(length).await })
.map_err(|e| PyRuntimeError::new_err(format!("blob read failed: {e}")))?;
Ok(PyBytes::new(self_.py(), bytes.as_ref()).unbind())
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}")))?;
Ok(PyBytes::new(py, bytes.as_ref()).unbind())
}
}