fix: use SQL for all remote materialized view APIs (#4673)

Remote materialized-view APIs now use SQL for the full lifecycle. Create
and refresh already submit SQL; this change moves describe, paginated
listing, and drop onto `DESCRIBE MATERIALIZED VIEW`, `SHOW MATERIALIZED
VIEWS`, and `DROP MATERIALIZED VIEW`.

Async create, refresh, and drop return the existing Job handle with
`job_type = "query"`. The handle has one `id`: the connection-scoped SQL
query ID, also accepted by `describe_query`. Submission starts the
query, the job drains its result once, and repeated waits reuse the
completed result. Synchronous drop waits for catalog removal; physical
cleanup proceeds in the background.

Listing uses namespace-scoped keyset pagination with opaque resume
tokens. Streamed and deferred SQL responses share the same path. Adds
transport coverage and updates Python/TypeScript documentation for
query-backed drop jobs.

---------

Co-authored-by: Xuanwo <github@xuanwo.io>

docs/src/js/classes/Connection.md

@@ -436,8 +436,8 @@ abstract dropMaterializedView(name, namespacePath?): Promise<void>
 
 Drop the materialized view named `name`.
 
-The view may become unavailable before physical cleanup finishes. Use
+Remote drops wait for the SQL command to remove the catalog binding.
-[dropMaterializedViewAsync](Connection.md#dropmaterializedviewasync) to retain and wait for the cleanup job.
+Physical cleanup continues in the background.
 
 Rejects a table that exists but is not a materialized view.
 
@@ -459,8 +459,9 @@ Rejects a table that exists but is not a materialized view.
 abstract dropMaterializedViewAsync(name, namespacePath?): Promise<Job>
 ```
 
-Start dropping the materialized view named `name` and return its cleanup
+Start dropping the materialized view named `name` and return its job.
-job without waiting for completion.
+Remote jobs have type `query` and use the SQL query ID as `id`. Waiting
+confirms catalog removal; physical cleanup continues in the background.
 
 Rejects a table that exists but is not a materialized view.
 

docs/web/basics/jobs.mdx

@@ -72,10 +72,12 @@ SQL-backed job handles follow the query lifecycle described below.
 
 ### SQL-backed job handles
 
-Remote materialized-view CREATE and REFRESH use the SQL query backend. Python's
+Remote materialized-view CREATE, REFRESH, and DROP use the SQL query backend.
-`create_materialized_view_async()` and `mv.refresh_async()` still return job
+Python's `create_materialized_view_async()`, `mv.refresh_async()`, and
-handles with `status()`, `wait()` and `cancel()`. Synchronous creation and refresh
+`drop_materialized_view_async()` return job handles with `status()`, `wait()`,
-wait on the same SQL execution before returning.
+and `cancel()`. Synchronous creation, refresh, and drop wait on the same SQL
+execution before returning. A completed DROP means the catalog binding has been
+removed; physical cleanup continues in the background.
 
 The existing `job.job_type` identifies query-backed handles as `"query"` from
 submission, and `job.id` contains the connection-scoped SQL query UUID string.

nodejs/__test__/remote.test.ts

@@ -481,23 +481,6 @@ describe("remote connection", () => {
     expect(insertRequests).toBe(2);
   });
 
-  it("lists materialized views through the namespace route", async () => {
-    await withMockDatabase(
-      (req, res) => {
-        expect(req.method).toBe("GET");
-        expect(req.url).toBe("/v1/namespace/$/materialized_view/list");
-        res
-          .writeHead(200, { "content-type": "application/json" })
-          .end(JSON.stringify({ views: ["daily_sales"] }));
-      },
-      async (db) => {
-        expect(await collect(db.listMaterializedViews())).toEqual([
-          "daily_sales",
-        ]);
-      },
-    );
-  });
-
   it("creates a view and decodes the schema it resolved to", async () => {
     await withMockDatabase(
       (req, res) => {
@@ -1725,14 +1708,13 @@ async function collect<T>(items: AsyncIterable<T>): Promise<T[]> {
 }
 
 describe("lazy resource listings", () => {
-  for (const kind of ["views", "materializedViews", "jobs"] as const) {
+  for (const kind of ["views", "jobs"] as const) {
     const listing = (
       db: Connection,
       options: { pageToken?: string; pageLimit?: number } = {},
     ) => {
       if (kind === "jobs") return db.listJobs(options);
-      if (kind === "views") return db.listViews(["team"], options);
+      return db.listViews(["team"], options);
-      return db.listMaterializedViews(options);
     };
     const page = (names: string[], token?: string) => {
       const items =
@@ -1807,11 +1789,7 @@ describe("lazy resource listings", () => {
                 limit: 2,
               });
             } else {
-              expect(url.pathname).toBe(
+              expect(url.pathname).toBe("/v1/namespace/team/view/list");
-                kind === "views"
-                  ? "/v1/namespace/team/view/list"
-                  : "/v1/namespace/$/materialized_view/list",
-              );
               expect(url.searchParams.get("page_token")).toBe(token);
               expect(url.searchParams.get("limit")).toBe("2");
             }

nodejs/lancedb/connection.ts

@@ -308,8 +308,8 @@ export abstract class Connection {
   /**
    * Drop the materialized view named `name`.
    *
-   * The view may become unavailable before physical cleanup finishes. Use
+   * Remote drops wait for the SQL command to remove the catalog binding.
-   * {@link dropMaterializedViewAsync} to retain and wait for the cleanup job.
+   * Physical cleanup continues in the background.
    *
    * Rejects a table that exists but is not a materialized view.
    */
@@ -319,8 +319,9 @@ export abstract class Connection {
   ): Promise<void>;
 
   /**
-   * Start dropping the materialized view named `name` and return its cleanup
+   * Start dropping the materialized view named `name` and return its job.
-   * job without waiting for completion.
+   * Remote jobs have type `query` and use the SQL query ID as `id`. Waiting
+   * confirms catalog removal; physical cleanup continues in the background.
    *
    * Rejects a table that exists but is not a materialized view.
    */

python/python/lancedb/db.py

@@ -721,9 +721,8 @@ class DBConnection(EnforceOverrides):
     ) -> None:
         """Drop a materialized view.
 
-        The view may become unavailable before physical cleanup finishes. Use
+        Remote drops wait for the SQL command to remove the catalog binding.
-        :meth:`drop_materialized_view_async` to retain and wait for the cleanup
+        Physical cleanup continues in the background.
-        job.
         """
         raise NotImplementedError(
             "materialized views are not supported on this connection type"
@@ -732,11 +731,11 @@ class DBConnection(EnforceOverrides):
     def drop_materialized_view_async(
         self, name: str, namespace_path: Optional[List[str]] = None
     ) -> Job[None]:
-        """Start dropping a materialized view and return its cleanup job.
+        """Start dropping a materialized view and return its job.
 
-        The job may already be complete for a local database. On LanceDB Cloud
+        The job may already be complete for a local database. Remote jobs have
-        and Enterprise, its ``id`` is the server job identifier from the
+        type ``query`` and use the SQL query ID as ``id``. Waiting confirms
-        ``202 Accepted`` drop response.
+        catalog removal; physical cleanup continues in the background.
         """
         raise NotImplementedError(
             "materialized views are not supported on this connection type"
@@ -2946,9 +2945,8 @@ class AsyncConnection(object):
     ) -> None:
         """Drop a materialized view.
 
-        The view may become unavailable before physical cleanup finishes. Use
+        Remote drops wait for the SQL command to remove the catalog binding.
-        :meth:`drop_materialized_view_async` to retain and wait for the cleanup
+        Physical cleanup continues in the background.
-        job.
         """
         if namespace_path is None:
             namespace_path = []
@@ -2960,10 +2958,11 @@ class AsyncConnection(object):
         *,
         namespace_path: Optional[List[str]] = None,
     ) -> AsyncJob[None]:
-        """Start dropping a materialized view and return its cleanup job.
+        """Start dropping a materialized view and return its job.
 
-        Await :meth:`AsyncJob.wait` before assuming physical cleanup has
+        For remote connections, the job has type ``query`` and its ID identifies
-        finished.
+        the SQL query. Await :meth:`AsyncJob.wait` for catalog removal; physical
+        cleanup continues in the background.
         """
         if namespace_path is None:
             namespace_path = []

python/python/tests/test_listing.py

@@ -66,7 +66,7 @@ def response_page(kind, names, token):
         ]
     else:
         items = names
-    key = "views" if kind == "materialized_views" else kind
+    key = kind
     return {key: items, "page_token": token}
 
 
@@ -85,7 +85,7 @@ async def advance(items, asynchronous):
     return await items.__anext__() if asynchronous else next(items)
 
 
-KINDS = ["secrets", "views", "jobs", "functions", "materialized_views"]
+KINDS = ["secrets", "views", "jobs", "functions"]
 
 
 @pytest.mark.asyncio
@@ -236,9 +236,7 @@ async def test_listing_lazy_pages_and_resume(listing_server, kind, asynchronous)
     )
     db = await connection(endpoint, asynchronous)
     method = getattr(db, "list_" + kind)
-    kwargs = (
+    kwargs = {} if kind == "jobs" else {"namespace_path": ["team"]}
-        {} if kind in ("jobs", "materialized_views") else {"namespace_path": ["team"]}
-    )
     items = method(page_token="start/token", page_limit=2, **kwargs)
     assert requests == []
     assert items.page_token() == "start/token"
@@ -273,7 +271,7 @@ async def test_listing_lazy_pages_and_resume(listing_server, kind, asynchronous)
             assert path.path == "/v1/jobs/list"
             assert body == {"page_token": token, "limit": 2}
         else:
-            namespace = "$" if kind == "materialized_views" else "team"
+            namespace = "team"
             resource = kind.removesuffix("s")
             assert path.path == f"/v1/namespace/{namespace}/{resource}/list"
             expected = {"page_token": [token], "limit": ["2"]}

python/python/tests/test_materialized_views.py

@@ -1,6 +1,7 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright The LanceDB Authors
 
+
 import contextlib
 import http.server
 import json
@@ -30,89 +31,6 @@ 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_drop():
-    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))
-            assert self.path.endswith("/drop")
-            job_id = "mv-drop-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 list(db.list_materialized_views()) == ["daily_sales"]
-    assert requests == ["/v1/namespace/$/materialized_view/list"]
-
-
-def test_remote_drop_async_returns_server_job():
-    with mock_remote_materialized_view_drop() 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", {})]
-
-
 @contextlib.contextmanager
 def mock_remote_materialized_view_refresh():
     requests = []
@@ -123,11 +41,8 @@ def mock_remote_materialized_view_refresh():
 
         def do_POST(self):
             requests.append(self.path)
-            if self.path == "/v1/materialized_view/view/describe":
+            assert self.path == "/v1/table/view/describe/"
-                response = {"query": "SELECT * FROM source"}
+            response = {"version": 42, "schema": {"fields": []}}
-            else:
-                assert self.path == "/v1/table/view/describe/"
-                response = {"version": 42, "schema": {"fields": []}}
             encoded = json.dumps(response).encode()
             self.send_response(200)
             self.send_header("Content-Type", "application/json")
@@ -159,6 +74,16 @@ async def test_remote_refresh_reports_unknown_metrics(sync, background, full):
 
         def do_get(self, context, ticket):
             self.commands.append(ticket.ticket)
+            if b"DESCRIBE MATERIALIZED VIEW" in ticket.ticket:
+                return flight.RecordBatchStream(
+                    pa.table(
+                        {
+                            "name": ["view"],
+                            "query": ["SELECT * FROM source"],
+                        }
+                    )
+                )
+            assert b"REFRESH MATERIALIZED VIEW" in ticket.ticket
             # SQL completion carries no refresh statistics.
             return flight.RecordBatchStream(pa.table({"status": ["ok"]}))
 
@@ -199,9 +124,24 @@ async def test_remote_refresh_reports_unknown_metrics(sync, background, full):
             "RefreshMaterializedViewResult(mode=None, rows_written=None, "
             "source_version=None, version=42)"
         )
-        (command,) = server.commands
+        describes = [
-        assert b'REFRESH MATERIALIZED VIEW "view"' in command
+            command
-        assert (b"WITH (full = true)" in command) is full
+            for command in server.commands
+            if b"DESCRIBE MATERIALIZED VIEW" in command
+        ]
+        refreshes = [
+            command
+            for command in server.commands
+            if b"REFRESH MATERIALIZED VIEW" in command
+        ]
+        assert len(describes) == 2
+        assert len(refreshes) == 1
+        assert all(
+            b'DESCRIBE MATERIALIZED VIEW "view"' in command for command in describes
+        )
+        (refresh,) = refreshes
+        assert b'REFRESH MATERIALIZED VIEW "view"' in refresh
+        assert (b"WITH (full = true)" in refresh) is full
         assert not any("count_rows" in path for path in requests)
 
 
@@ -639,8 +579,6 @@ async def test_async_namespace_connection_materialized_views(tmp_path):
 
 
 def test_stored_queries_and_legacy_layouts_are_read():
-    import json
-
     import pyarrow as pa
 
     from lancedb.materialized_view import _definition_from_schema
@@ -795,13 +733,11 @@ def test_embedded_create_refuses_automatic_refresh_before_creating_view(
 
 @pytest.mark.parametrize("seconds", [0, -1, float("nan"), float("inf")])
 def test_remote_create_rejects_invalid_interval_without_request(seconds):
-    with mock_remote_materialized_view_drop() as (host, requests):
+    db = lancedb.connect("db://dev", api_key="fake", host_override="http://127.0.0.1:1")
-        db = lancedb.connect("db://dev", api_key="fake", host_override=host)
+    with pytest.raises(ValueError, match="positive, finite"):
-        with pytest.raises(ValueError, match="positive, finite"):
+        db.create_materialized_view_async(
-            db.create_materialized_view_async(
+            "mv", "source", auto_refresh_interval_seconds=seconds
-                "adults", "people", auto_refresh_interval_seconds=seconds
+        )
-            )
-    assert requests == []
 
 
 @pytest.mark.experimental_functions

rust/lancedb/src/materialized_view.rs

@@ -1987,8 +1987,8 @@ impl Connection {
     /// Drop a materialized view.
     ///
     /// The view may become unavailable before its physical data is removed.
-    /// Use [`Connection::drop_materialized_view_async`] to retain the cleanup
+    /// Remote drops wait for the SQL command to remove the catalog binding;
-    /// job and wait for it explicitly.
+    /// physical cleanup proceeds in the background.
     pub async fn drop_materialized_view(
         &self,
         name: impl AsRef<str>,
@@ -1999,8 +1999,9 @@ impl Connection {
             return self
                 .database()
                 .drop_materialized_view_async(name, namespace_path)
-                .await
+                .await?
-                .map(|_| ());
+                .wait()
+                .await;
         }
         let table = self
             .open_table(name)
@@ -2011,12 +2012,12 @@ impl Connection {
         self.drop_table(name, namespace_path).await
     }
 
-    /// Start dropping a materialized view and return its cleanup job.
+    /// Start dropping a materialized view and return its job.
     ///
     /// This validates that the named resource is a materialized view rather
-    /// than an ordinary table. Call [`Job::wait`] before assuming physical
+    /// than an ordinary table. For remote connections the job has type `query`
-    /// cleanup has finished. When the backend performs cleanup inline, the
+    /// and its ID identifies the SQL query. Waiting completes once the catalog
-    /// returned job is already finished and has no job ID.
+    /// binding is removed; physical cleanup continues in the background.
     ///
     /// ```no_run
     /// # use lancedb::Connection;

rust/lancedb/src/remote/db.rs

@@ -47,6 +47,7 @@ use super::client::{
     Sender,
 };
 use super::sql::SqlClient;
+use super::sql::materialized_view::quote_identifier;
 use super::table::RemoteTable;
 use super::util::parse_server_version;
 use super::{ARROW_STREAM_CONTENT_TYPE, extract_job_id};
@@ -929,36 +930,26 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         name: &str,
         namespace_path: &[String],
     ) -> Result<Job> {
-        let identifier = build_table_identifier(name, namespace_path)?;
+        validate_table_name(name)?;
-        let request = self
+        validate_namespace(namespace_path)?;
-            .client
+        let client = self
-            .post(&format!("/v1/materialized_view/{identifier}/drop"));
+            .sql_client
-        let (request_id, response) = self.client.send(request).await?;
+            .as_ref()
-        let response = self.client.check_response(&request_id, response).await?;
+            .ok_or_else(|| Error::NotSupported {
-        let status = response.status();
+                message: "SQL is unavailable for this remote database client".into(),
-        let body = response.text().await.err_to_http(request_id.clone())?;
+            })?;
-        let job = match status {
+        let cache = self.table_cache.clone();
-            StatusCode::OK => Job::new_done(),
+        let key = build_cache_key(name, namespace_path);
-            StatusCode::ACCEPTED => {
+        client
-                let job_id = extract_job_id(&body).ok_or_else(|| Error::Http {
+            .submit_as_job(
-                    source: "materialized-view drop response did not contain a valid job_id".into(),
+                format!("DROP MATERIALIZED VIEW {}", quote_identifier(name)),
-                    request_id,
+                namespace_path.to_vec(),
-                    status_code: Some(status),
+                move || async move {
-                })?;
+                    cache.remove(&key).await;
-                Job::new(Box::new(RemoteJob::new(self.client.clone(), job_id)))
+                    Ok(())
-            }
+                },
-            _ => {
+            )
-                return Err(Error::Http {
+            .await
-                    source: "materialized-view drop must return 200 OK or 202 Accepted".into(),
-                    request_id,
-                    status_code: Some(status),
-                });
-            }
-        };
-        self.table_cache
-            .remove(&build_cache_key(name, namespace_path))
-            .await;
-        Ok(job)
     }
 
     async fn list_materialized_views(
@@ -966,23 +957,13 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         namespace_path: &[String],
         options: ListingOptions,
     ) -> Result<ListingPage<String>> {
-        let namespace_id = build_namespace_identifier(namespace_path)?;
+        self.sql_client
-        let path = format!("/v1/namespace/{namespace_id}/materialized_view/list");
+            .as_ref()
-        let mut req = self.client.get(&path);
+            .ok_or_else(|| Error::NotSupported {
-        if let Some(token) = options.page_token {
+                message: "SQL is unavailable for this remote database client".into(),
-            req = req.query(&[("page_token", token)]);
+            })?
-        }
+            .list_materialized_views(namespace_path, options)
-        if let Some(limit) = options.page_limit {
+            .await
-            req = req.query(&[("limit", limit)]);
-        }
-        let (request_id, response) = self.client.send(req).await?;
-        let response = self.client.check_response(&request_id, response).await?;
-        let response: RemoteListViewsResponse = response.json().await.err_to_http(request_id)?;
-        let items = response.views;
-        Ok(ListingPage {
-            items,
-            page_token: response.page_token,
-        })
     }
 
     async fn create_function_async(
@@ -2065,27 +2046,6 @@ mod tests {
         }
     }
 
-    #[tokio::test]
-    async fn test_drop_materialized_view_uses_item_route_and_job() {
-        let db = super::RemoteDatabase::new_mock(|request| {
-            assert_eq!(request.method(), "POST");
-            assert_eq!(
-                request.url().path(),
-                "/v1/materialized_view/analytics$adults/drop"
-            );
-            assert!(request.body().is_none());
-            http::Response::builder()
-                .status(202)
-                .body(serde_json::json!({"job_id": "j1-mv-drop"}).to_string())
-                .unwrap()
-        });
-        let job = db
-            .drop_materialized_view_async("adults", &["analytics".into()])
-            .await
-            .unwrap();
-        assert_eq!(job.id(), Some("j1-mv-drop"));
-    }
-
     #[tokio::test]
     async fn test_drop_function_async_returns_job() {
         let db = super::RemoteDatabase::new_mock(|request| {
@@ -2137,100 +2097,6 @@ mod tests {
         }
     }
 
-    #[tokio::test]
-    async fn test_drop_materialized_view_completed_inline() {
-        let db = super::RemoteDatabase::new_mock(|request| {
-            assert_eq!(request.method(), "POST");
-            assert_eq!(request.url().path(), "/v1/materialized_view/adults/drop");
-            http::Response::builder().status(200).body("{}").unwrap()
-        });
-        let job = db
-            .drop_materialized_view_async("adults", &[])
-            .await
-            .unwrap();
-        assert_eq!(job.id(), None);
-        assert_eq!(job.status().await.unwrap(), "finished");
-        job.wait().await.unwrap();
-    }
-
-    #[tokio::test]
-    async fn test_drop_materialized_view_rejects_incomplete_acceptance() {
-        for body in ["{}", r#"{"job_id":""}"#, r#"{"job_id":null}"#] {
-            let db = super::RemoteDatabase::new_mock(move |_| {
-                http::Response::builder().status(202).body(body).unwrap()
-            });
-            let error = db
-                .drop_materialized_view_async("adults", &[])
-                .await
-                .err()
-                .unwrap();
-            assert!(error.to_string().contains("valid job_id"));
-        }
-    }
-
-    #[tokio::test]
-    async fn test_drop_materialized_view_rejects_unexpected_success_status() {
-        let db = super::RemoteDatabase::new_mock(|_| {
-            http::Response::builder().status(204).body("").unwrap()
-        });
-        let error = db
-            .drop_materialized_view_async("adults", &[])
-            .await
-            .err()
-            .unwrap();
-        assert!(error.to_string().contains("200 OK or 202 Accepted"));
-    }
-
-    #[tokio::test]
-    async fn test_list_materialized_views_follows_empty_pages() {
-        let page = Arc::new(AtomicUsize::new(0));
-        let db = super::RemoteDatabase::new_mock({
-            let page = page.clone();
-            move |request| {
-                assert_eq!(request.method(), "GET");
-                assert_eq!(
-                    request.url().path(),
-                    "/v1/namespace/analytics/materialized_view/list"
-                );
-                match page.fetch_add(1, Ordering::SeqCst) {
-                    0 => {
-                        assert!(request.url().query().is_none());
-                        http::Response::builder()
-                            .status(200)
-                            .body(
-                                serde_json::json!({"views": [], "page_token": "next"}).to_string(),
-                            )
-                            .unwrap()
-                    }
-                    1 => {
-                        assert_eq!(request.url().query(), Some("page_token=next"));
-                        http::Response::builder()
-                            .status(200)
-                            .body(serde_json::json!({"views": ["adults"]}).to_string())
-                            .unwrap()
-                    }
-                    _ => panic!("listing requested too many pages"),
-                }
-            }
-        });
-        assert_eq!(
-            Listing::new(Default::default(), {
-                let db = Arc::new(db);
-                move |options| {
-                    let db = db.clone();
-                    async move {
-                        db.list_materialized_views(&["analytics".into()], options)
-                            .await
-                    }
-                }
-            })
-            .try_collect::<Vec<_>>()
-            .await
-            .unwrap(),
-            ["adults"]
-        );
-    }
-
     #[tokio::test]
     async fn test_retries() {
         // We'll record the request_id here, to check it matches the one in the error.
@@ -3912,7 +3778,7 @@ mod tests {
             assert!(items.try_next().await.unwrap().is_none());
             assert!(items.try_next().await.unwrap().is_none());
         }
-        for resource in ["secret", "view", "function", "materialized_view", "jobs"] {
+        for resource in ["secret", "view", "function", "jobs"] {
             let count = Arc::new(AtomicUsize::new(0));
             let conn = Connection::new_with_handler({
                 let count = count.clone();
@@ -3927,11 +3793,7 @@ mod tests {
                                 .unwrap();
                         assert_eq!(body, serde_json::json!({"page_token": token, "limit": 2}));
                     } else {
-                        let ns = if resource == "materialized_view" {
+                        let ns = "team";
-                            "$"
-                        } else {
-                            "team"
-                        };
                         assert_eq!(
                             request.url().path(),
                             format!("/v1/namespace/{ns}/{resource}/list")
@@ -3971,7 +3833,6 @@ mod tests {
                 "secret" => check(conn.list_secrets(&namespace, options), count).await,
                 "view" => check(conn.list_views(&namespace, options), count).await,
                 "function" => check(conn.list_functions(&namespace, options), count).await,
-                "materialized_view" => check(conn.list_materialized_views(options), count).await,
                 _ => check(conn.list_jobs(options), count).await,
             }
         }

rust/lancedb/src/remote/sql.rs

@@ -35,6 +35,7 @@ use crate::remote::client::{ClientConfig, TlsConfig};
 use crate::remote::retry::ResolvedRetryConfig;
 use crate::sql::{Query, QueryDescription, QueryHandle, QueryStatus};
 
+pub(super) mod materialized_view;
 mod stream;
 
 use stream::{StreamingQuery, StreamingQueryHandle, Submission};

rust/lancedb/src/remote/sql/materialized_view.rs

@@ -0,0 +1,129 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+use arrow_array::{Array, RecordBatch, StringArray};
+use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
+use futures::TryStreamExt;
+
+use super::SqlClient;
+use crate::database::ExecuteQueryRequest;
+use crate::error::{Error, Result};
+use crate::listing::{ListingOptions, ListingPage};
+use crate::materialized_view::{MaterializedViewDefinition, MaterializedViewInfo};
+use crate::utils::{validate_namespace, validate_table_name};
+
+pub(in crate::remote) fn quote_identifier(name: &str) -> String {
+    format!("\"{}\"", name.replace('"', "\"\""))
+}
+
+fn strings(batch: &RecordBatch, name: &str) -> Result<Vec<String>> {
+    let values = batch
+        .column_by_name(name)
+        .and_then(|column| column.as_any().downcast_ref::<StringArray>())
+        .ok_or_else(|| Error::Runtime {
+            message: format!("materialized-view SQL result is missing UTF-8 column '{name}'"),
+        })?;
+    if values.null_count() != 0 {
+        return Err(Error::Runtime {
+            message: format!("materialized-view SQL result contains null '{name}'"),
+        });
+    }
+    Ok(values
+        .iter()
+        .map(|value| value.unwrap().to_owned())
+        .collect())
+}
+
+impl SqlClient {
+    pub(in crate::remote) async fn describe_materialized_view(
+        &self,
+        name: &str,
+        namespace: &[String],
+    ) -> Result<MaterializedViewInfo> {
+        validate_table_name(name)?;
+        validate_namespace(namespace)?;
+        let query = self
+            .execute(ExecuteQueryRequest {
+                query: format!("DESCRIBE MATERIALIZED VIEW {}", quote_identifier(name)),
+                default_namespace_path: namespace.to_vec(),
+                parameters: None,
+            })
+            .await?;
+        let mut reader = query.reader().await?;
+        let mut definitions = Vec::new();
+        while let Some(batch) = reader.try_next().await? {
+            definitions.extend(strings(&batch, "query")?);
+        }
+        let [definition_sql] =
+            <[String; 1]>::try_from(definitions).map_err(|_| Error::Runtime {
+                message: "DESCRIBE MATERIALIZED VIEW must return exactly one definition".into(),
+            })?;
+        let parsed_definition = MaterializedViewDefinition::from_sql(&definition_sql).ok();
+        Ok(MaterializedViewInfo {
+            definition_sql,
+            parsed_definition,
+        })
+    }
+
+    pub(in crate::remote) async fn list_materialized_views(
+        &self,
+        namespace: &[String],
+        options: ListingOptions,
+    ) -> Result<ListingPage<String>> {
+        validate_namespace(namespace)?;
+        let limit = options.page_limit.unwrap_or(100);
+        if limit == 0 || limit > i32::MAX as u32 {
+            return Err(Error::InvalidInput {
+                message: "page limit must be between 1 and i32::MAX".into(),
+            });
+        }
+        let invalid_token = || Error::InvalidInput {
+            message: "invalid materialized-view page token for this database or namespace".into(),
+        };
+        let after = if let Some(token) = options.page_token.filter(|token| !token.is_empty()) {
+            let decoded = URL_SAFE_NO_PAD.decode(token).map_err(|_| invalid_token())?;
+            let (database, prefix, path, last): (String, Option<String>, Vec<String>, String) =
+                serde_json::from_slice(&decoded).map_err(|_| invalid_token())?;
+            if database != self.inner.database
+                || prefix != self.inner.database_prefix
+                || path != namespace
+            {
+                return Err(invalid_token());
+            }
+            format!(" WHERE name > '{}'", last.replace('\'', "''"))
+        } else {
+            String::new()
+        };
+        let query = self
+            .execute(ExecuteQueryRequest {
+                query: format!(
+                    "SHOW MATERIALIZED VIEWS{after} LIMIT {}",
+                    u64::from(limit) + 1
+                ),
+                default_namespace_path: namespace.to_vec(),
+                parameters: None,
+            })
+            .await?;
+        let mut reader = query.reader().await?;
+        let mut items = Vec::new();
+        while let Some(batch) = reader.try_next().await? {
+            items.extend(strings(&batch, "name")?);
+        }
+        let page_token = if items.len() > limit as usize {
+            items.truncate(limit as usize);
+            let cursor = serde_json::to_vec(&(
+                &self.inner.database,
+                &self.inner.database_prefix,
+                namespace,
+                items.last().unwrap(),
+            ))
+            .map_err(|error| Error::Runtime {
+                message: error.to_string(),
+            })?;
+            Some(URL_SAFE_NO_PAD.encode(cursor))
+        } else {
+            None
+        };
+        Ok(ListingPage { items, page_token })
+    }
+}

rust/lancedb/src/remote/sql_test.rs

@@ -1028,7 +1028,21 @@ async fn remote_refresh_omits_unknown_metrics_and_forwards_full() {
         let (socket, _) = listener.accept().await?;
         Ok::<_, std::io::Error>(Some((socket, listener)))
     });
-    let service = TestSqlService::default();
+    let result = RecordBatch::try_from_iter(vec![
+        (
+            "name",
+            Arc::new(StringArray::from(vec!["view"])) as ArrayRef,
+        ),
+        (
+            "query",
+            Arc::new(StringArray::from(vec!["SELECT * FROM source"])) as ArrayRef,
+        ),
+    ])
+    .unwrap();
+    let service = TestSqlService {
+        result,
+        ..Default::default()
+    };
     let statements = service.statements.clone();
     let headers = service.headers.clone();
     let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
@@ -1057,7 +1071,6 @@ async fn remote_refresh_omits_unknown_metrics_and_forwards_full() {
             .unwrap_or_default();
         captured.lock().unwrap().push(body["version"].clone());
         let text = match request.url().path() {
-            "/v1/materialized_view/ns$view/describe" => r#"{"query": "SELECT * FROM source"}"#,
             "/v1/table/ns$view/describe/" => r#"{"version": 42, "schema": {"fields": []}}"#,
             other => panic!("unexpected REST request: {other}"),
         };
@@ -1111,6 +1124,7 @@ async fn remote_refresh_omits_unknown_metrics_and_forwards_full() {
     assert_eq!(
         *statements.lock().unwrap(),
         [
+            "DESCRIBE MATERIALIZED VIEW \"view\"",
             "REFRESH MATERIALIZED VIEW \"view\"",
             "REFRESH MATERIALIZED VIEW \"view\" WITH (full = true)",
         ]
@@ -1994,3 +2008,151 @@ async fn malformed_adaptive_decision_never_resubmits_the_statement() {
         assert_eq!(polls.load(Ordering::SeqCst), 0);
     }
 }
+
+#[tokio::test]
+async fn remote_drop_materialized_view_uses_a_query_job() {
+    for defer in [false, true] {
+        let service = TestSqlService {
+            adaptive: true,
+            defer,
+            ..Default::default()
+        };
+        let statements = service.statements.clone();
+        let headers = service.headers.clone();
+        let (client, shutdown) = start_sql_service(service, ClientConfig::default()).await;
+        let db = RemoteDatabase::new_mock(|request| -> http::Response<String> {
+            panic!("unexpected REST request: {}", request.url().path());
+        })
+        .with_sql_client(client.clone());
+        let job = db
+            .drop_materialized_view_async("daily_sales", &["team".into(), "reports".into()])
+            .await
+            .unwrap();
+        assert_eq!(job.job_type().as_deref(), Some("query"));
+        let id = Uuid::parse_str(job.id().unwrap()).unwrap();
+        job.wait().await.unwrap();
+        job.wait().await.unwrap();
+        assert_eq!(
+            client.describe(id).await.unwrap().status,
+            QueryStatus::Finished
+        );
+        assert_eq!(
+            *statements.lock().unwrap(),
+            ["DROP MATERIALIZED VIEW \"daily_sales\""]
+        );
+        assert!(
+            headers
+                .lock()
+                .unwrap()
+                .iter()
+                .all(|header| header.namespace_path == "team$reports")
+        );
+        shutdown.send(()).unwrap();
+    }
+}
+
+#[tokio::test]
+async fn remote_mv_listing_uses_scoped_sql_keyset_pagination() {
+    for defer in [false, true] {
+        let result = RecordBatch::try_from_iter(vec![(
+            "name",
+            Arc::new(StringArray::from(vec!["a", "b'c", "z"])) as ArrayRef,
+        )])
+        .unwrap();
+        let service = TestSqlService {
+            adaptive: true,
+            defer,
+            result,
+            ..Default::default()
+        };
+        let statements = service.statements.clone();
+        let headers = service.headers.clone();
+        let (client, shutdown) = start_sql_service(service, ClientConfig::default()).await;
+        let db = RemoteDatabase::new_mock(|request| -> http::Response<String> {
+            panic!("unexpected REST request: {}", request.url().path());
+        })
+        .with_sql_client(client);
+        let namespace = vec!["team".to_owned(), "reports".to_owned()];
+        let first = db
+            .list_materialized_views(
+                &namespace,
+                crate::listing::ListingOptions::default().page_limit(2),
+            )
+            .await
+            .unwrap();
+        assert_eq!(first.items, ["a", "b'c"]);
+        let token = first.page_token.unwrap();
+        db.list_materialized_views(
+            &namespace,
+            crate::listing::ListingOptions::default()
+                .page_limit(2)
+                .page_token(&token),
+        )
+        .await
+        .unwrap();
+        assert_eq!(
+            *statements.lock().unwrap(),
+            [
+                "SHOW MATERIALIZED VIEWS LIMIT 3",
+                "SHOW MATERIALIZED VIEWS WHERE name > 'b''c' LIMIT 3"
+            ]
+        );
+        assert!(
+            headers
+                .lock()
+                .unwrap()
+                .iter()
+                .all(|header| header.namespace_path == "team$reports")
+        );
+        for (path, token) in [(vec![], token), (namespace, "invalid".into())] {
+            assert!(matches!(
+                db.list_materialized_views(
+                    &path,
+                    crate::listing::ListingOptions::default().page_token(token)
+                )
+                .await,
+                Err(Error::InvalidInput { .. })
+            ));
+        }
+        assert_eq!(statements.lock().unwrap().len(), 2);
+        shutdown.send(()).unwrap();
+    }
+}
+
+#[tokio::test]
+async fn remote_describe_materialized_view_preserves_the_sql_definition() {
+    for defer in [false, true] {
+        let definition = "SELECT s.x, d.label FROM source s JOIN dim d ON s.id = d.id";
+        let result = RecordBatch::try_from_iter(vec![(
+            "query",
+            Arc::new(StringArray::from(vec![definition])) as ArrayRef,
+        )])
+        .unwrap();
+        let service = TestSqlService {
+            adaptive: true,
+            defer,
+            result,
+            ..Default::default()
+        };
+        let statements = service.statements.clone();
+        let (client, shutdown) = start_sql_service(service, ClientConfig::default()).await;
+        let table = RemoteTable::new_with_sql_client(
+            client_with_handler(|request| -> http::Response<String> {
+                panic!("unexpected REST request: {}", request.url().path());
+            }),
+            "daily_sales".into(),
+            vec!["analytics".into()],
+            "analytics$daily_sales".into(),
+            Default::default(),
+            Some(client),
+        );
+        let info = table.materialized_view_info().await.unwrap();
+        assert_eq!(info.definition_sql, definition);
+        assert!(info.parsed_definition.is_none());
+        assert_eq!(
+            *statements.lock().unwrap(),
+            ["DESCRIBE MATERIALIZED VIEW \"daily_sales\""]
+        );
+        shutdown.send(()).unwrap();
+    }
+}

rust/lancedb/src/remote/table.rs

@@ -18,9 +18,7 @@ use crate::index::IndexStatistics;
 use crate::index::scalar::FtsQuery;
 use crate::index::waiter::{MAX_WAIT, wait_for_index};
 use crate::job::Job;
-use crate::materialized_view::{
+use crate::materialized_view::{MaterializedViewInfo, RefreshMaterializedViewResult};
-    MaterializedViewDefinition, MaterializedViewInfo, RefreshMaterializedViewResult,
-};
 use crate::query::wal_fusion::PkFusionMemory; // WAL-PK-FUSION: delete.
 use crate::query::{QueryFilter, QueryRequest, Select, VectorQueryRequest};
 use crate::remote::job::RemoteJob;
@@ -2201,24 +2199,13 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
         &self.identifier
     }
     async fn materialized_view_info(&self) -> Result<MaterializedViewInfo> {
-        #[derive(Deserialize)]
+        self.sql_client
-        struct DescribeMaterializedViewResponse {
+            .as_ref()
-            query: String,
+            .ok_or_else(|| Error::NotSupported {
-        }
+                message: "SQL is unavailable for this remote table client".into(),
-
+            })?
-        let request = self.client.post(&format!(
+            .describe_materialized_view(&self.name, &self.namespace)
-            "/v1/materialized_view/{}/describe",
+            .await
-            self.identifier
-        ));
-        let (request_id, response) = self.send(request, true).await?;
-        let response = self.check_table_response(&request_id, response).await?;
-        let response: DescribeMaterializedViewResponse =
-            response.json().await.err_to_http(request_id)?;
-        let parsed_definition = MaterializedViewDefinition::from_sql(&response.query).ok();
-        Ok(MaterializedViewInfo {
-            definition_sql: response.query,
-            parsed_definition,
-        })
     }
 
     async fn refresh_materialized_view_async(
@@ -14610,26 +14597,15 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_materialized_view_describe_and_refresh_requires_sql() {
+    async fn test_materialized_view_requires_sql() {
-        const QUERY: &str =
+        let table = Table::new_with_handler("my_table", |request| -> http::Response<String> {
-            "SELECT s.x, d.label FROM analytics.source s JOIN analytics.dim d ON s.id = d.id";
+            panic!("unexpected REST request: {}", request.url().path())
-        let table = Table::new_with_handler("my_table", |request| match request.url().path() {
-            "/v1/materialized_view/my_table/describe" => http::Response::builder()
-                .status(200)
-                .body(
-                    json!({
-                        "name": "my_table",
-                        "query": QUERY
-                    })
-                    .to_string(),
-                )
-                .unwrap(),
-            path => panic!("unexpected request: {path}"),
         });
-        let view = crate::MaterializedView::from_table(table).await.unwrap();
+        let error = table
-        assert_eq!(view.definition_sql(), QUERY);
+            .base_table()
-        assert!(view.definition().is_err());
+            .materialized_view_info()
-        let error = view.refresh().execute().await.unwrap_err();
+            .await
+            .unwrap_err();
         assert!(error.to_string().contains("SQL is unavailable"));
     }