# 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``. Local creation and refresh require the ``experimental-functions`` Cargo feature in a source build. Published packages disable it. Opening, reading, listing and dropping existing views and remote Cloud/Enterprise operations remain available. """ from __future__ import annotations from typing import TYPE_CHECKING, Dict, List, Optional, Sequence, Tuple, Union import json from dataclasses import dataclass from .background_loop import LOOP from .job import AsyncJob, Job, _typed_job 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, ] MaterializedViewSource = str DEFINITION_FORMAT = 2 """The newest stored layout this version reads: ``{"format": N, "query": ""}``, format 2 being a query with ``GROUP BY``. A ``kind`` key beside it is for readers older than the format number.""" @dataclass class MaterializedViewDefinition: """The query that defines a materialized view, as stored:: SELECT columns FROM [ns.]table [, function(args) AS alias | , UNNEST(column) AS alias] [WHERE predicate] [GROUP BY expr, ...] [LIMIT n] A Function in ``FROM`` position yields one row per element it returns. """ query: str """The defining query, in the canonical spelling the server stores.""" 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") return _definition_from_value(json.loads(raw), name) def _definition_from_json(raw: str, name: str = "") -> MaterializedViewDefinition: """Parse the definition native code hands over, in its stored layout.""" return _definition_from_value(json.loads(raw), name) def _definition_from_value(value: dict, name: str) -> MaterializedViewDefinition: fmt = value.get("format") if fmt is not None: # A newer writer's layout is reported, never guessed at. if not isinstance(fmt, int) or fmt > DEFINITION_FORMAT: raise NotImplementedError( f"materialized view '{name}' is stored in format {fmt}, which " "this version of lancedb cannot refresh" ) return MaterializedViewDefinition(query=value["query"]) # The structured layout written before the format number. kind = value.get("kind") if kind not in ("select", "namespaced_select"): raise NotImplementedError( f"materialized view '{name}' is stored in format kind '{kind}', " "which this version of lancedb cannot refresh" ) return MaterializedViewDefinition(query=_legacy_query(value)) def _legacy_ident(name: str) -> str: if name and all(c == "_" or c.islower() or c.isdigit() for c in name): return name return _quote_identifier(name) def _legacy_query(value: dict) -> str: """Render the pre-format structured layout as the query it described.""" projections = value.get("projections", []) if projections: items = [] for p in projections: output, expression = p["output"], p["expression"] if expression in (output, _quote_identifier(output)): items.append(expression) else: items.append(f"{expression} AS {_legacy_ident(output)}") columns = ", ".join(items) else: columns = "*" table = ".".join( _legacy_ident(part) for part in [*value.get("source_namespace", []), value["source_table"]] ) query = f"SELECT {columns} FROM {table}" if value.get("filter") is not None: query += f" WHERE {value['filter']}" if value.get("limit") is not None: query += f" LIMIT {value['limit']}" return query 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.""" raw = await self._table._inner.materialized_view_definition() return _definition_from_json(raw) async def refresh( self, *, full: bool = False, source_version: Optional[int] = None ) -> "RefreshMaterializedViewResult": """Recompute the view from its source. ``full=True`` forces a remote rebuild; otherwise the server chooses the refresh mode. Local views always rebuild, and ``source_version`` selects a source version instead of the latest. Remote SQL refreshes do not support source-version pinning. Remote results leave ``mode``, ``rows_written``, and ``source_version`` as ``None`` because the SQL job adapter does not expose these metrics. """ return await self._table._inner.refresh_materialized_view( source_version=source_version, full=full ) async def refresh_async( self, *, full: bool = False, source_version: Optional[int] = None ) -> "AsyncJob[RefreshMaterializedViewResult]": """Submit a refresh and return its job without waiting. The job may already be complete for a local view. Remote refreshes are submitted through the SQL service. Options and result metrics follow [refresh][lancedb.materialized_view.AsyncMaterializedView.refresh]. """ from ._lancedb import RefreshMaterializedViewResult return _typed_job( await self._table._inner.refresh_materialized_view_async( source_version=source_version, full=full ), RefreshMaterializedViewResult.from_json, ) 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.""" return LOOP.run(self._async.definition()) 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)) def refresh_async( self, *, full: bool = False, source_version: Optional[int] = None ) -> "Job[RefreshMaterializedViewResult]": """Submit a refresh and return its job without waiting. See [AsyncMaterializedView.refresh_async][lancedb.materialized_view.AsyncMaterializedView.refresh_async]. """ return Job( LOOP.run( self._async.refresh_async(full=full, source_version=source_version) ) )