diff --git a/docs/src/python/python.md b/docs/src/python/python.md index 41d3e16d7..02ccced2c 100644 --- a/docs/src/python/python.md +++ b/docs/src/python/python.md @@ -538,12 +538,64 @@ instead of being materialized with the rest of the row. `lancedb.BlobType` is `lance.blob.BlobType` when pylance is installed. Without pylance, LanceDB uses a matching `lance.blob.v2` extension type so blob columns -still work. Queries return descriptors. Call +still work. Native table queries return descriptors. Call [`fetch_blob_files`][lancedb.table.Table.fetch_blob_files] for lazy reads or [`fetch_blobs`][lancedb.table.Table.fetch_blobs] for eager bytes. One missing or deleted row ID rejects the entire blob-read batch; the APIs do not return partial results. +Remote SQL returns blob bytes by default. Select `blob_descriptor(video)` to +return an openable reference instead, then use +[`open_blob_file`][lancedb.table.Table.open_blob_file] or the lazy iterable +[`open_blob_files`][lancedb.table.Table.open_blob_files]: + +```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"]) +``` + +This example uses a remote SQL connection that produces source-bearing +descriptors. The descriptor contains `kind`, `position`, `size`, `blob_id`, +`blob_uri`, and a `source` struct with `uri`, `version`, `committed_at_micros`, +`branch`, `column`, `row_reference`, and `row_is_address`. The opener verifies +the source table and commit and reads the captured version without changing +the table handle's checkout. Raw native storage descriptors omit `source`; +use `fetch_blob_files(column, row_ids)` for those query results. + +The singular API accepts a dictionary or Arrow struct scalar. The iterable +accepts Arrow arrays/chunked arrays or another iterable, opens handles as +iteration advances, and preserves order, duplicates, and nulls as `None`. +Close each non-null handle after use. Async tables provide +`await table.open_blob_file(...)` and +`async for file in table.open_blob_files(...)`. + +```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"]) +``` + +For an async table, the iterable is consumed with `async for`: + +```python +async for video in async_videos.open_blob_files(rows["video"]): + if video is not None: + with video: + payload = await video.aread() +``` + +Payload bytes load on reads. Opening an external blob with unspecified size +can issue a metadata-only HEAD request. Remote reads use authenticated range +requests that validate the captured URI and commit timestamp. + Blobs inside lists can be written and queried as descriptors. They are omitted from `blob_columns()` because row ids do not identify individual list elements, so the fetch APIs cannot read their bytes yet. diff --git a/python/README.md b/python/README.md index 907f7c099..0b640d27c 100644 --- a/python/README.md +++ b/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"]) +``` diff --git a/python/python/lancedb/_blob.py b/python/python/lancedb/_blob.py index 1e9110dd0..46462553c 100644 --- a/python/python/lancedb/_blob.py +++ b/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`. diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 9e602cce8..0b4fc82f1 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/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: diff --git a/python/python/lancedb/remote/table.py b/python/python/lancedb/remote/table.py index 53591182e..dd22dea48 100644 --- a/python/python/lancedb/remote/table.py +++ b/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. diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index c08c6390e..665520479 100644 --- a/python/python/lancedb/table.py +++ b/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. diff --git a/python/python/tests/test_blob.py b/python/python/tests/test_blob.py index c2dc2a265..da362f053 100644 --- a/python/python/tests/test_blob.py +++ b/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"] diff --git a/python/src/table.rs b/python/src/table.rs index e77c58578..e58b8a139 100644 --- a/python/src/table.rs +++ b/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> { + 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( diff --git a/rust/lancedb/src/blob.rs b/rust/lancedb/src/blob.rs index 05650d7c2..142b2264e 100644 --- a/rust/lancedb/src/blob.rs +++ b/rust/lancedb/src/blob.rs @@ -20,11 +20,120 @@ use arrow_schema::{DataType, Field, Schema}; use lance::dataset::{BlobRangeRequest as LanceBlobRangeRequest, Dataset, WriteParams}; use lance_arrow::FieldExt; use lance_file::version::{ConcreteFileVersion, LanceFileVersion}; -use lance_io::object_store::ObjectStore; +use lance_io::object_store::{ObjectStore, uri_to_url}; use object_store::path::Path; +use serde::{Deserialize, Serialize}; use crate::error::{Error, Result}; +/// A stored Blob V2 descriptor with the source needed to reopen its snapshot. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct BlobDescriptor { + /// Canonical Blob V2 storage kind (0 through 5). + pub kind: u8, + /// Stored offset within the backing object. + pub position: u64, + /// Stored length; zero on an external blob can mean unspecified. + pub size: u64, + /// Stored sidecar or base identifier. + pub blob_id: u32, + /// Stored external URI or managed path. + pub blob_uri: String, + /// Source identity captured by the query. + pub source: BlobDescriptorSource, +} + +/// The pinned source of a descriptor returned by SQL. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct BlobDescriptorSource { + /// URI of the dataset on the source branch. + pub uri: String, + /// Exact table version read by the query. + pub version: u64, + /// Commit timestamp in microseconds, distinguishing recreated tables. + pub committed_at_micros: u64, + /// Source branch; `None` identifies main. + pub branch: Option, + /// Blob column path in the source table. + pub column: String, + /// Row ID or row address, as specified by `row_is_address`. + pub row_reference: u64, + /// Whether the row reference is a physical address. + pub row_is_address: bool, +} + +impl BlobDescriptor { + pub(crate) fn validate(&self) -> Result<()> { + if self.kind > 5 + || self.source.version == 0 + || self.source.uri.is_empty() + || self.source.column.is_empty() + { + return Err(Error::InvalidInput { + message: "invalid stored blob descriptor or missing pinned source".into(), + }); + } + Ok(()) + } +} + +pub(crate) async fn open_native_blob_file( + dataset: &Dataset, + descriptor: &BlobDescriptor, +) -> Result { + descriptor.validate()?; + let source = &descriptor.source; + let expected_uri = dataset + .branch_location() + .find_branch(source.branch.as_deref())? + .uri; + if uri_to_url(&expected_uri)? != uri_to_url(&source.uri)? { + return Err(Error::InvalidInput { + message: "blob descriptor belongs to a different table".into(), + }); + } + let dataset = Arc::new( + dataset + .checkout_version(lance::dataset::refs::Ref::Version( + source.branch.clone(), + Some(source.version), + )) + .await?, + ); + if uri_to_url(dataset.uri())? != uri_to_url(&source.uri)? { + return Err(Error::InvalidInput { + message: "blob descriptor belongs to a different table".into(), + }); + } + if dataset.manifest().timestamp_nanos / 1_000 != u128::from(source.committed_at_micros) { + return Err(Error::InvalidInput { + message: "blob descriptor source snapshot has been replaced".into(), + }); + } + ensure_blob_v2_column(dataset.schema(), &source.column)?; + let mut files = if source.row_is_address { + dataset + .take_blobs_by_addresses(&[source.row_reference], &source.column) + .await? + } else { + dataset + .take_blobs(&[source.row_reference], &source.column) + .await? + }; + if files.len() != 1 { + return Err(Error::InvalidInput { + message: "blob descriptor does not resolve to one stored value".into(), + }); + } + files + .pop() + .flatten() + .map(Into::into) + .ok_or_else(|| Error::InvalidInput { + message: "non-null blob descriptor resolved to a null value".into(), + }) +} + /// Seekable handle for one blob value, backed by local storage or a remote /// HTTP byte-range endpoint. #[derive(Debug)] diff --git a/rust/lancedb/src/remote/client.rs b/rust/lancedb/src/remote/client.rs index fe4a2072e..f54e2fe3e 100644 --- a/rust/lancedb/src/remote/client.rs +++ b/rust/lancedb/src/remote/client.rs @@ -809,6 +809,11 @@ impl RestfulLanceDbClient { self.client.get(full_uri) } + pub fn head(&self, uri: &str) -> RequestBuilder { + let full_uri = format!("{}{}", self.host, uri); + self.client.head(full_uri) + } + pub fn post(&self, uri: &str) -> RequestBuilder { let full_uri = format!("{}{}", self.host, uri); self.client.post(full_uri) diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 4cbba04df..181ce61f9 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -2927,6 +2927,10 @@ impl BaseTable for RemoteTable { self.fetch_blob_files_impl(column, row_ids).await } + async fn open_blob_file(&self, descriptor: &crate::blob::BlobDescriptor) -> Result { + self.open_blob_file_impl(descriptor).await + } + async fn explain_plan(&self, query: &AnyQuery, verbose: bool) -> Result { if let AnyQuery::Query(request) = query && let Some(offsets) = &request.take_offsets diff --git a/rust/lancedb/src/remote/table/blobs.rs b/rust/lancedb/src/remote/table/blobs.rs index bcaf47765..e3cdfc6a6 100644 --- a/rust/lancedb/src/remote/table/blobs.rs +++ b/rust/lancedb/src/remote/table/blobs.rs @@ -21,7 +21,7 @@ use reqwest::{Response, StatusCode, header}; use tokio::sync::{Mutex, OwnedSemaphorePermit}; use crate::Error; -use crate::blob::{BlobFile, BlobRangeRequest}; +use crate::blob::{BlobDescriptor, BlobDescriptorSource, BlobFile, BlobRangeRequest}; use crate::error::Result; use crate::query::{QueryFilter, QueryRequest, Select}; use crate::remote::client::{HttpSend, RequestResultExt, RestfulLanceDbClient}; @@ -49,6 +49,7 @@ fn is_fetch_blobs_byte_limit_error(error: &Error) -> bool { enum RangeRequestMode { SizeProbe, DataRead, + MetadataRead, } #[async_trait::async_trait] @@ -86,6 +87,9 @@ struct TableBlobRangeRequester { path: String, version: Option, branch: Option, + source_uri: Option, + source_commit_micros: Option, + row_is_address: bool, freshness: Arc>, parent_freshness: Arc>, parent_freshness_request: FreshnessHeaders, @@ -109,15 +113,27 @@ impl BlobRangeRequester for TableBlobRangeRequester { .expect("blob request semaphore is never closed"); let freshness_request = freshness_headers_snapshot(&self.freshness, self.read_consistency_interval); - let mut request = freshness_request - .apply(self.client.get(&self.path)) - .header(header::RANGE, range_header); + let mut request = if matches!(mode, RangeRequestMode::MetadataRead) { + freshness_request.apply(self.client.head(&self.path)) + } else { + freshness_request + .apply(self.client.get(&self.path)) + .header(header::RANGE, range_header) + }; if let Some(version) = version.or(self.version) { request = request.query(&[("version", version)]); } if let Some(branch) = &self.branch { request = request.query(&[("branch", branch)]); } + if let Some(source_uri) = &self.source_uri { + request = request + .query(&[("source_uri", source_uri)]) + .query(&[("row_is_address", self.row_is_address)]); + } + if let Some(timestamp) = self.source_commit_micros { + request = request.query(&[("source_commit_micros", timestamp)]); + } let (request_id, response) = self.client.send_with_retry(request, None, true).await?; // Preserve 416 size-probe responses so the caller can detect empty blobs. if response.status() == StatusCode::RANGE_NOT_SATISFIABLE @@ -131,8 +147,10 @@ impl BlobRangeRequester for TableBlobRangeRequester { } let response = self.client.check_response(&request_id, response).await?; freshness_request.observe_headers(&self.freshness, response.headers()); - self.parent_freshness_request - .observe_headers(&self.parent_freshness, response.headers()); + if self.source_uri.is_none() { + self.parent_freshness_request + .observe_headers(&self.parent_freshness, response.headers()); + } Ok(BlobRangeResponse { request_id, response, @@ -630,6 +648,51 @@ impl RemoteTable { Ok(files) } + pub(super) async fn open_blob_file_impl( + &self, + descriptor: &BlobDescriptor, + ) -> Result { + descriptor.validate()?; + self.ensure_blob_files_supported()?; + let source = &descriptor.source; + let snapshot = ReadSnapshot { + version: Some(source.version), + freshness_state: FreshnessState { + pinned: true, + ..Default::default() + }, + ..Default::default() + }; + let requester = self.blob_range_requester_with_source( + &source.column, + source.row_reference, + &snapshot, + Some(source), + ); + let size = if descriptor.kind == 3 && descriptor.size == 0 { + let metadata = requester + .request_range("", RangeRequestMode::MetadataRead, Some(source.version)) + .await?; + if metadata.response.status() != StatusCode::OK { + return Err(Error::InvalidInput { + message: "non-null blob descriptor did not resolve to a stored value".into(), + }); + } + metadata + .response + .headers() + .get(header::CONTENT_LENGTH) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()) + .ok_or_else(|| Error::InvalidInput { + message: "blob metadata response has no valid Content-Length".into(), + })? + } else { + descriptor.size + }; + Ok(RemoteBlobFile::new(requester, size, Some(source.version)).into()) + } + /// Probe each row once for its size. /// /// Serves callers that must not issue a query, such as a single blob that @@ -793,6 +856,16 @@ impl RemoteTable { column: &str, row_id: u64, read_snapshot: &ReadSnapshot, + ) -> Arc { + self.blob_range_requester_with_source(column, row_id, read_snapshot, None) + } + + fn blob_range_requester_with_source( + &self, + column: &str, + row_id: u64, + read_snapshot: &ReadSnapshot, + source: Option<&BlobDescriptorSource>, ) -> Arc { let path = format!( "/v1/table/{}/blob/{}/{row_id}/bytes", @@ -803,7 +876,10 @@ impl RemoteTable { client: self.client.clone(), path, version: read_snapshot.version, - branch: self.branch.clone(), + branch: source.map_or_else(|| self.branch.clone(), |source| source.branch.clone()), + source_uri: source.map(|source| source.uri.clone()), + source_commit_micros: source.map(|source| source.committed_at_micros), + row_is_address: source.is_some_and(|source| source.row_is_address), freshness: Arc::new(std::sync::Mutex::new(read_snapshot.freshness_state)), parent_freshness: self.freshness.clone(), parent_freshness_request: read_snapshot.freshness, @@ -1024,6 +1100,99 @@ mod tests { .unwrap() } + #[tokio::test] + async fn open_descriptor_files_are_lazy_and_pin_ranges_for_every_storage_kind() { + for kind in 0..=5 { + let requests = Arc::new(StdMutex::new(Vec::new())); + let captured = requests.clone(); + let table = RemoteTable::new_mock( + "videos".to_string(), + move |request| { + for name in [ + "x-lancedb-min-version", + "x-lancedb-min-read-version", + "x-lancedb-min-timestamp", + ] { + assert!(request.headers().get(name).is_none()); + } + assert_eq!( + request.url().path(), + "/v1/table/videos/blob/video/12884901889/bytes" + ); + let parameters = request + .url() + .query_pairs() + .into_owned() + .collect::>(); + assert_eq!(parameters["version"], "7"); + assert_eq!(parameters["source_commit_micros"], "1234"); + assert_eq!(parameters["branch"], "dev"); + assert_eq!(parameters["row_is_address"], "true"); + assert_eq!( + parameters["source_uri"], + "s3://bucket/videos.lance/tree/dev" + ); + captured.lock().unwrap().push(request.method().clone()); + if request.method() == reqwest::Method::HEAD { + assert!(request.headers().get(header::RANGE).is_none()); + http::Response::builder() + .status(200) + .header(header::CONTENT_LENGTH, PAYLOAD.len().to_string()) + .body(Vec::new()) + .unwrap() + } else { + range_response(&request, PAYLOAD) + } + }, + Some(Version::new(0, 5, 0)), + ); + { + let mut freshness = table.freshness.lock().unwrap(); + freshness.min_version = Some(99); + freshness.min_read_version = Some(2); + } + let descriptor = BlobDescriptor { + kind, + position: 0, + size: if kind == 3 { 0 } else { PAYLOAD.len() as u64 }, + blob_id: 0, + blob_uri: String::new(), + source: BlobDescriptorSource { + uri: "s3://bucket/videos.lance/tree/dev".into(), + version: 7, + committed_at_micros: 1234, + branch: Some("dev".into()), + column: "video".into(), + row_reference: 12884901889, + row_is_address: true, + }, + }; + let file = table.open_blob_file(&descriptor).await.unwrap(); + let expected = if kind == 3 { + vec![reqwest::Method::HEAD] + } else { + vec![] + }; + assert_eq!(*requests.lock().unwrap(), expected); + assert_eq!(file.size(), PAYLOAD.len() as u64); + file.seek(4).await.unwrap(); + assert_eq!(file.read_up_to(5).await.unwrap().as_ref(), &PAYLOAD[4..9]); + assert_eq!(file.tell().await.unwrap(), 9); + assert_eq!(table.freshness.lock().unwrap().min_read_version, Some(2)); + assert_eq!(file.read_range(0..3).await.unwrap().as_ref(), &PAYLOAD[..3]); + assert_eq!(file.tell().await.unwrap(), 9); + assert_eq!( + requests + .lock() + .unwrap() + .iter() + .filter(|method| **method == reqwest::Method::GET) + .count(), + 2 + ); + } + } + fn mock_remote_blob_table( requests: Arc>>, ) -> RemoteTable { diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 3123eb776..3fcf4bdb7 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -3,7 +3,7 @@ //! LanceDB Table APIs -use crate::blob::BlobFile; +use crate::blob::{BlobDescriptor, BlobFile}; use arrow_array::{LargeBinaryArray, RecordBatch, RecordBatchReader}; use arrow_schema::{Schema, SchemaRef}; use async_trait::async_trait; @@ -796,6 +796,12 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync { message: "fetch_blob_files is not supported on this table type".into(), }) } + /// Open a descriptor at its pinned source version. See [`Table::open_blob_file`]. + async fn open_blob_file(&self, _descriptor: &BlobDescriptor) -> Result { + Err(Error::NotSupported { + message: "open_blob_file is not supported on this table type".into(), + }) + } /// Gets the table tag manager. async fn tags(&self) -> Result>; /// Optimize the dataset. @@ -1377,6 +1383,26 @@ impl Table { self.inner.fetch_blob_files(column.as_ref(), row_ids).await } + /// Open a seekable file from a SQL `blob_descriptor` result. + /// + /// The descriptor carries its table version, branch, column and row identity. + /// Opening does not materialize the payload or change this table's checkout. + /// Remote reads use the table's authenticated HTTP byte-range endpoint. + /// + /// ``` + /// use lancedb::{Table, blob::BlobDescriptor}; + /// + /// # async fn read_header(videos: &Table, descriptor: &BlobDescriptor) -> lancedb::Result<()> { + /// let video = videos.open_blob_file(descriptor).await?; + /// let header = video.read_range(0..video.size().min(4096)).await?; + /// video.close().await?; + /// # Ok(()) + /// # } + /// ``` + pub async fn open_blob_file(&self, descriptor: &BlobDescriptor) -> Result { + self.inner.open_blob_file(descriptor).await + } + /// Insert new records into this Table /// /// # Arguments @@ -3764,6 +3790,11 @@ impl BaseTable for NativeTable { crate::blob::take_blob_files_aligned(&dataset, column, row_ids).await } + async fn open_blob_file(&self, descriptor: &BlobDescriptor) -> Result { + let dataset = self.dataset.get().await?; + crate::blob::open_native_blob_file(&dataset, descriptor).await + } + /// Delete rows from the table async fn delete(&self, predicate: Predicate<'_>) -> Result { let result = delete::execute_delete(self, predicate).await?;