feat(python): open lazy blob files from SQL descriptors (#4668)

SQL blob descriptors need pinned source identity to reopen a value after
the query finishes. Adds `open_blob_file(descriptor)` and
`open_blob_files(descriptors)` so Python can read the original blob
incrementally without materializing it in the SQL result.

### Python experience

```python
rows = db.execute_query("""
    SELECT f.timestamp, blob_descriptor(v.video) AS video
    FROM videos v JOIN frames f ON v.id = f.video_id
""").read_all()

videos = db.open_table("videos")
for row in rows.to_pylist():
    if row["video"] is not None:
        with videos.open_blob_file(row["video"]) as video:
            frame = extract_frame(video, row["timestamp"])
```

The singular API accepts the descriptor dictionary or an Arrow struct
scalar directly and returns a seekable `BlobFile`.
`videos.open_blob_files(rows["video"])` accepts an Arrow array/chunked
array or another iterable, opens handles as iteration advances, and
preserves order, duplicates, and nulls as `None`. Async tables provide
`await videos.open_blob_file(...)` and `async for video in
videos.open_blob_files(...)`.

The iterable API can be used directly with the Arrow result column. Zip
it with the rows to keep each frame timestamp:

```python
for row, video in zip(rows.to_pylist(), videos.open_blob_files(rows["video"])):
    if video is not None:
        with video:
            frame = extract_frame(video, row["timestamp"])
```

Each iteration opens one handle. Null descriptors yield `None`; handles
preserve input order and duplicates, and the `with` block closes each
non-null file.

For an async table:

```python
async for video in async_videos.open_blob_files(rows["video"]):
    if video is not None:
        with video:
            payload = await video.aread()
```

### Execution

The descriptor retains the canonical `kind`, `position`, `size`,
`blob_id`, and `blob_uri` fields and carries a `source` struct with its
dataset URI, version, commit timestamp, branch, column, row reference,
and whether that reference is a physical address. Native opening
resolves that exact snapshot without changing the table handle's
checkout and rejects a descriptor belonging to another table. Remote
handles send authenticated byte-range requests with the captured
version/branch, commit timestamp, and source URI, so an old physical
address is never interpreted against the latest table.

Known-size remote descriptors open without a payload request. External
blobs with an unspecified size use a metadata-only HEAD request. Payload
bytes load only when the file is read; full reads still allocate the
full value. Existing row-ID-based `fetch_blob_files` remains available
when a descriptor is not already present.

Coverage includes native snapshot preservation after overwrite, both row
IDs and addresses, dictionary/Arrow inputs, null/empty values, reordered
and duplicate descriptors, lazy iterable behavior, async iteration, and
remote range pinning for all six storage kinds.

Descriptor reads isolate their pinned snapshot from the live table
handle's freshness state. Both native and remote reads verify the source
commit timestamp to reject a replaced snapshot after table recreation.

python/README.md

@@ -101,3 +101,52 @@ print(results)
 ### Development
 
 See [CONTRIBUTING.md](./CONTRIBUTING.md) for information on how to contribute to LanceDB.
+
+### Opening SQL blob descriptors
+
+A SQL `blob_descriptor(video)` result keeps the Blob V2 storage metadata and a
+`source` struct containing its table URI, version, commit timestamp, branch,
+column and row identity. This example uses a remote SQL connection whose SQL
+service produces those source-bearing descriptors. Open the reference on the
+source table to read incrementally:
+
+```python
+rows = db.execute_query("""
+    SELECT f.timestamp, blob_descriptor(v.video) AS video
+    FROM videos v JOIN frames f ON v.id = f.video_id
+""").read_all()
+
+videos = db.open_table("videos")
+for row in rows.to_pylist():
+    if row["video"] is not None:
+        with videos.open_blob_file(row["video"]) as video:
+            frame = extract_frame(video, row["timestamp"])
+```
+
+`open_blob_file` accepts the dictionary or an Arrow struct scalar directly. It
+keeps the query's version and branch without changing the table handle's checkout.
+Payload bytes load on reads. An external blob with an unspecified size may need a
+metadata-only HEAD request to open.
+
+`videos.open_blob_files(rows["video"])` returns an iterator over the same handles,
+preserving order, duplicates and nulls (`None`). It accepts Arrow arrays/chunked
+arrays or another iterable and opens each handle as iteration advances. Close each
+non-null handle after use. Async tables offer `await videos.open_blob_file(...)`
+and `async for video in videos.open_blob_files(...)`.
+
+```python
+for video in videos.open_blob_files(rows["video"]):
+    if video is not None:
+        with video:
+            header = video.read(4096)
+```
+
+To extract a frame while retaining each row's timestamp, iterate over the
+handles and rows together:
+
+```python
+for row, video in zip(rows.to_pylist(), videos.open_blob_files(rows["video"])):
+    if video is not None:
+        with video:
+            frame = extract_frame(video, row["timestamp"])
+```

python/python/lancedb/_blob.py

@@ -7,7 +7,7 @@ from __future__ import annotations
 
 import io
 import operator
-from collections.abc import Awaitable, Callable, Iterable
+from collections.abc import Awaitable, Callable, Iterable, Mapping
 from typing import TYPE_CHECKING, Optional, Union
 
 import pyarrow as pa
@@ -31,6 +31,18 @@ FetchBlobsSync = Callable[[str, list[int]], pa.Array | pa.ChunkedArray]
 FetchBlobsAsync = Callable[[str, list[int]], Awaitable[pa.Array | pa.ChunkedArray]]
 
 
+def _normalize_blob_descriptor(value):
+    if isinstance(value, pa.Scalar):
+        value = value.as_py()
+    if value is None:
+        return None
+    if not isinstance(value, Mapping):
+        raise TypeError(
+            "blob descriptor must be a mapping, an Arrow struct scalar, or None"
+        )
+    return dict(value)
+
+
 class BlobFile(io.RawIOBase):
     """Seekable lazy handle from :meth:`~lancedb.table.Table.fetch_blob_files`.
 

python/python/lancedb/_lancedb.pyi

@@ -580,6 +580,7 @@ class Table:
     async def fetch_blob_files(
         self, column: str, row_ids: list[int]
     ) -> list[Optional[BlobFile]]: ...
+    async def open_blob_file(self, descriptor: str) -> BlobFile: ...
     def vector_search(self) -> VectorQuery: ...
 
 class Tags:

python/python/lancedb/remote/table.py

@@ -1221,6 +1221,11 @@ class RemoteTable(Table):
     ) -> "list[Optional[BlobFile]]":
         return LOOP.run(self._table.fetch_blob_files(column, row_ids))
 
+    def open_blob_file(
+        self, descriptor: Union[dict, pa.StructScalar, None]
+    ) -> Optional[BlobFile]:
+        return LOOP.run(self._table.open_blob_file(descriptor))
+
     def head(self, n=5) -> pa.Table:
         """
         Return the first `n` rows of the table.

python/python/lancedb/table.py

@@ -15,9 +15,11 @@ from functools import cached_property
 from typing import (
     TYPE_CHECKING,
     Any,
+    AsyncIterator,
     Callable,
     Dict,
     Iterable,
+    Iterator,
     List,
     Literal,
     Optional,
@@ -33,6 +35,7 @@ from lancedb.scannable import _hf_batches, _register_optional_converters, to_sca
 from . import __version__
 from ._blob import (
     BlobFile,
+    _normalize_blob_descriptor,
     _normalize_blob_row_ids,
     _wrap_blob_files,
     strip_auto_row_ids,
@@ -2289,6 +2292,44 @@ class Table(ABC):
         handles are returned.
         """
 
+    @abstractmethod
+    def open_blob_file(
+        self, descriptor: Union[dict, pa.StructScalar, None]
+    ) -> Optional[BlobFile]:
+        """Open a seekable file from a SQL ``blob_descriptor`` result.
+
+        Accepts the descriptor dictionary or an Arrow struct scalar. Its source
+        pins the table version, branch, column and row identity, so opening does
+        not change this table's checkout. Null descriptors return ``None``.
+        Payload bytes load only when the returned handle is read. An external
+        blob with an unspecified size may require a metadata request on opening.
+
+        ```python
+        rows = db.execute_query(
+            "SELECT blob_descriptor(video) AS video FROM videos"
+        ).read_all()
+        with videos.open_blob_file(rows["video"][0]) as video:
+            header = video.read(4096)
+        ```
+        """
+
+    def open_blob_files(self, descriptors: Iterable) -> Iterator[Optional[BlobFile]]:
+        """Iterate over lazy files from descriptor dictionaries or Arrow scalars.
+
+        Accepts an Arrow array/chunked array or any iterable. Opens each handle
+        as iteration advances, preserving input order and nulls as ``None``.
+        Close each non-null handle after use, normally with a ``with`` block.
+
+        ```python
+        for video in videos.open_blob_files(rows["video"]):
+            if video is not None:
+                with video:
+                    header = video.read(4096)
+        ```
+        """
+        for descriptor in descriptors:
+            yield self.open_blob_file(descriptor)
+
     @abstractmethod
     def _execute_query(
         self,
@@ -3239,6 +3280,11 @@ class LanceTable(Table):
     ) -> "list[Optional[BlobFile]]":
         return LOOP.run(self._table.fetch_blob_files(column, row_ids))
 
+    def open_blob_file(
+        self, descriptor: Union[dict, pa.StructScalar, None]
+    ) -> Optional[BlobFile]:
+        return LOOP.run(self._table.open_blob_file(descriptor))
+
     @property
     def tags(self) -> Tags:
         """Tag management for the table.
@@ -7451,6 +7497,36 @@ class AsyncTable:
         )
         return _wrap_blob_files(handles)
 
+    async def open_blob_file(
+        self, descriptor: Union[dict, pa.StructScalar, None]
+    ) -> Optional[BlobFile]:
+        """Open a snapshot-bound descriptor as a seekable file.
+
+        See [Table.open_blob_file][lancedb.table.Table.open_blob_file].
+        """
+        descriptor = _normalize_blob_descriptor(descriptor)
+        if descriptor is None:
+            return None
+        return BlobFile(await self._inner.open_blob_file(json.dumps(descriptor)))
+
+    async def open_blob_files(
+        self, descriptors: Iterable
+    ) -> AsyncIterator[Optional[BlobFile]]:
+        """Asynchronously iterate over files in descriptor order, preserving nulls.
+
+        Use ``async for file in table.open_blob_files(descriptors)``. Each
+        non-null file must be closed after use.
+
+        ```python
+        async for video in videos.open_blob_files(rows["video"]):
+            if video is not None:
+                with video:
+                    payload = await video.aread()
+        ```
+        """
+        for descriptor in descriptors:
+            yield await self.open_blob_file(descriptor)
+
     @property
     def tags(self) -> AsyncTags:
         """Tag management for the dataset.

python/python/tests/test_blob.py

@@ -2696,3 +2696,135 @@ def test_create_table_with_python_blob_dict_and_object():
         b"from-dict",
         b"from-blob",
     ]
+
+
+def _openable_descriptors(table, *, by_address):
+    hits = (
+        table.search()
+        .select(["id", "image", "_rowaddr"])
+        .with_row_id(True)
+        .limit(100)
+        .to_arrow(blob_mode="descriptions")
+    )
+    version = table.version
+    committed_at = next(
+        item["timestamp"]
+        for item in table.list_versions()
+        if item["version"] == version
+    )
+    descriptors = {}
+    for row in hits.to_pylist():
+        descriptor = row["image"]
+        if descriptor is not None and descriptor.get("kind") is not None:
+            descriptor = dict(descriptor)
+            descriptor["source"] = {
+                "uri": table.to_lance().uri,
+                "version": version,
+                "committed_at_micros": round(committed_at.timestamp() * 1_000_000),
+                "branch": None,
+                "column": "image",
+                "row_reference": row["_rowaddr" if by_address else "_rowid"],
+                "row_is_address": by_address,
+            }
+        else:
+            descriptor = None
+        descriptors[row["id"]] = descriptor
+    return descriptors
+
+
+@pytest.mark.parametrize("by_address", [False, True])
+def test_open_blob_files_preserve_snapshot_order_nulls_and_empty_values(
+    tmp_path, by_address
+):
+    db = lancedb.connect(str(tmp_path))
+    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
+    table = db.create_table("videos", schema=schema)
+    table.add(
+        [
+            {"id": 1, "image": b"alpha"},
+            {"id": 2, "image": None},
+            {"id": 3, "image": b"beta"},
+            {"id": 4, "image": b""},
+        ]
+    )
+    descriptors = _openable_descriptors(table, by_address=by_address)
+    table.add([{"id": 1, "image": b"replacement"}], mode="overwrite")
+    latest_version = table.version
+    values = pa.array(
+        [descriptors[3], descriptors[2], descriptors[1], descriptors[3], descriptors[4]]
+    )
+    files = table.open_blob_files(
+        pa.chunked_array([values.slice(0, 2), values.slice(2)])
+    )
+    assert iter(files) is files
+    actual = []
+    for file in files:
+        if file is None:
+            actual.append(None)
+        else:
+            with file:
+                actual.append(file.read())
+            assert file.closed
+    assert actual == [b"beta", None, b"alpha", b"beta", b""]
+    with table.open_blob_file(descriptors[1]) as file:
+        file.seek(2)
+        assert file.read_range(0, 2) == b"al"
+        assert file.tell() == 2
+        assert file.read(2) == b"ph"
+    assert table.version == latest_version
+    assert table.open_blob_file(pa.scalar(None, type=values.type)) is None
+    other = db.create_table("other", schema=schema)
+    with pytest.raises(ValueError, match="different table"):
+        other.open_blob_file(descriptors[1])
+    stale = dict(descriptors[1], source=dict(descriptors[1]["source"]))
+    stale["source"]["committed_at_micros"] += 1
+    with pytest.raises(ValueError, match="snapshot has been replaced"):
+        table.open_blob_file(stale)
+    raw = {key: value for key, value in descriptors[1].items() if key != "source"}
+    with pytest.raises(ValueError, match="source"):
+        table.open_blob_file(raw)
+    with pytest.raises(TypeError, match="descriptor"):
+        table.open_blob_file("not a descriptor")
+
+
+def test_open_blob_files_does_not_open_ahead(monkeypatch):
+    table = _blob_table("lazy_descriptor_iteration", [{"id": 1, "image": b"x"}])
+    opened = []
+
+    def open_file(descriptor):
+        opened.append(descriptor)
+        return descriptor
+
+    monkeypatch.setattr(table, "open_blob_file", open_file)
+    files = table.open_blob_files(iter([1, None, 2]))
+    assert opened == []
+    assert next(files) == 1
+    assert opened == [1]
+    assert list(files) == [None, 2]
+    assert opened == [1, None, 2]
+
+
+@pytest.mark.asyncio
+async def test_open_blob_files_async_iterates_snapshot_handles(tmp_path):
+    db = lancedb.connect(str(tmp_path))
+    table = db.create_table(
+        "videos",
+        schema=pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")]),
+    )
+    table.add([{"id": 1, "image": b"alpha"}, {"id": 2, "image": None}])
+    descriptors = _openable_descriptors(table, by_address=True)
+    async_db = await lancedb.connect_async(str(tmp_path))
+    videos = await async_db.open_table("videos")
+    file = await videos.open_blob_file(pa.scalar(descriptors[1]))
+    with file:
+        assert file.read(2) == b"al"
+    actual = []
+    async for file in videos.open_blob_files(
+        [descriptors[2], descriptors[1], descriptors[1]]
+    ):
+        if file is None:
+            actual.append(None)
+        else:
+            with file:
+                actual.append(await file.aread())
+    assert actual == [None, b"alpha", b"alpha"]

python/src/table.rs

@@ -24,7 +24,7 @@ use arrow::{
     ffi_stream::ArrowArrayStreamReader,
     pyarrow::{FromPyArrow, PyArrowType, ToPyArrow},
 };
-use lancedb::blob::{BlobFile, BlobRangeRequest};
+use lancedb::blob::{BlobDescriptor, BlobFile, BlobRangeRequest};
 use lancedb::index::scalar::FtsIndexBuilder;
 use lancedb::table::{
     AddDataMode, ColumnAlteration, Duration, FieldMetadataUpdate, FtsToken as LanceDbFtsToken,
@@ -1551,6 +1551,23 @@ impl Table {
         })
     }
 
+    #[pyo3(signature = (descriptor))]
+    pub fn open_blob_file<'py>(
+        self_: PyRef<'py, Self>,
+        descriptor: &str,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let descriptor: BlobDescriptor = serde_json::from_str(descriptor)
+            .map_err(|error| PyValueError::new_err(format!("invalid blob descriptor: {error}")))?;
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            let file = inner.open_blob_file(&descriptor).await.infer_error()?;
+            Ok(PyBlobFile {
+                inner: Arc::new(file),
+                closed: AtomicBool::new(false),
+            })
+        })
+    }
+
     /// Optimize the on-disk data by compacting and pruning old data, for better performance.
     #[pyo3(signature = (cleanup_since_ms=None, delete_unverified=None))]
     pub fn optimize(