diff --git a/docs/src/python/python.md b/docs/src/python/python.md index 3dd6f59f4..8176c6ca7 100644 --- a/docs/src/python/python.md +++ b/docs/src/python/python.md @@ -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 diff --git a/python/python/lancedb/__init__.py b/python/python/lancedb/__init__.py index 235049f97..2677b3e7e 100644 --- a/python/python/lancedb/__init__.py +++ b/python/python/lancedb/__init__.py @@ -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", diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 22878fd85..a8ef15fcf 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -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 diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index 14b6c0b0d..5924030dc 100644 --- a/python/python/lancedb/db.py +++ b/python/python/lancedb/db.py @@ -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, diff --git a/python/python/lancedb/materialized_view.py b/python/python/lancedb/materialized_view.py new file mode 100644 index 000000000..5abb44dc0 --- /dev/null +++ b/python/python/lancedb/materialized_view.py @@ -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)) diff --git a/python/python/lancedb/namespace.py b/python/python/lancedb/namespace.py index 0e60bd218..f2e553321 100644 --- a/python/python/lancedb/namespace.py +++ b/python/python/lancedb/namespace.py @@ -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: diff --git a/python/python/lancedb/remote/db.py b/python/python/lancedb/remote/db.py index 16ad65dcb..820a7a321 100644 --- a/python/python/lancedb/remote/db.py +++ b/python/python/lancedb/remote/db.py @@ -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. diff --git a/python/python/tests/test_materialized_views.py b/python/python/tests/test_materialized_views.py new file mode 100644 index 000000000..8cc48ba46 --- /dev/null +++ b/python/python/tests/test_materialized_views.py @@ -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 diff --git a/python/src/connection.rs b/python/src/connection.rs index dbda29ba6..df92bcf0c 100644 --- a/python/src/connection.rs +++ b/python/src/connection.rs @@ -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>, + filter: Option, + limit: Option, + ) -> PyResult> { + 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> { + 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::>()) + }) + } + #[pyo3(signature = (name, namespace_path=None))] pub fn drop_table( self_: PyRef<'_, Self>, diff --git a/python/src/lib.rs b/python/src/lib.rs index a19bf172d..8d3eab787 100644 --- a/python/src/lib.rs +++ b/python/src/lib.rs @@ -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::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/python/src/table.rs b/python/src/table.rs index 35ee92dc4..6c2009ccc 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -441,6 +441,41 @@ impl From 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 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, + ) -> PyResult> { + 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,