diff --git a/docs/src/js/classes/Connection.md b/docs/src/js/classes/Connection.md index ffe7106b7..2496bc2a9 100644 --- a/docs/src/js/classes/Connection.md +++ b/docs/src/js/classes/Connection.md @@ -436,8 +436,8 @@ abstract dropMaterializedView(name, namespacePath?): Promise Drop the materialized view named `name`. -The view may become unavailable before physical cleanup finishes. Use -[dropMaterializedViewAsync](Connection.md#dropmaterializedviewasync) to retain and wait for the cleanup job. +Remote drops wait for the SQL command to remove the catalog binding. +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 ``` -Start dropping the materialized view named `name` and return its cleanup -job without waiting for completion. +Start dropping the materialized view named `name` and return its job. +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. diff --git a/docs/web/basics/jobs.mdx b/docs/web/basics/jobs.mdx index a87663a51..26dd4dd8b 100644 --- a/docs/web/basics/jobs.mdx +++ b/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 -`create_materialized_view_async()` and `mv.refresh_async()` still return job -handles with `status()`, `wait()` and `cancel()`. Synchronous creation and refresh -wait on the same SQL execution before returning. +Remote materialized-view CREATE, REFRESH, and DROP use the SQL query backend. +Python's `create_materialized_view_async()`, `mv.refresh_async()`, and +`drop_materialized_view_async()` return job handles with `status()`, `wait()`, +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. diff --git a/nodejs/__test__/remote.test.ts b/nodejs/__test__/remote.test.ts index d443537b0..80befcc67 100644 --- a/nodejs/__test__/remote.test.ts +++ b/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(items: AsyncIterable): Promise { } 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.listMaterializedViews(options); + return db.listViews(["team"], options); }; const page = (names: string[], token?: string) => { const items = @@ -1807,11 +1789,7 @@ describe("lazy resource listings", () => { limit: 2, }); } else { - expect(url.pathname).toBe( - kind === "views" - ? "/v1/namespace/team/view/list" - : "/v1/namespace/$/materialized_view/list", - ); + expect(url.pathname).toBe("/v1/namespace/team/view/list"); expect(url.searchParams.get("page_token")).toBe(token); expect(url.searchParams.get("limit")).toBe("2"); } diff --git a/nodejs/lancedb/connection.ts b/nodejs/lancedb/connection.ts index 80fe2c722..62f218e16 100644 --- a/nodejs/lancedb/connection.ts +++ b/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 - * {@link dropMaterializedViewAsync} to retain and wait for the cleanup job. + * Remote drops wait for the SQL command to remove the catalog binding. + * 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; /** - * Start dropping the materialized view named `name` and return its cleanup - * job without waiting for completion. + * Start dropping the materialized view named `name` and return its job. + * 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. */ diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index f38608188..fe38f17f0 100644 --- a/python/python/lancedb/db.py +++ b/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 - :meth:`drop_materialized_view_async` to retain and wait for the cleanup - job. + Remote drops wait for the SQL command to remove the catalog binding. + Physical cleanup continues in the background. """ 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 - and Enterprise, its ``id`` is the server job identifier from the - ``202 Accepted`` drop response. + The job may already be complete for a local database. Remote jobs have + type ``query`` and use the SQL query ID as ``id``. Waiting confirms + 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 - :meth:`drop_materialized_view_async` to retain and wait for the cleanup - job. + Remote drops wait for the SQL command to remove the catalog binding. + Physical cleanup continues in the background. """ 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 - finished. + For remote connections, the job has type ``query`` and its ID identifies + the SQL query. Await :meth:`AsyncJob.wait` for catalog removal; physical + cleanup continues in the background. """ if namespace_path is None: namespace_path = [] diff --git a/python/python/tests/test_listing.py b/python/python/tests/test_listing.py index b0b973bbf..2d1e6497a 100644 --- a/python/python/tests/test_listing.py +++ b/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 = ( - {} if kind in ("jobs", "materialized_views") else {"namespace_path": ["team"]} - ) + kwargs = {} if kind == "jobs" 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"]} diff --git a/python/python/tests/test_materialized_views.py b/python/python/tests/test_materialized_views.py index 04dbe2fec..0966297f7 100644 --- a/python/python/tests/test_materialized_views.py +++ b/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": - response = {"query": "SELECT * FROM source"} - else: - assert self.path == "/v1/table/view/describe/" - response = {"version": 42, "schema": {"fields": []}} + 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 - assert b'REFRESH MATERIALIZED VIEW "view"' in command - assert (b"WITH (full = true)" in command) is full + describes = [ + command + 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=host) - with pytest.raises(ValueError, match="positive, finite"): - db.create_materialized_view_async( - "adults", "people", auto_refresh_interval_seconds=seconds - ) - assert requests == [] + db = lancedb.connect("db://dev", api_key="fake", host_override="http://127.0.0.1:1") + with pytest.raises(ValueError, match="positive, finite"): + db.create_materialized_view_async( + "mv", "source", auto_refresh_interval_seconds=seconds + ) @pytest.mark.experimental_functions diff --git a/rust/lancedb/src/materialized_view.rs b/rust/lancedb/src/materialized_view.rs index 5a7169b7c..266cdab59 100644 --- a/rust/lancedb/src/materialized_view.rs +++ b/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 - /// job and wait for it explicitly. + /// Remote drops wait for the SQL command to remove the catalog binding; + /// physical cleanup proceeds in the background. pub async fn drop_materialized_view( &self, name: impl AsRef, @@ -1999,8 +1999,9 @@ impl Connection { return self .database() .drop_materialized_view_async(name, namespace_path) - .await - .map(|_| ()); + .await? + .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 - /// cleanup has finished. When the backend performs cleanup inline, the - /// returned job is already finished and has no job ID. + /// than an ordinary table. For remote connections the job has type `query` + /// and its ID identifies the SQL query. Waiting completes once the catalog + /// binding is removed; physical cleanup continues in the background. /// /// ```no_run /// # use lancedb::Connection; diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs index 5c299dc5a..394dc4af2 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/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 Database for RemoteDatabase { name: &str, namespace_path: &[String], ) -> Result { - let identifier = build_table_identifier(name, namespace_path)?; - let request = self - .client - .post(&format!("/v1/materialized_view/{identifier}/drop")); - let (request_id, response) = self.client.send(request).await?; - let response = self.client.check_response(&request_id, response).await?; - let status = response.status(); - let body = response.text().await.err_to_http(request_id.clone())?; - let job = match status { - StatusCode::OK => Job::new_done(), - StatusCode::ACCEPTED => { - let job_id = extract_job_id(&body).ok_or_else(|| Error::Http { - source: "materialized-view drop response did not contain a valid job_id".into(), - request_id, - status_code: Some(status), - })?; - Job::new(Box::new(RemoteJob::new(self.client.clone(), job_id))) - } - _ => { - return Err(Error::Http { - 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) + validate_table_name(name)?; + validate_namespace(namespace_path)?; + let client = self + .sql_client + .as_ref() + .ok_or_else(|| Error::NotSupported { + message: "SQL is unavailable for this remote database client".into(), + })?; + let cache = self.table_cache.clone(); + let key = build_cache_key(name, namespace_path); + client + .submit_as_job( + format!("DROP MATERIALIZED VIEW {}", quote_identifier(name)), + namespace_path.to_vec(), + move || async move { + cache.remove(&key).await; + Ok(()) + }, + ) + .await } async fn list_materialized_views( @@ -966,23 +957,13 @@ impl Database for RemoteDatabase { namespace_path: &[String], options: ListingOptions, ) -> Result> { - let namespace_id = build_namespace_identifier(namespace_path)?; - let path = format!("/v1/namespace/{namespace_id}/materialized_view/list"); - let mut req = self.client.get(&path); - if let Some(token) = options.page_token { - req = req.query(&[("page_token", token)]); - } - if let Some(limit) = options.page_limit { - 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, - }) + self.sql_client + .as_ref() + .ok_or_else(|| Error::NotSupported { + message: "SQL is unavailable for this remote database client".into(), + })? + .list_materialized_views(namespace_path, options) + .await } 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::>() - .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" { - "$" - } else { - "team" - }; + let ns = "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, } } diff --git a/rust/lancedb/src/remote/sql.rs b/rust/lancedb/src/remote/sql.rs index 846e852f5..207a0cbec 100644 --- a/rust/lancedb/src/remote/sql.rs +++ b/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}; diff --git a/rust/lancedb/src/remote/sql/materialized_view.rs b/rust/lancedb/src/remote/sql/materialized_view.rs new file mode 100644 index 000000000..46c0a7bd2 --- /dev/null +++ b/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> { + let values = batch + .column_by_name(name) + .and_then(|column| column.as_any().downcast_ref::()) + .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 { + 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> { + 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, Vec, 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 }) + } +} diff --git a/rust/lancedb/src/remote/sql_test.rs b/rust/lancedb/src/remote/sql_test.rs index 260471612..6a0421fa6 100644 --- a/rust/lancedb/src/remote/sql_test.rs +++ b/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 { + 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 { + 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 { + 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(); + } +} diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 214e06d4e..7be8db77c 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/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::{ - MaterializedViewDefinition, MaterializedViewInfo, RefreshMaterializedViewResult, -}; +use crate::materialized_view::{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 BaseTable for RemoteTable { &self.identifier } async fn materialized_view_info(&self) -> Result { - #[derive(Deserialize)] - struct DescribeMaterializedViewResponse { - query: String, - } - - let request = self.client.post(&format!( - "/v1/materialized_view/{}/describe", - 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, - }) + self.sql_client + .as_ref() + .ok_or_else(|| Error::NotSupported { + message: "SQL is unavailable for this remote table client".into(), + })? + .describe_materialized_view(&self.name, &self.namespace) + .await } async fn refresh_materialized_view_async( @@ -14610,26 +14597,15 @@ mod tests { } #[tokio::test] - async fn test_materialized_view_describe_and_refresh_requires_sql() { - const QUERY: &str = - "SELECT s.x, d.label FROM analytics.source s JOIN analytics.dim d ON s.id = d.id"; - 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}"), + async fn test_materialized_view_requires_sql() { + let table = Table::new_with_handler("my_table", |request| -> http::Response { + panic!("unexpected REST request: {}", request.url().path()) }); - let view = crate::MaterializedView::from_table(table).await.unwrap(); - assert_eq!(view.definition_sql(), QUERY); - assert!(view.definition().is_err()); - let error = view.refresh().execute().await.unwrap_err(); + let error = table + .base_table() + .materialized_view_info() + .await + .unwrap_err(); assert!(error.to_string().contains("SQL is unavailable")); }