rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-09-01 02:58:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into gatekeeper/fix-2325-1

Browse Source
This commit is contained in:
Xuanwo
2026-08-27 18:13:07 +08:00
committed by GitHub
parent 10151b2dc8 0dd9dfdfc7
commit 670bda8725
131 changed files with 11656 additions and 1725 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lancedb-python"
version = "0.38.0-beta.3"
version = "0.38.0-beta.11"
publish = false
edition.workspace = true
description = "Python bindings for LanceDB"
python/pyproject.toml
+4 -1
View File
@@ -101,9 +101,12 @@ azure = ["adlfs>=2024.2.0"]
[tool.maturin]
python-source = "python"
module-name = "lancedb._lancedb"
# uv installs the project as an editable package before `uv run`, so keep that
# bootstrap build consistent with `maturin develop`.
editable-profile = "dev"
[build-system]
requires = ["maturin>=1.9.4"]
requires = ["maturin>=1.10"]
build-backend = "maturin"
[tool.ruff.lint]
python/python/lancedb/__init__.py
+22
View File
@@ -29,6 +29,7 @@ from .functions import (
FunctionRegistrationRequest as FunctionRegistrationRequest,
FunctionVersion as FunctionVersion,
PythonRuntimeSpec as PythonRuntimeSpec,
RefreshColumnResult as RefreshColumnResult,
UdfDefinition as UdfDefinition,
udf as udf,
)
@@ -178,6 +179,18 @@ def connect(
... },
... )
For Azure Blob Storage, credentials can be passed directly without setting
environment variables:
>>> azure_storage_options = {
... "account_name": "some-account",
... "account_key": "some-key",
... }
>>> db = lancedb.connect( # doctest: +SKIP
... "az://my-container/my-database",
... storage_options=azure_storage_options,
... )
For tests and temporary data, use an in-memory database:
>>> db = lancedb.connect("memory://")
@@ -464,6 +477,10 @@ async def connect_async(
--------
>>> import lancedb
>>> azure_storage_options = {
... "account_name": "some-account",
... "account_key": "some-key",
... }
>>> async def doctest_example():
... # For a local directory, provide a path to the database
... db = await lancedb.connect_async("~/.lancedb")
@@ -471,6 +488,11 @@ async def connect_async(
... db = await lancedb.connect_async("s3://my-bucket/lancedb",
... storage_options={
... "aws_access_key_id": "***"})
... # Azure credentials can also be passed directly
... db = await lancedb.connect_async(
... "az://my-container/my-database",
... storage_options=azure_storage_options,
... )
... # For tests and temporary data, use an in-memory database
... db = await lancedb.connect_async("memory://")
... # Connect to LanceDB cloud
python/python/lancedb/_lancedb.pyi
+3 -9
View File
@@ -147,7 +147,7 @@ class Connection(object):
limit: Optional[int],
) -> list[str]: ... # Deprecated: Use list_tables instead
def job(self, job_id: str) -> Job: ...
async def create_function_async(self, request_json: str) -> FunctionJob: ...
async def create_function_async(self, request_json: str) -> Job: ...
async def get_function(self, name: str, version: str) -> str: ...
async def list_jobs(self) -> List[JobInfo]: ...
async def get_job(self, job_id: str) -> Optional[JobDescription]: ...
@@ -234,14 +234,7 @@ class Job:
@property
def id(self) -> Optional[str]: ...
async def status(self) -> str: ...
async def wait(self) -> None: ...
async def cancel(self) -> None: ...
class FunctionJob:
@property
def id(self) -> Optional[str]: ...
async def status(self) -> str: ...
async def wait(self) -> str: ...
async def wait(self) -> Optional[str]: ...
async def cancel(self) -> None: ...
class JobInfo:
@@ -290,6 +283,7 @@ class Table:
mode: Literal["append", "overwrite"],
progress: Optional[Any] = None,
write_parallelism: Optional[int] = None,
allow_external_blob_outside_bases: bool = False,
) -> AddResult: ...
async def update(
self, updates: Dict[str, str], where: Optional[str]
python/python/lancedb/db.py
+2 -2
View File
@@ -46,7 +46,7 @@ from lance_namespace.errors import NamespaceNotEmptyError, TableNotFoundError
from . import __version__
from ._lancedb import connect as lancedb_connect # type: ignore
from .functions import FunctionVersion, UdfDefinition
from .job import AsyncJob, Job, _function_job
from .job import AsyncJob, Job, _typed_job
from .materialized_view import (
AsyncMaterializedView,
MaterializedView,
@@ -2237,7 +2237,7 @@ class AsyncConnection(object):
inner = await self._inner.create_function_async(
definition.registration_request.to_canonical_json()
)
return _function_job(inner)
return _typed_job(inner, FunctionVersion.from_json)
async def get_function(self, name: str, *, version: str) -> FunctionVersion:
"""Open one exact immutable Function version from the remote catalog."""
python/python/lancedb/functions.py
+231 -104
View File
@@ -1,26 +1,30 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
"""Canonical values exchanged with LanceDB Enterprise Function services.
"""Canonical Function values exchanged with LanceDB Enterprise services.
These immutable models contain client/wire state only. Catalog persistence,
environment bake, secret resolution, and execution are owned by Sophon.
environment bake, and execution are owned by Sophon.
``RefreshColumnResult`` is also the backend-neutral result of a local
expression-backed refresh job.
"""
from __future__ import annotations
import ast
import builtins
import base64
import functools
import hashlib
import importlib
import inspect
import symtable
import json
import math
import re
import sys
import textwrap
import types
import uuid
from collections.abc import Mapping
from datetime import date, datetime
from typing import (
@@ -218,13 +222,14 @@ class PythonEnvironmentSpec(_RemoteValue):
kind: str
packages: tuple[str, ...] = ()
channels: tuple[str, ...] = ()
path: Optional[str] = None
modules: tuple[str, ...] = ()
image: Optional[str] = None
class PythonRuntimeSpec(_RemoteValue):
"""Remote runtime definition with non-secret environment values.
"""Remote runtime definition with environment values.
V1 supports ``kind="python"``. Newer runtime kinds remain readable, while
their unknown payload fields are intentionally not retained by the client.
@@ -263,7 +268,6 @@ class FunctionVersion(_RemoteValue):
runtime: PythonRuntimeSpec
runtime_digest: str
environment_digest: str
required_secrets: tuple[str, ...] = ()
created_at: str
def __call__(self, **inputs: Any) -> FunctionApplication:
@@ -271,7 +275,7 @@ class FunctionVersion(_RemoteValue):
Every input must be a direct [lancedb.col][lancedb.expr.col]
reference. The returned application is immutable and retains a
named-struct output as one sibling group, so every row's sibling values
named-struct output as one binding, so every row's sibling values
come from one logical Function evaluation. Map result fields to table
columns with
[FunctionApplication.rename][lancedb.functions.FunctionApplication.rename],
@@ -321,22 +325,16 @@ class FunctionVersion(_RemoteValue):
function=FunctionVersionRef(name=self.name, version=self.version),
inputs=tuple(bindings),
output=self.signature.output,
group_id=f"fg_{uuid.uuid4().hex}",
)
class FunctionRegistrationRequest(_RemoteValue):
"""Stable remote registration envelope produced by :func:`udf`.
Only secret names are represented. Secret values are resolved inside the
remote service and have no client request field.
"""
"""Stable remote registration envelope produced by :func:`udf`."""
name: str
artifact: FunctionArtifactRequest
signature: FunctionSignature
runtime: PythonRuntimeSpec
required_secrets: tuple[str, ...] = ()
class FunctionVersionRef(_OpenRemoteValue):
@@ -365,7 +363,7 @@ class ApplicationInput(_OpenRemoteValue):
class FunctionApplication(_OpenRemoteValue):
"""Immutable pre-declaration application of an exact Function version.
A named-struct output remains one grouped application through table
A named-struct output remains one application through table
declaration and execution.
[FunctionApplication.rename][lancedb.functions.FunctionApplication.rename]
records the result-field to table-column mapping without splitting sibling
@@ -375,7 +373,6 @@ class FunctionApplication(_OpenRemoteValue):
function: FunctionVersionRef
inputs: tuple[ApplicationInput, ...]
output: FunctionOutput
group_id: str
columns: Mapping[str, str] = Field(default_factory=dict)
def _known_dict(self) -> dict[str, Any]:
@@ -447,12 +444,10 @@ class OutputMapping(_RemoteValue):
class FunctionBinding(_RemoteValue):
"""Immutable grouped binding persisted by the Enterprise table service."""
"""Immutable Function binding persisted by the Enterprise table service."""
binding_id: str
revision: _UInt64
function: FunctionVersionRef
group_id: str
inputs: tuple[InputBinding, ...]
outputs: tuple[OutputMapping, ...]
input_schema: Optional[Mapping[str, Any]] = None
@@ -460,7 +455,11 @@ class FunctionBinding(_RemoteValue):
class RefreshColumnResult(_RemoteValue):
"""Terminal result of a remote Function-column refresh Job."""
"""Terminal result of an expression-backed or Function-backed refresh Job.
Local jobs produce this value in process. LanceDB Cloud and Enterprise
decode the same value from the durable server-job terminal payload.
"""
rows_assigned: _UInt64
rows_failed: _UInt64
@@ -480,62 +479,60 @@ class RefreshColumnResult(_RemoteValue):
_FUNCTION_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_.-]*$")
_SECRET_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
_GRAMMAR_PRIMITIVES = (
(pa.bool_(), "bool"),
(pa.int8(), "int8"),
(pa.int16(), "int16"),
(pa.int32(), "int32"),
(pa.int64(), "int64"),
(pa.uint8(), "uint8"),
(pa.uint16(), "uint16"),
(pa.uint32(), "uint32"),
(pa.uint64(), "uint64"),
(pa.float16(), "float16"),
(pa.float32(), "float32"),
(pa.float64(), "float64"),
(pa.string(), "utf8"),
(pa.binary(), "binary"),
(pa.date32(), "date32"),
(pa.date64(), "date64"),
)
def _canonical_arrow_type(data_type: pa.DataType) -> str:
primitive_types = (
(pa.bool_(), "bool"),
(pa.int8(), "int8"),
(pa.int16(), "int16"),
(pa.int32(), "int32"),
(pa.int64(), "int64"),
(pa.uint8(), "uint8"),
(pa.uint16(), "uint16"),
(pa.uint32(), "uint32"),
(pa.uint64(), "uint64"),
(pa.float16(), "float16"),
(pa.float32(), "float32"),
(pa.float64(), "float64"),
(pa.string(), "utf8"),
(pa.large_utf8(), "large_utf8"),
(pa.binary(), "binary"),
(pa.large_binary(), "large_binary"),
(pa.date32(), "date32"),
(pa.date64(), "date64"),
)
for candidate, name in primitive_types:
"""The server's V1 Function type grammar. Anything outside it is rejected
here rather than at registration."""
for candidate, name in _GRAMMAR_PRIMITIVES:
if data_type == candidate:
return name
if pa.types.is_fixed_size_binary(data_type):
return f"fixed_size_binary[{data_type.byte_width}]"
if pa.types.is_list(data_type):
return f"list<{_canonical_arrow_type(data_type.value_type)}>"
if pa.types.is_large_list(data_type):
return f"large_list<{_canonical_arrow_type(data_type.value_type)}>"
if pa.types.is_fixed_size_list(data_type):
if pa.types.is_list(data_type) or pa.types.is_large_list(data_type):
prefix = "list" if pa.types.is_list(data_type) else "large_list"
return f"{prefix}<{_canonical_list_item(data_type)}>"
if pa.types.is_fixed_size_list(data_type) and data_type.list_size > 0:
return (
f"fixed_size_list<{_canonical_arrow_type(data_type.value_type)}>"
f"[{data_type.list_size}]"
f"fixed_size_list<{_canonical_list_item(data_type)}, {data_type.list_size}>"
)
if pa.types.is_struct(data_type):
fields = ",".join(
f"{field.name}:{_canonical_arrow_type(field.type)}" for field in data_type
)
return f"struct<{fields}>"
if pa.types.is_timestamp(data_type):
timezone = f",tz={data_type.tz}" if data_type.tz is not None else ""
return f"timestamp[{data_type.unit}{timezone}]"
if pa.types.is_time32(data_type) or pa.types.is_time64(data_type):
return f"time[{data_type.unit}]"
if pa.types.is_duration(data_type):
return f"duration[{data_type.unit}]"
if pa.types.is_decimal(data_type):
bit_width = data_type.bit_width
return f"decimal{bit_width}({data_type.precision},{data_type.scale})"
raise TypeError(f"unsupported Arrow type for Function signature: {data_type}")
def _canonical_list_item(data_type: pa.DataType) -> str:
"""The grammar names only the item type; it always means a non-nullable
child called `item`, so any other child metadata cannot be represented."""
child = data_type.value_field
if child.name != "item" or child.nullable or child.metadata:
raise TypeError(
"unsupported Arrow type for Function signature: list items must be a "
f"non-nullable field named 'item', got {child}"
)
return _canonical_arrow_type(child.type)
def _list_of(item: pa.DataType) -> pa.DataType:
return pa.list_(pa.field("item", item, nullable=False))
def _annotation_type(annotation: Any) -> tuple[pa.DataType, bool]:
nullable = False
origin = get_origin(annotation)
@@ -583,7 +580,7 @@ def _annotation_type(annotation: Any) -> tuple[pa.DataType, bool]:
value_type, value_nullable = _annotation_type(arguments[0])
if value_nullable:
raise TypeError("nullable Function list elements are not supported")
return pa.list_(value_type), nullable
return _list_of(value_type), nullable
raise TypeError(f"unsupported Function annotation: {annotation!r}")
@@ -730,6 +727,104 @@ def _literal_source(value: Any) -> str:
)
_DYNAMIC_NAMESPACE_ACCESS = frozenset(
{"globals", "locals", "vars", "eval", "exec", "compile", "__import__"}
)
# Modules that hand out namespaces (`sys.modules`, `builtins`, importers,
# introspection). The artifact's module namespace holds only the names it was
# packaged with, so reaching around it cannot be represented.
_NAMESPACE_MODULES = frozenset(
{"sys", "builtins", "importlib", "inspect", "gc", "ctypes", "types"}
)
def _namespace_acquisition(
definition: ast.FunctionDef, references: set[str]
) -> list[str]:
found = set(references & _DYNAMIC_NAMESPACE_ACCESS)
for node in ast.walk(definition):
if isinstance(node, ast.Import):
found.update(
alias.name
for alias in node.names
if alias.name.split(".")[0] in _NAMESPACE_MODULES
)
elif isinstance(node, ast.ImportFrom) and node.module:
if node.module.split(".")[0] in _NAMESPACE_MODULES:
found.add(node.module)
return sorted(found)
def _module_references(module_source: str) -> set[str]:
"""Names any scope in `module_source` binds or loads at module scope.
Python's own scope analysis on the exact text that ships: free variables
belong to an enclosing scope inside the function, and postponed
annotations are not runtime loads."""
def visit(table: symtable.SymbolTable, found: set[str]) -> None:
for symbol in table.get_symbols():
if symbol.is_global() and (
symbol.is_referenced() or symbol.is_declared_global()
):
found.add(symbol.get_name())
for child in table.get_children():
visit(child, found)
found: set[str] = set()
for table in symtable.symtable(module_source, "<udf>", "exec").get_children():
visit(table, found)
return found
def _global_source(name: str, value: Any) -> str:
"""One module-level line that rebinds `name` to `value` in the artifact:
an import for modules and importable classes/functions, a literal otherwise."""
if isinstance(value, types.ModuleType):
if value.__name__.split(".")[0] in _NAMESPACE_MODULES:
raise ValueError(
f"@udf cannot package dynamic namespace access: {value.__name__!r}"
)
try:
imported = importlib.import_module(value.__name__)
except ImportError:
imported = None
if imported is not value:
raise TypeError(
f"Function source references module {name!r} that does not import "
f"as {value.__name__!r}"
)
return f"import {value.__name__} as {name}"
module_name = getattr(value, "__module__", None)
qualname = getattr(value, "__qualname__", None)
if (
isinstance(module_name, str)
and isinstance(qualname, str)
and module_name != "__main__"
and "." not in qualname
and "<" not in qualname
):
try:
imported = getattr(importlib.import_module(module_name), qualname)
except (ImportError, AttributeError):
imported = None
if imported is value:
return f"from {module_name} import {qualname} as {name}"
return f"{name} = {_literal_source(value)}"
def _is_recursive_reference(function: Callable[..., Any], name: str) -> bool:
"""`name` inside the body means the function itself unless the module has
since bound it to something else."""
if name != function.__name__:
return False
bound = function.__globals__.get(name, function)
if bound is function:
return True
# The decorator's own result is the one wrapper known to call `function`
# unchanged; any other binding may behave differently from a self-call.
return type(bound) is UdfDefinition and bound._function is function
def _package_source(function: Callable[..., Any]) -> bytes:
if not inspect.isfunction(function) or inspect.iscoroutinefunction(function):
raise TypeError("@udf requires a synchronous Python function")
@@ -754,23 +849,46 @@ def _package_source(function: Callable[..., Any]) -> bytes:
closure = inspect.getclosurevars(function)
if closure.nonlocals:
raise ValueError("@udf cannot package functions that capture closure values")
if closure.unbound:
raise ValueError(
f"@udf source contains unresolved global names: {sorted(closure.unbound)!r}"
)
globals_source = []
for name, value in sorted(closure.globals.items()):
if isinstance(value, types.ModuleType):
globals_source.append(f"import {value.__name__} as {name}")
else:
globals_source.append(f"{name} = {_literal_source(value)}")
function_source = ast.unparse(definition)
parts = ["from __future__ import annotations"]
module_header = "from __future__ import annotations"
references = _module_references(f"{module_header}\n\n{function_source}\n")
dynamic = _namespace_acquisition(definition, references)
if dynamic:
raise ValueError(f"@udf cannot package dynamic namespace access: {dynamic!r}")
# Resolve every module-scope reference the way the interpreter would: the
# function's own globals first (a module global may shadow a builtin, and
# nested scopes are not visible to getclosurevars), then its builtins.
# The artifact runs under the standard builtins; only the exact mapping is
# provably equivalent (a subclass or copy can change lookups and hooks).
if function.__builtins__ is not vars(builtins):
raise ValueError("@udf cannot package a non-standard builtins environment")
globals_source = []
unresolved = []
for name in sorted(references):
if name == function.__name__:
if not _is_recursive_reference(function, name):
raise ValueError(
f"@udf cannot package {name!r}: the module binds that name to "
"another value, which the artifact's own definition would shadow"
)
continue
if name in function.__globals__:
globals_source.append(_global_source(name, function.__globals__[name]))
elif hasattr(builtins, name):
pass
else:
unresolved.append(name)
if unresolved:
raise ValueError(
f"@udf source contains unresolved global names: {unresolved!r}"
)
parts = [module_header]
if globals_source:
parts.extend(["", *globals_source])
parts.extend(["", function_source, ""])
return "\n".join(parts).encode("utf-8")
packaged = "\n".join(parts)
return packaged.encode("utf-8")
class UdfDefinition:
@@ -791,33 +909,32 @@ class UdfDefinition:
output_schema: Optional[pa.DataType | pa.Field | pa.Schema],
pip: tuple[str, ...],
env: Mapping[str, str],
secrets: tuple[str, ...],
python_version: Optional[str],
conda: tuple[str, ...] = (),
conda_channels: tuple[str, ...] = (),
):
function_name = name or function.__name__
if not _FUNCTION_NAME.fullmatch(function_name):
raise ValueError(f"invalid Function name: {function_name!r}")
packages = tuple(sorted(set(pip)))
if pip and conda:
raise ValueError("a Function environment is pip or conda, not both")
if conda_channels and not conda:
raise ValueError("conda_channels requires conda packages")
packages = tuple(sorted(set(conda if conda else pip)))
if any(not package or package != package.strip() for package in packages):
raise ValueError("pip requirements must be non-empty and trimmed")
raise ValueError("package requirements must be non-empty and trimmed")
if conda:
environment_spec = PythonEnvironmentSpec(
kind="conda", packages=packages, channels=tuple(conda_channels)
)
else:
environment_spec = PythonEnvironmentSpec(kind="pip", packages=packages)
environment = dict(env)
if any(
not isinstance(key, str) or not isinstance(value, str)
for key, value in environment.items()
):
raise TypeError("Function env keys and values must be strings")
required_secrets = tuple(sorted(set(secrets)))
invalid_secrets = [
secret for secret in required_secrets if not _SECRET_NAME.fullmatch(secret)
]
if invalid_secrets:
raise ValueError(f"invalid Function secret names: {invalid_secrets!r}")
overlap = set(environment) & set(required_secrets)
if overlap:
raise ValueError(
f"Function env and secret names must be disjoint: {sorted(overlap)!r}"
)
signature = _infer_signature(function, input_schema, output_schema)
source = _package_source(function)
digest = f"sha256:{hashlib.sha256(source).hexdigest()}"
@@ -825,7 +942,7 @@ class UdfDefinition:
kind="python",
python_version=python_version
or f"{sys.version_info.major}.{sys.version_info.minor}",
environment=PythonEnvironmentSpec(kind="pip", packages=packages),
environment=environment_spec,
env=environment,
)
self._function = function
@@ -846,7 +963,6 @@ class UdfDefinition:
),
signature=signature,
runtime=runtime,
required_secrets=required_secrets,
)
functools.update_wrapper(self, function)
@@ -872,8 +988,9 @@ def udf(
output_schema: Optional[pa.DataType | pa.Field | pa.Schema] = None,
pip: tuple[str, ...] | list[str] = (),
env: Optional[Mapping[str, str]] = None,
secrets: tuple[str, ...] | list[str] = (),
python_version: Optional[str] = None,
conda: tuple[str, ...] | list[str] = (),
conda_channels: tuple[str, ...] | list[str] = (),
) -> Callable[[Callable[..., Any]], UdfDefinition]: ...
@@ -885,8 +1002,9 @@ def udf(
output_schema: Optional[pa.DataType | pa.Field | pa.Schema] = None,
pip: tuple[str, ...] | list[str] = (),
env: Optional[Mapping[str, str]] = None,
secrets: tuple[str, ...] | list[str] = (),
python_version: Optional[str] = None,
conda: tuple[str, ...] | list[str] = (),
conda_channels: tuple[str, ...] | list[str] = (),
):
"""Prepare a scalar Python callable for remote Function registration.
@@ -909,14 +1027,22 @@ def udf(
provided together with ``input_schema``.
pip : sequence of str, optional
Pip requirements for the remote environment.
conda : sequence of str, optional
Conda packages for the remote environment, instead of ``pip``.
conda_channels : sequence of str, optional
Conda channels in priority order; requires ``conda``.
env : mapping of str to str, optional
Non-secret environment variables. Use ``secrets`` for credentials.
secrets : sequence of str, optional
Names of secrets resolved by the remote service. Secret values are not
accepted by this API or included in the registration request.
Environment variables included in the Function definition.
python_version : str, optional
Remote Python major/minor version. Defaults to the client version.
The packaged artifact is a snapshot: the function source plus exactly
the module-level names it references (modules as imports, importable
classes and functions as imports, literals inline). Code that reaches the
module namespace another way -- ``globals()``/``eval``, ``sys.modules``,
``builtins`` -- is rejected where it can be seen and otherwise
unsupported; closures and a non-standard ``__builtins__`` are rejected.
Returns
-------
UdfDefinition
@@ -928,7 +1054,7 @@ def udf(
Examples
--------
>>> from lancedb import udf
>>> @udf(pip=["numpy==2.2.0"], secrets=["MODEL_TOKEN"])
>>> @udf(pip=["numpy==2.2.0"])
... def score(value: float) -> float:
... return value * 2
>>> score(1.5)
@@ -943,8 +1069,9 @@ def udf(
output_schema=output_schema,
pip=tuple(pip),
env={} if env is None else env,
secrets=tuple(secrets),
python_version=python_version,
conda=tuple(conda),
conda_channels=tuple(conda_channels),
)
if function is None:
python/python/lancedb/index.py
+12
View File
@@ -7,6 +7,7 @@ from typing import List, Literal, Optional
from ._lancedb import (
IndexConfig,
)
from .query import DocumentGranularity
from .types import BaseTokenizerType
lang_mapping = {
@@ -121,6 +122,11 @@ class FTS:
>>> config = FTS(block_size=256)
Create an index that treats each deepest-list element as one document:
>>> from lancedb.query import DocumentGranularity
>>> config = FTS(document_granularity=DocumentGranularity.LIST_ELEMENT)
Attributes
----------
with_position : bool, default False
@@ -172,6 +178,11 @@ class FTS:
roughly half of the available CPU cores. The effective value is
limited by the available compute capacity. This build-only setting is
not persisted with the index and does not apply to remote tables.
document_granularity : DocumentGranularity, default ROW
``ROW`` treats the selected text in one table row as one document.
``LIST_ELEMENT`` treats each element of the deepest list on the indexed
field path as one document and returns its physical coordinates in
``_doc_index`` for matching queries.
Notes
-----
@@ -196,6 +207,7 @@ class FTS:
custom_stop_words: Optional[List[str]] = None
memory_limit: Optional[int] = None
num_workers: Optional[int] = None
document_granularity: DocumentGranularity = DocumentGranularity.ROW
@dataclass
python/python/lancedb/job.py
+28 -30
View File
@@ -5,12 +5,11 @@
import asyncio
from datetime import timedelta
from typing import Any, Generic, Optional, TypeVar, cast
from typing import Any, Callable, Generic, Optional, TypeVar, cast
from lancedb.background_loop import LOOP
from . import _lancedb
from .functions import FunctionVersion
T = TypeVar("T")
@@ -18,11 +17,18 @@ T = TypeVar("T")
class AsyncJob(Generic[T]):
"""A handle to an operation that may still be running.
The operation may already be complete when the handle is created.
The operation may already be complete when the handle is created. ``T``
is the endpoint's terminal result type; unit-result jobs resolve to
``None``.
"""
def __init__(self, inner: Optional[Any]):
def __init__(
self,
inner: Optional[Any],
result_decoder: Optional[Callable[[Any], T]] = None,
):
self._inner = inner
self._result_decoder = result_decoder
@property
def id(self) -> Optional[str]:
@@ -50,17 +56,21 @@ class AsyncJob(Generic[T]):
async def wait(self, timeout: Optional[timedelta] = None) -> T:
"""Wait until the operation reaches a terminal state.
Returns the endpoint's typed result, or ``None`` for a unit-result
job.
Raises `JobFailedError` if the operation failed, `JobCancelledError`
if it was cancelled, and `TimeoutError` if `timeout` elapses first.
"""
if self._inner is None:
return cast(T, None)
if timeout is None:
return cast(T, await self._inner.wait())
return cast(
T,
await asyncio.wait_for(self._inner.wait(), timeout.total_seconds()),
)
result = await self._inner.wait()
else:
result = await asyncio.wait_for(self._inner.wait(), timeout.total_seconds())
if self._result_decoder is not None:
return self._result_decoder(result)
return cast(T, result)
async def cancel(self):
"""Request cancellation. Cancelling a finished operation is a no-op."""
@@ -70,7 +80,7 @@ class AsyncJob(Generic[T]):
class Job(Generic[T]):
"""Synchronous counterpart of `AsyncJob`."""
"""Synchronous counterpart of `AsyncJob` with the same result type."""
def __init__(self, inner: Optional[AsyncJob[T]]):
self._inner = inner
@@ -96,6 +106,9 @@ class Job(Generic[T]):
def wait(self, timeout: Optional[timedelta] = None) -> T:
"""Block until the operation reaches a terminal state.
Returns the endpoint's typed result, or ``None`` for a unit-result
job.
Raises `JobFailedError` if the operation failed, `JobCancelledError`
if it was cancelled, and `TimeoutError` if `timeout` elapses first.
"""
@@ -110,23 +123,8 @@ class Job(Generic[T]):
LOOP.run(self._inner.cancel())
class _FunctionJobAdapter:
def __init__(self, inner: "_lancedb.FunctionJob"):
self._inner = inner
@property
def id(self) -> Optional[str]:
return self._inner.id
async def status(self) -> str:
return await self._inner.status()
async def wait(self) -> FunctionVersion:
return FunctionVersion.from_json(await self._inner.wait())
async def cancel(self):
await self._inner.cancel()
def _function_job(inner: "_lancedb.FunctionJob") -> AsyncJob[FunctionVersion]:
return AsyncJob(_FunctionJobAdapter(inner))
def _typed_job(
inner: "_lancedb.Job", result_decoder: Callable[[str], T]
) -> AsyncJob[T]:
"""Bind an internal JSON-producing job to its public result model."""
return AsyncJob(inner, result_decoder)
python/python/lancedb/permutation.py
+21 -5
View File
@@ -391,6 +391,15 @@ def _table_to_pickle_state(table: Table) -> dict[str, Any]:
}
def _drop_base_version(permutation_data: pa.Table) -> pa.Table:
"""Strip the recorded base version so the reader leaves the base table unpinned."""
metadata = dict(permutation_data.schema.metadata or {})
if metadata.pop(b"base_version", None) is None:
return permutation_data
metadata.pop(b"base_branch", None)
return permutation_data.replace_schema_metadata(metadata)
def _table_from_pickle_state(state: dict[str, Any]) -> Table:
from . import connect
@@ -679,11 +688,15 @@ class Permutation:
from . import connect
connection_factory = state["connection_factory"]
rebuilt_base = False
if connection_factory is not None:
base_table = connection_factory(state["base_table_name"])
elif "base_table_state" in state:
base_table = _table_from_pickle_state(state["base_table_state"])
base_state = state["base_table_state"]
rebuilt_base = base_state["kind"] == "memory"
base_table = _table_from_pickle_state(base_state)
elif "base_table_data" in state:
rebuilt_base = True
# In-memory base table inlined into the pickle; rebuild the same
# way we rebuild the in-memory permutation table.
mem_db = connect("memory://")
@@ -701,11 +714,14 @@ class Permutation:
)
permutation_table: Optional[Table] = None
if state["permutation_data"] is not None:
permutation_data = state["permutation_data"]
if permutation_data is not None:
if rebuilt_base:
# The base table was materialized from Arrow, so it is a fresh
# single-version dataset and the recorded pin cannot resolve on it.
permutation_data = _drop_base_version(permutation_data)
mem_db = connect("memory://")
permutation_table = mem_db.create_table(
"permutation", state["permutation_data"]
)
permutation_table = mem_db.create_table("permutation", permutation_data)
self.base_table = base_table
self.permutation_table = permutation_table
python/python/lancedb/query.py
+29 -5
View File
@@ -375,6 +375,13 @@ class FullTextOperator(str, Enum):
OR = "OR"
class DocumentGranularity(str, Enum):
"""The unit treated as one full-text-search document."""
ROW = "row"
LIST_ELEMENT = "list_element"
class Occur(str, Enum):
SHOULD = "SHOULD"
MUST = "MUST"
@@ -478,6 +485,10 @@ class MatchQuery(FullTextQuery):
prefix_length : int, optional
The number of beginning characters being unchanged for fuzzy matching.
This is useful to achieve prefix matching.
document_granularity : DocumentGranularity, optional
Explicitly select row or deepest-list-element documents. If omitted,
the indexed granularity is inferred. When both granularities are indexed
for the field, this must be specified. With no index, row granularity is used.
"""
query: str
@@ -487,6 +498,9 @@ class MatchQuery(FullTextQuery):
max_expansions: int = pydantic.Field(50, kw_only=True)
operator: FullTextOperator = pydantic.Field(FullTextOperator.OR, kw_only=True)
prefix_length: int = pydantic.Field(0, kw_only=True)
document_granularity: Optional[DocumentGranularity] = pydantic.Field(
None, kw_only=True
)
def query_type(self) -> FullTextQueryType:
return FullTextQueryType.MATCH
@@ -503,11 +517,20 @@ class PhraseQuery(FullTextQuery):
The query string to match against.
column : str
The name of the column to match against.
slop : int, default 0
The maximum number of intervening positions permitted in the phrase.
document_granularity : DocumentGranularity, optional
Explicitly select row or deepest-list-element documents. If omitted,
the indexed granularity is inferred. When both granularities are indexed
for the field, this must be specified. With no index, row granularity is used.
"""
query: str
column: str
slop: int = pydantic.Field(0, kw_only=True)
document_granularity: Optional[DocumentGranularity] = pydantic.Field(
None, kw_only=True
)
def query_type(self) -> FullTextQueryType:
return FullTextQueryType.MATCH_PHRASE
@@ -3378,9 +3401,10 @@ class AsyncQuery(AsyncStandardQuery):
pass in multiple vectors. When multiple vectors are passed in, if the vector
column is with multivector type, then the vectors will be treated as a single
query. Or the vectors will be treated as multiple queries, this can be useful
if you want to find the nearest vectors to multiple query vectors.
This is not expected to be faster than making multiple queries concurrently;
it is just a convenience method. If multiple vectors are passed in then
if you want to find the nearest vectors to multiple query vectors. Flat
searches share one table scan across the query vectors, avoiding the scan
and memory amplification of making multiple queries concurrently. If
multiple vectors are passed in then
an additional column `query_index` will be added to the results. This column
will contain the index of the query vector that the result is nearest to.
"""
@@ -3509,8 +3533,8 @@ class AsyncFTSQuery(AsyncStandardQuery):
Typically, a single vector is passed in as the query. However, you can also
pass in multiple vectors. This can be useful if you want to find the nearest
vectors to multiple query vectors. This is not expected to be faster than
making multiple queries concurrently; it is just a convenience method.
vectors to multiple query vectors. Flat searches share one table scan across
the query vectors instead of issuing concurrent full scans.
If multiple vectors are passed in then an additional column `query_index`
will be added to the results. This column will contain the index of the
query vector that the result is nearest to.
python/python/lancedb/remote/table.py
+9 -2
View File
@@ -49,7 +49,7 @@ from lancedb.index import (
LabelList,
)
from lancedb.job import Job
from lancedb.functions import FunctionApplication
from lancedb.functions import FunctionApplication, RefreshColumnResult
from lancedb.remote.db import LOOP
from lancedb.table import IndexConfigType, KNOWN_METRICS
import pyarrow as pa
@@ -61,6 +61,7 @@ from lancedb.table import _normalize_progress
from ..query import (
AnalyzePlanDistributedMetrics,
DocumentGranularity,
LanceQueryBuilder,
LanceTakeQueryBuilder,
LanceVectorQueryBuilder,
@@ -358,6 +359,7 @@ class RemoteTable(Table):
ngram_max_length: int = 3,
prefix_only: bool = False,
block_size: int = 128,
document_granularity: DocumentGranularity = DocumentGranularity.ROW,
name: Optional[str] = None,
):
"""Create a full-text search index on a column.
@@ -380,6 +382,7 @@ class RemoteTable(Table):
ngram_max_length=ngram_max_length,
prefix_only=prefix_only,
block_size=block_size,
document_granularity=document_granularity,
)
LOOP.run(
self._table.create_index(
@@ -619,6 +622,7 @@ class RemoteTable(Table):
fill_value: float = 0.0,
progress: Optional[Union[bool, Callable, Any]] = None,
write_parallelism: Optional[int] = None,
allow_external_blob_outside_bases: bool = False,
) -> AddResult:
"""Add more data to the [Table][lancedb.table.Table].
@@ -651,6 +655,8 @@ class RemoteTable(Table):
data in flight. Defaults to an estimate based on the data size,
capped at the number of CPU cores. Lower this if bulk ingestion is
using too much memory.
allow_external_blob_outside_bases: bool, default False
Not supported on LanceDB Cloud. Setting this raises.
Returns
-------
@@ -667,6 +673,7 @@ class RemoteTable(Table):
fill_value=fill_value,
progress=progress,
write_parallelism=write_parallelism,
allow_external_blob_outside_bases=allow_external_blob_outside_bases,
)
)
finally:
@@ -982,7 +989,7 @@ class RemoteTable(Table):
def refresh_column(self, column: str):
return LOOP.run(self._table.refresh_column(column))
def refresh_column_async(self, column: str) -> Job:
def refresh_column_async(self, column: str) -> Job[RefreshColumnResult]:
return Job(LOOP.run(self._table.refresh_column_async(column)))
def alter_columns(
python/python/lancedb/streaming.py
+1037 -91
View File
File diff suppressed because it is too large Load Diff
python/python/lancedb/table.py
+75 -12
View File
@@ -41,7 +41,7 @@ from ._blob import (
from .types import BlobMode
from lancedb.arrow import peek_reader
from lancedb.background_loop import LOOP, embedding_executor
from lancedb.job import AsyncJob, Job
from lancedb.job import AsyncJob, Job, _typed_job
from .dependencies import (
_check_for_hugging_face,
_check_for_lance,
@@ -73,7 +73,10 @@ from .index import (
FTS,
)
from .expr import Expr
from .functions import FunctionApplication
from .functions import (
FunctionApplication,
RefreshColumnResult as RefreshColumnJobResult,
)
from .merge import LanceMergeInsertBuilder
from .pydantic import LanceModel, model_to_dict
from .query import (
@@ -83,6 +86,7 @@ from .query import (
AsyncQuery,
AsyncTakeQuery,
AsyncVectorQuery,
DocumentGranularity,
FullTextQuery,
LanceEmptyQueryBuilder,
LanceFtsQueryBuilder,
@@ -1248,6 +1252,7 @@ class Table(ABC):
ngram_max_length: int = 3,
prefix_only: bool = False,
block_size: int = 128,
document_granularity: DocumentGranularity = DocumentGranularity.ROW,
wait_timeout: Optional[timedelta] = None,
name: Optional[str] = None,
):
@@ -1326,6 +1331,11 @@ class Table(ABC):
The number of documents per compressed posting block. Must be 128
or 256. A value of 256 uses the experimental FTS V3 format and
may introduce breaking changes.
document_granularity: DocumentGranularity, default ROW
``ROW`` treats the selected text in one table row as one document.
``LIST_ELEMENT`` treats each element of the deepest list on the field
path as one document and returns its physical coordinates in
``_doc_index`` for matching queries.
wait_timeout: timedelta, optional
The timeout to wait if indexing is asynchronous.
name: str, optional
@@ -1349,6 +1359,7 @@ class Table(ABC):
fill_value: float = 0.0,
progress: Optional[Union[bool, Callable, Any]] = None,
write_parallelism: Optional[int] = None,
allow_external_blob_outside_bases: bool = False,
) -> AddResult:
"""Add more data to the [Table][lancedb.table.Table].
@@ -1400,6 +1411,10 @@ class Table(ABC):
data in flight. Defaults to an estimate based on the data size,
capped at the number of CPU cores. Lower this if bulk ingestion is
using too much memory.
allow_external_blob_outside_bases: bool, default False
Store blob URIs that sit outside registered blob bases. The row
keeps a reference, so the object has to stay readable. Local
tables only.
Returns
-------
@@ -2058,7 +2073,7 @@ class Table(ABC):
A mapping with one ``FunctionApplication`` value keeps its scalar
or named-struct result in the named table column. A bare
named-struct application expands its ordered result fields as one
atomic sibling group; aliases come from ``rename(columns=...)``.
atomic binding; aliases come from ``rename(columns=...)``.
Function columns are supported only on LanceDB Cloud and
Enterprise.
computed: Dict[str, str], optional
@@ -2128,7 +2143,7 @@ class Table(ABC):
"""
@abstractmethod
def refresh_column_async(self, column: str) -> Job:
def refresh_column_async(self, column: str) -> Job[RefreshColumnJobResult]:
"""
Like :meth:`refresh_column`, but returns a handle to the refresh job
instead of blocking until it completes.
@@ -2139,6 +2154,12 @@ class Table(ABC):
than failing the job. On local tables the job runs in-process; on
LanceDB Cloud and Enterprise it is the server's backfill job.
Returns
-------
Job[RefreshColumnResult]
A job whose successful ``wait`` returns row counts plus the source
and published table versions.
Examples
--------
>>> import lancedb
@@ -2147,7 +2168,9 @@ class Table(ABC):
>>> table.add_columns(computed={"doubled": "x * 2"})
AddColumnsResult(version=2)
>>> job = table.refresh_column_async("doubled")
>>> job.wait()
>>> result = job.wait()
>>> result.rows_assigned
2
>>> job.status()
'finished'
"""
@@ -2193,12 +2216,25 @@ class Table(ABC):
----------
updates : dict
One or more dicts, each with:
- "path": str — dot-path to the field (e.g. "embedding" or "a.b.c").
- "metadata": dict[str, str | None] — keys to set; a value of ``None``
deletes that key.
- "replace": bool, optional — replace the field's whole metadata map
instead of merging (default False).
The following keys are treated specially, by convention, and should
be used when appropriate:
- "lancedb:description": for a human-readable description of a field.
- ``"lancedb:tag:<name>"`` for a user-defined key-value tag, where the
suffix names the tag category; e.g. "lancedb:tag:model": "clip".
- "lancedb:logical-column" for a column grouping; e.g. "feature_v1"
and "feature_v2" might be in the same logical column.
- "lancedb:status" for status options ("production", "candidate",
"deprecated", "archived") to designate the current life cycle
state of this column.
Returns
-------
UpdateFieldMetadataResult
@@ -3346,6 +3382,7 @@ class LanceTable(Table):
ngram_max_length: int = 3,
prefix_only: bool = False,
block_size: int = 128,
document_granularity: DocumentGranularity = DocumentGranularity.ROW,
name: Optional[str] = None,
):
"""Create a full-text search index on a column.
@@ -3397,7 +3434,11 @@ class LanceTable(Table):
tokenizer_configs = self.infer_tokenizer_configs(tokenizer_name)
tokenizer_configs["custom_stop_words"] = custom_stop_words
config = FTS(block_size=block_size, **tokenizer_configs)
config = FTS(
block_size=block_size,
document_granularity=document_granularity,
**tokenizer_configs,
)
try:
LOOP.run(
@@ -3487,6 +3528,7 @@ class LanceTable(Table):
fill_value: float = 0.0,
progress: Optional[Union[bool, Callable, Any]] = None,
write_parallelism: Optional[int] = None,
allow_external_blob_outside_bases: bool = False,
) -> AddResult:
"""Add data to the table.
If vector columns are missing and the table
@@ -3514,6 +3556,9 @@ class LanceTable(Table):
data in flight. Defaults to an estimate based on the data size,
capped at the number of CPU cores. Lower this if bulk ingestion is
using too much memory.
allow_external_blob_outside_bases: bool, default False
Allow blob URIs outside registered bases. See :meth:`Table.add`.
Local tables only.
Returns
-------
@@ -3530,6 +3575,7 @@ class LanceTable(Table):
fill_value=fill_value,
progress=progress,
write_parallelism=write_parallelism,
allow_external_blob_outside_bases=allow_external_blob_outside_bases,
)
)
finally:
@@ -4178,7 +4224,7 @@ class LanceTable(Table):
[`AsyncTable.refresh_column`][lancedb.AsyncTable.refresh_column]."""
return LOOP.run(self._table.refresh_column(column))
def refresh_column_async(self, column: str) -> Job:
def refresh_column_async(self, column: str) -> Job[RefreshColumnJobResult]:
"""Fill a computed column's unfilled rows, returning a handle to the
refresh job. See
[`Table.refresh_column_async`][lancedb.table.Table.refresh_column_async].
@@ -5450,6 +5496,7 @@ class AsyncTable:
fill_value: Optional[float] = None,
progress: Optional[Union[bool, Callable, Any]] = None,
write_parallelism: Optional[int] = None,
allow_external_blob_outside_bases: bool = False,
) -> AddResult:
"""Add more data to the [AsyncTable][lancedb.table.AsyncTable].
@@ -5480,6 +5527,9 @@ class AsyncTable:
data in flight. Defaults to an estimate based on the data size,
capped at the number of CPU cores. Lower this if bulk ingestion is
using too much memory.
allow_external_blob_outside_bases: bool, default False
Allow blob URIs outside registered bases. See :meth:`Table.add`.
Local tables only.
"""
schema = await self.schema()
@@ -5516,6 +5566,7 @@ class AsyncTable:
mode or "append",
progress=progress,
write_parallelism=write_parallelism,
allow_external_blob_outside_bases=allow_external_blob_outside_bases,
)
except RuntimeError as e:
if "Cast error" in str(e):
@@ -6123,7 +6174,7 @@ class AsyncTable:
A mapping with one ``FunctionApplication`` value keeps its scalar
or named-struct result in the named table column. A bare
named-struct application expands its ordered result fields as one
atomic sibling group; aliases come from ``rename(columns=...)``.
atomic binding; aliases come from ``rename(columns=...)``.
Function columns are supported only on LanceDB Cloud and
Enterprise.
computed: Dict[str, str], optional
@@ -6160,7 +6211,7 @@ class AsyncTable:
isinstance(value, FunctionApplication) for value in transforms.values()
):
raise ValueError(
"one add_columns call declares exactly one Function sibling group"
"one add_columns call declares exactly one Function binding"
)
function_output_name, function_application = next(iter(transforms.items()))
@@ -6218,7 +6269,9 @@ class AsyncTable:
"""
return await self._inner.refresh_column(column)
async def refresh_column_async(self, column: str) -> AsyncJob:
async def refresh_column_async(
self, column: str
) -> AsyncJob[RefreshColumnJobResult]:
"""
Like :meth:`refresh_column`, but returns a handle to the refresh job
instead of blocking until it completes.
@@ -6230,6 +6283,12 @@ class AsyncTable:
in-process; on LanceDB Cloud and Enterprise it is the server's
backfill job.
Returns
-------
AsyncJob[RefreshColumnResult]
A job whose successful ``wait`` returns row counts plus the source
and published table versions.
Examples
--------
>>> import asyncio
@@ -6239,12 +6298,16 @@ class AsyncTable:
... table = await db.create_table("computed_job_async_demo", [{"x": 1}])
... await table.add_columns(computed={"doubled": "x * 2"})
... job = await table.refresh_column_async("doubled")
... await job.wait()
... result = await job.wait()
... assert result.rows_assigned == 1
... return await job.status()
>>> asyncio.run(refresh_in_background())
'finished'
"""
return AsyncJob(await self._inner.refresh_column_async(column))
return _typed_job(
await self._inner.refresh_column_async(column),
RefreshColumnJobResult.from_json,
)
async def alter_columns(
self, *alterations: Iterable[dict[str, Any]]
python/python/tests/docs/test_basic.py
+2 -2
View File
@@ -105,7 +105,7 @@ def test_quickstart(tmp_path):
tbl.create_index(num_sub_vectors=1)
# --8<-- [end:create_index]
# --8<-- [start:delete_rows]
tbl.delete('item = "fizz"')
tbl.delete("item = 'fizz'")
# --8<-- [end:delete_rows]
# --8<-- [start:drop_table]
db.drop_table("my_table")
@@ -201,7 +201,7 @@ async def test_quickstart_async(tmp_path):
await tbl.create_index("vector")
# --8<-- [end:create_index_async]
# --8<-- [start:delete_rows_async]
await tbl.delete('item = "fizz"')
await tbl.delete("item = 'fizz'")
# --8<-- [end:delete_rows_async]
# --8<-- [start:drop_table_async]
await db.drop_table("my_table_async")
python/python/tests/docs/test_guide_tables.py
+2 -2
View File
@@ -266,7 +266,7 @@ def test_table():
tbl.add(pydantic_model_items)
# --8<-- [end:add_table_from_pydantic]
# --8<-- [start:delete_row]
tbl.delete('item = "fizz"')
tbl.delete("item = 'fizz'")
# --8<-- [end:delete_row]
# --8<-- [start:delete_specific_row]
data = [
@@ -538,7 +538,7 @@ async def test_table_async():
await async_tbl.add(pydantic_model_items)
# --8<-- [end:add_table_async_from_pydantic]
# --8<-- [start:delete_row_async]
await async_tbl.delete('item = "fizz"')
await async_tbl.delete("item = 'fizz'")
# --8<-- [end:delete_row_async]
# --8<-- [start:delete_specific_row_async]
data = [
python/python/tests/test_blob.py
+68
View File
@@ -617,3 +617,71 @@ def test_fetch_blobs_nested_path_survives_sort_after_query():
def _identifiable_payload(size: int) -> bytes:
block = 256
return b"".join(bytes([i % 256]) * block for i in range(size // block))
def _external_uri_blob_array(uris):
blob_type = lancedb.blob("image").type
storage_type = blob_type.storage_type
child_names = [field.name for field in storage_type]
assert "uri" in child_names, "blob layout no longer has a uri child"
children = [
pa.array(uris if field.name == "uri" else [None] * len(uris), type=field.type)
for field in storage_type
]
storage = pa.StructArray.from_arrays(children, fields=list(storage_type))
return pa.ExtensionArray.from_storage(blob_type, storage)
def _external_uri_table_and_rows(name, uris):
db = lancedb.connect("memory:///")
schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
table = db.create_table(name, schema=schema)
rows = pa.Table.from_arrays(
[
pa.array(range(len(uris)), type=pa.int64()),
_external_uri_blob_array(uris),
],
schema=schema,
)
return table, rows
def test_add_external_uri_struct_round_trips_with_flag(tmp_path):
payload = b"external-uri-bytes"
blob_path = tmp_path / "payload.bin"
blob_path.write_bytes(payload)
table, rows = _external_uri_table_and_rows("external_struct", [blob_path.as_uri()])
table.add(rows, allow_external_blob_outside_bases=True)
hits = table.search().to_arrow()
blobs = table.fetch_blobs("image", hits)
assert blobs[0].as_py() == payload
def test_add_external_uri_without_flag_raises(tmp_path):
blob_path = tmp_path / "payload.bin"
blob_path.write_bytes(b"unreachable")
table, rows = _external_uri_table_and_rows("external_no_flag", [blob_path.as_uri()])
with pytest.raises(ValueError, match="allow_external_blob_outside_bases"):
table.add(rows)
assert table.count_rows() == 0
def test_add_external_uri_string_round_trips_with_flag(tmp_path):
payload = b"external-uri-bytes"
blob_path = tmp_path / "payload.bin"
blob_path.write_bytes(payload)
db = lancedb.connect("memory:///")
schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
table = db.create_table("external_string", schema=schema)
table.add(
[{"id": 1, "image": blob_path.as_uri()}],
allow_external_blob_outside_bases=True,
)
hits = table.search().to_arrow()
blobs = table.fetch_blobs("image", hits)
assert blobs[0].as_py() == payload
python/python/tests/test_db.py
+2 -2
View File
@@ -774,7 +774,7 @@ def test_drop_table_async(tmp_db: lancedb.DBConnection):
job = tmp_db.drop_table_async("test")
assert job.id is None
assert job.status() == "finished"
job.wait()
assert job.wait() is None
assert tmp_db.table_names() == []
tmp_db.create_table("test", data=data)
@@ -790,7 +790,7 @@ async def test_drop_table_async_connection(tmp_db_async: lancedb.AsyncConnection
job = await tmp_db_async.drop_table_async("test")
assert job.id is None
assert await job.status() == "finished"
await job.wait()
assert await job.wait() is None
assert await tmp_db_async.table_names() == []
python/python/tests/test_elastic_dataloader.py
+1050 -2
View File
File diff suppressed because it is too large Load Diff
python/python/tests/test_first_class_function_slice1.py
+5 -36
View File
@@ -37,21 +37,6 @@ def job_result(name: str) -> dict:
return json.loads(fixture(name))["result"]
def assert_no_secret_values(value):
if isinstance(value, dict):
for key, child in value.items():
assert key not in {
"secret_value",
"secret_values",
"resolved_secret",
"resolved_secrets",
}
assert_no_secret_values(child)
elif isinstance(value, list):
for child in value:
assert_no_secret_values(child)
def test_public_function_values_are_in_api_reference():
docs = Path(__file__).parents[3] / "docs" / "src" / "python" / "python.md"
rendered = docs.read_text()
@@ -109,7 +94,6 @@ def test_function_version_identity_is_immutable_and_exact():
version = FunctionVersion.from_json(json.dumps(value))
assert version.name == "embed"
assert version.version == "fv_01K3EXACT"
assert version.required_secrets == ("HF_TOKEN",)
with pytest.raises((TypeError, ValueError)):
version.version = "fv_changed"
@@ -121,7 +105,7 @@ def test_function_version_identity_is_immutable_and_exact():
assert FunctionVersion(**changed) != version
def test_function_version_binds_named_columns_as_one_immutable_group():
def test_function_version_binds_named_columns_as_one_immutable_application():
version = FunctionVersion.from_json(
json.dumps(job_result("remote_function_job.json"))
)
@@ -131,13 +115,10 @@ def test_function_version_binds_named_columns_as_one_immutable_group():
assert application.function.name == version.name
assert application.function.version == version.version
assert application.output is version.signature.output
assert application.group_id.startswith("fg_")
assert [
(value.parameter, value.kind, value.value["path"])
for value in application.inputs
] == [("text", "column", "documents.body")]
with pytest.raises((TypeError, ValueError)):
application.group_id = "fg_changed"
def test_function_version_binding_validates_names_and_direct_columns():
@@ -156,7 +137,7 @@ def test_function_version_binding_validates_names_and_direct_columns():
def test_function_version_keeps_named_struct_outputs_in_one_application():
value = job_result("remote_function_job.json")
value["name"] = "text_features"
value["version"] = "fv_grouped"
value["version"] = "fv_multi_output"
value["signature"] = {
"inputs": [
{"name": "title", "arrow_type": "utf8", "nullable": True},
@@ -221,7 +202,6 @@ def test_function_application_uses_rename_columns_only():
assert application.columns["normalized_text"] == "search_text"
assert renamed.columns["normalized_text"] == "body_normalized"
assert renamed.function == application.function
assert renamed.group_id == application.group_id
assert not hasattr(application, "rename_outputs")
with pytest.raises(TypeError, match="immutable"):
renamed.columns["normalized_text"] = "changed"
@@ -242,7 +222,6 @@ def test_function_application_uses_rename_columns_only():
def test_binding_and_refresh_result_keep_stable_remote_fields():
binding = FunctionBinding.from_json(fixture("remote_function_binding.json"))
assert binding.revision == 3
assert binding.function.version == "fv_01K3TEXT"
assert [output.output_ordinal for output in binding.outputs] == [0, 1]
assert binding.input_schema is not None
@@ -297,15 +276,6 @@ def test_refresh_result_rejects_non_u64_values(field):
RefreshColumnResult.from_json(json.dumps(value))
def test_canonical_client_values_contain_secret_names_only():
version = FunctionVersion.from_json(
json.dumps(job_result("remote_function_job.json"))
)
canonical = json.loads(version.to_canonical_json())
assert canonical["required_secrets"] == ["HF_TOKEN"]
assert_no_secret_values(canonical)
class _FunctionDeclarationInner:
def __init__(self):
self.calls = []
@@ -322,7 +292,7 @@ def known_application() -> FunctionApplication:
@pytest.mark.asyncio
async def test_add_columns_routes_struct_as_one_and_grouped_expansion_atomically():
async def test_add_columns_routes_struct_as_one_and_multi_output_binding_atomically():
inner = _FunctionDeclarationInner()
table = AsyncTable(inner)
application = known_application()
@@ -343,12 +313,12 @@ async def test_add_columns_routes_struct_as_one_and_grouped_expansion_atomically
@pytest.mark.asyncio
async def test_add_columns_rejects_mixed_groups_and_unknown_newer_application():
async def test_add_columns_rejects_multiple_bindings_and_unknown_newer_application():
inner = _FunctionDeclarationInner()
table = AsyncTable(inner)
application = known_application()
with pytest.raises(ValueError, match="exactly one Function sibling group"):
with pytest.raises(ValueError, match="exactly one Function binding"):
await table.add_columns({"a": application, "b": application})
future = json.loads(fixture("remote_function_application.json"))
@@ -376,7 +346,6 @@ def test_rename_requires_named_struct_and_keeps_partial_mapping_immutable():
"arrow_type": "list<float32>",
"nullable": False,
},
"group_id": "fg_scalar",
}
)
)
python/python/tests/test_first_class_function_slice2.py
+417 -31
View File
@@ -3,7 +3,12 @@
from __future__ import annotations
import base64
import contextlib
import functools
import importlib.util
import types
from datetime import date
import http.server
import json
from pathlib import Path
@@ -16,6 +21,9 @@ import pytest
import lancedb
from lancedb.functions import UdfDefinition, udf
THRESHOLD = 20
_CACHE = None
FIXTURES = (
Path(__file__).parents[3]
@@ -31,28 +39,12 @@ FIXTURES = (
@udf(
pip=["numpy>=2"],
env={"MODE": "test"},
secrets=["API_TOKEN"],
python_version="3.12",
)
def normalize_score(value: float) -> float:
return value / 100.0
def _assert_no_secret_values(value):
if isinstance(value, dict):
for key, child in value.items():
assert key not in {
"secret_value",
"secret_values",
"resolved_secret",
"resolved_secrets",
}
_assert_no_secret_values(child)
elif isinstance(value, list):
for child in value:
_assert_no_secret_values(child)
def test_scalar_udf_matches_shared_registration_golden_and_remains_callable():
assert isinstance(normalize_score, UdfDefinition)
assert normalize_score(25.0) == 0.25
@@ -67,13 +59,417 @@ def test_scalar_udf_matches_shared_registration_golden_and_remains_callable():
"kind": "scalar_to_arrow_batch",
"version": 1,
}
assert request["required_secrets"] == ["API_TOKEN"]
_assert_no_secret_values(request)
def _run_packaged(definition, *args):
"""Execute the shipped artifact in a fresh namespace, as a worker would."""
source = base64.b64decode(definition.registration_request.artifact.content.data)
namespace: dict = {}
exec(compile(source, "<udf>", "exec"), namespace)
return namespace[definition.registration_request.artifact.entrypoint](*args)
def test_udf_conda_environment():
@udf(conda=["scipy", "numpy"], conda_channels=["conda-forge", "defaults"])
def halve(value: float) -> float:
return value / 2
request = json.loads(halve.registration_request.to_canonical_json())
assert request["runtime"]["environment"] == {
"kind": "conda",
"packages": ["numpy", "scipy"],
"channels": ["conda-forge", "defaults"],
}
pip_request = json.loads(normalize_score.registration_request.to_canonical_json())
assert "channels" not in pip_request["runtime"]["environment"]
with pytest.raises(ValueError, match="not both"):
udf(name="both", pip=["numpy"], conda=["numpy"])(lambda value: value)
with pytest.raises(ValueError, match="requires conda"):
udf(name="channels", conda_channels=["conda-forge"])(lambda value: value)
def test_udf_packages_attribute_access_and_body_imports():
@udf
def word_norm(body: str) -> float:
import numpy as np
try:
words = body.split()
except AttributeError as error:
raise ValueError(str(error)) from error
return float(np.linalg.norm([len(w) for w in words]))
assert _run_packaged(word_norm, "aa bb") == pytest.approx(8**0.5)
def test_udf_packages_module_globals_and_global_caches():
@udf
def label(value: int) -> str:
return "big" if value >= THRESHOLD else "small"
assert _run_packaged(label, 21) == "big"
@udf
def cached(value: int) -> int:
global _CACHE
if _CACHE is None:
_CACHE = 40
return _CACHE + value
assert _run_packaged(cached, 2) == 42
def test_udf_annotations_are_not_runtime_names():
@udf
def identity(value: date) -> date:
return value
assert _run_packaged(identity, date(2026, 8, 25)) == date(2026, 8, 25)
def test_udf_nested_scopes_resolve_lexically():
@udf
def score(value: int) -> int:
offset = 2
def add_offset() -> int:
return value + offset
return add_offset() + sum(v for v in [0])
assert _run_packaged(score, 3) == 5
def test_udf_resolves_module_globals_before_builtins(tmp_path):
module_path = tmp_path / "shadowing_udfs.py"
module_path.write_text(
"max = 7\n"
"len = lambda _: 99\n"
"\n"
"def uses_literal_shadow(value: int) -> int:\n"
" def nested() -> int:\n"
" return max\n"
" return nested() + value\n"
"\n"
"def uses_callable_shadow(value: int) -> int:\n"
" def nested() -> int:\n"
" return len([1])\n"
" return nested() + value\n"
)
spec = importlib.util.spec_from_file_location("shadowing_udfs", module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
# The module's `max = 7` is what the interpreter would use, so it ships.
assert _run_packaged(udf(module.uses_literal_shadow), 1) == 8
# A callable global cannot ship; it must not be silently swapped for the builtin.
with pytest.raises(TypeError, match="unsupported global value of type function"):
udf(module.uses_callable_shadow)
def test_canonical_arrow_type_is_exactly_the_grammar():
from lancedb.functions import _GRAMMAR_PRIMITIVES, _canonical_arrow_type
golden = json.loads(
(
Path(__file__).parents[3]
/ "rust/lancedb/tests/fixtures/first_class_functions/v1/arrow_types.json"
).read_text()
)
primitives = [
case["arrow_type"] for case in golden["valid"] if "<" not in case["arrow_type"]
]
assert [name for _, name in _GRAMMAR_PRIMITIVES] == primitives
for outside in [
pa.timestamp("us"),
pa.decimal128(10, 2),
pa.large_string(),
pa.large_binary(),
pa.binary(4),
pa.duration("s"),
pa.struct([pa.field("a", pa.int32())]),
pa.list_(pa.float32(), 0),
pa.list_(pa.timestamp("us")),
]:
with pytest.raises(TypeError, match="unsupported Arrow type"):
_canonical_arrow_type(outside)
def test_udf_nested_annotations_are_postponed_in_the_artifact():
@udf
def score(value: int) -> int:
def identity(item: date) -> date:
return item
identity(date(2026, 8, 25))
return value
assert _run_packaged(score, 3) == 3
def test_udf_ships_globals_the_body_deletes():
@udf
def clear(value: int) -> int:
global _CACHE
del _CACHE
return value
assert _run_packaged(clear, 3) == 3
def test_udf_rejects_a_module_global_that_does_not_import_as_itself(tmp_path):
module_path = tmp_path / "fake_module_udfs.py"
module_path.write_text(
"import types\n"
"np = types.ModuleType('numpy')\n"
"np.sqrt = lambda x: 0\n"
"\n"
"def score(value: int) -> int:\n"
" return int(np.sqrt(value))\n"
)
spec = importlib.util.spec_from_file_location("fake_module_udfs", module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
with pytest.raises(TypeError, match="does not import as 'numpy'"):
udf(module.score)
def test_udf_rejects_a_module_level_namespace_alias(tmp_path):
module_path = tmp_path / "aliasing_udfs.py"
module_path.write_text(
"import builtins as b\n"
"THRESHOLD = 5\n"
"\n"
"def score(value: int) -> int:\n"
" return value + b.vars(b.__import__('aliasing_udfs'))['THRESHOLD']\n"
)
spec = importlib.util.spec_from_file_location("aliasing_udfs", module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
with pytest.raises(ValueError, match="dynamic namespace access"):
udf(module.score)
@pytest.mark.parametrize(
"access",
[
"globals()['THRESHOLD']",
"eval('THRESHOLD')",
"(lambda g: g()['THRESHOLD'])(globals)",
"__import__('sys').modules[__name__].THRESHOLD",
"sys.modules[__name__].THRESHOLD",
],
)
def test_udf_rejects_dynamic_namespace_access(access):
namespace: dict = {}
exec(
f"def score(value: int) -> int:\n return value + {access}\n",
{"THRESHOLD": 5},
namespace,
)
with pytest.raises(ValueError, match="dynamic namespace access"):
_package_from_text(
"def score(value: int) -> int:\n"
" import sys\n"
f" return value + {access}\n"
)
def _package_from_text(source: str, module_globals: dict | None = None):
"""Load `source` as a real module file so the packager can inspect it."""
import tempfile
directory = tempfile.mkdtemp()
path = Path(directory) / "generated_udf_module.py"
path.write_text(source)
spec = importlib.util.spec_from_file_location(f"generated_udf_{id(source)}", path)
module = importlib.util.module_from_spec(spec)
if module_globals:
module.__dict__.update(module_globals)
spec.loader.exec_module(module)
functions = [
value
for value in vars(module).values()
if callable(value) and getattr(value, "__module__", None) == module.__name__
]
return udf(functions[0])
def test_udf_rejects_a_non_standard_builtins_environment():
def score(value: int) -> int:
return len([1]) + value
score.__globals__ # noqa: B018 -- real function, real globals
import builtins
patched = types.FunctionType(
score.__code__,
{"__builtins__": {**vars(builtins), "len": lambda _: 99}},
"score",
)
patched.__annotations__ = score.__annotations__
assert patched(3) == 102
with pytest.raises(ValueError, match="non-standard builtins environment"):
udf(patched)
class ReportingDict(dict): # reports standard entries, resolves differently
def __missing__(self, key):
return vars(builtins)[key]
disguised = types.FunctionType(
score.__code__, {"__builtins__": ReportingDict(len=lambda _: 99)}, "score"
)
disguised.__annotations__ = score.__annotations__
assert disguised(3) == 102
with pytest.raises(ValueError, match="non-standard builtins environment"):
udf(disguised)
hooked = types.FunctionType(
score.__code__,
{"__builtins__": {**vars(builtins), "__import__": lambda *a, **k: None}},
"score",
)
hooked.__annotations__ = score.__annotations__
with pytest.raises(ValueError, match="non-standard builtins environment"):
udf(hooked)
def test_udf_recursion_versus_a_rebound_module_name(tmp_path):
module_path = tmp_path / "rebound_udfs.py"
module_path.write_text(
"def fact(value: int) -> int:\n"
" return 1 if value <= 1 else value * fact(value - 1)\n"
"\n"
"def score(value: int) -> int:\n"
" return score + value\n"
)
spec = importlib.util.spec_from_file_location("rebound_udfs", module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
assert _run_packaged(udf(module.fact), 5) == 120
raw = module.score
module.score = 10
with pytest.raises(ValueError, match="binds that name to another value"):
udf(raw)
# A wrapper that merely exposes __wrapped__ is not the function.
module.score = functools.wraps(raw)(lambda value: 41)
with pytest.raises(ValueError, match="binds that name to another value"):
udf(raw)
# The decorator's own result is; a subclass of it is not.
module.fact = udf(module.fact)
assert _run_packaged(module.fact, 4) == 24
class Twisted(UdfDefinition):
def __call__(self, *args, **kwargs):
return 41
raw_fact = module.fact._function
module.fact = Twisted(
raw_fact,
name=None,
input_schema=None,
output_schema=None,
pip=(),
env={},
python_version=None,
)
with pytest.raises(ValueError, match="binds that name to another value"):
udf(raw_fact)
def test_canonical_arrow_type_rejects_unrepresentable_list_children():
from lancedb.functions import _canonical_arrow_type
for outside in [
pa.list_(pa.float32()), # pyarrow default: nullable child
pa.list_(pa.field("custom", pa.float32(), nullable=False)),
pa.list_(pa.field("item", pa.float32(), nullable=False, metadata={"k": "v"})),
pa.list_(pa.field("item", pa.float32(), nullable=False), 0),
]:
with pytest.raises(TypeError, match="unsupported Arrow type"):
_canonical_arrow_type(outside)
assert (
_canonical_arrow_type(
pa.list_(pa.field("item", pa.float32(), nullable=False), 3)
)
== "fixed_size_list<float32, 3>"
)
def _calls_missing(value: int) -> int:
return missing(value) # noqa: F821
def _shadows_missing_in_a_comprehension(value: int) -> int:
return missing(value) + sum(missing for missing in ()) # noqa: F821
def _shadows_missing_in_a_lambda(value: int) -> int:
return (lambda missing: missing)(value) + missing # noqa: F821
@pytest.mark.parametrize(
"function",
[_calls_missing, _shadows_missing_in_a_comprehension, _shadows_missing_in_a_lambda],
)
def test_udf_rejects_a_truly_unresolved_global(function):
with pytest.raises(ValueError, match=r"unresolved global names: \['missing'\]"):
udf(function)
def _arrow_type_from_golden(spec: dict) -> pa.DataType:
kind = spec["type"]
if kind in ("list", "large_list", "fixed_size_list"):
item = _arrow_type_from_golden(spec["fields"][0]["type"])
field = pa.field("item", item, nullable=False)
if kind == "list":
return pa.list_(field)
if kind == "large_list":
return pa.large_list(field)
return pa.list_(field, spec["length"])
return {
"null": pa.null(),
"bool": pa.bool_(),
"utf8": pa.string(),
"binary": pa.binary(),
"float16": pa.float16(),
"float32": pa.float32(),
"float64": pa.float64(),
"date32": pa.date32(),
"date64": pa.date64(),
}.get(kind) or getattr(pa, kind)()
def test_arrow_type_grammar_matches_the_shared_golden():
golden = json.loads(
(
Path(__file__).parents[3]
/ "rust/lancedb/tests/fixtures/first_class_functions/v1/arrow_types.json"
).read_text()
)
from lancedb.functions import _canonical_arrow_type
emitted = {
case["arrow_type"]: _canonical_arrow_type(_arrow_type_from_golden(case["json"]))
for case in golden["valid"]
}
assert emitted == {
case["arrow_type"]: case["arrow_type"] for case in golden["valid"]
}
assert not set(emitted) & set(golden["invalid"])
for case in golden["server_only"]:
with pytest.raises(TypeError, match="unsupported Arrow type"):
_canonical_arrow_type(_arrow_type_from_golden(case["json"]))
def test_explicit_arrow_schema_is_deterministic():
input_schema = pa.schema([pa.field("value", pa.float32(), nullable=True)])
output_schema = pa.field("embedding", pa.list_(pa.float32(), 3), nullable=False)
output_schema = pa.field(
"embedding",
pa.list_(pa.field("item", pa.float32(), nullable=False), 3),
nullable=False,
)
@udf(input_schema=input_schema, output_schema=output_schema)
def explicit(value):
@@ -82,7 +478,7 @@ def test_explicit_arrow_schema_is_deterministic():
signature = explicit.registration_request.signature
assert signature.inputs[0].arrow_type == "float32"
assert signature.inputs[0].nullable is True
assert signature.output.arrow_type == "fixed_size_list<float32>[3]"
assert signature.output.arrow_type == "fixed_size_list<float32, 3>"
assert signature.output.nullable is False
@@ -130,14 +526,6 @@ def test_annotation_and_explicit_schema_validation_fail_closed():
return value
def test_environment_rejects_secret_value_overlap():
with pytest.raises(ValueError, match="must be disjoint"):
@udf(env={"TOKEN": "plaintext"}, secrets=["TOKEN"])
def overlapping(value: int) -> int:
return value
def test_local_function_catalog_operations_are_not_supported(tmp_path):
db = lancedb.connect(tmp_path)
message = "Function catalog operations are not supported by this database"
@@ -174,7 +562,6 @@ def _mock_remote_function_catalog():
"runtime": body["runtime"],
"runtime_digest": "sha256:runtime",
"environment_digest": "sha256:environment",
"required_secrets": body.get("required_secrets", []),
"created_at": "2026-08-21T00:00:00Z",
}
response = {"job_id": "job-register"}
@@ -187,7 +574,7 @@ def _mock_remote_function_catalog():
"job_state": "DONE",
"result": state["version"],
}
elif self.path == "/v1/functions/get":
elif self.path == "/v1/functions/describe":
assert body == {
"name": "normalize_score",
"version": "fv_exact",
@@ -233,7 +620,6 @@ def test_remote_registration_job_and_exact_version_reopen_round_trip():
assert create_request == json.loads(
normalize_score.registration_request.to_canonical_json()
)
_assert_no_secret_values(create_request)
def test_blocking_remote_registration_returns_function_version():
python/python/tests/test_fts.py
+77
View File
@@ -25,6 +25,7 @@ from lancedb.db import DBConnection
from lancedb.index import FTS
from lancedb.query import (
BoostQuery,
DocumentGranularity,
MatchQuery,
MultiMatchQuery,
PhraseQuery,
@@ -245,6 +246,55 @@ def test_create_inverted_index_rejects_invalid_block_size(table):
table.create_index("text", config=FTS(block_size=129))
def test_list_element_document_granularity(tmp_path):
docs_type = pa.list_(pa.struct([pa.field("content", pa.string())]))
docs = pa.array(
[
[
{"content": "alpha beta"},
None,
{"content": ""},
{"content": "the and"},
{"content": "alpha beta"},
]
],
type=docs_type,
)
table = ldb.connect(tmp_path).create_table(
"list_element_docs", pa.table({"id": [0], "docs": docs})
)
row_table = ldb.connect(tmp_path).create_table(
"row_docs", pa.table({"id": [0], "docs": docs})
)
row_table.create_index("docs.content", config=FTS())
row_result = row_table.search(MatchQuery("alpha", "docs.content")).to_arrow()
assert row_result.num_rows == 1
assert "_doc_index" not in row_result.column_names
granularity = DocumentGranularity.LIST_ELEMENT
table.create_index(
"docs.content",
config=FTS(with_position=True, document_granularity=granularity),
)
assert table.list_indices()[0].columns == ["docs.content"]
def coordinates(query):
result = table.search(query).limit(10).to_arrow()
doc_index_type = result.schema.field("_doc_index").type
assert pa.types.is_list(doc_index_type)
assert doc_index_type.value_type == pa.uint32()
return sorted(result["_doc_index"].to_pylist())
assert coordinates(
MatchQuery("alpha", "docs.content", document_granularity=granularity)
) == [[0], [4]]
assert coordinates(
PhraseQuery("alpha beta", "docs.content", document_granularity=granularity)
) == [[0], [4]]
assert coordinates(MatchQuery("alpha", "docs.content")) == [[0], [4]]
assert FTS().document_granularity is DocumentGranularity.ROW
def test_create_inverted_index_respects_build_memory_limit(table):
with pytest.raises(ValueError, match="exceeds worker memory limit"):
table.create_index(
@@ -1089,6 +1139,20 @@ def test_fts_query_to_json():
)
assert json_str == expected
# Test MatchQuery with list-element document granularity
match_query = MatchQuery(
"hello world",
"text",
document_granularity=DocumentGranularity.LIST_ELEMENT,
)
json_str = match_query.to_json()
expected = (
'{"match":{"column":"text","terms":"hello world","boost":1.0,'
'"fuzziness":0,"max_expansions":50,"operator":"Or","prefix_length":0,'
'"document_granularity":"list_element"}}'
)
assert json_str == expected
# Test MatchQuery with options
match_query = MatchQuery("puppy", "text", fuzziness=2, boost=1.5, prefix_length=3)
json_str = match_query.to_json()
@@ -1098,6 +1162,19 @@ def test_fts_query_to_json():
)
assert json_str == expected
# Test PhraseQuery with list-element document granularity
phrase_query = PhraseQuery(
"quick brown fox",
"title",
document_granularity=DocumentGranularity.LIST_ELEMENT,
)
json_str = phrase_query.to_json()
expected = (
'{"phrase":{"column":"title","terms":"quick brown fox","slop":0,'
'"document_granularity":"list_element"}}'
)
assert json_str == expected
# Test PhraseQuery
phrase_query = PhraseQuery("quick brown fox", "title")
json_str = phrase_query.to_json()
python/python/tests/test_index.py
+1 -1
View File
@@ -88,7 +88,7 @@ async def binary_table(db_async):
async def test_create_index_async_returns_done_job(some_table: AsyncTable):
job = await some_table.create_index_async("id", config=BTree())
assert job.id is None
await job.wait()
assert await job.wait() is None
assert len(await some_table.list_indices()) == 1
await job.cancel()
python/python/tests/test_permutation.py
+25
View File
@@ -56,6 +56,31 @@ def test_execute_does_not_reenter_background_loop(tmp_path, monkeypatch):
assert permutation_tbl._conn.read_consistency_interval is None
def test_pickled_permutation_reads_pinned_version(tmp_path):
"""An unpickled copy must still read the pinned version, which also covers the
version surviving the ``to_arrow()`` round trip in ``__getstate__``."""
import pickle
db = connect(tmp_path)
tbl = db.create_table("base", pa.table({"idx": range(20)}))
permutation_tbl = permutation_builder(tbl).execute()
perm = Permutation.from_tables(tbl, permutation_tbl)
payload = pickle.dumps(perm)
# Compact so the stored row addresses no longer describe these rows at latest.
tbl.delete("true")
tbl.optimize()
assert tbl.count_rows() == 0
# Unpickle after the mutation: __setstate__ reopens at latest, so this only
# passes if the recorded version is applied on reopen.
restored = pickle.loads(payload)
assert len(restored) == 20
rows = restored.__getitems__(list(range(20)))
assert sorted(row["idx"] for row in rows) == list(range(20))
def test_split_random_counts(mem_db):
"""Test random splitting with absolute counts."""
tbl = mem_db.create_table(
python/python/tests/test_query.py
+32
View File
@@ -675,6 +675,21 @@ def test_distance_range(table: lancedb.table.Table):
assert res["_distance"].to_pylist() == [min_dist, max_dist]
@pytest.mark.parametrize("expression", ["1 - _distance", "1.0 - _distance"])
def test_select_arithmetic_with_distance(table, expression):
result = (
table.search([10, 10])
.select({"similarity": expression, "_distance": "_distance"})
.distance_type("cosine")
.to_arrow()
)
assert result.schema.field("similarity").type == pa.float32()
assert result["similarity"].to_pylist() == pytest.approx(
[1 - distance for distance in result["_distance"].to_pylist()]
)
@pytest.mark.asyncio
async def test_distance_range_async(table_async: AsyncTable):
q = [0, 0]
@@ -897,6 +912,23 @@ def test_query_builder_batches(table):
assert rs_list["id"][1] == 2
def test_batch_vector_query_shares_filtered_flat_scan(table):
query = (
table.search([[1.0, 2.0], [3.0, 4.0]])
.where("id > 0", prefilter=True)
.limit(1)
.select(["id"])
)
plan = query.explain_plan(verbose=True)
assert "KNNVectorDistance: queries=2" in plan
assert "UnionExec" not in plan
results = query.to_arrow()
assert len(results) == 2
assert results["query_index"].to_pylist() == [0, 1]
def test_dynamic_projection(table):
rs = (
LanceVectorQueryBuilder(table, [0, 0], "vector")
python/python/tests/test_remote_db.py
+118 -1
View File
@@ -875,11 +875,85 @@ def test_remote_create_index_async_returns_job():
table = db.create_table("test", [{"id": 1}])
job = table.create_index_async("id", config=BTree())
assert job.id == "job-1"
job.wait(timeout=timedelta(seconds=30))
assert job.wait(timeout=timedelta(seconds=30)) is None
assert len(describe_calls) == 2
job.cancel()
def test_remote_refresh_async_returns_typed_terminal_result():
terminal_result = {
"rows_assigned": 12,
"rows_failed": 0,
"rows_remaining": 0,
"source_version": 7,
"published_version": 8,
}
def handler(request):
content_len = int(request.headers.get("Content-Length", 0))
body = request.rfile.read(content_len) if content_len > 0 else b""
if request.path == "/v1/table/test/backfill_column":
assert json.loads(body)["column"] == "derived"
request.send_response(202)
request.send_header("Content-Type", "application/json")
request.end_headers()
request.wfile.write(b'{"job_id": "refresh-1"}')
elif request.path == "/v1/jobs/describe":
assert json.loads(body)["job_id"] == "refresh-1"
request.send_response(200)
request.send_header("Content-Type", "application/json")
request.end_headers()
request.wfile.write(
json.dumps(
{
"job_id": "refresh-1",
"job_type": "function_refresh",
"job_state": "DONE",
"result": terminal_result,
}
).encode()
)
elif request.path == "/v1/table/test/create/?mode=create":
request.send_response(200)
request.send_header("Content-Type", "application/json")
request.end_headers()
request.wfile.write(b"{}")
elif request.path == "/v1/table/test/describe/":
request.send_response(200)
request.send_header("Content-Type", "application/json")
request.end_headers()
request.wfile.write(
json.dumps(
{
"version": 1,
"schema": {
"fields": [
{
"name": "id",
"type": {"type": "int64"},
"nullable": False,
}
]
},
}
).encode()
)
else:
request.send_response(404)
request.end_headers()
with mock_lancedb_connection(handler) as db:
table = db.create_table("test", [{"id": 1}])
job = table.refresh_column_async("derived")
assert job.id == "refresh-1"
result = job.wait(timeout=timedelta(seconds=30))
assert isinstance(result, lancedb.RefreshColumnResult)
assert result.model_dump() == terminal_result
assert result.rows_filled == 12
assert result.version == 8
def test_remote_job_wait_raises_on_failure():
from lancedb.exceptions import JobFailedError
from lancedb.index import BTree
@@ -1544,6 +1618,49 @@ def test_query_sync_fts():
)
def test_query_sync_fts_document_granularity():
from lancedb.query import DocumentGranularity, MatchQuery
def handler(body):
assert body == {
"full_text_query": {
"query": {
"match": {
"column": "docs.content",
"terms": "alpha",
"boost": 1.0,
"fuzziness": 0,
"max_expansions": 50,
"operator": "Or",
"prefix_length": 0,
"document_granularity": "list_element",
}
}
},
"k": 10,
"prefilter": True,
"vector": [],
"version": None,
}
return pa.table(
{
"id": [1, 1],
"_doc_index": pa.array([[0], [4]], type=pa.list_(pa.uint32())),
}
)
with query_test_table(handler, server_version=Version("0.6.0")) as table:
result = table.search(
MatchQuery(
"alpha",
"docs.content",
document_granularity=DocumentGranularity.LIST_ELEMENT,
)
).to_arrow()
assert result["_doc_index"].to_pylist() == [[0], [4]]
def test_query_sync_hybrid():
def handler(body):
if "full_text_query" in body:
python/python/tests/test_s3.py
+20
View File
@@ -4,6 +4,7 @@
import asyncio
import copy
from concurrent.futures import ThreadPoolExecutor
from datetime import timedelta
import threading
@@ -86,6 +87,25 @@ def test_s3_lifecycle(s3_bucket: str):
asyncio.run(test())
@pytest.mark.s3_test
def test_concurrent_open_table(s3_bucket: str):
uri = f"s3://{s3_bucket}/test_concurrent_open_table"
db = lancedb.connect(uri, storage_options=copy.copy(CONFIG))
db.create_table("test", pa.table({"x": [1, 2, 3]}))
num_workers = 32
barrier = threading.Barrier(num_workers)
def open_and_count(_):
barrier.wait()
return db.open_table("test").count_rows()
with ThreadPoolExecutor(max_workers=num_workers) as pool:
row_counts = list(pool.map(open_and_count, range(num_workers)))
assert row_counts == [3] * num_workers
@pytest.fixture()
def kms_key():
kms = get_boto3_client("kms", endpoint_url=CONFIG["aws_endpoint"])
python/python/tests/test_table.py
+18 -3
View File
@@ -1467,7 +1467,7 @@ def test_create_index_async_returns_done_job(mem_db: DBConnection):
table = mem_db.create_table("job_test", [{"id": i} for i in range(10)])
job = table.create_index_async("id", config=BTree())
assert job.id is None
job.wait()
assert job.wait() is None
assert len(table.list_indices()) == 1
job.cancel()
@@ -4041,10 +4041,21 @@ def test_refresh_column_async_returns_job(tmp_path):
job = table.refresh_column_async("doubled")
assert job.id is None # in-process jobs have no server id
assert job.wait() is None
result = job.wait()
assert isinstance(result, lancedb.RefreshColumnResult)
assert result.rows_assigned == 2
assert result.rows_failed == 0
assert result.rows_remaining == 0
assert result.source_version == 2
assert result.published_version == 3
assert job.status() == "finished"
assert sorted(table.to_arrow()["doubled"].to_pylist()) == [2, 4]
no_op = table.refresh_column_async("doubled").wait()
assert no_op.rows_assigned == 0
assert no_op.source_version == 3
assert no_op.published_version is None
# Bad input raises at the call, not through the job.
with pytest.raises(Exception, match="not a computed column"):
table.refresh_column_async("x")
@@ -4057,6 +4068,10 @@ async def test_refresh_column_async_job_async_table(tmp_path):
await table.add_columns(computed={"tripled": "x * 3"})
job = await table.refresh_column_async("tripled")
assert await job.wait() is None
result = await job.wait()
assert isinstance(result, lancedb.RefreshColumnResult)
assert result.rows_assigned == 1
assert result.source_version == 2
assert result.published_version == 3
assert await job.status() == "finished"
assert (await table.to_arrow())["tripled"].to_pylist() == [9]
python/src/connection.rs
+1 -1
View File
@@ -609,7 +609,7 @@ impl Connection {
.create_function_async(request)
.await
.infer_error()
.map(crate::job::FunctionJob::new)
.map(crate::job::Job::new_typed)
})
}
python/src/index.rs
+8 -2
View File
@@ -8,7 +8,7 @@ use lancedb::index::vector::{
};
use lancedb::index::{
Index as LanceDbIndex,
scalar::{BTreeIndexBuilder, FmIndexBuilder, FtsIndexBuilder},
scalar::{BTreeIndexBuilder, DocumentGranularity, FmIndexBuilder, FtsIndexBuilder},
};
use pyo3::IntoPyObject;
use pyo3::types::PyStringMethods;
@@ -60,7 +60,11 @@ pub fn extract_index_params(source: &Option<Bound<'_, PyAny>>) -> PyResult<Lance
.ngram_min_length(params.ngram_min_length)
.ngram_max_length(params.ngram_max_length)
.ngram_prefix_only(params.prefix_only)
.custom_stop_words(params.custom_stop_words);
.custom_stop_words(params.custom_stop_words)
.document_granularity(
DocumentGranularity::try_from(params.document_granularity.as_str())
.map_err(|err| PyValueError::new_err(err.to_string()))?,
);
if let Some(memory_limit) = params.memory_limit {
inner_opts = inner_opts.memory_limit_mb(memory_limit);
}
@@ -221,6 +225,7 @@ struct FtsParams {
block_size: usize,
memory_limit: Option<u64>,
num_workers: Option<usize>,
document_granularity: String,
}
#[derive(FromPyObject)]
@@ -481,6 +486,7 @@ mod tests {
block_size = 128
memory_limit = 2048
num_workers = 7
document_granularity = 'row'
config = FTS()",
None,
python/src/job.rs
+18 -55
View File
@@ -5,72 +5,33 @@ use std::sync::Arc;
use crate::runtime::future_into_py;
use pyo3::{Bound, PyAny, PyRef, PyResult, pyclass, pymethods};
use serde::Serialize;
use crate::error::PythonErrorExt;
#[pyclass]
pub struct Job {
inner: Arc<lancedb::Job>,
}
/// Python bridge for a typed remote Function registration job.
///
/// The public Python layer decodes the canonical JSON returned by `wait`
/// into its immutable `FunctionVersion` model.
#[pyclass]
pub struct FunctionJob {
inner: Arc<lancedb::Job<lancedb::function::FunctionVersion>>,
}
impl FunctionJob {
pub(crate) fn new(inner: lancedb::Job<lancedb::function::FunctionVersion>) -> Self {
Self {
inner: Arc::new(inner),
}
}
inner: Arc<lancedb::Job<std::result::Result<Option<String>, String>>>,
}
impl Job {
pub(crate) fn new(inner: lancedb::Job) -> Self {
Self {
inner: Arc::new(inner),
inner: Arc::new(inner.map(|()| Ok(None))),
}
}
}
#[pymethods]
impl FunctionJob {
#[getter]
pub fn id(&self) -> Option<String> {
self.inner.id().map(str::to_string)
}
pub fn status(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner.clone();
future_into_py(
self_.py(),
async move { inner.status().await.infer_error() },
)
}
pub fn wait(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner.clone();
future_into_py(self_.py(), async move {
inner
.wait()
.await
.infer_error()?
.to_canonical_json()
.infer_error()
})
}
pub fn cancel(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner.clone();
future_into_py(self_.py(), async move {
inner.cancel().await.infer_error()?;
Ok(())
})
pub(crate) fn new_typed<T>(inner: lancedb::Job<T>) -> Self
where
T: Clone + Serialize + Send + Sync + 'static,
{
Self {
inner: Arc::new(inner.map(|result| {
serde_json::to_string(&result)
.map(Some)
.map_err(|error| format!("failed to serialize typed job result: {error}"))
})),
}
}
}
@@ -92,8 +53,10 @@ impl Job {
pub fn wait(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner.clone();
future_into_py(self_.py(), async move {
inner.wait().await.infer_error()?;
Ok(None::<()>)
let result = inner.wait().await.infer_error()?;
result
.map_err(|message| lancedb::Error::Runtime { message })
.infer_error()
})
}
python/src/lib.rs
-1
View File
@@ -47,7 +47,6 @@ pub fn _lancedb(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<Session>()?;
m.add_class::<Table>()?;
m.add_class::<crate::job::Job>()?;
m.add_class::<crate::job::FunctionJob>()?;
m.add_class::<crate::job::JobInfo>()?;
m.add_class::<crate::job::JobDescription>()?;
m.add_class::<crate::job::JobFailureInfo>()?;
python/src/query.rs
+45 -12
View File
@@ -16,8 +16,8 @@ use arrow::pyarrow::FromPyArrow;
use arrow::pyarrow::IntoPyArrow;
use arrow::pyarrow::ToPyArrow;
use lancedb::index::scalar::{
BooleanQuery, BoostQuery, FtsQuery, FullTextSearchQuery, MatchQuery, MultiMatchQuery, Occur,
Operator, PhraseQuery,
BooleanQuery, BoostQuery, DocumentGranularity, FtsQuery, FullTextSearchQuery, MatchQuery,
MultiMatchQuery, Occur, Operator, PhraseQuery,
};
use lancedb::query::AnalyzePlanDistributedMetrics;
use lancedb::query::QueryBase;
@@ -76,8 +76,16 @@ impl<'a, 'py> FromPyObject<'a, 'py> for PyLanceDB<FtsQuery> {
let max_expansions = ob.getattr("max_expansions")?.extract()?;
let operator = ob.getattr("operator")?.extract::<String>()?;
let prefix_length = ob.getattr("prefix_length")?.extract()?;
let document_granularity = ob
.getattr("document_granularity")?
.extract::<Option<String>>()?
.map(|value| {
DocumentGranularity::try_from(value.as_str())
.map_err(|err| PyValueError::new_err(err.to_string()))
})
.transpose()?;
Ok(Self(
let mut query =
MatchQuery::new(query)
.with_column(Some(column))
.with_boost(boost)
@@ -86,21 +94,32 @@ impl<'a, 'py> FromPyObject<'a, 'py> for PyLanceDB<FtsQuery> {
.with_operator(Operator::try_from(operator.as_str()).map_err(|e| {
PyValueError::new_err(format!("Invalid operator: {}", e))
})?)
.with_prefix_length(prefix_length)
.into(),
))
.with_prefix_length(prefix_length);
if let Some(document_granularity) = document_granularity {
query = query.with_document_granularity(document_granularity);
}
Ok(Self(query.into()))
}
"PhraseQuery" => {
let query = ob.getattr("query")?.extract()?;
let column = ob.getattr("column")?.extract()?;
let slop = ob.getattr("slop")?.extract()?;
let document_granularity = ob
.getattr("document_granularity")?
.extract::<Option<String>>()?
.map(|value| {
DocumentGranularity::try_from(value.as_str())
.map_err(|err| PyValueError::new_err(err.to_string()))
})
.transpose()?;
Ok(Self(
PhraseQuery::new(query)
.with_column(Some(column))
.with_slop(slop)
.into(),
))
let mut query = PhraseQuery::new(query)
.with_column(Some(column))
.with_slop(slop);
if let Some(document_granularity) = document_granularity {
query = query.with_document_granularity(document_granularity);
}
Ok(Self(query.into()))
}
"BoostQuery" => {
let positive: Self = ob.getattr("positive")?.extract()?;
@@ -167,6 +186,13 @@ impl<'py> IntoPyObject<'py> for PyLanceDB<FtsQuery> {
kwargs.set_item("max_expansions", query.max_expansions)?;
kwargs.set_item::<_, &str>("operator", query.operator.into())?;
kwargs.set_item("prefix_length", query.prefix_length)?;
if let Some(document_granularity) = query.document_granularity {
let value = match document_granularity {
DocumentGranularity::Row => "row",
DocumentGranularity::ListElement => "list_element",
};
kwargs.set_item("document_granularity", value)?;
}
namespace
.getattr(intern!(py, "MatchQuery"))?
.call((query.terms, query.column.unwrap()), Some(&kwargs))
@@ -174,6 +200,13 @@ impl<'py> IntoPyObject<'py> for PyLanceDB<FtsQuery> {
FtsQuery::Phrase(query) => {
let kwargs = PyDict::new(py);
kwargs.set_item("slop", query.slop)?;
if let Some(document_granularity) = query.document_granularity {
let value = match document_granularity {
DocumentGranularity::Row => "row",
DocumentGranularity::ListElement => "list_element",
};
kwargs.set_item("document_granularity", value)?;
}
namespace
.getattr(intern!(py, "PhraseQuery"))?
.call((query.terms, query.column.unwrap()), Some(&kwargs))
python/src/table.rs
+7 -3
View File
@@ -903,15 +903,19 @@ impl Table {
})
}
#[pyo3(signature = (data, mode, progress=None, write_parallelism=None))]
#[pyo3(signature = (data, mode, progress=None, write_parallelism=None, allow_external_blob_outside_bases=false))]
pub fn add<'a>(
self_: PyRef<'a, Self>,
data: PyScannable,
mode: String,
progress: Option<Py<PyAny>>,
write_parallelism: Option<usize>,
allow_external_blob_outside_bases: bool,
) -> PyResult<Bound<'a, PyAny>> {
let mut op = self_.inner_ref()?.add(data);
let mut op = self_
.inner_ref()?
.add(data)
.allow_external_blob_outside_bases(allow_external_blob_outside_bases);
if mode == "append" {
op = op.mode(AddDataMode::Append);
} else if mode == "overwrite" {
@@ -1744,7 +1748,7 @@ impl Table {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let job = inner.refresh_column_async(column).await.infer_error()?;
Ok(crate::job::Job::new(job))
Ok(crate::job::Job::new_typed(job))
})
}