From ac7930fa772c5c51db3e0753717262ff6a69c1ab Mon Sep 17 00:00:00 2001 From: Jack Ye Date: Fri, 30 Oct 2026 11:09:54 -0700 Subject: [PATCH] feat: use SQL for remote view lifecycle Stacked on #4675. Remote view create, describe, list, and drop now execute SQL. CREATE returns the stored definition, resolution defaults, and Arrow schema from the same execution; DESCRIBE VIEW returns that metadata and SHOW VIEWS supports scoped pagination. `drop_view_async()` returns a `job_type="query"` handle whose single `job.id` is the SQL query UUID. Waiting drains the query result once and caches completion; `describe_query(job.id)` works on the submitting connection. Completion means catalog removal, with physical cleanup continuing separately. Query handles have no registry history and cannot be reopened through `open_job()`. --- docs/src/js/classes/Connection.md | 13 +- docs/web/basics/jobs.mdx | 5 + docs/web/sql/ddl/view.mdx | 26 +- nodejs/__test__/remote.test.ts | 97 +----- nodejs/lancedb/connection.ts | 13 +- python/python/lancedb/db.py | 26 +- python/python/tests/test_listing.py | 2 +- python/python/tests/test_remote_db.py | 54 ---- rust/lancedb/src/connection.rs | 26 +- rust/lancedb/src/remote/db.rs | 405 +++----------------------- rust/lancedb/src/remote/sql.rs | 1 + rust/lancedb/src/remote/sql/view.rs | 158 ++++++++++ rust/lancedb/src/remote/sql_test.rs | 213 ++++++++++++++ 13 files changed, 477 insertions(+), 562 deletions(-) create mode 100644 rust/lancedb/src/remote/sql/view.rs diff --git a/docs/src/js/classes/Connection.md b/docs/src/js/classes/Connection.md index 14f283cc9..f14afaa46 100644 --- a/docs/src/js/classes/Connection.md +++ b/docs/src/js/classes/Connection.md @@ -558,10 +558,11 @@ on the returned job to know when cleanup has finished. abstract dropView(name, namespacePath?): Promise ``` -Drop the view named `name` and wait for its definition to be deleted. +Drop the view named `name` and wait for catalog removal. The tables it reads are untouched: a view holds no rows of its own. Use -[dropViewAsync](Connection.md#dropviewasync) to retain the cleanup job instead of waiting on it. +[dropViewAsync](Connection.md#dropviewasync) to retain its SQL query job. Completion confirms +catalog removal; physical cleanup continues in the background. #### Parameters @@ -581,11 +582,9 @@ The tables it reads are untouched: a view holds no rows of its own. Use abstract dropViewAsync(name, namespacePath?): Promise ``` -Start dropping the view named `name` and return the job deleting its -definition, without waiting for completion. - -The name is free before this resolves. When nothing was bound to it, the -returned job is already finished and has no id. +Drop a view and return its SQL query job. +The job has type `query` and its ID is the SQL query UUID. Waiting confirms +catalog removal; physical cleanup continues in the background. #### Parameters diff --git a/docs/web/basics/jobs.mdx b/docs/web/basics/jobs.mdx index baa1c2ae8..7b1a8cbb6 100644 --- a/docs/web/basics/jobs.mdx +++ b/docs/web/basics/jobs.mdx @@ -132,6 +132,11 @@ the credential as a parameter to `CREATE SECRET name AS $1` or completion status or Secret names and timestamps; they never return credential values. SQL `DESCRIBE SECRET` additionally exposes rotation history. +Remote view creation, description, listing, and drop also use SQL. Creation +returns the stored definition and schema from the same SQL execution. +`drop_view_async()` returns a query job: waiting confirms catalog removal while +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. Use `describe_query(job.id)` on the submitting connection to inspect the query. diff --git a/docs/web/sql/ddl/view.mdx b/docs/web/sql/ddl/view.mdx index c6c1f7ecb..b4ec8d823 100644 --- a/docs/web/sql/ddl/view.mdx +++ b/docs/web/sql/ddl/view.mdx @@ -1,7 +1,7 @@ --- title: "View" sidebarTitle: "View" -description: "CREATE VIEW and DROP VIEW: a saved query, evaluated when read." +description: "Create, describe, list, and drop saved queries with SQL." icon: eye keywords: ["create view", "drop view", "sql", "catalog"] availability: @@ -16,8 +16,10 @@ stores rows and is brought up to date by a refresh; both are [catalog objects](/basics/catalog) defined by a query, differing only in when that query runs. -There is no SDK method for a view, because the thing it stores *is* a SQL -statement. +The SDK methods `create_view`, `describe_view`, `list_views`, and `drop_view` +submit SQL through the connection's Flight endpoint. `drop_view_async` returns +a job of type `query` whose single `id` is the SQL query UUID. Waiting confirms +catalog removal; physical cleanup can continue afterward. ## CREATE VIEW {#create} @@ -28,8 +30,10 @@ FROM "documents" AS t1 INNER JOIN "categories" AS t2 ON t1.ints = t2.ints ``` -The statement returns the view's schema (column name, data type, nullability), -resolved from the query at creation time. Anything the +The statement returns one UTF-8 `definition` column containing JSON with the +view name, namespace, original query, default database and namespace, and its +Arrow schema resolved at creation time. `DESCRIBE VIEW "recent_matches"` +returns the same metadata shape. Anything the [dialect](/sql/functions) supports can define one: joins, predicates, aggregates, window functions. @@ -56,6 +60,18 @@ was none to begin with. ## Listing {#listing} +```sql +SHOW VIEWS +SHOW VIEWS IN "analytics" WHERE name > 'previous_view' LIMIT 100 +``` + +`SHOW VIEWS` returns ascending `name` values and a nullable `location` column. +Locations are populated only when the deployment exposes catalog locations. +The `name` cursor is scoped to the selected namespace. + +Object commands retain their small results in shared storage even when general +query-result copying is disabled, so asynchronous polling can use any query node. + Views and tables share a namespace, and both are things you can query. The catalog distinguishes them: diff --git a/nodejs/__test__/remote.test.ts b/nodejs/__test__/remote.test.ts index 8d5c21aa3..6f63b0fc6 100644 --- a/nodejs/__test__/remote.test.ts +++ b/nodejs/__test__/remote.test.ts @@ -483,98 +483,6 @@ describe("remote connection", () => { expect(insertRequests).toBe(2); }); - it("creates a view and decodes the schema it resolved to", async () => { - await withMockDatabase( - (req, res) => { - expect(req.method).toBe("POST"); - expect(req.url).toBe("/v1/view/analytics$adults/create"); - res.writeHead(200, { "content-type": "application/json" }).end( - JSON.stringify({ - name: "adults", - namespace: ["analytics"], - query: "SELECT name FROM people", - // biome-ignore lint/style/useNamingConvention: the wire field is snake_case - default_database: "db", - // biome-ignore lint/style/useNamingConvention: the wire field is snake_case - default_namespace: ["analytics"], - schema: { - fields: [ - { name: "name", nullable: true, type: { type: "utf8" } }, - ], - }, - }), - ); - }, - async (db) => { - const view = await db.createView("adults", "SELECT name FROM people", [ - "analytics", - ]); - expect(view.name).toBe("adults"); - expect(view.namespacePath).toEqual(["analytics"]); - expect(view.query).toBe("SELECT name FROM people"); - expect(view.defaultDatabase).toBe("db"); - expect(view.defaultNamespacePath).toEqual(["analytics"]); - expect(view.schema.fields.map((f) => f.name)).toEqual(["name"]); - }, - ); - }); - - it("lists and drops views through their own routes", async () => { - await withMockDatabase( - (req, res) => { - expect(req.url).toBe("/v1/namespace/$/view/list"); - res - .writeHead(200, { "content-type": "application/json" }) - .end(JSON.stringify({ views: ["adults"] })); - }, - async (db) => { - expect(await collect(db.listViews())).toEqual(["adults"]); - }, - ); - - await withMockDatabase( - (req, res) => { - expect(req.method).toBe("POST"); - expect(req.url).toBe("/v1/view/adults/drop"); - res.writeHead(200, { "content-type": "application/json" }).end("{}"); - }, - async (db) => { - await db.dropView("adults"); - }, - ); - }); - - it("reports the cleanup job when a view drop is accepted", async () => { - await withMockDatabase( - (req, res) => { - expect(req.method).toBe("POST"); - expect(req.url).toBe("/v1/view/adults/drop"); - res - .writeHead(202, { "content-type": "application/json" }) - .end('{"job_id": "j1-do-abc"}'); - }, - async (db) => { - const job = await db.dropViewAsync("adults"); - expect(job.id).toBe("j1-do-abc"); - }, - ); - }); - - it("reports a finished job when a view drop had nothing to delete", async () => { - await withMockDatabase( - (req, res) => { - expect(req.url).toBe("/v1/view/adults/drop"); - res.writeHead(200, { "content-type": "application/json" }).end("{}"); - }, - async (db) => { - // A 200 means the name was not bound, so there is no cleanup to wait on. - const job = await db.dropViewAsync("adults"); - expect(job.id).toBeNull(); - await job.wait(); - }, - ); - }); - it("should accept partial connection options", async () => { await connect("db://test", { apiKey: "fake", @@ -1710,13 +1618,12 @@ async function collect(items: AsyncIterable): Promise { } describe("lazy resource listings", () => { - for (const kind of ["views", "jobs"] as const) { + for (const kind of ["jobs"] as const) { const listing = ( db: Connection, options: { pageToken?: string; pageLimit?: number } = {}, ) => { - if (kind === "jobs") return db.listJobs(options); - return db.listViews(["team"], options); + return db.listJobs(options); }; const page = (names: string[], token?: string) => { const items = diff --git a/nodejs/lancedb/connection.ts b/nodejs/lancedb/connection.ts index d5a856894..45581be5a 100644 --- a/nodejs/lancedb/connection.ts +++ b/nodejs/lancedb/connection.ts @@ -356,19 +356,18 @@ export abstract class Connection { ): Promise; /** - * Drop the view named `name` and wait for its definition to be deleted. + * Drop the view named `name` and wait for catalog removal. * * The tables it reads are untouched: a view holds no rows of its own. Use - * {@link dropViewAsync} to retain the cleanup job instead of waiting on it. + * {@link dropViewAsync} to retain its SQL query job. Completion confirms + * catalog removal; physical cleanup continues in the background. */ abstract dropView(name: string, namespacePath?: string[]): Promise; /** - * Start dropping the view named `name` and return the job deleting its - * definition, without waiting for completion. - * - * The name is free before this resolves. When nothing was bound to it, the - * returned job is already finished and has no id. + * Drop a view and return its SQL query job. + * The job has type `query` and its ID is the SQL query UUID. Waiting confirms + * catalog removal; physical cleanup continues in the background. */ abstract dropViewAsync(name: string, namespacePath?: string[]): Promise; diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index c43e71744..a2988933a 100644 --- a/python/python/lancedb/db.py +++ b/python/python/lancedb/db.py @@ -1178,10 +1178,11 @@ class DBConnection(EnforceOverrides): def drop_view( self, name: str, *, namespace_path: Optional[List[str]] = None ) -> None: - """Drop a view and wait for its definition to be deleted. + """Drop a view and wait for catalog removal. The tables it reads are untouched: a view holds no rows of its own. Use - :meth:`drop_view_async` to get the cleanup job instead of waiting on it. + :meth:`drop_view_async` to retain its SQL query job. Completion confirms + catalog removal; physical cleanup continues in the background. LanceDB Embedded connections raise ``NotImplementedError``. """ raise NotImplementedError( @@ -1191,12 +1192,11 @@ class DBConnection(EnforceOverrides): def drop_view_async( self, name: str, *, namespace_path: Optional[List[str]] = None ) -> "Job[None]": - """Start dropping a view and return the job deleting its definition. + """Drop a view and return its SQL query job. - The name is free before this returns. Call :meth:`Job.wait` to wait for - the definition dataset to be deleted. When nothing was bound to the - name, the returned job is already finished and has no id. Local - connections raise ``NotImplementedError``. + The job has type ``query`` and its ID is the SQL query UUID. Call + :meth:`Job.wait` to confirm catalog removal. Physical cleanup continues + in the background. Local connections raise ``NotImplementedError``. """ raise NotImplementedError( "View operations are not supported for this connection type" @@ -3549,10 +3549,11 @@ class AsyncConnection(object): async def drop_view( self, name: str, *, namespace_path: Optional[List[str]] = None ) -> None: - """Drop a view and wait for its definition to be deleted. + """Drop a view and wait for catalog removal. The tables it reads are untouched. Use :meth:`drop_view_async` to get - the cleanup job instead of waiting on it. + its SQL query job. Completion confirms catalog removal; physical cleanup + continues in the background. """ await self._inner.drop_view(name, list(namespace_path or [])) @@ -3562,10 +3563,11 @@ class AsyncConnection(object): *, namespace_path: Optional[List[str]] = None, ) -> AsyncJob[None]: - """Start dropping a view and return the job deleting its definition. + """Drop a view and return its SQL query job. - The name is free before this returns. Await :meth:`AsyncJob.wait` before - assuming the definition dataset is gone. + The job has type ``query`` and its ID is the SQL query UUID. Await + :meth:`AsyncJob.wait` to confirm 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 380d6fae8..cc1c44e90 100644 --- a/python/python/tests/test_listing.py +++ b/python/python/tests/test_listing.py @@ -84,7 +84,7 @@ async def advance(items, asynchronous): return await items.__anext__() if asynchronous else next(items) -KINDS = ["views", "jobs"] +KINDS = ["jobs"] @pytest.mark.asyncio diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py index b59975571..1b4af3487 100644 --- a/python/python/tests/test_remote_db.py +++ b/python/python/tests/test_remote_db.py @@ -4570,57 +4570,3 @@ def test_remote_job_handle_reports_its_own_detail(): "limit": 500, "filter": "state = 'claim_complete'", } - - -def test_view_crud_addresses_its_own_routes(): - # The view verbs are their own routes, and the schema comes back in the - # namespace spec's JSON encoding, decoded into a pyarrow schema. - paths = [] - - def handler(request): - paths.append((request.command, request.path)) - if request.path.endswith("/view/list"): - body = {"views": ["adults"]} - elif request.path.endswith("/drop"): - body = {} - else: - body = { - "name": "adults", - "namespace": ["analytics"], - "query": "SELECT name FROM people", - "default_database": "dev", - "default_namespace": ["analytics"], - "schema": { - "fields": [ - {"name": "name", "nullable": True, "type": {"type": "utf8"}} - ] - }, - } - request.send_response(200) - request.send_header("Content-Type", "application/json") - request.end_headers() - request.wfile.write(json.dumps(body).encode()) - - with mock_lancedb_connection(handler) as db: - view = db.create_view( - "adults", "SELECT name FROM people", namespace_path=["analytics"] - ) - assert view.name == "adults" - assert view.namespace_path == ["analytics"] - assert view.query == "SELECT name FROM people" - assert view.default_database == "dev" - assert view.default_namespace_path == ["analytics"] - assert view.schema == pa.schema([pa.field("name", pa.utf8(), nullable=True)]) - - described = db.describe_view("adults", namespace_path=["analytics"]) - assert described.schema == view.schema - - assert list(db.list_views(namespace_path=["analytics"])) == ["adults"] - db.drop_view("adults", namespace_path=["analytics"]) - - assert paths == [ - ("POST", "/v1/view/analytics$adults/create"), - ("POST", "/v1/view/analytics$adults/describe"), - ("GET", "/v1/namespace/analytics/view/list"), - ("POST", "/v1/view/analytics$adults/drop"), - ] diff --git a/rust/lancedb/src/connection.rs b/rust/lancedb/src/connection.rs index c7559f999..d0f8c2f51 100644 --- a/rust/lancedb/src/connection.rs +++ b/rust/lancedb/src/connection.rs @@ -986,31 +986,23 @@ impl Connection { .await } - /// Drop a view and wait for its definition to be deleted. + /// Drop a view and wait for catalog removal. /// /// The tables it reads are untouched: a view holds no rows of its own. - /// Use [`Connection::drop_view_async`] to get the cleanup job instead of - /// waiting on it. LanceDB Embedded databases return [`Error::NotSupported`]. + /// Use [`Connection::drop_view_async`] to retain its SQL query job. + /// Completion confirms catalog removal; physical cleanup continues + /// separately. LanceDB Embedded databases return [`Error::NotSupported`]. pub async fn drop_view(&self, name: impl AsRef, namespace_path: &[String]) -> Result<()> { validate_view_reference(name.as_ref(), namespace_path)?; self.internal.drop_view(name.as_ref(), namespace_path).await } - /// Start dropping a view and return the job deleting its definition. + /// Drop a view and return its SQL query job. /// - /// The name is free before this returns; the definition dataset may still - /// be being deleted. Await [`Job::wait`][crate::job::Job::wait] to wait for - /// that. When nothing was bound to the name, the returned job is already - /// finished and has no id. LanceDB Embedded databases return [`Error::NotSupported`]. - /// - /// ```no_run - /// # use lancedb::Connection; - /// # async fn drop(conn: &Connection) -> lancedb::Result<()> { - /// let job = conn.drop_view_async("recent_orders", &[]).await?; - /// job.wait().await?; - /// # Ok(()) - /// # } - /// ``` + /// The job has type `query` and its ID is the SQL query UUID. Await + /// [`Job::wait`][crate::job::Job::wait] to confirm catalog removal; physical + /// cleanup continues separately. LanceDB Embedded databases return + /// [`Error::NotSupported`]. pub async fn drop_view_async( &self, name: impl AsRef, diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs index 22db05b06..b50c355d5 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/rust/lancedb/src/remote/db.rs @@ -15,8 +15,8 @@ use reqwest::header::CONTENT_TYPE; use lance_namespace::models::{ CreateNamespaceRequest, CreateNamespaceResponse, DescribeNamespaceRequest, - DescribeNamespaceResponse, DropNamespaceRequest, DropNamespaceResponse, JsonArrowSchema, - ListNamespacesRequest, ListNamespacesResponse, ListTablesRequest, ListTablesResponse, + DescribeNamespaceResponse, DropNamespaceRequest, DropNamespaceResponse, ListNamespacesRequest, + ListNamespacesResponse, ListTablesRequest, ListTablesResponse, }; use crate::Error; @@ -652,14 +652,6 @@ fn build_namespace_identifier(namespace: &[String]) -> Result { } } -/// Build a full path to a view. Note that we do not need to percent-encode -/// this when using it in a URL because the validation functions ensure that -/// only URL-safe characters are present. -fn build_view_identifier(name: &str, namespace: &[String]) -> Result { - validate_view_name(name)?; - build_compound_identifier(name, namespace) -} - /// Build a secure cache key using length prefixes. /// This format is completely unambiguous regardless of delimiter or content. /// Format: [u32_len][namespace1][u32_len][namespace2]...[u32_len][table_name] @@ -704,59 +696,6 @@ struct RemoteListJobsResponse { page_token: Option, } -/// Define a view from a query. The name and its namespace are the path -/// identifier, so neither appears here. -#[derive(serde::Serialize)] -struct RemoteCreateViewRequest<'a> { - query: &'a str, -} - -/// What the service reports about one view. The schema arrives as the -/// namespace spec's JSON encoding, which is what `describe_table` uses too. -#[derive(serde::Deserialize)] -struct RemoteViewDescription { - name: String, - #[serde(default)] - namespace: Vec, - query: String, - default_database: String, - /// A path, like `namespace`: the root is the absent field rather than a - /// spelling of its own. - #[serde(default)] - default_namespace: Vec, - schema: JsonArrowSchema, -} - -impl RemoteViewDescription { - fn into_description(self, request_id: String) -> Result { - let schema = - lance_namespace::schema::convert_json_arrow_schema(&self.schema).map_err(|source| { - Error::Http { - source: format!("View '{}' has an undecodable schema: {source}", self.name) - .into(), - request_id, - status_code: None, - } - })?; - Ok(ViewDescription { - name: self.name, - namespace_path: self.namespace, - query: self.query, - default_database: self.default_database, - default_namespace_path: self.default_namespace, - schema: Arc::new(schema), - }) - } -} - -#[derive(serde::Deserialize)] -struct RemoteListViewsResponse { - #[serde(default)] - views: Vec, - #[serde(default)] - page_token: Option, -} - #[async_trait] impl Database for RemoteDatabase { fn uri(&self) -> &str { @@ -1052,90 +991,45 @@ impl Database for RemoteDatabase { query: &str, namespace_path: &[String], ) -> Result { - let view_id = build_view_identifier(name, namespace_path)?; - let req = self - .client - .post(&format!("/v1/view/{view_id}/create")) - .json(&RemoteCreateViewRequest { query }); - let (request_id, response) = self.client.send(req).await?; - let response = self.client.check_response(&request_id, response).await?; - let description: RemoteViewDescription = - response.json().await.err_to_http(request_id.clone())?; - description.into_description(request_id) + validate_view_name(name)?; + validate_namespace(namespace_path)?; + self.sql_client()? + .create_view(name, query, namespace_path) + .await } - async fn describe_view( &self, name: &str, namespace_path: &[String], ) -> Result { - let view_id = build_view_identifier(name, namespace_path)?; - let req = self.client.post(&format!("/v1/view/{view_id}/describe")); - let (request_id, response) = self.client.send(req).await?; - let response = self.client.check_response(&request_id, response).await?; - let description: RemoteViewDescription = - response.json().await.err_to_http(request_id.clone())?; - description.into_description(request_id) + validate_view_name(name)?; + validate_namespace(namespace_path)?; + self.sql_client()?.describe_view(name, namespace_path).await } - async fn drop_view(&self, name: &str, namespace_path: &[String]) -> Result<()> { self.drop_view_async(name, namespace_path) .await? .wait() .await } - async fn drop_view_async(&self, name: &str, namespace_path: &[String]) -> Result { - let view_id = build_view_identifier(name, namespace_path)?; - let req = self.client.post(&format!("/v1/view/{view_id}/drop")); - let (request_id, response) = self.client.send(req).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())?; - match status { - // Nothing was bound to the name, so nothing is being deleted. - StatusCode::OK => Ok(Job::new_done()), - StatusCode::ACCEPTED => { - let job_id = extract_job_id(&body).ok_or_else(|| Error::Http { - source: "view drop response did not contain a valid job_id".into(), - request_id, - status_code: Some(status), - })?; - Ok(Job::new(Box::new(RemoteJob::new( - self.client.clone(), - job_id, - )))) - } - _ => Err(Error::Http { - source: "view drop must return 200 OK or 202 Accepted".into(), - request_id, - status_code: Some(status), - }), - } + validate_view_name(name)?; + validate_namespace(namespace_path)?; + self.sql_client()? + .submit_as_job( + format!("DROP VIEW {}", quote_identifier(name)), + namespace_path.to_vec(), + || async { Ok(()) }, + ) + .await } - async fn list_views( &self, namespace_path: &[String], options: ListingOptions, ) -> Result> { - let namespace_id = build_namespace_identifier(namespace_path)?; - let path = format!("/v1/namespace/{namespace_id}/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, - }) + validate_namespace(namespace_path)?; + self.sql_client()?.list_views(namespace_path, options).await } async fn open_job(&self, job_id: &str) -> Result { @@ -1916,26 +1810,6 @@ mod tests { assert_eq!(key1, key6, "Same inputs should produce same cache key"); } - #[tokio::test] - async fn test_create_materialized_view_validates_before_submission() { - let db = super::RemoteDatabase::new_mock(|_| -> http::Response { - panic!("materialized-view creation must not send REST requests") - }); - for seconds in [0.0, -1.0, f64::NAN, f64::INFINITY] { - let result = db - .create_materialized_view_async(CreateMaterializedViewRequest { - name: "adults".into(), - namespace_path: vec![], - query: "SELECT * FROM people".into(), - with_no_data: false, - auto_refresh_enabled: None, - auto_refresh_interval_seconds: Some(seconds), - }) - .await; - assert!(matches!(result, Err(Error::InvalidInput { .. }))); - } - } - #[tokio::test] async fn test_retries() { // We'll record the request_id here, to check it matches the one in the error. @@ -3617,7 +3491,7 @@ mod tests { assert!(items.try_next().await.unwrap().is_none()); assert!(items.try_next().await.unwrap().is_none()); } - for resource in ["view", "jobs"] { + for resource in ["jobs"] { let count = Arc::new(AtomicUsize::new(0)); let conn = Connection::new_with_handler({ let count = count.clone(); @@ -4101,222 +3975,6 @@ mod tests { )); } - /// A view description carries the schema in the namespace spec's JSON - /// encoding, so the body a test asserts against is the one that encoding - /// produces rather than a hand-written guess at it. - fn view_description_body(name: &str, namespace: &[&str], query: &str) -> String { - let schema = Schema::new(vec![Field::new("id", DataType::Int32, true)]); - serde_json::json!({ - "name": name, - "namespace": namespace, - "query": query, - "default_database": "db", - "default_namespace": namespace, - "schema": lance_namespace::schema::arrow_schema_to_json(&schema).unwrap(), - }) - .to_string() - } - - #[tokio::test] - async fn test_create_view_posts_the_query_and_returns_the_resolved_schema() { - let conn = Connection::new_with_handler(|request| { - assert_eq!(request.method(), &reqwest::Method::POST); - assert_eq!(request.url().path(), "/v1/view/analytics$adults/create"); - let body: serde_json::Value = - serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap(); - // The name and its namespace address the view in the path, so the - // body says only what they cannot. - assert_eq!(body, serde_json::json!({"query": "SELECT id FROM people"})); - http::Response::builder() - .status(200) - .body(view_description_body( - "adults", - &["analytics"], - "SELECT id FROM people", - )) - .unwrap() - }); - let view = conn - .create_view("adults", "SELECT id FROM people", &["analytics".into()]) - .await - .unwrap(); - assert_eq!(view.name, "adults"); - assert_eq!(view.namespace_path, vec!["analytics".to_string()]); - assert_eq!(view.query, "SELECT id FROM people"); - assert_eq!(view.default_database, "db"); - assert_eq!(view.default_namespace_path, vec!["analytics".to_string()]); - assert_eq!(view.schema.field(0).name(), "id"); - } - - #[tokio::test] - async fn test_describe_and_drop_address_the_view_in_the_path() { - let conn = Connection::new_with_handler(|request| { - assert_eq!(request.url().path(), "/v1/view/adults/describe"); - assert!(request.body().is_none(), "{:?}", request.body()); - http::Response::builder() - .status(200) - .body(view_description_body( - "adults", - &[], - "SELECT id FROM people", - )) - .unwrap() - }); - let view = conn.describe_view("adults", &[]).await.unwrap(); - assert!(view.namespace_path.is_empty()); - assert_eq!(view.schema.fields().len(), 1); - - let conn = Connection::new_with_handler(|request| { - assert_eq!(request.method(), &reqwest::Method::POST); - assert_eq!(request.url().path(), "/v1/view/analytics$adults/drop"); - assert!(request.body().is_none(), "{:?}", request.body()); - http::Response::builder().status(200).body("{}").unwrap() - }); - conn.drop_view("adults", &["analytics".into()]) - .await - .unwrap(); - } - - /// An accepted drop hands back the job so a caller can wait on the delete, - /// and the waiting `drop_view` does that for them. - #[tokio::test] - async fn test_drop_view_async_reports_the_cleanup_job() { - let db = super::RemoteDatabase::new_mock(|_| { - http::Response::builder() - .status(202) - .body(r#"{"job_id":"j1-do-abc"}"#) - .unwrap() - }); - let job = db.drop_view_async("adults", &[]).await.unwrap(); - assert_eq!(job.id(), Some("j1-do-abc")); - } - - /// Nothing was bound, so nothing is being deleted and the job is already done. - #[tokio::test] - async fn test_drop_view_async_reports_a_finished_job_when_nothing_was_bound() { - let db = super::RemoteDatabase::new_mock(|_| { - http::Response::builder().status(200).body("{}").unwrap() - }); - let job = db.drop_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_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_view_async("adults", &[]).await.err().unwrap(); - assert!(error.to_string().contains("valid job_id"), "{error}"); - } - } - - #[tokio::test] - async fn test_drop_view_rejects_unexpected_success_status() { - let db = super::RemoteDatabase::new_mock(|_| { - http::Response::builder().status(204).body("").unwrap() - }); - let error = db.drop_view_async("adults", &[]).await.err().unwrap(); - assert!( - error.to_string().contains("200 OK or 202 Accepted"), - "{error}" - ); - } - - /// A schema the client cannot decode is a broken response, not a view - /// with no columns: reporting it as an error keeps a caller from reading - /// an empty schema as the truth about the view. - #[tokio::test] - async fn test_describe_view_rejects_an_undecodable_schema() { - let conn = Connection::new_with_handler(|_| { - http::Response::builder() - .status(200) - .body( - r#"{"name":"adults","query":"SELECT 1","default_database":"db", - "schema":{"fields":[{"name":"id","type":{"type":"nonesuch"}, - "nullable":true}]}}"#, - ) - .unwrap() - }); - let error = conn.describe_view("adults", &[]).await.unwrap_err(); - assert!(error.to_string().contains("undecodable schema"), "{error}"); - } - - #[tokio::test] - async fn test_list_views_walks_pages() { - let conn = Connection::new_with_handler(|request| { - assert_eq!(request.method(), &reqwest::Method::GET); - assert_eq!(request.url().path(), "/v1/namespace/analytics/view/list"); - let page = request - .url() - .query_pairs() - .find(|(key, _)| key == "page_token") - .map(|(_, value)| value.into_owned()); - let body = match page.as_deref() { - None => r#"{"views":[],"page_token":"p2"}"#, - Some("p2") => r#"{"views":["adults"]}"#, - Some(other) => panic!("unexpected page token: {other}"), - }; - http::Response::builder().status(200).body(body).unwrap() - }); - assert_eq!( - conn.list_views(&["analytics".into()], Default::default()) - .try_collect::>() - .await - .unwrap(), - vec!["adults".to_string()] - ); - } - - /// A server that keeps handing back the same token would otherwise spin - /// forever. - #[tokio::test] - async fn test_list_views_rejects_a_repeated_page_token() { - let conn = Connection::new_with_handler(|_| { - http::Response::builder() - .status(200) - .body(r#"{"views":["adults"],"page_token":"same"}"#) - .unwrap() - }); - let error = conn - .list_views(&[], Default::default()) - .try_collect::>() - .await - .unwrap_err(); - assert!( - error.to_string().contains("repeated a page_token"), - "{error}" - ); - } - - /// A name carrying the delimiter would split back apart as a different - /// view, so it is refused before it reaches a route. - #[tokio::test] - async fn test_view_names_that_would_resplit_are_refused() { - let conn = Connection::new_with_handler(|_| -> http::Response { - panic!("an invalid identifier must not reach the service") - }); - for (name, namespace) in [ - ("analytics$adults", vec![]), - ("adults", vec!["ana$lytics".to_string()]), - ("", vec![]), - ("..", vec![]), - ] { - let error = match conn.describe_view(name, &namespace).await { - Ok(_) => panic!("accepted {name:?} in {namespace:?}"), - Err(error) => error.to_string(), - }; - // A view is not a table, so the refusal says so. - assert!( - error.contains("view name") || error.contains("namespace component"), - "{error}" - ); - } - } - #[tokio::test] async fn test_conn_job_waits_to_done() { let polls = Arc::new(AtomicUsize::new(0)); @@ -4347,4 +4005,23 @@ mod tests { assert_eq!(job.status().await.unwrap(), "finished"); assert!(polls.load(Ordering::SeqCst) >= 4); } + #[tokio::test] + async fn test_create_materialized_view_validates_before_submission() { + let db = super::RemoteDatabase::new_mock(|_| -> http::Response { + panic!("materialized-view creation must not send REST requests") + }); + for seconds in [0.0, -1.0, f64::NAN, f64::INFINITY] { + let result = db + .create_materialized_view_async(CreateMaterializedViewRequest { + name: "adults".into(), + namespace_path: vec![], + query: "SELECT * FROM people".into(), + with_no_data: false, + auto_refresh_enabled: None, + auto_refresh_interval_seconds: Some(seconds), + }) + .await; + assert!(matches!(result, Err(Error::InvalidInput { .. }))); + } + } } diff --git a/rust/lancedb/src/remote/sql.rs b/rust/lancedb/src/remote/sql.rs index ebe1cab49..7e3a42ac7 100644 --- a/rust/lancedb/src/remote/sql.rs +++ b/rust/lancedb/src/remote/sql.rs @@ -40,6 +40,7 @@ pub(super) mod materialized_view; mod objects; mod secret; mod stream; +mod view; use stream::{StreamingQuery, StreamingQueryHandle, Submission}; diff --git a/rust/lancedb/src/remote/sql/view.rs b/rust/lancedb/src/remote/sql/view.rs new file mode 100644 index 000000000..993d7dc56 --- /dev/null +++ b/rust/lancedb/src/remote/sql/view.rs @@ -0,0 +1,158 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +use super::{ + SqlClient, + materialized_view::quote_identifier, + objects::{definition, strings}, +}; +use crate::{ + database::ExecuteQueryRequest, + error::{Error, Result}, + listing::{ListingOptions, ListingPage}, + utils::{validate_namespace, validate_view_name}, + view::ViewDescription, +}; +use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD}; +use futures::TryStreamExt; +use lance_namespace::models::JsonArrowSchema; +use std::sync::Arc; +#[derive(Clone, serde::Serialize, serde::Deserialize)] +struct ViewDefinition { + name: String, + #[serde(default)] + namespace: Vec, + query: String, + default_database: String, + #[serde(default)] + default_namespace: Vec, + schema: JsonArrowSchema, +} +impl ViewDefinition { + fn into_description(self) -> Result { + let schema = + lance_namespace::schema::convert_json_arrow_schema(&self.schema).map_err(|error| { + Error::Runtime { + message: format!("View '{}' has an undecodable schema: {error}", self.name), + } + })?; + Ok(ViewDescription { + name: self.name, + namespace_path: self.namespace, + query: self.query, + default_database: self.default_database, + default_namespace_path: self.default_namespace, + schema: Arc::new(schema), + }) + } +} +impl SqlClient { + pub(in crate::remote) async fn create_view( + &self, + name: &str, + query: &str, + namespace: &[String], + ) -> Result { + validate_view_name(name)?; + validate_namespace(namespace)?; + let definition: ViewDefinition = self + .object_job( + ExecuteQueryRequest { + query: format!("CREATE VIEW {} AS {query}", quote_identifier(name)), + default_namespace_path: namespace.to_vec(), + parameters: None, + }, + definition, + ) + .await? + .wait() + .await?; + definition.into_description() + } + pub(in crate::remote) async fn describe_view( + &self, + name: &str, + namespace: &[String], + ) -> Result { + validate_view_name(name)?; + validate_namespace(namespace)?; + let definition: ViewDefinition = self + .object_job( + ExecuteQueryRequest { + query: format!("DESCRIBE VIEW {}", quote_identifier(name)), + default_namespace_path: namespace.to_vec(), + parameters: None, + }, + definition, + ) + .await? + .wait() + .await?; + definition.into_description() + } + pub(in crate::remote) async fn list_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 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 (kind, database, prefix, path, last): ( + String, + String, + Option, + Vec, + String, + ) = serde_json::from_slice(&decoded).map_err(|_| invalid_token())?; + if kind != "view" + || 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 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(&( + "view", + &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 61f807d37..3a701d2e6 100644 --- a/rust/lancedb/src/remote/sql_test.rs +++ b/rust/lancedb/src/remote/sql_test.rs @@ -14,6 +14,7 @@ use arrow_flight::{ FlightInfo, HandshakeRequest, HandshakeResponse, PollInfo, PutResult, SchemaResult, Ticket, }; use arrow_schema::{DataType, Field, Schema}; +use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD}; use futures::stream::BoxStream; use futures::{StreamExt, TryStreamExt}; use tonic::metadata::MetadataMap; @@ -21,6 +22,7 @@ use tonic::{Request, Response, Status, Streaming}; use super::stream::{PROTOCOL_HEADER, PROTOCOL_VERSION, QueryResponse}; use super::*; +use crate::Connection; use crate::database::Database; use crate::materialized_view::CreateMaterializedViewRequest; use crate::remote::RemoteCatalogOptions; @@ -2674,3 +2676,214 @@ async fn function_listing_uses_bounded_sql_and_scoped_version_cursors() { assert!(seen[1].contains("after_version = '1'")); shutdown.send(()).unwrap(); } + +#[tokio::test] +async fn view_create_and_describe_return_metadata_through_sql() { + let definition = r#"{"name":"adults","namespace":["team"],"query":"SELECT name FROM people","default_database":"db","default_namespace":["team"],"schema":{"fields":[{"name":"name","nullable":true,"type":{"type":"utf8"}}]}}"#; + for defer in [false, true] { + let result = RecordBatch::try_from_iter(vec![( + "definition", + 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 db = RemoteDatabase::new_mock(|_| -> http::Response { + panic!("unexpected REST request"); + }) + .with_sql_client(client); + let created = db + .create_view("adults", "SELECT name FROM people", &["team".into()]) + .await + .unwrap(); + let described = db.describe_view("adults", &["team".into()]).await.unwrap(); + assert_eq!(created, described); + assert_eq!(created.schema.fields()[0].name(), "name"); + assert_eq!(created.default_namespace_path, ["team"]); + assert_eq!( + statements.lock().unwrap().as_slice(), + [ + "CREATE VIEW \"adults\" AS SELECT name FROM people", + "DESCRIBE VIEW \"adults\"" + ] + ); + shutdown.send(()).unwrap(); + } +} +#[tokio::test] +async fn view_drop_uses_a_repeatable_query_job() { + let service = TestSqlService { + adaptive: true, + defer: true, + ..Default::default() + }; + let statements = service.statements.clone(); + let (client, shutdown) = start_sql_service(service, ClientConfig::default()).await; + let db = RemoteDatabase::new_mock(|_| -> http::Response { + panic!("unexpected REST request"); + }) + .with_sql_client(client.clone()); + let job = db + .drop_view_async("adults", &["team".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().as_slice(), + ["DROP VIEW \"adults\""] + ); + shutdown.send(()).unwrap(); +} + +#[tokio::test] +async fn view_listing_uses_bounded_sql_and_scoped_cursors() { + for defer in [false, true] { + let first_result = RecordBatch::try_from_iter(vec![( + "name", + Arc::new(StringArray::from(vec!["adults", "hf'prod", "openai-prod"])) as ArrayRef, + )]) + .unwrap(); + let second_result = RecordBatch::try_from_iter(vec![( + "name", + Arc::new(StringArray::from(vec!["openai-prod"])) as ArrayRef, + )]) + .unwrap(); + let result_overrides = Arc::new(std::sync::Mutex::new(HashMap::from([ + ("SHOW VIEWS LIMIT 3".to_string(), first_result), + ( + "SHOW VIEWS WHERE name > 'hf''prod' LIMIT 3".to_string(), + second_result, + ), + ]))); + let service = TestSqlService { + adaptive: true, + defer, + result_overrides, + ..Default::default() + }; + let statements = service.statements.clone(); + let headers = service.headers.clone(); + let (client, shutdown) = start_sql_service_with_prefix( + service, + ClientConfig::default(), + Some("tenant/production"), + ) + .await; + let db = RemoteDatabase::new_mock(|_| -> http::Response { + panic!("unexpected REST request"); + }) + .with_sql_client(client); + let connection = Connection::new( + Arc::new(db), + Arc::new(crate::embeddings::MemoryRegistry::new()), + ); + let namespace = vec!["team".to_owned(), "reports".to_owned()]; + + let first = connection + .list_views( + &namespace, + crate::listing::ListingOptions::default().page_limit(2), + ) + .next_page() + .await + .unwrap() + .unwrap(); + assert_eq!(first.items, ["adults", "hf'prod"]); + let token = first.page_token.expect("a third row must produce a token"); + + let second = connection + .list_views( + &namespace, + crate::listing::ListingOptions::default() + .page_limit(2) + .page_token(&token), + ) + .next_page() + .await + .unwrap() + .unwrap(); + assert_eq!(second.items, ["openai-prod"]); + assert!(second.page_token.is_none()); + + assert_eq!( + *statements.lock().unwrap(), + [ + "SHOW VIEWS LIMIT 3", + "SHOW VIEWS WHERE name > 'hf''prod' LIMIT 3" + ] + ); + { + let headers = headers.lock().unwrap(); + assert!(!headers.is_empty()); + for header in headers.iter() { + assert_eq!(header.database, "analytics"); + assert_eq!(header.database_prefix, "tenant/production"); + assert_eq!(header.namespace_path, "team$reports"); + } + } + + let encode_token = + |kind: &str, database: &str, prefix: Option<&str>, path: &[String], last: &str| { + URL_SAFE_NO_PAD.encode( + serde_json::to_vec(&(kind, database, prefix, path.to_vec(), last)).unwrap(), + ) + }; + for forged in [ + encode_token( + "secret", + "analytics", + Some("tenant/production"), + &namespace, + "hf'prod", + ), + encode_token( + "view", + "billing", + Some("tenant/production"), + &namespace, + "hf'prod", + ), + encode_token("view", "analytics", Some("other"), &namespace, "hf'prod"), + encode_token( + "view", + "analytics", + Some("tenant/production"), + &["other".to_owned()], + "hf'prod", + ), + "not-a-json-token".to_string(), + ] { + let mut listing = connection.list_views( + &namespace, + crate::listing::ListingOptions::default().page_token(forged), + ); + assert!(matches!( + listing.next_page().await, + Err(Error::InvalidInput { .. }) + )); + } + let mut listing = connection.list_views( + &["other".to_owned()], + crate::listing::ListingOptions::default().page_token(&token), + ); + assert!(matches!( + listing.next_page().await, + Err(Error::InvalidInput { .. }) + )); + assert_eq!(statements.lock().unwrap().len(), 2); + shutdown.send(()).unwrap(); + } +}