feat(remote): support materialized view APIs (#4180)

## Summary

Align the experimental materialized-view HTTP transport with the
equivalent Table API shape and add remote materialized-view support
across Rust, Python, and TypeScript. This is an intentional breaking
change to the experimental materialized-view surface.

Materialized-view creation performs an initial refresh by default. The
create endpoint returns `202 Accepted` with `{ "job_id": "..." }`;
blocking SDK creation waits for that job before returning a populated
view. `with_no_data` / `withNoData` explicitly creates only the
definition and empty backing table.

## Route comparison

| Operation | Materialized-view API | Equivalent Table API |
| --- | --- | --- |
| Create | `POST /v1/materialized_view/{id}/create` | `POST
/v1/table/{id}/create` |
| Describe/open | `POST /v1/materialized_view/{id}/describe` | `POST
/v1/table/{id}/describe` |
| List | `GET /v1/namespace/{id}/materialized_view/list` | `GET
/v1/namespace/{id}/table/list` |
| Refresh | `POST /v1/materialized_view/{id}/refresh` | asynchronous
Table mutation pattern |
| Drop | `POST /v1/materialized_view/{id}/drop` | `POST
/v1/table/{id}/drop` |

Create, describe, refresh, and drop identify the target in the singular
item path instead of duplicating it in the request body. Create and drop
require `202 Accepted` with a valid job ID. List is a namespace-scoped
GET with opaque pagination tokens. The Rust list API now returns view
names, matching Table listing and the existing Python and TypeScript
APIs.

## Python API changes

| Operation | Synchronous API | Asynchronous API | Table/job pattern |
| --- | --- | --- | --- |
| Create and wait | `DBConnection.create_materialized_view(...)` |
`await AsyncConnection.create_materialized_view(...)` | Returns a
materialized-view handle after its initial-population job finishes |
| Submit create | `DBConnection.create_materialized_view_async(...) ->
Job[None]` | `await AsyncConnection.create_materialized_view_async(...)
-> AsyncJob[None]` | Matches job-returning Table mutations such as
`create_index_async` |
| Open | `DBConnection.open_materialized_view(...)` | `await
AsyncConnection.open_materialized_view(...)` | Opens the backing Table
plus its definition |
| List | `DBConnection.list_materialized_views()` | `await
AsyncConnection.list_materialized_views()` | Returns names like Table
listing |
| Refresh and wait | `MaterializedView.refresh(...)` | `await
AsyncMaterializedView.refresh(...)` | Returns the typed refresh result
after the job finishes |
| Submit refresh | `MaterializedView.refresh_async(...) ->
Job[RefreshMaterializedViewResult]` | `await
AsyncMaterializedView.refresh_async(...) ->
AsyncJob[RefreshMaterializedViewResult]` | Matches
`Table.refresh_column_async`; remote job handles expose the server job
ID |
| Drop | `DBConnection.drop_materialized_view(...)` | `await
AsyncConnection.drop_materialized_view(...)` | Matches blocking
`drop_table` |
| Submit drop | `DBConnection.drop_materialized_view_async(...) ->
Job[None]` | `await AsyncConnection.drop_materialized_view_async(...) ->
AsyncJob[None]` | Matches `drop_table_async`; remote handles expose the
server cleanup job ID |

The materialized-view handle exposes its backing Table through `.table`,
so normal Table query, search, and index APIs apply. Definition lookup
and refresh are backend-aware rather than depending on local schema
metadata. TypeScript exposes the equivalent blocking/job drop pair as
`dropMaterializedView` and `dropMaterializedViewAsync`.

python/python/tests/test_materialized_views.py

@@ -1,9 +1,15 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright The LanceDB Authors
 
+import contextlib
+import http.server
+import json
+import threading
+
 import lancedb
 import pytest
 from lancedb.materialized_view import MaterializedViewDefinition
+from lancedb.remote.db import RemoteDBConnection
 
 
 STABLE_ROW_IDS = {"new_table_enable_stable_row_ids": "true"}
@@ -22,6 +28,152 @@ def make_db(tmp_path):
     return db
 
 
+@contextlib.contextmanager
+def mock_remote_materialized_views():
+    requests = []
+
+    class Handler(http.server.BaseHTTPRequestHandler):
+        def log_message(self, *args):
+            pass
+
+        def do_GET(self):
+            requests.append(self.path)
+            encoded = json.dumps({"views": ["daily_sales"]}).encode()
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(encoded)))
+            self.end_headers()
+            self.wfile.write(encoded)
+
+    with http.server.HTTPServer(("localhost", 0), Handler) as server:
+        thread = threading.Thread(target=server.serve_forever)
+        thread.start()
+        try:
+            yield f"http://localhost:{server.server_address[1]}", requests
+        finally:
+            server.shutdown()
+            thread.join()
+
+
+@contextlib.contextmanager
+def mock_remote_materialized_view_create():
+    requests = []
+
+    class Handler(http.server.BaseHTTPRequestHandler):
+        def log_message(self, *args):
+            pass
+
+        def do_POST(self):
+            length = int(self.headers.get("Content-Length", 0))
+            body = json.loads(self.rfile.read(length) or b"{}")
+            requests.append((self.path, body))
+            job_id = "mv-drop-123" if self.path.endswith("/drop") else "mv-create-123"
+            encoded = json.dumps({"job_id": job_id}).encode()
+            self.send_response(202)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(encoded)))
+            self.end_headers()
+            self.wfile.write(encoded)
+
+    with http.server.HTTPServer(("localhost", 0), Handler) as server:
+        thread = threading.Thread(target=server.serve_forever)
+        thread.start()
+        try:
+            yield f"http://localhost:{server.server_address[1]}", requests
+        finally:
+            server.shutdown()
+            thread.join()
+
+
+def test_remote_list_uses_namespace_route():
+    with mock_remote_materialized_views() as (host, requests):
+        db = lancedb.connect(
+            "db://dev",
+            api_key="fake",
+            host_override=host,
+            client_config={"retry_config": {"retries": 0}},
+        )
+        assert db.list_materialized_views() == ["daily_sales"]
+    assert requests == ["/v1/namespace/$/materialized_view/list"]
+
+
+def test_remote_create_async_returns_server_job():
+    with mock_remote_materialized_view_create() as (host, requests):
+        db = lancedb.connect(
+            "db://dev",
+            api_key="fake",
+            host_override=host,
+            client_config={"retry_config": {"retries": 0}},
+        )
+        job = db.create_materialized_view_async("adults", "people", where="age >= 18")
+        assert job.id == "mv-create-123"
+    assert requests == [
+        (
+            "/v1/materialized_view/adults/create",
+            {"query": 'SELECT * FROM "people" WHERE age >= 18', "with_no_data": False},
+        )
+    ]
+
+
+def test_remote_drop_async_returns_server_job():
+    with mock_remote_materialized_view_create() as (host, requests):
+        db = lancedb.connect(
+            "db://dev",
+            api_key="fake",
+            host_override=host,
+            client_config={"retry_config": {"retries": 0}},
+        )
+        job = db.drop_materialized_view_async("adults")
+        assert job.id == "mv-drop-123"
+    assert requests == [("/v1/materialized_view/adults/drop", {})]
+
+
+def test_sync_remote_create_uses_public_async_connection():
+    calls = []
+
+    class StubAsyncTable:
+        name = "adults"
+
+    class StubAsyncMaterializedView:
+        table = StubAsyncTable()
+
+    class StrictAsyncConnection:
+        async def create_materialized_view(
+            self,
+            name,
+            source,
+            *,
+            select=None,
+            where=None,
+            limit=None,
+            with_no_data=False,
+        ):
+            calls.append((name, source, select, where, limit, with_no_data))
+            return StubAsyncMaterializedView()
+
+        async def drop_materialized_view(self, name, *, namespace_path=None):
+            calls.append(("drop", name, namespace_path))
+
+    db = RemoteDBConnection.__new__(RemoteDBConnection)
+    db._conn = StrictAsyncConnection()
+    db.db_name = "example"
+    db.serialize = lambda: "{}"
+
+    view = db.create_materialized_view(
+        "adults",
+        "people",
+        select=["name"],
+        where="age >= 18",
+        limit=10,
+        with_no_data=True,
+    )
+    assert view.name == "adults"
+    assert calls == [("adults", "people", ["name"], "age >= 18", 10, True)]
+
+    db.drop_materialized_view("adults", namespace_path=["analytics"])
+    assert calls[-1] == ("drop", "adults", ["analytics"])
+
+
 def test_create_refresh_and_query(tmp_path):
     db = make_db(tmp_path)
     view = db.create_materialized_view(
@@ -31,16 +183,34 @@ def test_create_refresh_and_query(tmp_path):
         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
+    assert view.table.count_rows() == 2
 
     rows = view.table.search().to_list()
     assert sorted(row["shout"] for row in rows) == ["ADA", "GRACE"]
 
 
+def test_create_and_refresh_jobs(tmp_path):
+    db = make_db(tmp_path)
+    create_job = db.create_materialized_view_async(
+        "adults", "people", where="age >= 18", with_no_data=True
+    )
+    assert create_job.id is None
+    assert create_job.wait() is None
+
+    view = db.open_materialized_view("adults")
+    refresh_job = view.refresh_async()
+    assert refresh_job.id is None
+    result = refresh_job.wait()
+    assert result.mode == "rebuild"
+    assert result.rows_written == 2
+    assert view.table.count_rows() == 2
+
+    drop_job = db.drop_materialized_view_async("adults")
+    assert drop_job.id is None
+    assert drop_job.wait() is None
+    assert "adults" not in db.list_materialized_views()
+
+
 def test_definition_round_trips(tmp_path):
     db = make_db(tmp_path)
     db.create_materialized_view("adults", "people", where="age >= 18")
@@ -56,7 +226,7 @@ def test_definition_round_trips(tmp_path):
 
 def test_incremental_refresh_after_append(tmp_path):
     db = make_db(tmp_path)
-    view = db.create_materialized_view("copy", "people")
+    view = db.create_materialized_view("copy", "people", with_no_data=True)
     view.refresh()
 
     db.open_table("people").add([{"name": "alan", "age": 41}])
@@ -70,7 +240,7 @@ def test_incremental_refresh_after_append(tmp_path):
 
 def test_incremental_refresh_after_update(tmp_path):
     db = make_db(tmp_path)
-    view = db.create_materialized_view("copy", "people")
+    view = db.create_materialized_view("copy", "people", with_no_data=True)
     view.refresh()
 
     db.open_table("people").update(where="name = 'kid'", values={"age": 8})
@@ -87,7 +257,7 @@ def test_legacy_storage_source_update_rebuilds(tmp_path):
         storage_options={**STABLE_ROW_IDS, "new_table_data_storage_version": "legacy"},
     )
     db.create_table("people", [{"name": "ada", "age": 36}, {"name": "kid", "age": 7}])
-    view = db.create_materialized_view("copy", "people")
+    view = db.create_materialized_view("copy", "people", with_no_data=True)
     view.refresh()
 
     db.open_table("people").update(where="name = 'kid'", values={"age": 8})
@@ -104,6 +274,11 @@ def test_list_and_not_a_view(tmp_path):
     assert db.list_materialized_views() == ["adults"]
     with pytest.raises(ValueError, match="not a materialized view"):
         db.open_materialized_view("people")
+    with pytest.raises(ValueError, match="not a materialized view"):
+        db.drop_materialized_view("people")
+
+    db.drop_materialized_view("adults")
+    assert db.list_materialized_views() == []
 
 
 def test_invalid_expression_fails_at_create(tmp_path):
@@ -119,7 +294,10 @@ async def test_async_create_refresh_and_open(tmp_path):
     await db.create_table("people", [{"name": "ada", "age": 36}])
 
     view = await db.create_materialized_view(
-        "shouts", "people", select=[("shout", "upper(name)")]
+        "shouts",
+        "people",
+        select=[("shout", "upper(name)")],
+        with_no_data=True,
     )
     result = await view.refresh()
     assert result.mode == "rebuild"
@@ -131,11 +309,35 @@ async def test_async_create_refresh_and_open(tmp_path):
     assert await db.list_materialized_views() == ["shouts"]
 
 
+@pytest.mark.asyncio
+async def test_async_create_and_refresh_jobs(tmp_path):
+    db = await lancedb.connect_async(tmp_path, storage_options=STABLE_ROW_IDS)
+    await db.create_table("people", [{"name": "ada", "age": 36}])
+
+    create_job = await db.create_materialized_view_async(
+        "adults", "people", with_no_data=True
+    )
+    assert create_job.id is None
+    assert await create_job.wait() is None
+
+    view = await db.open_materialized_view("adults")
+    refresh_job = await view.refresh_async()
+    assert refresh_job.id is None
+    result = await refresh_job.wait()
+    assert result.mode == "rebuild"
+    assert result.rows_written == 1
+
+    drop_job = await db.drop_materialized_view_async("adults")
+    assert drop_job.id is None
+    assert await drop_job.wait() is None
+    assert "adults" not in await db.list_materialized_views()
+
+
 @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")
+    view = await db.create_materialized_view("copy", "people", with_no_data=True)
     await view.refresh()
 
     table = await db.open_table("people")
@@ -157,7 +359,10 @@ def test_bare_select_names_are_quoted(tmp_path):
     db.create_table("odd_names", [{"order item": "widget", "select": 2}])
 
     view = db.create_materialized_view(
-        "quoted", "odd_names", select=["order item", "select"]
+        "quoted",
+        "odd_names",
+        select=["order item", "select"],
+        with_no_data=True,
     )
     result = view.refresh()
     assert result.rows_written == 1
@@ -166,19 +371,6 @@ def test_bare_select_names_are_quoted(tmp_path):
     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")
@@ -235,6 +427,17 @@ def test_namespace_connection_materialized_views(tmp_path):
     with pytest.raises(ValueError, match="not a materialized view"):
         db.open_materialized_view("people")
 
+    create_job = db.create_materialized_view_async(
+        "job_view", "people", with_no_data=True
+    )
+    assert create_job.wait() is None
+    refresh_job = db.open_materialized_view("job_view").refresh_async()
+    assert refresh_job.wait().rows_written == 2
+
+    assert db.drop_materialized_view_async("job_view").wait() is None
+    db.drop_materialized_view("adults")
+    assert db.list_materialized_views() == []
+
 
 @pytest.mark.asyncio
 async def test_async_namespace_connection_materialized_views(tmp_path):
@@ -267,6 +470,19 @@ async def test_async_namespace_connection_materialized_views(tmp_path):
         )
         assert handle._namespace_path == through_namespace._namespace_path
 
+    create_job = await db.create_materialized_view_async(
+        "job_view", "people", with_no_data=True
+    )
+    assert await create_job.wait() is None
+    job_view = await db.open_materialized_view("job_view")
+    refresh_job = await job_view.refresh_async()
+    assert (await refresh_job.wait()).rows_written == 2
+
+    drop_job = await db.drop_materialized_view_async("job_view")
+    assert await drop_job.wait() is None
+    await db.drop_materialized_view("adults")
+    assert await db.list_materialized_views() == []
+
 
 def test_namespaced_select_kind_is_read_and_unknown_kinds_are_refused():
     import json