feat: record the source namespace in a materialized view definition (#4098)

A view definition recorded its source by bare name and refresh resolved
that name at the root, so declaring a view over a namespaced source was
refused outright -- materialized views were root-only for every caller.

The definition now carries `source_namespace`, and refresh opens the
source at that coordinate. `plan` takes the namespace too: refresh
re-plans the stored definition and persists the result when it migrates,
so defaulting it there would strand the view on its next rebuild.

The stored kind is the version boundary. Root definitions keep the
`select` form byte-for-byte, so everything written before this change
reads exactly as it always did. A namespaced source is stored as
`namespaced_select`: released readers drop unknown fields and resolve a
`select` source at the root, so keeping the old kind would let a
rolled-back worker refresh a view from a same-name root table -- the new
kind routes them to their existing unrecognized-kind refusal instead.
The Python and Node definition parsers learn the new kind alongside the
Rust core.

python/python/lancedb/materialized_view.py

@@ -42,6 +42,8 @@ class MaterializedViewDefinition:
     """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."""
 
 
 def _definition_from_schema(
@@ -53,7 +55,8 @@ def _definition_from_schema(
         raise ValueError(f"Table '{name}' is not a materialized view")
     value = json.loads(raw)
     kind = value.get("kind")
-    if kind != "select":
+    # "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"
@@ -66,6 +69,7 @@ def _definition_from_schema(
         filter=value.get("filter"),
         limit=value.get("limit"),
         inputs=value.get("inputs", []),
+        source_namespace=value.get("source_namespace", []),
     )
 
 

python/python/tests/test_materialized_views.py

@@ -266,3 +266,38 @@ async def test_async_namespace_connection_materialized_views(tmp_path):
             handle._route_pushdown_to_rust == through_namespace._route_pushdown_to_rust
         )
         assert handle._namespace_path == through_namespace._namespace_path
+
+
+def test_namespaced_select_kind_is_read_and_unknown_kinds_are_refused():
+    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(
+            {b"mv.definition": json.dumps(definition).encode()}
+        )
+
+    # "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(
+            {
+                "kind": "namespaced_select",
+                "source_table": "people",
+                "source_namespace": ["ns"],
+                "projections": [{"output": "name", "expression": "name"}],
+            }
+        ),
+        "v",
+    )
+    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"
+        )