diff --git a/docs/web/basics/lance.mdx b/docs/web/basics/lance.mdx index 0341ee333..84996ffe9 100644 --- a/docs/web/basics/lance.mdx +++ b/docs/web/basics/lance.mdx @@ -49,9 +49,10 @@ tbl = db.open_table("frames") # opens /data/frames.lance; version=N pins a vers Use `lancedb` for search, retrieval, data curation, multimodal blob access and application code. Install `pylance` (`import lance`) only when building query engines or connectors, calling format-level fragment APIs directly, or -converting plain-scan lazy or legacy blob columns with `to_pandas()` (see -[Multimodal data and blobs](/management/table#multimodal) for `pylance`-free -conversions). +converting plain-scan lazy blob handles with `to_pandas()`. Legacy plain-scan +byte conversions use LanceDB's native path without `pylance` (see +[Multimodal data and blobs](/management/table#multimodal) for conversion +boundaries). | `pylance` (`import lance`) | `lancedb` (`import lancedb`) | | --- | --- | diff --git a/docs/web/management/table.mdx b/docs/web/management/table.mdx index 99c92f4e2..2a05b4a0f 100644 --- a/docs/web/management/table.mdx +++ b/docs/web/management/table.mdx @@ -465,9 +465,11 @@ methods `to_pandas()`, `to_arrow()`, `to_list()`, and `to_batches()` accept a For supported local v2 blob projections, `to_pandas(blob_mode="bytes")` and `to_pandas(blob_mode="descriptions")` use native Arrow conversion without -pylance. Pandas conversions that need the Lance scanner, including plain-scan -`"lazy"` handles and legacy blobs, still require pylance. `to_arrow()`, -`to_list()`, `to_batches()`, and the fetch methods do not require pylance. +pylance. Eligible local plain scans also convert row-addressable legacy blob +bytes without pylance. Pandas conversions that need the Lance scanner, +including plain-scan `"lazy"` handles and scanner-only options, still require +pylance. `to_arrow()`, `to_list()`, `to_batches()`, and the fetch methods do +not require pylance for supported local queries. `blob_mode="bytes"` requires a LanceDB Embedded table snapshot, so Enterprise and in-memory tables raise `NotImplementedError`. On an Enterprise table, @@ -531,10 +533,11 @@ PyArrow. ###### Stream blob results {#multimodal-blob-batches} -`to_batches(blob_mode="bytes")` fetches blob v2 payloads one batch at a time -and returns them as Arrow `large_binary` values. Iterate over the reader and -call `batch.to_pandas()` to convert each batch to a DataFrame without collecting -all payloads at once. +`to_batches(blob_mode="bytes")` fetches row-addressable blob v2 payloads and +eligible local plain-scan legacy v1 payloads one batch at a time and returns +them as Arrow `large_binary` values. Iterate over the reader and call +`batch.to_pandas()` to convert each batch to a DataFrame without collecting all +payloads at once. For sync queries, call `query.to_batches(batch_size=64, blob_mode="bytes")` and iterate with `for`. For async queries, call @@ -542,17 +545,18 @@ and iterate with `for`. For async queries, call with `async for`. Sync full-text and hybrid queries do not support `to_batches()`. -The default `"lazy"` mode returns descriptors and keeps automatic `_rowid` -columns visible for manual fetches. `"descriptions"` returns the five-field -descriptors. Both `"bytes"` and `"descriptions"` hide automatic row IDs; -explicitly requested row IDs remain visible. +The default `"lazy"` mode returns blob v2 descriptors and keeps their automatic +`_rowid` columns visible for manual fetches. `"descriptions"` returns the +five-field descriptors. Both `"bytes"` and `"descriptions"` hide automatic row +IDs; explicitly requested row IDs remain visible. Bytes mode has the same restrictions as `to_arrow(blob_mode="bytes")`, -including no legacy blobs, blobs inside lists, Enterprise tables, or namespace -`QueryTable` pushdown. If concurrent writers can overwrite the table, use a -separate reader checked out to a fixed version before querying. Otherwise, -an overwrite can invalidate the row IDs used to fetch bytes and raise -`ValueError`. +including no blobs inside lists, Enterprise tables, namespace `QueryTable` +pushdown, or query shapes that cannot preserve their row results. Legacy v1 +bytes are limited to eligible local plain scans. If concurrent writers can +overwrite the table, use a separate reader checked out to a fixed version +before querying. Otherwise, an overwrite can invalidate the row IDs used to +fetch bytes and raise `ValueError`. ###### Optional pandas examples {#multimodal-blob-pandas} diff --git a/docs/web/quickstart.mdx b/docs/web/quickstart.mdx index 5bd9a3efe..ef462cbf7 100644 --- a/docs/web/quickstart.mdx +++ b/docs/web/quickstart.mdx @@ -59,8 +59,9 @@ uv add --prerelease allow --index https://pypi.fury.io/lancedb/ lancedb Most applications only need `lancedb`. Install the optional `pylance` extra and include the Lance preview index only when building query engines or connectors, -calling format-level `lance` APIs directly, or converting plain-scan lazy or -legacy blob columns with `to_pandas()`: +calling format-level `lance` APIs directly, or converting plain-scan lazy blob +handles with `to_pandas()`. Legacy plain-scan byte conversions use LanceDB's +native path and do not require `pylance`: ```bash Python (pip) icon="terminal" diff --git a/python/python/lancedb/_blob.py b/python/python/lancedb/_blob.py index 010c92009..2c545a17e 100644 --- a/python/python/lancedb/_blob.py +++ b/python/python/lancedb/_blob.py @@ -1,7 +1,7 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright The LanceDB Authors -"""Blob fetch API and v2 projection helpers.""" +"""Blob fetch API and projection helpers.""" from __future__ import annotations @@ -13,7 +13,11 @@ from typing import TYPE_CHECKING, Optional, Union import pyarrow as pa from .expr import Expr -from .schema import blob_column_paths, row_addressable_blob_v2_paths +from .schema import ( + blob_column_paths, + row_addressable_blob_paths, + row_addressable_blob_v2_paths, +) from .types import BlobMode, QueryProjection, QueryProjectionSpec if TYPE_CHECKING: @@ -165,6 +169,25 @@ def blob_v2_projection_sources( return _project_blob_paths(row_addressable_blob_v2_paths(schema), projection) +def blob_projection_sources( + schema: pa.Schema, + projection: QueryProjection, + *, + include_legacy: bool = False, +) -> dict[str, str]: + """Map projected blob outputs to row-addressable source columns. + + Legacy v1 columns are included only when the caller can safely fetch them + through the same row-id path as v2. + """ + paths = ( + row_addressable_blob_paths(schema) + if include_legacy + else row_addressable_blob_v2_paths(schema) + ) + return _project_blob_paths(paths, projection) + + def reject_unfetchable_blob_columns( schema: pa.Schema, projection: QueryProjection, blob_sources: dict[str, str] ) -> None: @@ -177,8 +200,9 @@ def reject_unfetchable_blob_columns( if unfetchable: raise ValueError( f"blob_mode='bytes' cannot fetch blob columns {unfetchable}. Bytes are " - "fetched by row id, which covers blob v2 columns with one blob per " - "row but not legacy blob columns or blobs inside lists. Use " + "fetched by row id for legacy and v2 columns in local plain scans, " + "but not for blobs inside lists or query shapes whose result columns " + "cannot be matched back to the source. Use " "blob_mode='descriptions' or remove these columns from the projection." ) @@ -269,7 +293,7 @@ def finalize_blob_query_table( return stash_auto_row_ids(tbl, blob_paths) -async def replace_v2_blob_columns_with_bytes( +async def replace_blob_columns_with_bytes( tbl: pa.Table, blob_sources: dict[str, str], fetch_blobs: FetchBlobsAsync, @@ -289,7 +313,7 @@ async def replace_v2_blob_columns_with_bytes( return tbl -def replace_v2_blob_columns_with_bytes_sync( +def replace_blob_columns_with_bytes_sync( tbl: pa.Table, blob_sources: dict[str, str], fetch_blobs: FetchBlobsSync, diff --git a/python/python/lancedb/query.py b/python/python/lancedb/query.py index ac15f013c..bab40be95 100644 --- a/python/python/lancedb/query.py +++ b/python/python/lancedb/query.py @@ -48,12 +48,13 @@ from ._blob import ( FetchBlobsAsync, FetchBlobsSync, blob_auto_row_id_for_scan, + blob_projection_sources, blob_v2_projection_sources, finalize_blob_query_table, reject_unfetchable_blob_columns, reject_unmatched_blob_outputs, - replace_v2_blob_columns_with_bytes, - replace_v2_blob_columns_with_bytes_sync, + replace_blob_columns_with_bytes, + replace_blob_columns_with_bytes_sync, strip_auto_row_ids, validate_blob_mode, ) @@ -245,6 +246,17 @@ def _query_is_plain_scan(query: Query) -> bool: ) +def _query_is_plain_blob_scan(query: Query) -> bool: + """Whether native batch conversion can preserve this query's row result.""" + return ( + query.vector is None + and query.full_text_query is None + and query.take_offsets is None + and query.take_row_ids is None + and not query.postfilter + ) + + def _filter_to_sql(filter: Optional[Union[str, Expr]]) -> Optional[str]: if filter is None: return None @@ -421,7 +433,7 @@ def _finish_plain_scan_pandas( ) -> pd.DataFrame: if blob_sources: tbl = _scanner_to_table(scanner) - tbl = replace_v2_blob_columns_with_bytes_sync(tbl, blob_sources, fetch_blobs) + tbl = replace_blob_columns_with_bytes_sync(tbl, blob_sources, fetch_blobs) if strip_auto_row_id and "_rowid" in tbl.column_names: tbl = tbl.drop_columns(["_rowid"]) if flatten is not None: @@ -475,7 +487,7 @@ async def _finish_plain_scan_pandas_async( ) -> pd.DataFrame: if blob_sources: tbl = _scanner_to_table(scanner) - tbl = await replace_v2_blob_columns_with_bytes(tbl, blob_sources, fetch_blobs) + tbl = await replace_blob_columns_with_bytes(tbl, blob_sources, fetch_blobs) if strip_auto_row_id and "_rowid" in tbl.column_names: tbl = tbl.drop_columns(["_rowid"]) if flatten is not None: @@ -1281,10 +1293,10 @@ class LanceQueryBuilder(ABC): blob_mode: str, default "lazy" Controls how blob columns are returned. Local blob v2 queries support ``"bytes"`` and ``"descriptions"`` without pylance. Eligible local - plain scans use pylance's scanner for lazy blob handles in ``"lazy"`` - mode and for legacy blob columns. Other local blob v2 - queries return descriptors in ``"lazy"`` and ``"descriptions"`` - modes, or payloads in ``"bytes"`` mode, through :meth:`to_arrow`. + plain scans also support ``"bytes"`` for row-addressable legacy blob + columns without pylance. Other local blob v2 queries return + descriptors in ``"lazy"`` and ``"descriptions"`` modes, or payloads + in ``"bytes"`` mode, through :meth:`to_arrow`. LanceDB Enterprise blob queries support only ``"descriptions"`` until a stable table snapshot can bind the query and blob fetches. @@ -1352,8 +1364,8 @@ class LanceQueryBuilder(ABC): blob_mode: str, default "lazy" ``"lazy"`` returns blob v2 descriptors that carry the row id ``fetch_blobs`` reads. ``"descriptions"`` returns them without it. - ``"bytes"`` returns payload bytes and raises for legacy blob columns - and blobs inside lists. + ``"bytes"`` returns payload bytes for supported row-addressable blob + columns and raises for blobs inside lists. """ raise NotImplementedError @@ -1370,7 +1382,7 @@ class LanceQueryBuilder(ABC): Execute the query and return the results as a pyarrow [RecordBatchReader](https://arrow.apache.org/docs/python/generated/pyarrow.RecordBatchReader.html) - For v2 blob projections, the default ``blob_mode="lazy"`` keeps the + For blob v2 projections, the default ``blob_mode="lazy"`` keeps the auto ``_rowid`` column visible so batch consumers can call ``fetch_blobs``. The ``"bytes"`` and ``"descriptions"`` modes hide automatic row ids. @@ -1390,10 +1402,11 @@ class LanceQueryBuilder(ABC): The timeout also covers reading the batches, so consuming them slowly can exceed it. blob_mode: str, default "lazy" - ``"lazy"`` keeps blob descriptors and auto row ids for manual fetches. - ``"descriptions"`` omits auto row ids. ``"bytes"`` fetches blob v2 - payloads one batch at a time, without requiring pylance. Bytes mode - has the same restrictions as :meth:`to_arrow`. + ``"lazy"`` keeps blob v2 descriptors and auto row ids for manual + fetches. ``"descriptions"`` omits auto row ids. ``"bytes"`` fetches + supported row-addressable blob payloads one batch at a time, without + requiring pylance. Bytes mode has the same restrictions as + :meth:`to_arrow`. Examples -------- @@ -1599,20 +1612,41 @@ class LanceQueryBuilder(ABC): def _scan_needs_row_id(self) -> bool: return self._user_requested_row_id() or self._blob_auto_row_id_enabled() - def _query_for_scan(self) -> Query: + def _query_for_scan(self, blob_sources: Optional[dict[str, str]] = None) -> Query: query = self.to_query_object() - if self._scan_needs_row_id(): + if self._scan_needs_row_id() or blob_sources: query.with_row_id = True return query + def _can_fetch_plain_legacy_blobs(self, blob_mode: BlobMode) -> bool: + if blob_mode != "bytes": + return False + try: + query = self.to_query_object() + except NotImplementedError: + return False + return ( + _query_is_plain_blob_scan(query) + and getattr(self, "_reranker", None) is None + and not _query_needs_lance_scanner(self) + and _supports_local_blob_fetch( + self._table._table, + getattr(self._table, "_pushdown_operations", ()), + ) + ) + def _validated_blob_sources(self, blob_mode: BlobMode) -> dict[str, str]: - """Map projected blob v2 outputs to their source columns. + """Map projected blob outputs to their source columns. Bytes mode is checked here, before the query runs. """ validate_blob_mode(blob_mode) schema = self._table.schema - blob_sources = blob_v2_projection_sources(schema, self._columns) + blob_sources = blob_projection_sources( + schema, + self._columns, + include_legacy=self._can_fetch_plain_legacy_blobs(blob_mode), + ) if blob_mode != "bytes": return blob_sources reject_unfetchable_blob_columns(schema, self._columns, blob_sources) @@ -1633,7 +1667,7 @@ class LanceQueryBuilder(ABC): return tbl if blob_mode == "descriptions": return strip_auto_row_ids(tbl, blob_sources) - return replace_v2_blob_columns_with_bytes_sync( + return replace_blob_columns_with_bytes_sync( tbl, blob_sources, self._table.fetch_blobs ) @@ -1645,18 +1679,23 @@ class LanceQueryBuilder(ABC): blob_sources: dict[str, str], ) -> pa.Table: blob_auto_row_id = self._blob_auto_row_id_enabled() - finalized = finalize_blob_query_table( - tbl, - user_requested_row_id=self._user_requested_row_id(), - blob_auto_row_id=blob_auto_row_id, - blob_paths=blob_sources.keys() if blob_auto_row_id else (), - ) + if blob_mode == "bytes" and blob_sources: + # Bytes conversion reads _rowid directly, including when the + # projection mixes legacy and v2 columns. + finalized = tbl + else: + finalized = finalize_blob_query_table( + tbl, + user_requested_row_id=self._user_requested_row_id(), + blob_auto_row_id=blob_auto_row_id, + blob_paths=blob_sources.keys() if blob_auto_row_id else (), + ) finalized = self._apply_arrow_blob_mode(finalized, blob_mode, blob_sources) # with_row_id(False) does not reach the scan, which adds _rowid anyway. # Lazy keeps it because fetch_blobs(hits) reads it there. if ( blob_mode != "lazy" - and self._with_row_id is False + and not self._user_requested_row_id() and "_rowid" in finalized.column_names ): finalized = finalized.drop_columns(["_rowid"]) @@ -1896,18 +1935,29 @@ class LanceQueryBuilder(ABC): query = self.to_query_object() if not _query_is_plain_scan(query): return None + blob_sources = ( + blob_projection_sources( + self._table.schema, + query.columns, + include_legacy=self._can_fetch_plain_legacy_blobs(blob_mode), + ) + if blob_mode == "bytes" + else {} + ) # Remote and in-memory tables have no Lance dataset to scan. if not self._table._table._inner._is_native() or ( get_uri_scheme(self._table._dataset_path) == "memory" ): return None - # Blob v2 payloads and descriptions use LanceDB's native fetch path. - # Only lazy handles, legacy blobs, and scanner-only options need pylance. - if ( - blob_mode != "lazy" - and not _query_needs_lance_scanner(self) - and blob_v2_projection_sources(self._table.schema, query.columns) + # Bytes conversions use LanceDB's fetch path for both blob versions. + # Descriptions for v2 do too; lazy handles still need the scanner. + if not _query_needs_lance_scanner(self) and ( + (blob_mode == "bytes" and blob_sources) + or ( + blob_mode == "descriptions" + and blob_v2_projection_sources(self._table.schema, query.columns) + ) ): return None try: @@ -1915,12 +1965,7 @@ class LanceQueryBuilder(ABC): except ImportError as err: raise _unsupported_blob_pandas_error(str(err)) from err blob_auto_row_id = self._blob_auto_row_id_enabled() - blob_sources = ( - blob_v2_projection_sources(self._table.schema, query.columns) - if blob_mode == "bytes" - else {} - ) - if blob_sources: + if blob_mode == "bytes": reject_unfetchable_blob_columns( self._table.schema, query.columns, blob_sources ) @@ -2254,8 +2299,8 @@ class LanceVectorQueryBuilder(LanceQueryBuilder): blob_mode: str, default "lazy" ``"lazy"`` returns blob v2 descriptors that carry the row id ``fetch_blobs`` reads. ``"descriptions"`` returns them without it. - ``"bytes"`` returns payload bytes and raises for legacy blob columns - and blobs inside lists. + ``"bytes"`` returns payload bytes for supported row-addressable blob + columns and raises for blobs inside lists. """ blob_sources = self._validated_blob_sources(blob_mode) return self._finalize_blob_query_table( @@ -2333,7 +2378,7 @@ class LanceVectorQueryBuilder(LanceQueryBuilder): vector = self._query if isinstance(self._query, list) else self._query.tolist() if isinstance(vector[0], np.ndarray): vector = [v.tolist() for v in vector] - query = self._query_for_scan() + query = self._query_for_scan(blob_sources) result_set = self._table._execute_query( query, batch_size=batch_size, timeout=timeout ) @@ -2551,7 +2596,7 @@ class LanceFtsQueryBuilder(LanceQueryBuilder): blob_sources = self._validated_blob_sources(blob_mode) self._table._ensure_no_legacy_fts_index() - query = self._query_for_scan() + query = self._query_for_scan(blob_sources) results = self._table._execute_query(query, timeout=timeout).read_all() if self._reranker is not None: results = self._reranker.rerank_fts(self._query, results) @@ -2597,6 +2642,8 @@ class LanceEmptyQueryBuilder(LanceQueryBuilder): blob_mode: BlobMode = "lazy", ) -> pa.Table: blob_sources = self._validated_blob_sources(blob_mode) + if blob_mode == "bytes" and blob_sources: + return self.to_batches(timeout=timeout, blob_mode=blob_mode).read_all() return self._finalize_blob_query_table( self.to_batches(timeout=timeout).read_all(), blob_mode=blob_mode, @@ -2630,7 +2677,7 @@ class LanceEmptyQueryBuilder(LanceQueryBuilder): blob_mode: BlobMode = "lazy", ) -> pa.RecordBatchReader: blob_sources = self._validated_blob_sources(blob_mode) - query = self._query_for_scan() + query = self._query_for_scan(blob_sources) reader = self._table._execute_query( query, batch_size=batch_size, timeout=timeout ) @@ -3501,6 +3548,22 @@ class AsyncQueryBase(object): self._blob_paths = tuple(blob_v2_projection_sources(schema, projection).keys()) self._inner.with_row_id() + async def _can_fetch_plain_legacy_blobs(self, blob_mode: BlobMode) -> bool: + if blob_mode != "bytes" or self._table is None: + return False + try: + query = self.to_query_object() + except NotImplementedError: + return False + return ( + _query_is_plain_blob_scan(query) + and getattr(self, "_reranker", None) is None + and not _query_needs_lance_scanner(self) + and _supports_local_blob_fetch( + self._table, self._table._pushdown_operations + ) + ) + def select(self, columns: Union[List[str], dict[str, str]]) -> Self: """ Return only the specified columns. @@ -3608,10 +3671,11 @@ class AsyncQueryBase(object): The timeout also covers reading the batches, so consuming them slowly can exceed it. blob_mode: str, default "lazy" - ``"lazy"`` keeps blob descriptors and auto row ids for manual fetches. - ``"descriptions"`` omits auto row ids. ``"bytes"`` fetches blob v2 - payloads one batch at a time, without requiring pylance. Bytes mode - has the same restrictions as :meth:`to_arrow`. + ``"lazy"`` keeps blob v2 descriptors and auto row ids for manual + fetches. ``"descriptions"`` omits auto row ids. ``"bytes"`` fetches + supported row-addressable blob payloads one batch at a time, without + requiring pylance. Bytes mode has the same restrictions as + :meth:`to_arrow`. Examples -------- @@ -3620,13 +3684,16 @@ class AsyncQueryBase(object): """ blob_sources = await self._validated_blob_sources(blob_mode) reader = await self._execute_batches( - max_batch_length=max_batch_length, timeout=timeout + max_batch_length=max_batch_length, + timeout=timeout, + blob_sources=blob_sources, ) if blob_mode == "lazy": return reader async def convert(tbl: pa.Table) -> pa.Table: - tbl = self._finalize_blob_query_table(tbl) + if not (blob_mode == "bytes" and blob_sources): + tbl = self._finalize_blob_query_table(tbl) tbl = await self._apply_arrow_blob_mode(tbl, blob_mode, blob_sources) if not self._user_requested_row_id() and "_rowid" in tbl.column_names: tbl = tbl.drop_columns(["_rowid"]) @@ -3647,7 +3714,24 @@ class AsyncQueryBase(object): *, max_batch_length: Optional[int] = None, timeout: Optional[timedelta] = None, + blob_sources: Optional[dict[str, str]] = None, ) -> AsyncRecordBatchReader: + if ( + blob_sources + and not self._user_requested_row_id() + and await self._can_fetch_plain_legacy_blobs("bytes") + ): + # Build a separate query so the hidden row id needed for byte + # conversion cannot leak into later executions of this builder. + query = self.to_query_object() + query.with_row_id = True + if self._table is None: + raise RuntimeError("blob query is not bound to a table") + row_id_query = self._table._sync_query_to_async(query) + return await row_id_query._execute_batches( + max_batch_length=max_batch_length, + timeout=timeout, + ) await self._maybe_add_blob_row_id() return AsyncRecordBatchReader( await self._inner.execute( @@ -3664,7 +3748,7 @@ class AsyncQueryBase(object): return await self._inner.output_schema() async def _validated_blob_sources(self, blob_mode: BlobMode) -> dict[str, str]: - """Map projected blob v2 outputs to their source columns. + """Map projected blob outputs to their source columns. Bytes mode is checked here, before the query runs. """ @@ -3674,7 +3758,11 @@ class AsyncQueryBase(object): return {} schema = await self._table.schema() projection = _query_request_projection(self._inner.to_query_request()) - blob_sources = blob_v2_projection_sources(schema, projection) + blob_sources = blob_projection_sources( + schema, + projection, + include_legacy=await self._can_fetch_plain_legacy_blobs(blob_mode), + ) if blob_mode != "bytes": return blob_sources reject_unfetchable_blob_columns(schema, projection, blob_sources) @@ -3695,7 +3783,7 @@ class AsyncQueryBase(object): return tbl if blob_mode == "descriptions": return strip_auto_row_ids(tbl, blob_sources) - return await replace_v2_blob_columns_with_bytes( + return await replace_blob_columns_with_bytes( tbl, blob_sources, self._table.fetch_blobs ) @@ -3721,11 +3809,14 @@ class AsyncQueryBase(object): blob_mode: str, default "lazy" ``"lazy"`` returns blob v2 descriptors that carry the row id ``fetch_blobs`` reads. ``"descriptions"`` returns them without it. - ``"bytes"`` returns payload bytes and raises for legacy blob columns - and blobs inside lists. + ``"bytes"`` returns payload bytes for supported row-addressable blob + columns and raises for blobs inside lists. LanceDB Enterprise and namespace query pushdown refuse ``"bytes"``. """ blob_sources = await self._validated_blob_sources(blob_mode) + if blob_mode == "bytes" and blob_sources: + reader = await self.to_batches(timeout=timeout, blob_mode=blob_mode) + return pa.Table.from_batches(await reader.read_all(), schema=reader.schema) batch_iter = await self.to_batches(timeout=timeout) tbl = self._finalize_blob_query_table( pa.Table.from_batches(await batch_iter.read_all(), schema=batch_iter.schema) @@ -3798,10 +3889,10 @@ class AsyncQueryBase(object): blob_mode: str, default "lazy" Controls how blob columns are returned. Local blob v2 queries support ``"bytes"`` and ``"descriptions"`` without pylance. Eligible local - plain scans use pylance's scanner for lazy blob handles in ``"lazy"`` - mode and for legacy blob columns. Other local blob v2 - queries return descriptors in ``"lazy"`` and ``"descriptions"`` - modes, or payloads in ``"bytes"`` mode, through :meth:`to_arrow`. + plain scans also support ``"bytes"`` for row-addressable legacy blob + columns without pylance. Other local blob v2 queries return + descriptors in ``"lazy"`` and ``"descriptions"`` modes, or payloads + in ``"bytes"`` mode, through :meth:`to_arrow`. LanceDB Enterprise blob queries support only ``"descriptions"`` until a stable table snapshot can bind the query and blob fetches. @@ -3879,21 +3970,27 @@ class AsyncQueryBase(object): with_row_id=self._with_row_id, ) blob_sources = ( - blob_v2_projection_sources(schema, query.columns) + blob_projection_sources( + schema, + query.columns, + include_legacy=await self._can_fetch_plain_legacy_blobs(blob_mode), + ) if blob_mode == "bytes" else {} ) - if blob_sources: + if blob_mode == "bytes": reject_unfetchable_blob_columns(schema, query.columns, blob_sources) # Remote and in-memory tables have no Lance dataset to scan. if not self._table._inner._is_native() or ( get_uri_scheme(await self._table.uri()) == "memory" ): return None - if ( - blob_mode != "lazy" - and not _query_needs_lance_scanner(self) - and blob_v2_projection_sources(schema, query.columns) + if not _query_needs_lance_scanner(self) and ( + (blob_mode == "bytes" and blob_sources) + or ( + blob_mode == "descriptions" + and blob_v2_projection_sources(schema, query.columns) + ) ): return None try: @@ -4404,8 +4501,11 @@ class AsyncFTSQuery(AsyncStandardQuery): *, max_batch_length: Optional[int] = None, timeout: Optional[timedelta] = None, + blob_sources: Optional[dict[str, str]] = None, ) -> AsyncRecordBatchReader: - reader = await super()._execute_batches(timeout=timeout) + reader = await super()._execute_batches( + timeout=timeout, blob_sources=blob_sources + ) results = pa.Table.from_batches(await reader.read_all(), reader.schema) if self._reranker: results = self._reranker.rerank_fts(self.get_query(), results) @@ -4728,8 +4828,11 @@ class AsyncVectorQuery(AsyncStandardQuery, AsyncVectorQueryBase): *, max_batch_length: Optional[int] = None, timeout: Optional[timedelta] = None, + blob_sources: Optional[dict[str, str]] = None, ) -> AsyncRecordBatchReader: - reader = await super()._execute_batches(timeout=timeout) + reader = await super()._execute_batches( + timeout=timeout, blob_sources=blob_sources + ) results = pa.Table.from_batches(await reader.read_all(), reader.schema) if self._reranker: results = self._reranker.rerank_vector(self._query_string, results) @@ -4857,11 +4960,14 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase): *, max_batch_length: Optional[int] = None, timeout: Optional[timedelta] = None, + blob_sources: Optional[dict[str, str]] = None, ) -> AsyncRecordBatchReader: # WAL-PK-FUSION: without the fallback, this body is `_run_hybrid`'s. return await _wal_hybrid.with_pk_fallback_async( lambda: self._run_hybrid( - max_batch_length=max_batch_length, timeout=timeout + max_batch_length=max_batch_length, + timeout=timeout, + blob_sources=blob_sources, ), self._table, fused=lambda: self._pk_fusion is not None, @@ -4873,19 +4979,18 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase): *, max_batch_length: Optional[int] = None, timeout: Optional[timedelta] = None, + blob_sources: Optional[dict[str, str]] = None, ) -> AsyncRecordBatchReader: fts_query, vec_query, limit, offset = await self._create_child_queries() req = fts_query._inner.to_query_request() - blob_auto_row_id = False + blob_auto_row_id = bool(blob_sources) blob_paths: tuple[str, ...] = () if self._table is not None: schema = await self._table.schema() projection = _query_request_projection(req) - blob_auto_row_id = blob_auto_row_id_for_scan( - schema, - projection, - with_row_id=self._with_row_id, + blob_auto_row_id = blob_auto_row_id or blob_auto_row_id_for_scan( + schema, projection, with_row_id=self._with_row_id ) if blob_auto_row_id: blob_paths = tuple( @@ -5182,8 +5287,7 @@ class BaseQueryBuilder(object): blob_mode: str, default "lazy" ``"lazy"`` returns blob v2 descriptors that carry the row id ``fetch_blobs`` reads. ``"descriptions"`` returns them without it. - ``"bytes"`` returns payload bytes and raises for legacy blob columns - and blobs inside lists. + ``"bytes"`` is not supported for remote or pushdown blob queries. """ return LOOP.run(self._inner.to_arrow(timeout, blob_mode=blob_mode)) diff --git a/python/python/lancedb/schema.py b/python/python/lancedb/schema.py index d92f7f692..5e36ffc86 100644 --- a/python/python/lancedb/schema.py +++ b/python/python/lancedb/schema.py @@ -122,6 +122,15 @@ def blob_v2_column_paths(schema: pa.Schema) -> list[str]: return [path for path, _ in _collect_blob_paths(schema, is_blob_v2_field)] +def row_addressable_blob_paths(schema: pa.Schema) -> list[str]: + """Blob v1 and v2 paths with one blob addressable by table row id.""" + return [ + path + for path, has_list_ancestor in _collect_blob_paths(schema, is_blob_like_field) + if not has_list_ancestor + ] + + def row_addressable_blob_v2_paths(schema: pa.Schema) -> list[str]: """Blob v2 paths with one blob addressable by table row id. diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index 59a7384fd..f113baa05 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -1589,8 +1589,8 @@ class Table(ABC): blob_mode: str, default "lazy" ``"lazy"`` returns blob v2 descriptors that carry the row id ``fetch_blobs`` reads. ``"descriptions"`` returns them without it. - ``"bytes"`` returns payload bytes and raises for legacy blob columns - and blobs inside lists. + ``"bytes"`` returns payload bytes for supported row-addressable blob + columns and raises for blobs inside lists. Returns ------- @@ -4160,7 +4160,7 @@ class LanceTable(Table): ): return self.to_arrow().to_pandas(**kwargs) - if blob_mode == "bytes" and blob_v2_column_paths(self.schema): + if blob_mode == "bytes" and schema_has_blob_field(self.schema): return self.search().to_pandas(blob_mode=blob_mode, **kwargs) return self.to_lance().to_pandas(blob_mode=blob_mode, **kwargs) @@ -4173,8 +4173,8 @@ class LanceTable(Table): blob_mode: str, default "lazy" ``"lazy"`` returns blob v2 descriptors that carry the row id ``fetch_blobs`` reads. ``"descriptions"`` returns them without it. - ``"bytes"`` returns payload bytes and raises for legacy blob columns - and blobs inside lists. + ``"bytes"`` returns payload bytes for supported row-addressable blob + columns and raises for blobs inside lists. Returns ------- @@ -6640,7 +6640,7 @@ class AsyncTable: if blob_mode == "lazy" and get_uri_scheme(await self.uri()) == "memory": return (await self.to_arrow()).to_pandas(**kwargs) - if blob_mode == "bytes" and blob_v2_column_paths(schema): + if blob_mode == "bytes" and schema_has_blob_field(schema): return await self.query().to_pandas(blob_mode=blob_mode, **kwargs) return (await self.to_lance()).to_pandas(blob_mode=blob_mode, **kwargs) @@ -6652,8 +6652,8 @@ class AsyncTable: blob_mode: str, default "lazy" ``"lazy"`` returns blob v2 descriptors that carry the row id ``fetch_blobs`` reads. ``"descriptions"`` returns them without it. - ``"bytes"`` returns payload bytes and raises for legacy blob columns - and blobs inside lists. + ``"bytes"`` returns payload bytes for supported row-addressable blob + columns and raises for blobs inside lists. Returns ------- @@ -7460,7 +7460,10 @@ class AsyncTable: if query.offset is not None: async_query = async_query.offset(query.offset) if query.columns: - async_query = async_query.select(query.columns) + columns = query.columns + if isinstance(columns, list) and columns and isinstance(columns[0], tuple): + columns = dict(columns) + async_query = async_query.select(columns) if query.filter is not None: async_query = async_query.where(query.filter) if query.fast_search: diff --git a/python/python/tests/test_blob.py b/python/python/tests/test_blob.py index ebafdfbe9..bb946c37c 100644 --- a/python/python/tests/test_blob.py +++ b/python/python/tests/test_blob.py @@ -1651,16 +1651,41 @@ def test_bytes_mode_rejects_blobs_inside_lists(): def test_row_addressable_paths_exclude_list_children(): - from lancedb.schema import row_addressable_blob_v2_paths + from lancedb.schema import ( + row_addressable_blob_paths, + row_addressable_blob_v2_paths, + ) schema = pa.schema( [ pa.field("id", pa.int64()), - pa.field("info", pa.struct([lancedb.blob("blob")])), + pa.field( + "info", + pa.struct( + [ + lancedb.blob("blob"), + pa.field( + "legacy", + pa.large_binary(), + metadata={"lance-encoding:blob": "true"}, + ), + ] + ), + ), + pa.field( + "legacy", + pa.large_binary(), + metadata={"lance-encoding:blob": "true"}, + ), pa.field("images", pa.list_(lancedb.blob("image"))), ] ) assert blob_v2_column_paths(schema) == ["info.blob", "images.image"] + assert row_addressable_blob_paths(schema) == [ + "info.blob", + "info.legacy", + "legacy", + ] assert row_addressable_blob_v2_paths(schema) == ["info.blob"] @@ -2325,6 +2350,8 @@ async def test_async_to_batches_blob_modes(shape, blob_mode, monkeypatch): if blob_mode != "lazy": assert "_rowid" not in reader.schema.names assert "_lance_row_id" not in str(reader.schema) + else: + assert "_rowid" in reader.schema.names class _KeepAllScoresReranker(Reranker): @@ -2368,7 +2395,7 @@ def test_bytes_mode_rejects_legacy_blob_after_reranker_drops_metadata(tmp_path): .rerank(_KeepAllScoresReranker()) ) - with pytest.raises(ValueError, match=r"\['blob'\].*legacy blob columns"): + with pytest.raises(ValueError, match=r"\['blob'\].*local plain scans"): query.to_arrow(blob_mode="bytes") diff --git a/python/python/tests/test_blob_query_without_pylance.py b/python/python/tests/test_blob_query_without_pylance.py index a5ce6db44..7c503d1ff 100644 --- a/python/python/tests/test_blob_query_without_pylance.py +++ b/python/python/tests/test_blob_query_without_pylance.py @@ -8,12 +8,7 @@ import textwrap import pytest -@pytest.mark.parametrize("api", ["sync", "async"]) -@pytest.mark.parametrize("result", ["pandas", "batches", "lazy_error"]) -def test_blob_query_without_pylance(tmp_path, api, result): - if result == "pandas": - pytest.importorskip("pandas") - script = """\ +_WITHOUT_LANCE = """\ import importlib.abc import sys @@ -23,7 +18,15 @@ class MissingLance(importlib.abc.MetaPathFinder): raise ModuleNotFoundError(fullname, name="lance") sys.meta_path.insert(0, MissingLance()) -""" + textwrap.dedent( +""" + + +@pytest.mark.parametrize("api", ["sync", "async"]) +@pytest.mark.parametrize("result", ["pandas", "batches", "lazy_error"]) +def test_blob_query_without_pylance(tmp_path, api, result): + if result == "pandas": + pytest.importorskip("pandas") + script = _WITHOUT_LANCE + textwrap.dedent( """\ import asyncio import lancedb @@ -94,3 +97,187 @@ sys.meta_path.insert(0, MissingLance()) text=True, ) assert completed.returncode == 0, completed.stderr + + +@pytest.mark.parametrize("api", ["sync", "async"]) +@pytest.mark.parametrize("result", ["pandas", "batches", "arrow"]) +def test_plain_legacy_blob_query_without_pylance(tmp_path, api, result): + if result == "pandas": + pytest.importorskip("pandas") + script = _WITHOUT_LANCE + textwrap.dedent( + """\ + import asyncio + import lancedb + import pyarrow as pa + from lancedb.query import ColumnOrdering + + schema = pa.schema( + [ + pa.field("id", pa.int64()), + pa.field("filename", pa.string()), + pa.field( + "video_blob", + pa.large_binary(), + metadata={"lance-encoding:blob": "true"}, + ), + ] + ) + rows = [ + {"id": 1, "filename": "file-1.mp4", "video_blob": b"outside"}, + {"id": 2, "filename": "file-2.mp4", "video_blob": b""}, + {"id": 3, "filename": "file-3.mp4", "video_blob": None}, + {"id": 4, "filename": "file-4.mp4", "video_blob": b"four"}, + ] + expected = [ + {"filename": "file-2.mp4", "payload": b""}, + {"filename": "file-3.mp4", "payload": None}, + {"filename": "file-4.mp4", "payload": b"four"}, + ] + + def configure(query): + return ( + query.select({"filename": "filename", "payload": "video_blob"}) + .where("id >= 2") + .order_by( + [ColumnOrdering(column_name="id", ascending=True)] + ) + .limit(3) + ) + + async def main(): + storage_options = {"new_table_data_storage_version": "2.1"} + if sys.argv[2] == "async": + db = await lancedb.connect_async( + sys.argv[1], storage_options=storage_options + ) + table = await db.create_table("legacy", schema=schema) + await table.add(rows) + table = await db.open_table("legacy") + def query(): + return configure(table.query()) + def original_query(): + return table.query().select( + {"filename": "filename", "payload": "video_blob"} + ).limit(1) + def empty_query(): + return query().where("id < 0") + def batches(): + return query().to_batches( + max_batch_length=1, blob_mode="bytes" + ) + else: + db = lancedb.connect( + sys.argv[1], storage_options=storage_options + ) + table = db.create_table("legacy", schema=schema) + table.add(rows) + table = db.open_table("legacy") + def query(): + return configure(table.search()) + def original_query(): + return table.search().select( + {"filename": "filename", "payload": "video_blob"} + ).limit(1) + def empty_query(): + return query().where("id < 0") + def batches(): + return query().to_batches( + batch_size=1, blob_mode="bytes" + ) + + if sys.argv[2] == "async": + reused = table.query().select( + {"filename": "filename", "payload": "video_blob"} + ).limit(3) + byte_reader = await reused.to_batches(blob_mode="bytes") + await byte_reader.read_all() + lazy_reader = await reused.to_batches(blob_mode="lazy") + assert "_rowid" not in lazy_reader.schema.names + await lazy_reader.read_all() + reused.select(["filename"]) + nonblob_reader = await reused.to_batches(blob_mode="lazy") + assert nonblob_reader.schema.names == ["filename"] + await nonblob_reader.read_all() + + if sys.argv[3] == "pandas": + frame = query().to_pandas(blob_mode="bytes") + if sys.argv[2] == "async": + frame = await frame + assert frame.to_dict("records") == expected + original = original_query().to_pandas(blob_mode="bytes") + if sys.argv[2] == "async": + original = await original + assert original.to_dict("records") == [ + {"filename": "file-1.mp4", "payload": b"outside"} + ] + direct = table.to_pandas(blob_mode="bytes") + if sys.argv[2] == "async": + direct = await direct + assert direct.to_dict("records") == rows + empty = empty_query().to_pandas(blob_mode="bytes") + if sys.argv[2] == "async": + empty = await empty + assert empty.to_dict("records") == [] + elif sys.argv[3] == "batches": + reader = batches() + if sys.argv[2] == "async": + reader = await reader + rows_out = [batch async for batch in reader] + else: + rows_out = list(reader) + assert reader.schema.field("payload").type == pa.large_binary() + assert "_rowid" not in reader.schema.names + assert all(batch.num_rows == 1 for batch in rows_out) + assert all(batch.schema == reader.schema for batch in rows_out) + actual = [row for batch in rows_out for row in batch.to_pylist()] + assert actual == expected + original_reader = original_query().to_batches(blob_mode="bytes") + if sys.argv[2] == "async": + original_reader = await original_reader + original_rows = [batch async for batch in original_reader] + else: + original_rows = list(original_reader) + assert [ + row for batch in original_rows for row in batch.to_pylist() + ] == [{"filename": "file-1.mp4", "payload": b"outside"}] + empty_reader = empty_query().to_batches(blob_mode="bytes") + if sys.argv[2] == "async": + empty_reader = await empty_reader + empty_batches = [batch async for batch in empty_reader] + else: + empty_batches = list(empty_reader) + assert empty_batches == [] + assert empty_reader.schema.field("payload").type == pa.large_binary() + else: + table_out = query().to_arrow(blob_mode="bytes") + if sys.argv[2] == "async": + table_out = await table_out + assert table_out.schema.field("payload").type == pa.large_binary() + assert "_rowid" not in table_out.column_names + assert table_out.to_pylist() == expected + original = original_query().to_arrow(blob_mode="bytes") + if sys.argv[2] == "async": + original = await original + assert original.to_pylist() == [ + {"filename": "file-1.mp4", "payload": b"outside"} + ] + direct = table.to_arrow(blob_mode="bytes") + if sys.argv[2] == "async": + direct = await direct + assert direct.to_pylist() == rows + empty_table = empty_query().to_arrow(blob_mode="bytes") + if sys.argv[2] == "async": + empty_table = await empty_table + assert empty_table.num_rows == 0 + assert empty_table.schema.field("payload").type == pa.large_binary() + assert "lance" not in sys.modules + + asyncio.run(main()) + """ + ) + completed = subprocess.run( + [sys.executable, "-c", script, str(tmp_path), api, result], + capture_output=True, + text=True, + ) + assert completed.returncode == 0, completed.stderr diff --git a/python/python/tests/test_query.py b/python/python/tests/test_query.py index 95148dcf9..aee67d32a 100644 --- a/python/python/tests/test_query.py +++ b/python/python/tests/test_query.py @@ -348,7 +348,7 @@ def test_plain_scan_query_to_pandas_blob_bytes_projection(tmp_db, blob_schema): @pytest.mark.parametrize("blob_mode", ["bytes", "descriptions"]) -def test_plain_scan_query_to_pandas_blob_mode_does_not_collect_arrow( +def test_plain_scan_query_to_pandas_blob_mode_uses_native_path( tmp_db, monkeypatch, blob_mode ): pytest.importorskip("lance") @@ -362,7 +362,8 @@ def test_plain_scan_query_to_pandas_blob_mode_does_not_collect_arrow( def fail_to_arrow(*args, **kwargs): raise AssertionError("to_arrow should not be called before native pandas") - monkeypatch.setattr(query, "to_arrow", fail_to_arrow) + if blob_mode == "descriptions": + monkeypatch.setattr(query, "to_arrow", fail_to_arrow) df = query.to_pandas(blob_mode=blob_mode) @@ -488,7 +489,7 @@ async def test_async_plain_scan_query_to_pandas_blob_bytes_projection( @pytest.mark.asyncio @pytest.mark.parametrize("blob_mode", ["bytes", "descriptions"]) -async def test_async_plain_scan_query_to_pandas_blob_mode_does_not_collect_arrow( +async def test_async_plain_scan_query_to_pandas_blob_mode_uses_native_path( tmp_db_async, monkeypatch, blob_mode ): pytest.importorskip("lance") @@ -502,7 +503,8 @@ async def test_async_plain_scan_query_to_pandas_blob_mode_does_not_collect_arrow async def fail_to_arrow(*args, **kwargs): raise AssertionError("to_arrow should not be called before native pandas") - monkeypatch.setattr(query, "to_arrow", fail_to_arrow) + if blob_mode == "descriptions": + monkeypatch.setattr(query, "to_arrow", fail_to_arrow) df = await query.to_pandas(blob_mode=blob_mode) @@ -541,12 +543,14 @@ def test_vector_query_to_pandas_legacy_blob_descriptions_and_bytes(tmp_db): df = query.to_pandas(blob_mode="descriptions") assert df["blob"].tolist() == [{"position": 0, "size": 3}] - with pytest.raises(ValueError, match="legacy blob columns"): + with pytest.raises(ValueError, match=r"\['blob'\].*local plain scans"): query.to_pandas(blob_mode="bytes") - with pytest.raises(ValueError, match="legacy blob columns"): + with pytest.raises(ValueError, match=r"\['blob'\].*local plain scans"): query.to_arrow(blob_mode="bytes") - with pytest.raises(ValueError, match="legacy blob columns"): - table.to_arrow(blob_mode="bytes") + table_arrow = table.to_arrow(blob_mode="bytes") + assert table_arrow.schema.field("blob").type == pa.large_binary() + assert "_rowid" not in table_arrow.column_names + assert table_arrow["blob"].to_pylist() == [b"one", b"two", b"three", b"four"] assert table.to_pandas(blob_mode="bytes")["blob"].tolist() == [ b"one", b"two",