feat: define a materialized view by its query, with UNNEST for table-valued Functions

A view definition was a structured record under a `kind` tag, one kind
per query shape, and a Function returning `list<struct>` was about to
add a third. That names shapes instead of describing a relation.

A materialized view is now a relation defined by a query, stored as one
canonical SQL string under a format number:

    SELECT columns FROM [ns.]table [, UNNEST(column) AS alias]
      [WHERE predicate] [LIMIT n]

Any other clause is refused at parse time. Older readers report the view
as unrefreshable, the pre-format layouts still read, and a legacy view is
rewritten on its next refresh that commits, rebuilt only where its raw
text meant something else under lance's parser.

UNNEST is the first operator past project and filter: projections plan
against the source with the list column replaced by its element under
the alias, and refresh unnests, filters and projects each batch itself.
Row ids repeat per element, so eviction and incremental append are
unchanged. A limit is refused with UNNEST: a scan limit counts source rows.

python/python/lancedb/materialized_view.py

@@ -7,7 +7,7 @@ maintained by refresh. See ``DBConnection.create_materialized_view``."""
 from __future__ import annotations
 
 import json
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Dict, List, Optional, Sequence, Tuple, Union
 
 from .background_loop import LOOP
@@ -29,22 +29,21 @@ SelectArg = Union[
 ]
 
 
+DEFINITION_FORMAT = 1
+"""The stored layout this version reads: ``{"format": 1, "query": "<SQL>"}``.
+A ``kind`` key beside it is for readers older than the format number."""
+
+
 @dataclass
 class MaterializedViewDefinition:
-    """The query that defines a materialized view."""
+    """The query that defines a materialized view, as stored::
 
-    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."""
-    source_namespace: List[str] = field(default_factory=list)
-    """Namespace holding the source table; empty is the root namespace."""
+    SELECT columns FROM [ns.]table [, UNNEST(column) AS alias]
+      [WHERE predicate] [LIMIT n]
+    """
+
+    query: str
+    """The defining query, in the canonical spelling the server stores."""
 
 
 def _definition_from_schema(
@@ -54,38 +53,62 @@ def _definition_from_schema(
     raw = metadata.get(DEFINITION_META_KEY)
     if raw is None:
         raise ValueError(f"Table '{name}' is not a materialized view")
+    return _definition_from_json(raw, name)
+
+
+def _definition_from_json(raw: str, name: str) -> MaterializedViewDefinition:
+    """Parse the stored layout, which native code also returns for a view on
+    any connection."""
     value = json.loads(raw)
+    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")
-    # "namespaced_select" keeps older readers from resolving the source at root.
     if kind not in ("select", "namespaced_select"):
         raise NotImplementedError(
-            f"materialized view '{name}' is defined by '{kind}', which this "
-            "version of lancedb cannot refresh"
+            f"materialized view '{name}' is stored in format kind '{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", []),
-        source_namespace=value.get("source_namespace", []),
-    )
+    return MaterializedViewDefinition(query=_legacy_query(value))
 
 
-def _definition_from_json(raw: str) -> MaterializedViewDefinition:
-    value = json.loads(raw)
-    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", []),
-        source_namespace=value.get("source_namespace", []),
+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:
@@ -143,7 +166,7 @@ class AsyncMaterializedView:
     async def definition(self) -> MaterializedViewDefinition:
         """The query that defines the view."""
         raw = await self._table._inner.materialized_view_definition()
-        return _definition_from_json(raw)
+        return _definition_from_json(raw, self.name)
 
     async def refresh(
         self, *, full: bool = False, source_version: Optional[int] = None

python/python/tests/test_materialized_views.py

@@ -217,10 +217,7 @@ def test_definition_round_trips(tmp_path):
 
     view = db.open_materialized_view("adults")
     assert view.definition == MaterializedViewDefinition(
-        source_table="people",
-        projections=[("name", "`name`"), ("age", "`age`")],
-        filter="age >= 18",
-        inputs=["age", "name"],
+        query="SELECT name, age FROM people WHERE age >= 18"
     )
 
 
@@ -305,7 +302,7 @@ async def test_async_create_refresh_and_open(tmp_path):
 
     reopened = await db.open_materialized_view("shouts")
     definition = await reopened.definition()
-    assert definition.projections == [("shout", "upper(name)")]
+    assert definition.query == "SELECT upper(name) AS shout FROM people"
     assert await db.list_materialized_views() == ["shouts"]
 
 
@@ -423,7 +420,7 @@ def test_namespace_connection_materialized_views(tmp_path):
     assert db.list_materialized_views() == ["adults"]
 
     reopened = db.open_materialized_view("adults")
-    assert reopened.definition.source_table == "people"
+    assert reopened.definition.query.startswith("SELECT name, age FROM ")
     with pytest.raises(ValueError, match="not a materialized view"):
         db.open_materialized_view("people")
 
@@ -458,7 +455,7 @@ async def test_async_namespace_connection_materialized_views(tmp_path):
     assert await db.list_materialized_views() == ["adults"]
 
     reopened = await db.open_materialized_view("adults")
-    assert (await reopened.definition()).source_table == "people"
+    assert (await reopened.definition()).query.startswith("SELECT name, age FROM ")
 
     # The view's table came through the namespace, not straight from the
     # inner connection: a bare inner table carries no namespace context, so
@@ -484,36 +481,49 @@ async def test_async_namespace_connection_materialized_views(tmp_path):
     assert await db.list_materialized_views() == []
 
 
-def test_namespaced_select_kind_is_read_and_unknown_kinds_are_refused():
+def test_stored_queries_and_legacy_layouts_are_read():
     import json
 
     import pyarrow as pa
 
     from lancedb.materialized_view import _definition_from_schema
 
-    def schema_with(definition: dict) -> pa.Schema:
-        return pa.schema([pa.field("id", pa.int32())]).with_metadata(
+    def read(definition: dict) -> MaterializedViewDefinition:
+        schema = pa.schema([pa.field("id", pa.int32())]).with_metadata(
             {b"mv.definition": json.dumps(definition).encode()}
         )
+        return _definition_from_schema(schema, "v")
 
-    # "namespaced_select" is the namespaced form of "select": same shape,
-    # a separate kind so readers that predate it refuse instead of
-    # resolving the source at the root.
-    definition = _definition_from_schema(
-        schema_with(
+    query = "SELECT id, c.chunk FROM ns.docs, UNNEST(chunks) AS c WHERE id > 1"
+    assert read({"format": 1, "query": query}).query == query
+
+    # The structured layout written before the format number reads as the
+    # query it described, under either of its kind tags.
+    assert (
+        read(
             {
                 "kind": "namespaced_select",
                 "source_table": "people",
                 "source_namespace": ["ns"],
-                "projections": [{"output": "name", "expression": "name"}],
+                "projections": [
+                    {"output": "name", "expression": "`name`"},
+                    {"output": "Shout", "expression": "upper(name)"},
+                ],
+                "filter": "age >= 18",
+                "limit": 10,
             }
-        ),
-        "v",
+        ).query
+        == "SELECT `name`, upper(name) AS `Shout` FROM ns.people "
+        "WHERE age >= 18 LIMIT 10"
+    )
+    assert read({"kind": "select", "source_table": "people"}).query == (
+        "SELECT * FROM people"
     )
-    assert definition.source_table == "people"
-    assert definition.source_namespace == ["ns"]
 
-    with pytest.raises(NotImplementedError, match="cannot refresh"):
-        _definition_from_schema(
-            schema_with({"kind": "select_v3", "source_table": "people"}), "v"
-        )
+    # A newer writer's layout is reported, never guessed at.
+    for newer in (
+        {"format": 2, "query": query},
+        {"kind": "select_v3", "source_table": "people"},
+    ):
+        with pytest.raises(NotImplementedError, match="cannot refresh"):
+            read(newer)

python/src/table.rs

@@ -1684,7 +1684,7 @@ impl Table {
             let view = lancedb::MaterializedView::from_table(inner)
                 .await
                 .infer_error()?;
-            serde_json::to_string(view.definition()).map_err(|err| {
+            lancedb::materialized_view::definition_to_metadata(view.definition()).map_err(|err| {
                 PyRuntimeError::new_err(format!(
                     "failed to serialize materialized-view definition: {err}"
                 ))