rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-08-18 03:58:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(python): materialized view bindings

Browse Source 
Exposes materialized views to Python in both the async and sync clients:
create_materialized_view / open_materialized_view / list_materialized_views
on the connections, and MaterializedView / AsyncMaterializedView handles
carrying the parsed definition and refresh(full=, source_version=), which
returns the typed refresh result. select accepts column names, (alias,
expression) pairs, or a dict of the same; the definition reads back off the
stored schema, so a reopened handle needs no side channel. Remote
connections raise NotImplementedError up front rather than failing deep in
a request, matching the computed-column convention.
This commit is contained in:
Wyatt Alt
2026-08-17 10:42:35 -07:00
parent 60f5f6fef2
commit 9dd968466f
11 changed files with 819 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/src/python/python.md
+10
View File
@@ -52,6 +52,12 @@ listing a storage directory.
::: lancedb.table.Branches
## Materialized Views (Synchronous)
::: lancedb.materialized_view.MaterializedView
::: lancedb.materialized_view.MaterializedViewDefinition
## Expressions
Type-safe expression builder for filters and projections. Use these instead
@@ -244,6 +250,10 @@ Table hold your actual data as a collection of records / rows.
::: lancedb.table.AsyncBranches
## Materialized Views (Asynchronous)
::: lancedb.materialized_view.AsyncMaterializedView
## Indices (Asynchronous)
Indices can be created on a table to speed up queries. This section
python/python/lancedb/__init__.py
+8
View File
@@ -21,6 +21,11 @@ from .remote.db import RemoteDBConnection
from .expr import Expr, col, lit, func
from .schema import blob, vector, BlobType
from .job import AsyncJob, Job
from .materialized_view import (
AsyncMaterializedView,
MaterializedView,
MaterializedViewDefinition,
)
from .table import AsyncTable, Table
from .types import BaseTokenizerType
from ._lancedb import Session
@@ -495,6 +500,9 @@ async def connect_async(
__all__ = [
"AsyncMaterializedView",
"MaterializedView",
"MaterializedViewDefinition",
"connect",
"connect_async",
"tokenize",
python/python/lancedb/_lancedb.pyi
+18
View File
@@ -195,6 +195,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: ...
@@ -343,6 +352,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]]
@@ -692,6 +704,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
@@ -46,6 +46,12 @@ from lance_namespace.errors import NamespaceNotEmptyError, TableNotFoundError
from . import __version__
from ._lancedb import connect as lancedb_connect # type: ignore
from .job import AsyncJob, Job
from .materialized_view import (
AsyncMaterializedView,
MaterializedView,
SelectArg,
normalize_select,
)
from .table import (
AsyncTable,
LanceTable,
@@ -509,6 +515,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.
@@ -1110,6 +1180,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,
@@ -1871,6 +1993,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/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/remote/db.py
+27
View File
@@ -24,6 +24,7 @@ import pyarrow as pa
from ..common import DATA
from ..db import DBConnection, LOOP
from ..job import AsyncJob, Job
from ..materialized_view import MaterializedView, SelectArg
if TYPE_CHECKING:
from .._lancedb import JobDescription, JobInfo
@@ -647,6 +648,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_materialized_views.py
+252
View File
@@ -0,0 +1,252 @@
# 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_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/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/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;
@@ -59,6 +59,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/table.rs
+55
View File
@@ -441,6 +441,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 {
@@ -1570,6 +1605,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>,