rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-11-15 07:06:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(python): support legacy blob query bytes (#4947)

Browse Source 
## Summary

- Let local plain queries over row-addressable legacy v1 and v2 blob
columns use `blob_mode="bytes"` without importing `pylance` through
pandas, batch, or Arrow conversion.
- Preserve aliases, filters, ordering, limits, null and empty payloads,
empty result schemas, and the caller's `_rowid` choice.
- Build the hidden row-id query separately so reusing an async builder
for lazy reads or non-blob projections cannot retain an automatic
`_rowid`.
- Keep byte reads closed for blobs inside lists, remote or namespace
pushdown, and query shapes whose result rows cannot be matched back to
the source. Blob v2 lazy auto-row-id behavior is unchanged.

Fixes #4645

## Validation

- 26 focused Python tests passed with the test subprocess blocking all
`lance` imports. The persisted fixture writes and reopens a storage-v2.1
table, then covers sync and async pandas, batches, and Arrow conversion
plus async builder reuse.
- The existing unsupported vector and hybrid reranker regressions now
assert the current eligibility diagnostic, and the legacy table-level
Arrow regression now asserts successful bytes, schema, values, and
hidden row IDs. Both focused tests passed.
- The runtime receipt records no installed `pylance` distribution and no
importable `lance` module. The Python-only change was verified with an
existing compatible 0.42.0b9 extension; the relevant query and fetch
APIs are unchanged between that source and this 0.42.0b10 base.
- Ruff format and lint pass for every changed Python file.
This commit is contained in:
Xuanwo
2026-11-11 10:08:13 +08:00
committed by GitHub
parent 4d22c0ce2b
commit 1e5b6deee3
10 changed files with 493 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/python/lancedb/_blob.py
+30 -6
View File
@@ -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,
python/python/lancedb/query.py
+179 -75
View File
@@ -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))
python/python/lancedb/schema.py
+9
View File
@@ -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.
python/python/lancedb/table.py
+12 -9
View File
@@ -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:
python/python/tests/test_blob.py
+30 -3
View File
@@ -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")
python/python/tests/test_blob_query_without_pylance.py
+194 -7
View File
@@ -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
python/python/tests/test_query.py
+12 -8
View File
@@ -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",