rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-08-25 15:38:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge origin/main into gatekeeper/fix-2325-1

Browse Source 
# Conflicts:
#	python/python/lancedb/__init__.py
This commit is contained in:
Gatefixer
2026-08-22 07:07:52 +00:00
parent 638430fdb4 851fa16b47
commit 53b4c3b715
91 changed files with 8369 additions and 1285 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/Cargo.toml
+5 -6
View File
@@ -1,6 +1,6 @@
[package]
name = "lancedb-python"
version = "0.38.0-beta.2"
version = "0.38.0-beta.3"
publish = false
edition.workspace = true
description = "Python bindings for LanceDB"
@@ -26,7 +26,9 @@ lance-namespace-impls.workspace = true
lance-io.workspace = true
env_logger.workspace = true
log.workspace = true
pyo3 = { version = "0.28", features = ["extension-module", "abi3-py310", "chrono"] }
# Maturin enables extension-module mode for Python builds. Keeping it out of
# Cargo features lets Rust unit tests link against libpython.
pyo3 = { version = "0.28", features = ["abi3-py310", "chrono"] }
chrono.workspace = true
pyo3-async-runtimes = { version = "0.28", features = [
"attributes",
@@ -41,10 +43,7 @@ tokio.workspace = true
libc = "0.2"
[build-dependencies]
pyo3-build-config = { version = "0.28", features = [
"extension-module",
"abi3-py310",
] }
pyo3-build-config = { version = "0.28", features = ["abi3-py310"] }
[features]
default = ["remote", "lancedb/aws", "lancedb/gcs", "lancedb/azure", "lancedb/dynamodb", "lancedb/oss", "lancedb/huggingface", "lancedb/cos", "lancedb/goosefs", "lancedb/metrics-otel"]
python/README.md
+19
View File
@@ -38,6 +38,25 @@ Stable releases are created about every 2 weeks. For the latest features and bug
pip install --pre --extra-index-url https://pypi.fury.io/lancedb/ lancedb
```
### Threading in CPU-limited containers
LanceDB uses separate pools for compute work and storage I/O. On a container with
two visible CPUs, current releases intentionally use one compute worker by default;
no manual configuration is needed. If every query logs an I/O core reservation
warning on a two-CPU container, upgrade from LanceDB 0.21.1 or earlier.
The two commonly tuned environment variables control different resources:
- `LANCE_CPU_THREADS` overrides the number of compute workers. One worker is the
appropriate setting for a two-CPU container when an explicit override is needed.
- `LANCE_IO_THREADS` controls concurrent storage operations, not reserved CPU
cores. Its default can be greater than the number of CPUs because I/O workers
spend much of their time waiting for storage.
Keep the defaults unless measurements show that the workload benefits from an
override. See the [Lance threading model](https://lance.org/guide/performance/#threading-model)
for the current defaults and tuning guidance.
## Usage
### Basic Example
python/pyproject.toml
+1 -1
View File
@@ -103,7 +103,7 @@ python-source = "python"
module-name = "lancedb._lancedb"
[build-system]
requires = ["maturin>=1.4"]
requires = ["maturin>=1.9.4"]
build-backend = "maturin"
[tool.ruff.lint]
python/python/lancedb/__init__.py
+8
View File
@@ -32,6 +32,11 @@ from .functions import (
UdfDefinition as UdfDefinition,
udf as udf,
)
from .materialized_view import (
AsyncMaterializedView,
MaterializedView,
MaterializedViewDefinition,
)
from .table import AsyncTable, CompactionOptions, Table
from .types import BaseTokenizerType
from ._lancedb import Session
@@ -506,6 +511,9 @@ async def connect_async(
__all__ = [
"AsyncMaterializedView",
"MaterializedView",
"MaterializedViewDefinition",
"connect",
"connect_async",
"tokenize",
python/python/lancedb/_lancedb.pyi
+18
View File
@@ -197,6 +197,15 @@ class Connection(object):
cur_namespace_path: Optional[List[str]] = None,
new_namespace_path: Optional[List[str]] = None,
) -> None: ...
async def create_materialized_view(
self,
name: str,
source: str,
projections: Optional[List[Tuple[str, str]]] = None,
filter: Optional[str] = None,
limit: Optional[int] = None,
) -> Table: ...
async def list_materialized_views(self) -> List[str]: ...
async def drop_table(
self, name: str, namespace_path: Optional[List[str]] = None
) -> None: ...
@@ -355,6 +364,9 @@ class Table:
) -> AddColumnsResult: ...
async def refresh_column(self, column: str) -> RefreshColumnResult: ...
async def refresh_column_async(self, column: str) -> Job: ...
async def refresh_materialized_view(
self, full: bool = False, source_version: Optional[int] = None
) -> RefreshMaterializedViewResult: ...
async def add_columns_with_schema(self, schema: pa.Schema) -> AddColumnsResult: ...
async def alter_columns(
self, columns: list[dict[str, Any]]
@@ -705,6 +717,12 @@ class RefreshColumnResult:
rows_filled: int
version: int
class RefreshMaterializedViewResult:
mode: str
rows_written: int
source_version: int
version: int
class AlterColumnsResult:
version: int
python/python/lancedb/db.py
+166
View File
@@ -47,6 +47,12 @@ 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 .materialized_view import (
AsyncMaterializedView,
MaterializedView,
SelectArg,
normalize_select,
)
from .table import (
AsyncTable,
LanceTable,
@@ -510,6 +516,70 @@ class DBConnection(EnforceOverrides):
"""
raise NotImplementedError
def create_materialized_view(
self,
name: str,
source: str,
*,
select: SelectArg = None,
where: Optional[str] = None,
limit: Optional[int] = None,
) -> MaterializedView:
"""Define a materialized view named ``name`` over the table ``source``.
The view is created empty, with the query recorded in its schema
metadata; ``view.refresh()`` computes the rows. The view is a normal
table: it can be queried, indexed and searched, and it appears in
``table_names``. Local databases only.
The source table must have stable row ids (create it with the
``new_table_enable_stable_row_ids`` storage option): they keep the
view's provenance valid across source compactions, and cannot be
enabled after a table exists.
Parameters
----------
name: str
The name of the view.
source: str
The name of the source table, in this database.
select: list or dict, optional
The view's columns: column names, ``(alias, SQL expression)``
pairs, or a dict of the same. Omitting it selects every source
column, expanded against the source schema at creation time.
where: str, optional
SQL predicate; only matching source rows appear in the view.
limit: int, optional
Cap the view at this many rows, in materialization order.
Returns
-------
MaterializedView
"""
raise NotImplementedError(
"materialized views are not supported on this connection type"
)
def open_materialized_view(self, name: str) -> MaterializedView:
"""Open the materialized view named ``name``.
Raises ``ValueError`` if the table exists but is not a materialized
view.
"""
raise NotImplementedError(
"materialized views are not supported on this connection type"
)
def list_materialized_views(self) -> List[str]:
"""The names of the materialized views in this database.
Found by reading every table's schema, so this costs an open per
table.
"""
raise NotImplementedError(
"materialized views are not supported on this connection type"
)
def drop_table(self, name: str, namespace_path: Optional[List[str]] = None):
"""Drop a table from the database.
@@ -1136,6 +1206,58 @@ class LanceDBConnection(DBConnection):
tbl.checkout(version)
return tbl
@override
def create_materialized_view(
self,
name: str,
source: str,
*,
select: SelectArg = None,
where: Optional[str] = None,
limit: Optional[int] = None,
) -> MaterializedView:
"""Define a materialized view named ``name`` over the table ``source``.
See
[DBConnection.create_materialized_view][lancedb.DBConnection.create_materialized_view].
Examples
--------
>>> import lancedb
>>> db = lancedb.connect(
... "./.lancedb",
... storage_options={"new_table_enable_stable_row_ids": "true"},
... )
>>> data = [{"name": "ada", "age": 36}, {"name": "kid", "age": 7}]
>>> table = db.create_table("people", data)
>>> view = db.create_materialized_view(
... "adults",
... "people",
... select=["name", ("shout", "upper(name)")],
... where="age >= 18",
... )
>>> result = view.refresh()
>>> result.rows_written
1
"""
LOOP.run(
self._conn.create_materialized_view(
name, source, select=select, where=where, limit=limit
)
)
return MaterializedView(self.open_table(name))
@override
def open_materialized_view(self, name: str) -> MaterializedView:
"""Open the materialized view named ``name``."""
view = MaterializedView(self.open_table(name))
view.definition
return view
@override
def list_materialized_views(self) -> List[str]:
"""The names of the materialized views in this database."""
return LOOP.run(self._conn.list_materialized_views())
def clone_table(
self,
target_table_name: str,
@@ -1906,6 +2028,50 @@ class AsyncConnection(object):
await tbl.checkout(version)
return tbl
async def create_materialized_view(
self,
name: str,
source: str,
*,
select: SelectArg = None,
where: Optional[str] = None,
limit: Optional[int] = None,
) -> AsyncMaterializedView:
"""Define a materialized view named ``name`` over the table ``source``.
See
[DBConnection.create_materialized_view][lancedb.DBConnection.create_materialized_view].
"""
inner = await self._inner.create_materialized_view(
name,
source,
projections=normalize_select(select),
filter=where,
limit=limit,
)
return AsyncMaterializedView(AsyncTable(inner))
async def open_materialized_view(self, name: str) -> AsyncMaterializedView:
"""Open the materialized view named ``name``.
Raises ``ValueError`` if the table exists but is not a materialized
view.
"""
if self.uri.startswith("db://"):
raise NotImplementedError(
"materialized views are supported only on local databases"
)
view = AsyncMaterializedView(await self.open_table(name))
await view.definition()
return view
async def list_materialized_views(self) -> List[str]:
"""The names of the materialized views in this database.
Found by reading every table's schema, so this costs an open per
table.
"""
return await self._inner.list_materialized_views()
async def clone_table(
self,
target_table_name: str,
python/python/lancedb/expr.py
+3 -2
View File
@@ -85,8 +85,9 @@ class Expr:
# for dict keys / set membership.
__hash__ = None # type: ignore[assignment]
def __init__(self, inner: PyExpr) -> None:
def __init__(self, inner: PyExpr, *, column_path: str | None = None) -> None:
self._inner = inner
self._column_path = column_path
# ── comparisons ──────────────────────────────────────────────────────────
@@ -273,7 +274,7 @@ def col(name: str) -> Expr:
>>> col("age") > lit(18)
Expr((age > 18))
"""
return Expr(expr_col(name))
return Expr(expr_col(name), column_path=name)
def lit(value: Union[bool, int, float, str, bytes, date, datetime, Decimal]) -> Expr:
python/python/lancedb/functions.py
+67 -1
View File
@@ -20,6 +20,7 @@ import re
import sys
import textwrap
import types
import uuid
from collections.abc import Mapping
from datetime import date, datetime
from typing import (
@@ -265,6 +266,64 @@ class FunctionVersion(_RemoteValue):
required_secrets: tuple[str, ...] = ()
created_at: str
def __call__(self, **inputs: Any) -> FunctionApplication:
"""Bind this exact version to named table columns.
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
come from one logical Function evaluation. Map result fields to table
columns with
[FunctionApplication.rename][lancedb.functions.FunctionApplication.rename],
then pass the application to
[Table.add_columns][lancedb.table.Table.add_columns].
Examples
--------
>>> from lancedb import col
>>> application = function( # doctest: +SKIP
... title=col("title"),
... body=col("body"),
... ).rename(columns={
... "normalized_text": "search_text",
... "token_count": "search_token_count",
... })
>>> table.add_columns(application) # doctest: +SKIP
"""
from lancedb.expr import Expr
parameters = tuple(parameter.name for parameter in self.signature.inputs)
missing = [parameter for parameter in parameters if parameter not in inputs]
unknown = sorted(set(inputs) - set(parameters))
if missing or unknown:
details = []
if missing:
details.append(f"missing inputs: {missing!r}")
if unknown:
details.append(f"unknown inputs: {unknown!r}")
raise TypeError("invalid Function inputs (" + "; ".join(details) + ")")
bindings = []
for parameter in parameters:
value = inputs[parameter]
if not isinstance(value, Expr) or value._column_path is None:
raise TypeError(
f"Function input {parameter!r} must be a direct col(...) reference"
)
bindings.append(
ApplicationInput(
parameter=parameter,
kind="column",
value={"path": value._column_path},
)
)
return FunctionApplication(
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`.
@@ -304,7 +363,14 @@ class ApplicationInput(_OpenRemoteValue):
class FunctionApplication(_OpenRemoteValue):
"""Immutable pre-declaration application of an exact Function version."""
"""Immutable pre-declaration application of an exact Function version.
A named-struct output remains one grouped application through table
declaration and execution.
[FunctionApplication.rename][lancedb.functions.FunctionApplication.rename]
records the result-field to table-column mapping without splitting sibling
outputs into separate UDF calls.
"""
function: FunctionVersionRef
inputs: tuple[ApplicationInput, ...]
python/python/lancedb/index.py
+11
View File
@@ -163,6 +163,15 @@ class FTS:
The number of documents per compressed posting block. Supported values
are 128 and 256. A value of 256 uses the experimental FTS V3 format
and may introduce breaking changes.
memory_limit : int, optional
The total memory limit in MiB for the local FTS build stage. The limit
is divided evenly among indexing workers. This build-only setting is
not persisted with the index and does not apply to remote tables.
num_workers : int, optional
The number of workers for a local FTS build. By default Lance uses
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.
Notes
-----
@@ -185,6 +194,8 @@ class FTS:
prefix_only: bool = False
block_size: int = 128
custom_stop_words: Optional[List[str]] = None
memory_limit: Optional[int] = None
num_workers: Optional[int] = None
@dataclass
python/python/lancedb/materialized_view.py
+178
View File
@@ -0,0 +1,178 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
"""Materialized views: tables defined by a query over a source table and
maintained by refresh. See ``DBConnection.create_materialized_view``."""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Dict, List, Optional, Sequence, Tuple, Union
from .background_loop import LOOP
if TYPE_CHECKING:
import pyarrow as pa
from ._lancedb import RefreshMaterializedViewResult
from .table import AsyncTable, LanceTable
DEFINITION_META_KEY = b"mv.definition"
SelectArg = Union[
str,
Sequence[Union[str, Tuple[str, str]]],
Dict[str, str],
None,
]
@dataclass
class MaterializedViewDefinition:
"""The query that defines a materialized view."""
source_table: str
"""Name of the source table, in the same database as the view."""
projections: List[Tuple[str, str]]
"""``(output column, SQL expression)`` pairs, in view schema order."""
filter: Optional[str] = None
"""SQL predicate selecting the source rows the view holds."""
limit: Optional[int] = None
"""Cap on the number of rows the view holds."""
inputs: List[str] = field(default_factory=list)
"""Source columns the projections and filter read."""
def _definition_from_schema(
schema: "pa.Schema", name: str
) -> MaterializedViewDefinition:
metadata = schema.metadata or {}
raw = metadata.get(DEFINITION_META_KEY)
if raw is None:
raise ValueError(f"Table '{name}' is not a materialized view")
value = json.loads(raw)
kind = value.get("kind")
if kind != "select":
raise NotImplementedError(
f"materialized view '{name}' is defined by '{kind}', which this "
"version of lancedb cannot refresh"
)
return MaterializedViewDefinition(
source_table=value["source_table"],
projections=[
(p["output"], p["expression"]) for p in value.get("projections", [])
],
filter=value.get("filter"),
limit=value.get("limit"),
inputs=value.get("inputs", []),
)
def _quote_identifier(name: str) -> str:
"""Quote a column name as a Lance SQL identifier (backticks)."""
escaped = name.replace("`", "``")
return f"`{escaped}`"
def normalize_select(select: SelectArg) -> Optional[List[Tuple[str, str]]]:
"""``select`` items may be a column name, an ``(alias, expression)`` pair,
or a dict of the same. A bare name projects itself and is quoted, so any
valid column name works; dict and pair entries are kept verbatim because
their right side is an expression.
A lone string is one column, not a sequence of its characters."""
if select is None:
return None
if isinstance(select, str):
select = [select]
if isinstance(select, dict):
return list(select.items())
normalized = []
for item in select:
if isinstance(item, str):
normalized.append((item, _quote_identifier(item)))
else:
alias, expression = item
normalized.append((alias, expression))
return normalized
class AsyncMaterializedView:
"""A handle on a materialized view: its table plus its definition.
Obtained from ``AsyncConnection.create_materialized_view`` or
``AsyncConnection.open_materialized_view``.
"""
def __init__(self, table: "AsyncTable"):
self._table = table
def __repr__(self) -> str:
return f"AsyncMaterializedView(name={self.name!r})"
@property
def name(self) -> str:
return self._table.name
@property
def table(self) -> "AsyncTable":
"""The view, as the table it is. Queries, indexes and search all
apply; writes are not blocked, but a rebuild replaces them."""
return self._table
async def definition(self) -> MaterializedViewDefinition:
"""The query that defines the view, read from its stored schema."""
return _definition_from_schema(await self._table.schema(), self.name)
async def refresh(
self, *, full: bool = False, source_version: Optional[int] = None
) -> "RefreshMaterializedViewResult":
"""Recompute the view from its source.
The refresh is incremental when the source's changes can be
reconciled into the view -- rows added, changed or removed since the
last one -- and otherwise rebuilds. ``full=True`` forces a rebuild;
``source_version`` refreshes to that source version instead of the
latest.
Concurrent refreshes of one view do not duplicate its rows. Two that
plan the same source rows conflict on commit, and the loser raises
rather than writing them a second time.
"""
return await self._table._inner.refresh_materialized_view(
full=full, source_version=source_version
)
class MaterializedView:
"""Synchronous variant of
[AsyncMaterializedView][lancedb.materialized_view.AsyncMaterializedView]."""
def __init__(self, table: "LanceTable"):
self._table = table
self._async = AsyncMaterializedView(table._table)
def __repr__(self) -> str:
return f"MaterializedView(name={self.name!r})"
@property
def name(self) -> str:
return self._table.name
@property
def table(self) -> "LanceTable":
"""The view, as the table it is."""
return self._table
@property
def definition(self) -> MaterializedViewDefinition:
"""The query that defines the view, read from its stored schema."""
return _definition_from_schema(self._table.schema, self.name)
def refresh(
self, *, full: bool = False, source_version: Optional[int] = None
) -> "RefreshMaterializedViewResult":
"""Recompute the view from its source. See
[AsyncMaterializedView.refresh][lancedb.materialized_view.AsyncMaterializedView.refresh]."""
return LOOP.run(self._async.refresh(full=full, source_version=source_version))
python/python/lancedb/namespace.py
+68
View File
@@ -61,6 +61,11 @@ from lance_namespace import (
NamespaceExistsRequest,
TableExistsRequest,
)
from lancedb.materialized_view import (
AsyncMaterializedView,
MaterializedView,
SelectArg,
)
from lancedb.table import AsyncTable, LanceTable, Table
from lancedb.util import validate_table_name
from lancedb.common import DATA
@@ -619,6 +624,42 @@ class LanceNamespaceDBConnection(DBConnection):
tbl.checkout(version)
return tbl
@override
def create_materialized_view(
self,
name: str,
source: str,
*,
select: "SelectArg" = None,
where: Optional[str] = None,
limit: Optional[int] = None,
) -> "MaterializedView":
"""Define a materialized view over a table in the root namespace.
See
[DBConnection.create_materialized_view][lancedb.DBConnection.create_materialized_view].
"""
return MaterializedView(
self.open_table(
LOOP.run(
self._inner.create_materialized_view(
name, source, select=select, where=where, limit=limit
)
).name
)
)
@override
def open_materialized_view(self, name: str) -> "MaterializedView":
"""Open the materialized view named ``name``."""
view = MaterializedView(self.open_table(name))
view.definition
return view
@override
def list_materialized_views(self) -> List[str]:
"""The names of the materialized views in the root namespace."""
return LOOP.run(self._inner.list_materialized_views())
@override
def drop_table(self, name: str, namespace_path: Optional[List[str]] = None):
if namespace_path is None:
@@ -1141,6 +1182,33 @@ class AsyncLanceNamespaceDBConnection:
route_pushdown_to_rust=self._route_pushdown_to_rust,
)
async def create_materialized_view(
self,
name: str,
source: str,
*,
select: "SelectArg" = None,
where: Optional[str] = None,
limit: Optional[int] = None,
) -> "AsyncMaterializedView":
"""Define a materialized view over a table in the root namespace."""
view = await self._inner.create_materialized_view(
name, source, select=select, where=where, limit=limit
)
# Reopen through the namespace so the view's table carries the
# namespace client and pushdown configuration a bare inner table lacks.
return AsyncMaterializedView(await self.open_table(view.name))
async def open_materialized_view(self, name: str) -> "AsyncMaterializedView":
"""Open the materialized view named ``name``."""
view = AsyncMaterializedView(await self.open_table(name))
await view.definition()
return view
async def list_materialized_views(self) -> List[str]:
"""The names of the materialized views in the root namespace."""
return await self._inner.list_materialized_views()
async def drop_table(self, name: str, namespace_path: Optional[List[str]] = None):
"""Drop a table from the namespace."""
if namespace_path is None:
python/python/lancedb/permutation.py
+6 -12
View File
@@ -41,21 +41,15 @@ class PermutationBuilder:
The permutation is stored in memory and will be lost when the program exits.
"""
def __init__(self, table: LanceTable):
def __init__(self, table: Table):
"""
Creates a new permutation builder for the given table.
By default, the permutation builder will create a single split that contains all
rows in the same order as the base table.
Tables with an LSM write spec are rejected: unflushed rows have no row id.
"""
if not hasattr(table, "_inner"):
raise TypeError(
f"PermutationBuilder requires a local LanceTable, "
f"got {type(table).__name__}. "
"The permutation API is not supported on remote tables. "
"Remote tables connect to LanceDB Cloud or Enterprise and do not have "
"direct access to the underlying Lance dataset needed for permutations."
)
self._async = async_permutation_builder(table)
def split_random(
@@ -231,7 +225,7 @@ class PermutationBuilder:
return LOOP.run(do_execute())
def permutation_builder(table: LanceTable) -> PermutationBuilder:
def permutation_builder(table: Table) -> PermutationBuilder:
return PermutationBuilder(table)
@@ -248,7 +242,7 @@ class Permutations:
Attributes
----------
base_table: LanceTable
base_table: Table
The base table that the permutations are based on.
permutation_table: LanceTable
The permutation table that defines the splits.
@@ -282,7 +276,7 @@ class Permutations:
{'train': 0, 'test': 1}
"""
def __init__(self, base_table: LanceTable, permutation_table: LanceTable):
def __init__(self, base_table: Table, permutation_table: LanceTable):
self.base_table = base_table
self.permutation_table = permutation_table
python/python/lancedb/remote/db.py
+27
View File
@@ -25,6 +25,7 @@ from ..common import DATA
from ..db import DBConnection, LOOP
from ..functions import FunctionVersion, UdfDefinition
from ..job import AsyncJob, Job
from ..materialized_view import MaterializedView, SelectArg
if TYPE_CHECKING:
from .._lancedb import JobDescription, JobInfo
@@ -648,6 +649,32 @@ class RemoteDBConnection(DBConnection):
namespace_path=namespace_path,
)
@override
def create_materialized_view(
self,
name: str,
source: str,
*,
select: SelectArg = None,
where: Optional[str] = None,
limit: Optional[int] = None,
) -> MaterializedView:
raise NotImplementedError(
"materialized views are supported only on local databases"
)
@override
def open_materialized_view(self, name: str) -> MaterializedView:
raise NotImplementedError(
"materialized views are supported only on local databases"
)
@override
def list_materialized_views(self) -> List[str]:
raise NotImplementedError(
"materialized views are supported only on local databases"
)
@override
def drop_table(self, name: str, namespace_path: Optional[List[str]] = None):
"""Drop a table from the database.
python/python/tests/test_elastic_dataloader.py
+29
View File
@@ -37,6 +37,11 @@ from unittest.mock import patch
import lancedb
import pyarrow as pa
import pytest
from utils import (
MockPermutationServer,
assert_server_safe_row_id_requests,
mock_remote_table,
)
torch = pytest.importorskip("torch")
streaming = pytest.importorskip("lancedb.streaming")
@@ -2118,3 +2123,27 @@ def test_doc_example_checkpoint(lance_table):
assert sorted(consumed + remaining_original) == list(range(NUM_ROWS)), (
"Consumed + remaining must cover every row exactly once"
)
# ---------------------------------------------------------------------------
# Remote tables (LanceDB Cloud / Enterprise)
# ---------------------------------------------------------------------------
def test_streaming_dataset_over_remote_table():
"""StreamingDataset reads a remote table, with server-safe requests.
Builds a permutation over a remote table, then fetches batches from it by row id.
"""
server = MockPermutationServer()
with mock_remote_table(server) as table:
ds = StreamingDataset(table, num_splits=2, shuffle_seed=SHUFFLE_SEED)
ids = [row["id"] for row in ds]
assert sorted(ids) == list(range(server.num_rows)), (
"Every row of the remote table must be yielded exactly once"
)
assert len(server.scans) == 1, "the permutation is built with one row-id scan"
assert server.takes, "rows must be fetched with row-id takes"
assert_server_safe_row_id_requests(server)
python/python/tests/test_first_class_function_slice1.py
+78
View File
@@ -6,6 +6,7 @@ from pathlib import Path
import pytest
from lancedb import col
import lancedb.functions as functions
from lancedb.functions import (
FunctionApplication,
@@ -120,6 +121,83 @@ def test_function_version_identity_is_immutable_and_exact():
assert FunctionVersion(**changed) != version
def test_function_version_binds_named_columns_as_one_immutable_group():
version = FunctionVersion.from_json(
json.dumps(job_result("remote_function_job.json"))
)
application = version(text=col("documents.body"))
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():
version = FunctionVersion.from_json(
json.dumps(job_result("remote_function_job.json"))
)
with pytest.raises(TypeError, match=r"missing inputs: \['text'\]"):
version()
with pytest.raises(TypeError, match=r"unknown inputs: \['body'\]"):
version(text=col("text"), body=col("body"))
with pytest.raises(TypeError, match="direct col"):
version(text=col("text").lower())
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["signature"] = {
"inputs": [
{"name": "title", "arrow_type": "utf8", "nullable": True},
{"name": "body", "arrow_type": "utf8", "nullable": True},
],
"output": {
"kind": "named_struct",
"fields": [
{
"name": "normalized_text",
"arrow_type": "utf8",
"nullable": False,
},
{
"name": "token_count",
"arrow_type": "int64",
"nullable": False,
},
],
},
}
version = FunctionVersion(**value)
application = version(body=col("body"), title=col("title")).rename(
columns={
"normalized_text": "search_text",
"token_count": "search_token_count",
}
)
assert [value.parameter for value in application.inputs] == ["title", "body"]
assert [field.name for field in application.output.fields] == [
"normalized_text",
"token_count",
]
assert dict(application.columns) == {
"normalized_text": "search_text",
"token_count": "search_token_count",
}
def test_unknown_fields_and_discriminators_are_forward_decodable():
value = job_result("remote_function_job.json")
value["future_version_metadata"] = {"retention_class": "catalog"}
python/python/tests/test_first_class_function_slice2.py
+3 -3
View File
@@ -162,7 +162,7 @@ def _mock_remote_function_catalog():
body = json.loads(self.rfile.read(length) or b"{}")
state["requests"].append((self.path, body))
status = 200
if self.path == "/v1/function/create":
if self.path == "/v1/functions/create":
state["version"] = {
"name": body["name"],
"version": "fv_exact",
@@ -187,7 +187,7 @@ def _mock_remote_function_catalog():
"job_state": "DONE",
"result": state["version"],
}
elif self.path == "/v1/function/describe":
elif self.path == "/v1/functions/get":
assert body == {
"name": "normalize_score",
"version": "fv_exact",
@@ -249,6 +249,6 @@ def test_blocking_remote_registration_returns_function_version():
assert created.name == "normalize_score"
assert created.version == "fv_exact"
assert [path for path, _ in state["requests"]] == [
"/v1/function/create",
"/v1/functions/create",
"/v1/jobs/describe",
]
python/python/tests/test_fts.py
+8
View File
@@ -245,6 +245,14 @@ def test_create_inverted_index_rejects_invalid_block_size(table):
table.create_index("text", config=FTS(block_size=129))
def test_create_inverted_index_respects_build_memory_limit(table):
with pytest.raises(ValueError, match="exceeds worker memory limit"):
table.create_index(
"text",
config=FTS(memory_limit=0, num_workers=1),
)
def test_custom_stop_words_list(table):
table.create_index(
"text",
python/python/tests/test_materialized_views.py
+268
View File
@@ -0,0 +1,268 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
import lancedb
import pytest
from lancedb.materialized_view import MaterializedViewDefinition
STABLE_ROW_IDS = {"new_table_enable_stable_row_ids": "true"}
def make_db(tmp_path):
db = lancedb.connect(tmp_path, storage_options=STABLE_ROW_IDS)
db.create_table(
"people",
[
{"name": "ada", "age": 36},
{"name": "kid", "age": 7},
{"name": "grace", "age": 85},
],
)
return db
def test_create_refresh_and_query(tmp_path):
db = make_db(tmp_path)
view = db.create_materialized_view(
"adults",
"people",
select=["name", ("shout", "upper(name)")],
where="age >= 18",
)
assert view.name == "adults"
assert view.table.count_rows() == 0
result = view.refresh()
assert result.mode == "rebuild"
assert result.rows_written == 2
rows = view.table.search().to_list()
assert sorted(row["shout"] for row in rows) == ["ADA", "GRACE"]
def test_definition_round_trips(tmp_path):
db = make_db(tmp_path)
db.create_materialized_view("adults", "people", where="age >= 18")
view = db.open_materialized_view("adults")
assert view.definition == MaterializedViewDefinition(
source_table="people",
projections=[("name", "`name`"), ("age", "`age`")],
filter="age >= 18",
inputs=["age", "name"],
)
def test_incremental_refresh_after_append(tmp_path):
db = make_db(tmp_path)
view = db.create_materialized_view("copy", "people")
view.refresh()
db.open_table("people").add([{"name": "alan", "age": 41}])
result = view.refresh()
assert result.mode == "incremental"
assert result.rows_written == 1
assert view.table.count_rows() == 4
assert view.refresh().mode == "no_op"
def test_incremental_refresh_after_update(tmp_path):
db = make_db(tmp_path)
view = db.create_materialized_view("copy", "people")
view.refresh()
db.open_table("people").update(where="name = 'kid'", values={"age": 8})
result = view.refresh()
assert result.mode == "incremental"
assert result.rows_written == 1
rows = view.table.search().to_list()
assert sorted(row["age"] for row in rows) == [8, 36, 85]
def test_legacy_storage_source_update_rebuilds(tmp_path):
db = lancedb.connect(
tmp_path,
storage_options={**STABLE_ROW_IDS, "new_table_data_storage_version": "legacy"},
)
db.create_table("people", [{"name": "ada", "age": 36}, {"name": "kid", "age": 7}])
view = db.create_materialized_view("copy", "people")
view.refresh()
db.open_table("people").update(where="name = 'kid'", values={"age": 8})
result = view.refresh()
assert result.mode == "rebuild"
rows = view.table.search().to_list()
assert sorted(row["age"] for row in rows) == [8, 36]
def test_list_and_not_a_view(tmp_path):
db = make_db(tmp_path)
db.create_materialized_view("adults", "people", where="age >= 18")
assert db.list_materialized_views() == ["adults"]
with pytest.raises(ValueError, match="not a materialized view"):
db.open_materialized_view("people")
def test_invalid_expression_fails_at_create(tmp_path):
db = make_db(tmp_path)
with pytest.raises(Exception, match="missing"):
db.create_materialized_view("bad", "people", select=[("x", "missing + 1")])
assert "bad" not in db.list_tables().tables
@pytest.mark.asyncio
async def test_async_create_refresh_and_open(tmp_path):
db = await lancedb.connect_async(tmp_path, storage_options=STABLE_ROW_IDS)
await db.create_table("people", [{"name": "ada", "age": 36}])
view = await db.create_materialized_view(
"shouts", "people", select=[("shout", "upper(name)")]
)
result = await view.refresh()
assert result.mode == "rebuild"
assert result.rows_written == 1
reopened = await db.open_materialized_view("shouts")
definition = await reopened.definition()
assert definition.projections == [("shout", "upper(name)")]
assert await db.list_materialized_views() == ["shouts"]
@pytest.mark.asyncio
async def test_async_incremental(tmp_path):
db = await lancedb.connect_async(tmp_path, storage_options=STABLE_ROW_IDS)
await db.create_table("people", [{"name": "ada", "age": 36}])
view = await db.create_materialized_view("copy", "people")
await view.refresh()
table = await db.open_table("people")
await table.add([{"name": "alan", "age": 41}])
result = await view.refresh()
assert result.mode == "incremental"
assert result.rows_written == 1
def test_source_requires_stable_row_ids(tmp_path):
db = lancedb.connect(tmp_path)
db.create_table("plain", [{"x": 1}])
with pytest.raises(Exception, match="stable row ids"):
db.create_materialized_view("v", "plain")
def test_bare_select_names_are_quoted(tmp_path):
db = lancedb.connect(tmp_path, storage_options=STABLE_ROW_IDS)
db.create_table("odd_names", [{"order item": "widget", "select": 2}])
view = db.create_materialized_view(
"quoted", "odd_names", select=["order item", "select"]
)
result = view.refresh()
assert result.rows_written == 1
rows = view.table.search().to_list()
assert rows[0]["order item"] == "widget"
assert rows[0]["select"] == 2
@pytest.mark.asyncio
async def test_async_remote_is_refused_without_network():
db = await lancedb.connect_async(
"db://nowhere", api_key="sk_test", region="us-east-1"
)
with pytest.raises(NotImplementedError, match="local"):
await db.create_materialized_view("v", "src")
with pytest.raises(NotImplementedError, match="local"):
await db.open_materialized_view("v")
with pytest.raises(NotImplementedError, match="local"):
await db.list_materialized_views()
def test_scalar_select_is_one_column(tmp_path):
db = make_db(tmp_path)
view = db.create_materialized_view("just_name", "people", select="name")
view.refresh()
rows = view.table.search().to_list()
assert set(rows[0]) - {"__source_row_id"} == {"name"}
assert sorted(row["name"] for row in rows) == ["ada", "grace", "kid"]
@pytest.mark.asyncio
async def test_async_scalar_select_is_one_column(tmp_path):
db = await lancedb.connect_async(tmp_path, storage_options=STABLE_ROW_IDS)
await db.create_table("people", [{"name": "ada", "age": 36}])
view = await db.create_materialized_view("just_name", "people", select="name")
await view.refresh()
rows = await view.table.query().to_list()
assert set(rows[0]) - {"__source_row_id"} == {"name"}
def test_limit_above_i64_max_is_refused(tmp_path):
db = make_db(tmp_path)
with pytest.raises(ValueError, match="exceeds the maximum"):
db.create_materialized_view("too_big", "people", limit=2**63)
# The boundary is fine, and zero still means an empty view.
db.create_materialized_view("at_max", "people", limit=2**63 - 1)
empty = db.create_materialized_view("none", "people", limit=0)
empty.refresh()
assert empty.table.count_rows() == 0
def _namespace_db(tmp_path):
return lancedb.connect_namespace(
"dir",
{"root": str(tmp_path)},
storage_options=STABLE_ROW_IDS,
)
def test_namespace_connection_materialized_views(tmp_path):
db = _namespace_db(tmp_path)
db.create_table(
"people",
[{"name": "ada", "age": 36}, {"name": "kid", "age": 7}],
storage_options=STABLE_ROW_IDS,
)
view = db.create_materialized_view("adults", "people", where="age >= 18")
view.refresh()
assert view.table.count_rows() == 1
assert db.list_materialized_views() == ["adults"]
reopened = db.open_materialized_view("adults")
assert reopened.definition.source_table == "people"
with pytest.raises(ValueError, match="not a materialized view"):
db.open_materialized_view("people")
@pytest.mark.asyncio
async def test_async_namespace_connection_materialized_views(tmp_path):
db = lancedb.connect_namespace_async(
"dir",
{"root": str(tmp_path)},
storage_options=STABLE_ROW_IDS,
)
await db.create_table(
"people",
[{"name": "ada", "age": 36}, {"name": "kid", "age": 7}],
storage_options=STABLE_ROW_IDS,
)
view = await db.create_materialized_view("adults", "people", where="age >= 18")
await view.refresh()
assert await view.table.count_rows() == 1
assert await db.list_materialized_views() == ["adults"]
reopened = await db.open_materialized_view("adults")
assert (await reopened.definition()).source_table == "people"
# The view's table came through the namespace, not straight from the
# inner connection: a bare inner table carries no namespace context, so
# its pushdown routing differs from a table the namespace opened.
through_namespace = await db.open_table("adults")
for handle in (view.table, reopened.table):
assert (
handle._route_pushdown_to_rust == through_namespace._route_pushdown_to_rust
)
assert handle._namespace_path == through_namespace._namespace_path
python/python/tests/test_permutation.py
+59
View File
@@ -8,6 +8,11 @@ import pytest
from lancedb import DBConnection, Table, connect
from lancedb.background_loop import LOOP
from lancedb.permutation import Permutation, Permutations, permutation_builder
from utils import (
MockPermutationServer,
assert_server_safe_row_id_requests,
mock_remote_table,
)
def test_split_random_ratios(mem_db):
@@ -1214,3 +1219,57 @@ def test_remove_rowid_after_select(some_permutation: Permutation):
perm_without_rowid = perm_with_rowid.remove_columns(["_rowid"])
assert "_rowid" not in perm_without_rowid.column_names
assert perm_without_rowid.column_names == ["id"]
def test_permutation_is_stable_when_remote_scan_order_varies():
"""Splits are assigned by scan position, and every rank builds its own
permutation, so two ranks seeing different scan orders must still agree."""
server = MockPermutationServer(num_rows=16, vary_scan_order=True)
def split_of_each_row(permutation_tbl):
# Sequential splits are assigned by position, so a reversed scan would put
# the last rows in split 0. Compare the mapping rather than the table order,
# which the split-id sort does not pin down.
rows = permutation_tbl.search(None).to_arrow().to_pydict()
return dict(zip(rows["row_id"], rows["split_id"]))
with mock_remote_table(server) as table:
first = split_of_each_row(
permutation_builder(table).split_sequential(fixed=2).execute()
)
second = split_of_each_row(
permutation_builder(table).split_sequential(fixed=2).execute()
)
assert server.scan_calls == 2, "both builds must have scanned"
assert first == second
assert first[0] == 0 and first[server.num_rows - 1] == 1, first
def test_permutation_over_remote_table():
"""The permutation API accepts a remote table, addressing rows by `_rowid` just
as `take_row_ids` does. Also pins the request shapes sent to the server.
"""
server = MockPermutationServer()
with mock_remote_table(server) as table:
permutation_tbl = permutation_builder(table).split_sequential(fixed=2).execute()
assert permutation_tbl.count_rows() == server.num_rows
permutation = Permutation.from_tables(table, permutation_tbl, 0)
assert permutation.num_rows == server.num_rows // 2
# Compare against the permutation's own order; the split-id sort is not stable.
rows = permutation_tbl.search(None).to_arrow().to_pydict()
split0 = [
row_id
for row_id, split in zip(rows["row_id"], rows["split_id"])
if not split
]
# The mock table's `id` equals its `_rowid`.
assert permutation.take_offsets([2, 0]) == [
{"id": split0[2]},
{"id": split0[0]},
]
assert_server_safe_row_id_requests(server)
python/python/tests/test_table.py
+2 -2
View File
@@ -4041,7 +4041,7 @@ 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
job.wait()
assert job.wait() is None
assert job.status() == "finished"
assert sorted(table.to_arrow()["doubled"].to_pylist()) == [2, 4]
@@ -4057,6 +4057,6 @@ 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")
await job.wait()
assert await job.wait() is None
assert await job.status() == "finished"
assert (await table.to_arrow())["tripled"].to_pylist() == [9]
python/python/tests/utils.py
+206
View File
@@ -1,7 +1,17 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
import contextlib
import http.server
import json
import re
import threading
import lancedb
import pyarrow as pa
import pytest
ARROW_FILE_CONTENT_TYPE = "application/vnd.apache.arrow.file"
def exception_output(e_info: pytest.ExceptionInfo):
import traceback
@@ -9,3 +19,199 @@ def exception_output(e_info: pytest.ExceptionInfo):
# skip traceback part, since it's not worth checking in tests
lines = traceback.format_exception_only(e_info.type, e_info.value)
return "".join(lines).strip()
def parse_in_list(filter_sql: str) -> list[int]:
"""Pull the integers out of a `<col> IN (a, b, c)` predicate.
Scoped to the parenthesised list so a cast in the SQL adds no phantom values.
"""
match = re.search(r"\bIN\s*\(([^)]*)\)", filter_sql, re.IGNORECASE)
assert match is not None, f"expected an IN list, got: {filter_sql}"
return [int(m) for m in re.findall(r"-?\d+", match.group(1))]
def is_row_id_take(body) -> bool:
"""True when a query body fetches specific rows by row id."""
return "_rowid" in (body.get("filter") or "")
def arrow_file_bytes(table: pa.Table) -> bytes:
"""Serialize to the Arrow IPC *file* framing the /query/ route answers with."""
sink = pa.BufferOutputStream()
with pa.ipc.new_file(sink, table.schema) as writer:
writer.write_table(table)
return sink.getvalue().to_pybytes()
class MockPermutationServer:
"""A stand-in LanceDB server hosting one table whose ``id`` equals its ``_rowid``.
Records every ``/query/`` body so tests can assert on the request shapes sent to
the server, which is the part that has to stay compatible.
"""
def __init__(self, name="remote_data", num_rows=8, vary_scan_order=False):
self.name = name
self.num_rows = num_rows
self.query_bodies = []
# Stand in for a distributed scan that answers in no fixed order.
self.vary_scan_order = vary_scan_order
self.scan_calls = 0
def __call__(self, request):
path = request.path
if path == f"/v1/table/{self.name}/describe/":
return self._json(
request,
{
"version": 1,
"schema": {
"fields": [
{"name": "id", "type": {"type": "int64"}, "nullable": False}
]
},
},
)
if path == f"/v1/table/{self.name}/get_lsm_write_spec/":
self._read_body(request)
# Null spec: this table has no LSM write path.
return self._json(request, {"lsm_write_spec": None})
if path == f"/v1/table/{self.name}/count_rows/":
self._read_body(request)
return self._json(request, self.num_rows)
if path == f"/v1/table/{self.name}/query/":
return self._query(request, self._read_body(request))
# Drain first, so an unexpected route cannot desync a keep-alive connection.
self._read_body(request)
request.send_response(404)
request.end_headers()
@property
def scans(self):
"""Bodies of the permutation build scan: the row id column, nothing else."""
return [b for b in self.query_bodies if b.get("columns") == ["_rowid"]]
@property
def takes(self):
"""Bodies of the row-id takes the loader fetches batches with.
Keyed on `_rowid`, not "has a filter": the schema probe also has a predicate.
"""
return [b for b in self.query_bodies if is_row_id_take(b)]
@staticmethod
def _read_body(request):
content_len = int(request.headers.get("Content-Length") or 0)
return json.loads(request.rfile.read(content_len)) if content_len else {}
@staticmethod
def _json(request, payload):
request.send_response(200)
request.send_header("Content-Type", "application/json")
request.end_headers()
request.wfile.write(json.dumps(payload).encode())
@staticmethod
def _arrow(request, table):
body = arrow_file_bytes(table)
request.send_response(200)
request.send_header("Content-Type", ARROW_FILE_CONTENT_TYPE)
request.send_header("Content-Length", str(len(body)))
request.end_headers()
request.wfile.write(body)
def _query(self, request, body):
self.query_bodies.append(body)
if is_row_id_take(body):
# A row-id take. Answer ascending, so tests prove the client reorders.
row_ids = sorted(parse_in_list(body["filter"]))
return self._arrow(
request,
pa.table(
{
"id": pa.array(row_ids, pa.int64()),
"_rowid": pa.array(row_ids, pa.uint64()),
}
),
)
if body.get("columns") == ["_rowid"]:
# The permutation build scan: row ids and nothing else.
row_ids = list(range(self.num_rows))
if self.vary_scan_order and self.scan_calls % 2:
row_ids.reverse()
self.scan_calls += 1
return self._arrow(
request,
pa.table({"_rowid": pa.array(row_ids, pa.uint64())}),
)
# The schema probe: filtered to nothing, so it carries schema and no rows.
return self._arrow(request, pa.table({"id": pa.array([], pa.int64())}))
def _make_handler(serve):
class MockLanceDBHandler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
serve(self)
def do_POST(self):
serve(self)
def log_message(self, *args):
pass # keep pytest output readable
return MockLanceDBHandler
@contextlib.contextmanager
def mock_remote_table(server):
"""Run ``server`` on a local port and yield an open remote table against it.
Threading: the loader fans out fetch threads a single-threaded server would
serialize, hiding the prefetch overlap under test.
"""
with http.server.ThreadingHTTPServer(
("localhost", 0), _make_handler(server)
) as srv:
thread = threading.Thread(target=srv.serve_forever)
thread.start()
try:
db = lancedb.connect(
"db://dev",
api_key="fake",
host_override=f"http://localhost:{srv.server_address[1]}",
client_config={"timeout_config": {"connect_timeout": 5}},
)
yield db.open_table(server.name)
finally:
srv.shutdown()
thread.join()
def assert_server_safe_row_id_requests(server):
"""Assert the loader fetched rows by row id and bounded everything else.
`.get`, not `[...]`, so a dropped field reads as the assertion, not a KeyError.
"""
for body in server.takes:
# The fetch needs the row id back to restore the requested order.
assert body.get("with_row_id") is True, body
assert "_rowid" in body["filter"], body
# Only the one-off permutation scan may scan the whole table; the schema probe is
# built once per split per epoch. `k == 0` counts as unbounded: lance reads a zero
# limit as "no limit".
def is_unbounded(body):
if is_row_id_take(body):
return False
k = body.get("k")
return k is None or k == 0 or k > server.num_rows
unbounded = [b for b in server.query_bodies if is_unbounded(b)]
assert unbounded == server.scans, (
f"only the permutation scan may be unbounded, got {unbounded}"
)
python/src/connection.rs
+34
View File
@@ -333,6 +333,40 @@ impl Connection {
})
}
#[pyo3(signature = (name, source, projections=None, filter=None, limit=None))]
pub fn create_materialized_view(
self_: PyRef<'_, Self>,
name: String,
source: String,
projections: Option<Vec<(String, String)>>,
filter: Option<String>,
limit: Option<u64>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
future_into_py(self_.py(), async move {
let mut builder = inner.create_materialized_view(name, source);
if let Some(projections) = projections {
builder = builder.select(projections);
}
if let Some(filter) = filter {
builder = builder.only_if(filter);
}
if let Some(limit) = limit {
builder = builder.limit(limit);
}
let view = builder.execute().await.infer_error()?;
Ok(Table::new(view.table().clone()))
})
}
pub fn list_materialized_views(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
future_into_py(self_.py(), async move {
let views = inner.list_materialized_views().await.infer_error()?;
Ok(views.into_iter().map(|view| view.name).collect::<Vec<_>>())
})
}
#[pyo3(signature = (name, namespace_path=None))]
pub fn drop_table(
self_: PyRef<'_, Self>,
python/src/index.rs
+57 -1
View File
@@ -42,7 +42,7 @@ pub fn extract_index_params(source: &Option<Bound<'_, PyAny>>) -> PyResult<Lance
"Fm" => Ok(LanceDbIndex::Fm(FmIndexBuilder::default())),
"FTS" => {
let params = source.extract::<FtsParams>()?;
let inner_opts = FtsIndexBuilder::default()
let mut inner_opts = FtsIndexBuilder::default()
.base_tokenizer(params.base_tokenizer)
.language(&params.language)
.map_err(|_| {
@@ -61,6 +61,12 @@ pub fn extract_index_params(source: &Option<Bound<'_, PyAny>>) -> PyResult<Lance
.ngram_max_length(params.ngram_max_length)
.ngram_prefix_only(params.prefix_only)
.custom_stop_words(params.custom_stop_words);
if let Some(memory_limit) = params.memory_limit {
inner_opts = inner_opts.memory_limit_mb(memory_limit);
}
if let Some(num_workers) = params.num_workers {
inner_opts = inner_opts.num_workers(num_workers);
}
let inner_opts = inner_opts
.block_size(params.block_size)
.map_err(|err| PyValueError::new_err(err.to_string()))?;
@@ -213,6 +219,8 @@ struct FtsParams {
ngram_max_length: u32,
prefix_only: bool,
block_size: usize,
memory_limit: Option<u64>,
num_workers: Option<usize>,
}
#[derive(FromPyObject)]
@@ -444,3 +452,51 @@ impl IndexConfig {
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use pyo3::types::{PyDict, PyDictMethods};
use serde_json::json;
#[test]
fn fts_build_controls_are_forwarded() {
Python::initialize();
Python::attach(|py| {
let locals = PyDict::new(py);
py.run(
c"class FTS:
with_position = True
base_tokenizer = 'simple'
language = 'English'
max_token_length = None
lower_case = True
stem = False
remove_stop_words = False
custom_stop_words = None
ascii_folding = False
ngram_min_length = 3
ngram_max_length = 3
prefix_only = False
block_size = 128
memory_limit = 2048
num_workers = 7
config = FTS()",
None,
Some(&locals),
)
.unwrap();
let config = locals.get_item("config").unwrap().unwrap();
let index = extract_index_params(&Some(config)).unwrap();
let LanceDbIndex::FTS(params) = index else {
panic!("expected FTS index parameters");
};
let training_json = params.to_training_json().unwrap();
assert_eq!(training_json.get("memory_limit"), Some(&json!(2048)));
assert_eq!(training_json.get("num_workers"), Some(&json!(7)));
});
}
}
python/src/job.rs
+1 -1
View File
@@ -93,7 +93,7 @@ impl Job {
let inner = self_.inner.clone();
future_into_py(self_.py(), async move {
inner.wait().await.infer_error()?;
Ok(())
Ok(None::<()>)
})
}
python/src/lib.rs
+3 -2
View File
@@ -16,8 +16,8 @@ use query::{FTSQuery, HybridQuery, Query, VectorQuery};
use session::Session;
use table::{
AddColumnsResult, AddResult, AlterColumnsResult, DeleteResult, DropColumnsResult, FtsToken,
LsmWriteSpec, MergeResult, PyBlobFile, RefreshColumnResult, Table, UpdateFieldMetadataResult,
UpdateResult,
LsmWriteSpec, MergeResult, PyBlobFile, RefreshColumnResult, RefreshMaterializedViewResult,
Table, UpdateFieldMetadataResult, UpdateResult,
};
pub mod arrow;
@@ -60,6 +60,7 @@ pub fn _lancedb(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<RecordBatchStream>()?;
m.add_class::<AddColumnsResult>()?;
m.add_class::<RefreshColumnResult>()?;
m.add_class::<RefreshMaterializedViewResult>()?;
m.add_class::<AlterColumnsResult>()?;
m.add_class::<UpdateFieldMetadataResult>()?;
m.add_class::<AddResult>()?;
python/src/permutation.rs
+3 -1
View File
@@ -268,7 +268,9 @@ impl PyPermutationReader {
.await
.infer_error()?
} else {
PermutationReader::identity(base_table).await
PermutationReader::identity(base_table)
.await
.infer_error()?
};
Ok(Self::from_reader(reader))
})
python/src/table.rs
+55
View File
@@ -564,6 +564,41 @@ impl From<lancedb::table::RefreshColumnResult> for RefreshColumnResult {
}
}
#[pyclass(get_all, from_py_object)]
#[derive(Clone, Debug)]
pub struct RefreshMaterializedViewResult {
pub mode: String,
pub rows_written: u64,
pub source_version: u64,
pub version: u64,
}
#[pymethods]
impl RefreshMaterializedViewResult {
pub fn __repr__(&self) -> String {
format!(
"RefreshMaterializedViewResult(mode={}, rows_written={}, source_version={}, version={})",
self.mode, self.rows_written, self.source_version, self.version
)
}
}
impl From<lancedb::RefreshMaterializedViewResult> for RefreshMaterializedViewResult {
fn from(result: lancedb::RefreshMaterializedViewResult) -> Self {
let mode = match result.mode {
lancedb::RefreshMode::Rebuild => "rebuild",
lancedb::RefreshMode::Incremental => "incremental",
lancedb::RefreshMode::NoOp => "no_op",
};
Self {
mode: mode.to_string(),
rows_written: result.rows_written,
source_version: result.source_version,
version: result.version,
}
}
}
#[pymethods]
impl AddColumnsResult {
pub fn __repr__(&self) -> String {
@@ -1713,6 +1748,26 @@ impl Table {
})
}
#[pyo3(signature = (full=false, source_version=None))]
pub fn refresh_materialized_view(
self_: PyRef<'_, Self>,
full: bool,
source_version: Option<u64>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let view = lancedb::MaterializedView::from_table(inner)
.await
.infer_error()?;
let mut builder = view.refresh().full(full);
if let Some(version) = source_version {
builder = builder.source_version(version);
}
let result = builder.execute().await.infer_error()?;
Ok(RefreshMaterializedViewResult::from(result))
})
}
pub fn add_columns_with_schema(
self_: PyRef<'_, Self>,
schema: PyArrowType<Schema>,