rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-11-13 14:16:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: use SQL for remote view lifecycle

Browse Source 
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()`.
This commit is contained in:
Jack Ye
2026-10-31 02:09:54 +08:00
committed by GitHub
parent 45e00d4de7
commit ac7930fa77
13 changed files with 477 additions and 562 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/src/js/classes/Connection.md
+6 -7
View File
@@ -558,10 +558,11 @@ on the returned job to know when cleanup has finished.
abstract dropView(name, namespacePath?): Promise<void>
```
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<Job>
```
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
docs/web/basics/jobs.mdx
+5
View File
@@ -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.
docs/web/sql/ddl/view.mdx
+21 -5
View File
@@ -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:
nodejs/__test__/remote.test.ts
+2 -95
View File
@@ -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<T>(items: AsyncIterable<T>): Promise<T[]> {
}
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 =
nodejs/lancedb/connection.ts
+6 -7
View File
@@ -356,19 +356,18 @@ export abstract class Connection {
): Promise<ViewDescription>;
/**
* 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<void>;
/**
* 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<Job>;
python/python/lancedb/db.py
+14 -12
View File
@@ -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 = []
python/python/tests/test_listing.py
+1 -1
View File
@@ -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
python/python/tests/test_remote_db.py
-54
View File
@@ -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"),
]
rust/lancedb/src/connection.rs
+9 -17
View File
@@ -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<str>, 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<str>,
rust/lancedb/src/remote/db.rs
+41 -364
View File
@@ -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<String> {
}
}
/// 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<String> {
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<String>,
}
/// 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<String>,
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<String>,
schema: JsonArrowSchema,
}
impl RemoteViewDescription {
fn into_description(self, request_id: String) -> Result<ViewDescription> {
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<String>,
#[serde(default)]
page_token: Option<String>,
}
#[async_trait]
impl<S: HttpSend> Database for RemoteDatabase<S> {
fn uri(&self) -> &str {
@@ -1052,90 +991,45 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
query: &str,
namespace_path: &[String],
) -> Result<ViewDescription> {
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<ViewDescription> {
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<Job> {
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<ListingPage<String>> {
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<Job> {
@@ -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<String> {
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::<Vec<_>>()
.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::<Vec<_>>()
.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<String> {
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<String> {
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 { .. })));
}
}
}
rust/lancedb/src/remote/sql.rs
+1
View File
@@ -40,6 +40,7 @@ pub(super) mod materialized_view;
mod objects;
mod secret;
mod stream;
mod view;
use stream::{StreamingQuery, StreamingQueryHandle, Submission};
rust/lancedb/src/remote/sql/view.rs
+158
View File
@@ -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<String>,
query: String,
default_database: String,
#[serde(default)]
default_namespace: Vec<String>,
schema: JsonArrowSchema,
}
impl ViewDefinition {
fn into_description(self) -> Result<ViewDescription> {
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<ViewDescription> {
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<ViewDescription> {
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<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 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<String>,
Vec<String>,
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 })
}
}
rust/lancedb/src/remote/sql_test.rs
+213
View File
@@ -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<String> {
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<String> {
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<String> {
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();
}
}