rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-11-13 14:16:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(python): materialize remote blobs at the query snapshot (#4973)

Browse Source 
Python remote queries materialize Blob v2 payloads with
`blob_mode="bytes"` through synchronous and asynchronous `to_arrow`,
`to_list`, `to_pandas`, and `to_batches`. Deferred reads bind to the
actual producing dataset without changing the caller's live table
handle, including concurrent writes and same-name DROP/CREATE.

Trust only the authoritative `x-lancedb-query-version` plus
`x-lancedb-query-snapshot`. A numeric version alone cannot distinguish a
recreated table; the ordinary live watermark and user-supplied Arrow
metadata cannot establish provenance. An independent immutable handle
carries the original identity through every batch, byte-limit split, and
large-Blob Range fallback, and requires a verified response echo before
accepting payloads. Aliases, NULL/empty results, rerankers, and hybrid
legs retain consistent provenance. Missing capability, mixed identities,
or a serving node ignoring the precondition fails explicitly; ordinary
descriptor queries remain available on older servers.

The source docs distinguish Enterprise SQL payload bytes from SDK
descriptors. Server dependency:
https://github.com/lancedb/sophon/pull/10211.

Validation uses retained, independently audited local evidence: 724
Python cases passed (one optional PyAV skip), 67 focused SDK Rust cases
passed, 36 server checks passed, and 15 actual REST/SDK scenarios passed
across retained runs. These cover sync/async and Arrow file/stream
recreation, concurrent overwrite, deletion of a selected version, legacy
capabilities, forged IPC metadata, WAL exclusion, and actual mid-batch
recreation. The original aggregate experiment exited 101 because its
extra harness assumed a Python batch size would split the native result;
its 36 server checks and 13 completed scenarios remain valid. The
corrected native-batch experiment passed separately for sync and async.
No aggregate all-green claim is made.

The public batched-range operation has separate cross-partition identity
integration in https://github.com/lancedb/sophon/pull/10219 and
https://github.com/lancedb/lancedb/pull/4994. This change does not claim
to fix that loop. Realtime/MemWAL remains outside Blob v2 GA, without
disabling existing writes or compaction. No dependency pins changed.

Merge/rollout hold remains: upgrade every serving node before enabling
the new SDK capability, verify proxy forwarding of both authoritative
headers and the verification echo, and preserve queried versions for
later reads. Missing or inconsistent capability fails explicitly;
identity does not retain versions or authenticate storage paths. Actual
deployed proxy/Cloud/PE behavior and production retention remain
unverified. Old SDKs do not gain incarnation protection from a
server-only upgrade.

Published documentation retains the Early access badge and released
Enterprise instructions, including explicit fetches for copying. The new
Enterprise `blob_mode="bytes"` guidance, snapshot-bound copy example,
future streaming restrictions and GA wording remain source-only MDX
drafts until release. The correction distinguishing released Enterprise
SQL bytes from SDK descriptors remains visible. Documentation validation
checked comment boundaries, visible instructions, Python example syntax
and whitespace; the site was not built. This documentation-only revision
reuses the implementation evidence above.
This commit is contained in:
Xuanwo
2026-11-13 02:05:39 +08:00
committed by GitHub
parent da40aa6c44
commit 9b13e8f725
9 changed files with 876 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/python/lancedb/_lancedb.pyi
+6 -1
View File
@@ -749,7 +749,12 @@ class Table:
def take_row_ids(self, row_ids: list[int]) -> TakeQuery: ...
async def blob_columns(self) -> list[str]: ...
async def fetch_blobs(
self, column: str, row_ids: list[int]
self,
column: str,
row_ids: list[int],
*,
version: Optional[int] = None,
snapshot_identity: Optional[str] = None,
) -> pa.LargeBinaryArray: ...
async def fetch_blob_ranges(
self,
python/python/lancedb/query.py
+110 -28
View File
@@ -4,6 +4,7 @@
from __future__ import annotations
import asyncio
import re
from abc import ABC, abstractmethod
from concurrent.futures import ThreadPoolExecutor
from datetime import timedelta
@@ -194,20 +195,57 @@ def _unsupported_blob_pandas_error(reason: str) -> RuntimeError:
)
def _supports_local_blob_fetch(
def _supports_blob_fetch(
async_table: "AsyncTable", pushdown_operations: Iterable[str]
) -> bool:
# Remote tables and QueryTable pushdown answer queries on a server, so a
# fetch through this handle could read another version.
return async_table._inner._is_native() and "QueryTable" not in pushdown_operations
# Remote queries carry their producing snapshot in response metadata.
# Namespace QueryTable pushdown has no equivalent snapshot contract.
return (
not async_table._inner._is_native() or "QueryTable" not in pushdown_operations
)
def _remote_blob_snapshot(schema: pa.Schema) -> tuple[int, str]:
value = (schema.metadata or {}).get(b"lancedb:query_version", b"")
if not value.isdigit() or not 0 < int(value) < 2**64:
raise NotImplementedError(
"blob_mode='bytes' requires a stable table snapshot: the query response "
"has a missing, invalid, or inconsistent x-lancedb-query-version header. "
"Upgrade the server and use a table without Realtime/MemWAL, or use "
"blob_mode='descriptions'."
)
identity = (schema.metadata or {}).get(b"lancedb:query_snapshot", b"")
if not re.fullmatch(rb"v1:[0-9a-f]{64}", identity):
raise NotImplementedError(
"blob_mode='bytes' requires a trustworthy query snapshot identity: "
"missing or unsupported x-lancedb-query-snapshot. Upgrade the server "
"on every serving node, or use blob_mode='descriptions'."
)
return int(value), identity.decode("ascii")
def _merge_query_snapshot(result: pa.Table, *sources: pa.Table) -> pa.Table:
"""Preserve a snapshot only when every query leg used the same identity."""
keys = (b"lancedb:query_version", b"lancedb:query_snapshot")
snapshots = {
tuple((source.schema.metadata or {}).get(key) for key in keys)
for source in sources
}
metadata = dict(result.schema.metadata or {})
for key in keys:
metadata.pop(key, None)
snapshot = next(iter(snapshots)) if len(snapshots) == 1 else None
if snapshot is not None and all(value is not None for value in snapshot):
metadata.update(zip(keys, snapshot))
return result.replace_schema_metadata(metadata)
def _blob_bytes_without_snapshot_error() -> NotImplementedError:
return NotImplementedError(
"blob_mode='bytes' cannot safely materialize blob columns without a "
"stable table snapshot. LanceDB Enterprise and namespace QueryTable pushdown "
"run the query on a server, so the blob fetch could read another table "
"version. Use blob_mode='descriptions' to inspect descriptors. To read "
"stable table snapshot. Namespace QueryTable pushdown does not bind "
"the query and blob fetches to one table version. "
"Use blob_mode='descriptions' to inspect descriptors. To read "
"payloads, select row IDs with with_row_id(True), then call "
"table.fetch_blobs(column, row_ids) or table.fetch_blob_files(column, "
"row_ids). Keep the source unchanged between selection and fetching."
@@ -1297,9 +1335,9 @@ class LanceQueryBuilder(ABC):
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.
Enterprise queries support ``"bytes"`` at the query response's
snapshot and ``"descriptions"`` without payload reads. Remote
``"lazy"`` handles are not supported.
**kwargs
Forwarded to pyarrow.Table.to_pandas after query execution and
optional flattening.
@@ -1307,6 +1345,10 @@ class LanceQueryBuilder(ABC):
validate_blob_mode(blob_mode)
if _is_remote_query_table(self._table):
blob_sources = self._validated_blob_sources(blob_mode)
if blob_mode == "bytes":
return flatten_columns(
self.to_arrow(timeout=timeout, blob_mode=blob_mode), flatten
).to_pandas(**kwargs)
return _finish_remote_pandas(
self.to_arrow(timeout=timeout),
blob_mode=blob_mode,
@@ -1629,7 +1671,8 @@ class LanceQueryBuilder(ABC):
_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(
and self._table._table._inner._is_native()
and _supports_blob_fetch(
self._table._table,
getattr(self._table, "_pushdown_operations", ()),
)
@@ -1650,11 +1693,12 @@ class LanceQueryBuilder(ABC):
if blob_mode != "bytes":
return blob_sources
reject_unfetchable_blob_columns(schema, self._columns, blob_sources)
if blob_sources and not _supports_local_blob_fetch(
if blob_sources and not _supports_blob_fetch(
self._table._table, getattr(self._table, "_pushdown_operations", ())
):
raise _blob_bytes_without_snapshot_error()
reject_unmatched_blob_outputs(self._blob_output_schema(), blob_sources)
if not _is_remote_query_table(self._table):
reject_unmatched_blob_outputs(self._blob_output_schema(), blob_sources)
return blob_sources
def _blob_output_schema(self) -> pa.Schema:
@@ -1667,9 +1711,20 @@ class LanceQueryBuilder(ABC):
return tbl
if blob_mode == "descriptions":
return strip_auto_row_ids(tbl, blob_sources)
return replace_blob_columns_with_bytes_sync(
tbl, blob_sources, self._table.fetch_blobs
)
reject_unmatched_blob_outputs(tbl.schema, blob_sources)
if _is_remote_query_table(self._table) and blob_sources:
version, identity = _remote_blob_snapshot(tbl.schema)
async def fetch_at_version(column, row_ids):
return await self._table._table._inner.fetch_blobs(
column, row_ids, version=version, snapshot_identity=identity
)
def fetch_blobs(column, row_ids):
return LOOP.run(fetch_at_version(column, row_ids))
else:
fetch_blobs = self._table.fetch_blobs
return replace_blob_columns_with_bytes_sync(tbl, blob_sources, fetch_blobs)
def _finalize_blob_query_table(
self,
@@ -2386,6 +2441,7 @@ class LanceVectorQueryBuilder(LanceQueryBuilder):
rs_table = result_set.read_all()
result_set = self._reranker.rerank_vector(self._str_query, rs_table)
check_reranker_result(result_set)
result_set = _merge_query_snapshot(result_set, rs_table)
# convert result_set back to RecordBatchReader
result_set = pa.RecordBatchReader.from_batches(
result_set.schema, result_set.to_batches()
@@ -2599,8 +2655,10 @@ class LanceFtsQueryBuilder(LanceQueryBuilder):
query = self._query_for_scan(blob_sources)
results = self._table._execute_query(query, timeout=timeout).read_all()
if self._reranker is not None:
source = results
results = self._reranker.rerank_fts(self._query, results)
check_reranker_result(results)
results = _merge_query_snapshot(results, source)
return self._finalize_blob_query_table(
results, blob_mode=blob_mode, blob_sources=blob_sources
)
@@ -2847,6 +2905,7 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
with_row_ids=True,
offset=self._offset,
)
results = _merge_query_snapshot(results, fts_results, vector_results)
return self._finish_hybrid_results(
results, blob_mode=blob_mode, blob_sources=blob_sources
)
@@ -3559,9 +3618,8 @@ class AsyncQueryBase(object):
_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
)
and self._table._inner._is_native()
and _supports_blob_fetch(self._table, self._table._pushdown_operations)
)
def select(self, columns: Union[List[str], dict[str, str]]) -> Self:
@@ -3766,11 +3824,14 @@ class AsyncQueryBase(object):
if blob_mode != "bytes":
return blob_sources
reject_unfetchable_blob_columns(schema, projection, blob_sources)
if blob_sources and not _supports_local_blob_fetch(
if blob_sources and not _supports_blob_fetch(
self._table, self._table._pushdown_operations
):
raise _blob_bytes_without_snapshot_error()
reject_unmatched_blob_outputs(await self._blob_output_schema(), blob_sources)
if self._table._inner._is_native():
reject_unmatched_blob_outputs(
await self._blob_output_schema(), blob_sources
)
return blob_sources
async def _blob_output_schema(self) -> pa.Schema:
@@ -3783,9 +3844,21 @@ class AsyncQueryBase(object):
return tbl
if blob_mode == "descriptions":
return strip_auto_row_ids(tbl, blob_sources)
return await replace_blob_columns_with_bytes(
tbl, blob_sources, self._table.fetch_blobs
)
reject_unmatched_blob_outputs(tbl.schema, blob_sources)
if (
self._table is not None
and not self._table._inner._is_native()
and blob_sources
):
version, identity = _remote_blob_snapshot(tbl.schema)
async def fetch_blobs(column, row_ids):
return await self._table._inner.fetch_blobs(
column, row_ids, version=version, snapshot_identity=identity
)
else:
fetch_blobs = self._table.fetch_blobs
return await replace_blob_columns_with_bytes(tbl, blob_sources, fetch_blobs)
async def to_arrow(
self,
@@ -3811,7 +3884,8 @@ class AsyncQueryBase(object):
``fetch_blobs`` reads. ``"descriptions"`` returns them without it.
``"bytes"`` returns payload bytes for supported row-addressable blob
columns and raises for blobs inside lists.
LanceDB Enterprise and namespace query pushdown refuse ``"bytes"``.
Enterprise queries fetch bytes at the version reported by the query
response. Namespace query pushdown refuses ``"bytes"``.
"""
blob_sources = await self._validated_blob_sources(blob_mode)
if blob_mode == "bytes" and blob_sources:
@@ -3893,9 +3967,9 @@ class AsyncQueryBase(object):
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.
Enterprise queries support ``"bytes"`` at the query response's
snapshot and ``"descriptions"`` without payload reads. Remote
``"lazy"`` handles are not supported.
**kwargs
Forwarded to pyarrow.Table.to_pandas after query execution and
optional flattening.
@@ -3903,6 +3977,9 @@ class AsyncQueryBase(object):
validate_blob_mode(blob_mode)
if self._table is not None and not self._table._inner._is_native():
await self._validated_blob_sources(blob_mode)
if blob_mode == "bytes":
tbl = await self.to_arrow(timeout=timeout, blob_mode=blob_mode)
return flatten_columns(tbl, flatten).to_pandas(**kwargs)
return await self._remote_to_pandas(
flatten=flatten, timeout=timeout, blob_mode=blob_mode, **kwargs
)
@@ -4508,7 +4585,9 @@ class AsyncFTSQuery(AsyncStandardQuery):
)
results = pa.Table.from_batches(await reader.read_all(), reader.schema)
if self._reranker:
source = results
results = self._reranker.rerank_fts(self.get_query(), results)
results = _merge_query_snapshot(results, source)
return AsyncRecordBatchReader(results, max_batch_length=max_batch_length)
@@ -4835,7 +4914,9 @@ class AsyncVectorQuery(AsyncStandardQuery, AsyncVectorQueryBase):
)
results = pa.Table.from_batches(await reader.read_all(), reader.schema)
if self._reranker:
source = results
results = self._reranker.rerank_vector(self._query_string, results)
results = _merge_query_snapshot(results, source)
return AsyncRecordBatchReader(results, max_batch_length=max_batch_length)
@@ -5027,6 +5108,7 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
):
result = result.drop(["_rowid"])
result = _merge_query_snapshot(result, fts_results, vector_results)
return AsyncRecordBatchReader(result, max_batch_length=max_batch_length)
async def explain_plan(self, verbose: Optional[bool] = False):
python/python/tests/test_remote_db.py
+432 -25
View File
@@ -5462,14 +5462,34 @@ def blob_descriptor_take_table():
@contextlib.contextmanager
def blob_remote_table(*, server_version=Version("0.5.0"), location=None):
def blob_remote_table(
*,
server_version=Version("0.5.0"),
location=None,
version_header="1",
framing="file",
requests=None,
state=None,
empty=False,
schema_version=None,
search_scores=False,
snapshot_supported=True,
identity_header="v1:" + "a" * 64,
verify_identity=True,
):
if state is None:
state = {"version": 1}
def handler(request):
if request.path == "/v1/table/test/describe/":
content_len = int(request.headers.get("Content-Length", 0))
body = json.loads(request.rfile.read(content_len)) if content_len else {}
request.send_response(200)
request.send_header("Content-Type", "application/json")
request.send_header("phalanx-version", str(server_version))
request.end_headers()
response = dict(BLOB_DESCRIBE_RESPONSE)
response["version"] = body.get("version") or state["version"]
if location is not None:
response["location"] = str(location)
request.wfile.write(json.dumps(response).encode())
@@ -5502,9 +5522,22 @@ def blob_remote_table(*, server_version=Version("0.5.0"), location=None):
elif request.path == "/v1/table/test/query/":
content_len = int(request.headers.get("Content-Length", 0))
body = json.loads(request.rfile.read(content_len))
if requests is not None:
requests.append(("query", body))
columns = body.get("columns")
assert columns in (None, ["id", "image"], ["id"], ["image"])
if columns == ["id"]:
assert columns in (
None,
["id", "image"],
["id"],
["image"],
{"payload": "image"},
)
if isinstance(columns, dict):
response_table = blob_query_response_table().select(["image", "_rowid"])
response_table = response_table.rename_columns(["payload", "_rowid"])
if not body.get("with_row_id", False):
response_table = response_table.drop(["_rowid"])
elif columns == ["id"]:
response_table = blob_query_response_table().select(["id"])
else:
if columns == ["image"]:
@@ -5521,25 +5554,86 @@ def blob_remote_table(*, server_version=Version("0.5.0"), location=None):
response_table = blob_query_response_table()
if not body.get("with_row_id", False):
response_table = response_table.drop(["_rowid"])
if search_scores and not body.get("filter", "").startswith("_rowid IN"):
if body.get("vector"):
response_table = response_table.append_column(
"_distance", pa.array([0.1, 0.2, 0.3], type=pa.float32())
)
elif body.get("full_text_query"):
response_table = response_table.append_column(
"_score", pa.array([3.0, 2.0, 1.0], type=pa.float32())
)
if empty:
response_table = response_table.slice(0, 0)
if schema_version is not None:
response_table = response_table.replace_schema_metadata(
{
b"lancedb:query_version": schema_version.encode(),
b"lancedb:query_snapshot": ("v1:" + "f" * 64).encode(),
}
)
request.send_response(200)
request.send_header("Content-Type", "application/vnd.apache.arrow.file")
request.send_header("x-lancedb-version", "1")
request.send_header(
"Content-Type", f"application/vnd.apache.arrow.{framing}"
)
header = (
version_header(body) if callable(version_header) else version_header
)
if header is not None:
request.send_header("x-lancedb-version", header)
if snapshot_supported:
request.send_header("x-lancedb-query-version", header)
identity = (
identity_header(body)
if callable(identity_header)
else identity_header
)
if identity is not None:
request.send_header("x-lancedb-query-snapshot", identity)
request.end_headers()
with pa.ipc.new_file(request.wfile, response_table.schema) as writer:
writer.write_table(response_table)
writer_factory = pa.ipc.new_file if framing == "file" else pa.ipc.new_stream
with writer_factory(request.wfile, response_table.schema) as writer:
writer.write_table(response_table, max_chunksize=1)
elif request.path == "/v1/table/test/fetch_blobs/":
content_len = int(request.headers.get("Content-Length", 0))
body = json.loads(request.rfile.read(content_len))
assert body["column"] == "image"
assert body["row_ids"] == [10, 20, 30]
if requests is not None:
requests.append(("fetch_blobs", body))
payloads = {10: b"alpha", 20: None, 30: b"gamma", 40: b""}
if state["version"] > 1 and body.get("version") != 1:
payloads = {row: b"new-version" for row in payloads}
response_table = pa.table(
{"image": pa.array([b"alpha", None, b"gamma"], type=pa.large_binary())}
{
"image": pa.array(
[payloads[row] for row in body["row_ids"]],
type=pa.large_binary(),
)
}
)
request.send_response(200)
request.send_header("Content-Type", "application/vnd.apache.arrow.stream")
if verify_identity and request.headers.get("x-lancedb-query-snapshot"):
request.send_header(
"x-lancedb-query-snapshot",
request.headers["x-lancedb-query-snapshot"],
)
request.end_headers()
with pa.ipc.new_stream(request.wfile, response_table.schema) as writer:
writer.write_table(response_table)
elif request.path == "/v1/table/test/insert/":
if request.headers.get("Transfer-Encoding") == "chunked":
while size := int(request.rfile.readline().split(b";")[0], 16):
request.rfile.read(size)
assert request.rfile.read(2) == b"\r\n"
assert request.rfile.read(2) == b"\r\n"
else:
request.rfile.read(int(request.headers.get("Content-Length", 0)))
state["version"] += 1
request.send_response(200)
request.send_header("Content-Type", "application/json")
request.end_headers()
request.wfile.write(json.dumps({"version": state["version"]}).encode())
else:
request.send_response(404)
request.end_headers()
@@ -5548,6 +5642,222 @@ def blob_remote_table(*, server_version=Version("0.5.0"), location=None):
yield db.open_table("test")
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
@pytest.mark.parametrize("framing", ["file", "stream"])
@pytest.mark.parametrize("method", ["to_arrow", "to_list", "to_pandas", "to_batches"])
async def test_remote_blob_bytes_conversion(use_async, framing, method):
if method == "to_pandas":
pytest.importorskip("pandas")
requests = []
with blob_remote_table(framing=framing, requests=requests) as remote_table:
table = remote_table._table if use_async else remote_table
query = table.query() if use_async else table.search()
query = query.select(["id", "image"])
if method == "to_batches":
kwargs = {"max_batch_length" if use_async else "batch_size": 1}
reader = query.to_batches(blob_mode="bytes", **kwargs)
if use_async:
reader = await reader
batches = [batch async for batch in reader]
else:
batches = list(reader)
assert all(batch.num_rows == 1 for batch in batches)
result = pa.Table.from_batches(batches, schema=reader.schema).to_pylist()
else:
result = getattr(query, method)(blob_mode="bytes")
if use_async:
result = await result
if method == "to_arrow":
assert result.schema.field("image").type == pa.large_binary()
result = result.to_pylist()
elif method == "to_pandas":
result = result.to_dict("records")
assert result == [
{"id": 1, "image": b"alpha"},
{"id": 2, "image": None},
{"id": 3, "image": b"gamma"},
]
fetches = [body for path, body in requests if path == "fetch_blobs"]
assert fetches
assert all(body["version"] == 1 for body in fetches)
assert [row for body in fetches for row in body["row_ids"]] == [10, 20, 30]
descriptor_queries = [
body
for path, body in requests
if path == "query" and body.get("filter", "").startswith("_rowid IN")
]
assert all(body["version"] == 1 for body in descriptor_queries)
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
@pytest.mark.parametrize("framing", ["file", "stream"])
async def test_remote_blob_bytes_keeps_query_snapshot_during_writes(use_async, framing):
requests = []
state = {"version": 1}
with blob_remote_table(
framing=framing, requests=requests, state=state
) as remote_table:
table = remote_table._table if use_async else remote_table
query = table.query() if use_async else table.search()
kwargs = {"max_batch_length" if use_async else "batch_size": 1}
reader = query.select(["id", "image"]).to_batches(blob_mode="bytes", **kwargs)
if use_async:
reader = await reader
# The query has selected v1; commit v2 before consuming any payloads.
write = table.add([{"id": 4, "image": b"new-version"}])
if use_async:
await write
version = await table.version() if use_async else table.version
assert version == 2
if use_async:
rows = [row async for batch in reader for row in batch.to_pylist()]
else:
rows = [row for batch in reader for row in batch.to_pylist()]
assert [row["image"] for row in rows] == [b"alpha", None, b"gamma"]
fetches = [body for path, body in requests if path == "fetch_blobs"]
assert all(body["version"] == 1 for body in fetches)
# Snapshot fetches must not check out the caller's live handle.
version = await table.version() if use_async else table.version
assert version == 2
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
@pytest.mark.parametrize("version_header", [None, "invalid", "0", "-1", str(2**64)])
async def test_remote_blob_bytes_invalid_snapshot(use_async, version_header):
requests = []
with blob_remote_table(
version_header=version_header, requests=requests, schema_version="1"
) as remote_table:
table = remote_table._table if use_async else remote_table
query = table.query() if use_async else table.search()
with pytest.raises(NotImplementedError, match="x-lancedb-query-version"):
result = query.to_arrow(blob_mode="bytes")
if use_async:
await result
assert not any(path == "fetch_blobs" for path, _ in requests)
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
@pytest.mark.parametrize("framing", ["file", "stream"])
async def test_remote_blob_bytes_empty_results(use_async, framing):
requests = []
with blob_remote_table(
empty=True, framing=framing, requests=requests
) as remote_table:
table = remote_table._table if use_async else remote_table
query = table.query() if use_async else table.search()
result = query.to_arrow(blob_mode="bytes")
if use_async:
result = await result
assert result.num_rows == 0
assert result.schema.field("image").type == pa.large_binary()
assert result.column_names == ["id", "image"]
assert not any(path == "fetch_blobs" for path, _ in requests)
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
@pytest.mark.parametrize("kind", ["vector", "fts", "hybrid"])
@pytest.mark.parametrize("rerank", [False, True])
async def test_remote_blob_bytes_search_queries(use_async, kind, rerank):
with blob_remote_table(search_scores=True) as remote_table:
if use_async:
query = remote_table._table.query()
if kind in ("vector", "hybrid"):
query = query.nearest_to([1.0, 2.0])
if kind in ("fts", "hybrid"):
query = query.nearest_to_text("hello")
elif kind == "vector":
query = remote_table.search([1.0, 2.0], vector_column_name="vector")
elif kind == "fts":
query = remote_table.search("hello", query_type="fts")
else:
query = (
remote_table.search(query_type="hybrid", vector_column_name="vector")
.vector([1.0, 2.0])
.text("hello")
)
if rerank:
from lancedb.rerankers import RRFReranker
class StripMetadataReranker(RRFReranker):
def rerank_vector(self, query, vector_results):
return vector_results.append_column(
"_relevance_score", pa.array([3.0, 2.0, 1.0])
).replace_schema_metadata(None)
def rerank_fts(self, query, fts_results):
return self.rerank_vector(query, fts_results)
def rerank_hybrid(self, query, vector_results, fts_results):
return (
super()
.rerank_hybrid(query, vector_results, fts_results)
.replace_schema_metadata(None)
)
kwargs = {"query_string": "hello"} if kind == "vector" else {}
query = query.rerank(StripMetadataReranker(), **kwargs)
result = query.select(["id", "image"]).to_arrow(blob_mode="bytes")
if use_async:
result = await result
assert result["image"].to_pylist() == [b"alpha", None, b"gamma"]
assert "_rowid" not in result.column_names
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
async def test_remote_blob_bytes_hybrid_rejects_different_snapshots(use_async):
def header(body):
return "2" if body.get("full_text_query") else "1"
requests = []
with blob_remote_table(
search_scores=True, version_header=header, requests=requests
) as remote_table:
if use_async:
query = (
remote_table._table.query()
.nearest_to([1.0, 2.0])
.nearest_to_text("hello")
)
else:
query = (
remote_table.search(query_type="hybrid", vector_column_name="vector")
.vector([1.0, 2.0])
.text("hello")
)
with pytest.raises(
NotImplementedError, match="inconsistent x-lancedb-query-version"
):
result = query.select(["id", "image"]).to_arrow(blob_mode="bytes")
if use_async:
await result
assert not any(path == "fetch_blobs" for path, _ in requests)
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
@pytest.mark.parametrize("with_row_id", [False, True])
async def test_remote_blob_bytes_alias_and_explicit_row_ids(use_async, with_row_id):
with blob_remote_table() as remote_table:
query = remote_table._table.query() if use_async else remote_table.search()
query = query.select({"payload": "image"})
if with_row_id:
query = query.with_row_id() if use_async else query.with_row_id(True)
result = query.to_arrow(blob_mode="bytes")
if use_async:
result = await result
assert result["payload"].to_pylist() == [b"alpha", None, b"gamma"]
assert ("_rowid" in result.column_names) == with_row_id
if with_row_id:
assert result["_rowid"].to_pylist() == [10, 20, 30]
def test_remote_blob_columns_and_fetch():
with blob_remote_table() as table:
assert table.blob_columns() == ["image"]
@@ -5569,7 +5879,7 @@ def test_sync_remote_blob_pandas_avoids_storage_access(
raise AssertionError("remote storage must not be opened")
monkeypatch.setattr(lance, "dataset", fail_dataset)
with blob_remote_table(location=location) as table:
with blob_remote_table(location=location, version_header=None) as table:
monkeypatch.setattr(table, "to_lance", fail_dataset)
query = table.search().select(["id", "image"])
if blob_mode == "descriptions":
@@ -5582,8 +5892,11 @@ def test_sync_remote_blob_pandas_avoids_storage_access(
NotImplementedError, match="stable table snapshot"
) as exc:
query.to_pandas(blob_mode=blob_mode)
assert "fetch_blobs(column, row_ids)" in str(exc.value)
assert "with_row_id(True)" in str(exc.value)
if blob_mode == "bytes":
assert "x-lancedb-query-version" in str(exc.value)
else:
assert "fetch_blobs(column, row_ids)" in str(exc.value)
assert "with_row_id(True)" in str(exc.value)
@pytest.mark.asyncio
@@ -5667,7 +5980,7 @@ async def test_async_remote_blob_descriptions_to_pandas_uses_server(monkeypatch)
@pytest.mark.asyncio
@pytest.mark.parametrize("blob_mode", ["bytes", "lazy"])
async def test_async_remote_blob_pandas_rejects_unpinned_fetch(monkeypatch, blob_mode):
with blob_remote_table() as remote_table:
with blob_remote_table(version_header=None) as remote_table:
table = remote_table._table
async def fail_uri():
@@ -5692,8 +6005,8 @@ async def test_async_remote_blob_pandas_rejects_unpinned_fetch(monkeypatch, blob
@pytest.mark.parametrize("method", ["to_arrow", "to_list"])
def test_remote_search_bytes_refused_before_query(monkeypatch, method):
with blob_remote_table() as table:
def test_remote_search_bytes_requires_response_version(method):
with blob_remote_table(version_header=None) as table:
query = table.search().select(["id", "image"])
descriptions = getattr(query, method)(blob_mode="descriptions")
images = (
@@ -5703,19 +6016,15 @@ def test_remote_search_bytes_refused_before_query(monkeypatch, method):
)
assert "_lance_row_id" not in images[0]
def fail_query(*args, **kwargs):
raise AssertionError("bytes must be refused before the query runs")
monkeypatch.setattr(table, "_execute_query", fail_query)
with pytest.raises(NotImplementedError, match="stable table snapshot"):
with pytest.raises(NotImplementedError, match="x-lancedb-query-version"):
getattr(query, method)(blob_mode="bytes")
with pytest.raises(ValueError, match="blob_mode must be one of"):
table.to_pandas(blob_mode="handles")
@pytest.mark.asyncio
async def test_async_remote_to_arrow_bytes_refused_before_query(monkeypatch):
with blob_remote_table() as remote_table:
async def test_async_remote_to_arrow_bytes_requires_response_version(monkeypatch):
with blob_remote_table(version_header=None) as remote_table:
table = remote_table._table
async def fail_fetch(*args, **kwargs):
@@ -5723,8 +6032,7 @@ async def test_async_remote_to_arrow_bytes_refused_before_query(monkeypatch):
monkeypatch.setattr(table, "fetch_blobs", fail_fetch)
query = table.query().select(["id", "image"])
monkeypatch.setattr(query, "to_batches", fail_fetch)
with pytest.raises(NotImplementedError, match="stable table snapshot"):
with pytest.raises(NotImplementedError, match="x-lancedb-query-version"):
await query.to_arrow(blob_mode="bytes")
with pytest.raises(NotImplementedError, match="stable table snapshot"):
await table.to_arrow(blob_mode="bytes")
@@ -5821,7 +6129,7 @@ def test_remote_blob_query_to_pandas():
@pytest.mark.parametrize("blob_mode", ["default", "lazy", "bytes"])
def test_remote_blob_query_to_pandas_rejects_unpinned_fetch(monkeypatch, blob_mode):
with blob_remote_table() as table:
with blob_remote_table(version_header=None) as table:
def fail_fetch(*args, **kwargs):
raise AssertionError("blob fetch must not use an unpinned remote table")
@@ -6884,3 +7192,102 @@ def test_remote_optimize_rejects_explicit_pruning_before_submitting():
table.optimize(delete_unverified=True)
assert server.tickets == []
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
@pytest.mark.parametrize("framing", ["file", "stream"])
async def test_remote_blob_bytes_rejects_legacy_watermark(use_async, framing):
requests = []
with blob_remote_table(
framing=framing,
snapshot_supported=False,
version_header="1",
schema_version="1",
requests=requests,
) as remote_table:
table = remote_table._table if use_async else remote_table
query = table.query() if use_async else table.search()
descriptors = query.to_arrow(blob_mode="descriptions")
if use_async:
descriptors = await descriptors
assert descriptors.num_rows == 3
assert pa.types.is_struct(descriptors.schema.field("image").type)
query = table.query() if use_async else table.search()
with pytest.raises(NotImplementedError, match="x-lancedb-query-version"):
result = query.to_arrow(blob_mode="bytes")
if use_async:
await result
assert not any(path == "fetch_blobs" for path, _ in requests)
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
@pytest.mark.parametrize("identity_header", [None, "", "v2:" + "a" * 64, "v1:broken"])
async def test_remote_blob_bytes_requires_trustworthy_identity(
use_async, identity_header
):
requests = []
with blob_remote_table(
identity_header=identity_header, schema_version="1", requests=requests
) as remote_table:
table = remote_table._table if use_async else remote_table
query = table.query() if use_async else table.search()
# Ordinary descriptors still work, even with forged Arrow schema metadata.
result = query.to_arrow(blob_mode="descriptions")
if use_async:
result = await result
assert result.num_rows == 3
with pytest.raises(NotImplementedError, match="x-lancedb-query-snapshot"):
result = query.to_arrow(blob_mode="bytes")
if use_async:
await result
assert not any(path == "fetch_blobs" for path, _ in requests)
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
async def test_remote_blob_bytes_rejects_unverified_serving_node(use_async):
with blob_remote_table(verify_identity=False) as remote_table:
table = remote_table._table if use_async else remote_table
query = table.query() if use_async else table.search()
with pytest.raises(
NotImplementedError, match="did not verify the query snapshot identity"
):
result = query.to_arrow(blob_mode="bytes")
if use_async:
await result
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
async def test_remote_blob_bytes_hybrid_rejects_same_version_different_incarnations(
use_async,
):
requests = []
def identity(body):
return "v1:" + ("b" if body.get("full_text_query") else "a") * 64
with blob_remote_table(
search_scores=True, identity_header=identity, requests=requests
) as remote_table:
if use_async:
query = (
remote_table._table.query()
.nearest_to([1.0, 2.0])
.nearest_to_text("hello")
)
else:
query = (
remote_table.search(query_type="hybrid", vector_column_name="vector")
.vector([1.0, 2.0])
.text("hello")
)
with pytest.raises(
NotImplementedError, match="inconsistent x-lancedb-query-version"
):
result = query.select(["id", "image"]).to_arrow(blob_mode="bytes")
if use_async:
await result
assert not any(path == "fetch_blobs" for path, _ in requests)
python/src/table.rs
+15 -1
View File
@@ -1978,14 +1978,28 @@ impl Table {
}
/// Read blob bytes for `row_ids` from blob v2 column `column`.
#[pyo3(signature = (column, row_ids))]
#[pyo3(signature = (column, row_ids, *, version=None, snapshot_identity=None))]
pub fn fetch_blobs(
self_: PyRef<'_, Self>,
column: String,
row_ids: Vec<u64>,
version: Option<u64>,
snapshot_identity: Option<String>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let inner = match (version, snapshot_identity) {
(None, None) => inner,
(Some(version), Some(identity)) => inner
.query_snapshot_with_identity(version, &identity)
.await
.infer_error()?,
_ => {
return Err(PyValueError::new_err(
"deferred blob reads require both query version and snapshot identity",
));
}
};
let blobs: LargeBinaryArray = inner
.fetch_blobs(column.as_str(), &row_ids)
.await