rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-10-29 14:56:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

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

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

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

---------

Co-authored-by: Xuanwo <github@xuanwo.io>
This commit is contained in:
Jack Ye
2026-10-28 04:37:00 +08:00
committed by GitHub
co-authored by Xuanwo
parent f6a348ac5b
commit 45dbf5eff5
13 changed files with 420 additions and 375 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/src/js/classes/Connection.md
+5 -4
View File
@@ -436,8 +436,8 @@ abstract dropMaterializedView(name, namespacePath?): Promise<void>
Drop the materialized view named `name`.
The view may become unavailable before physical cleanup finishes. Use
[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<Job>
```
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.
docs/web/basics/jobs.mdx
+6 -4
View File
@@ -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.
nodejs/__test__/remote.test.ts
+3 -25
View File
@@ -481,23 +481,6 @@ describe("remote connection", () => {
expect(insertRequests).toBe(2);
});
it("lists materialized views through the namespace route", async () => {
await withMockDatabase(
(req, res) => {
expect(req.method).toBe("GET");
expect(req.url).toBe("/v1/namespace/$/materialized_view/list");
res
.writeHead(200, { "content-type": "application/json" })
.end(JSON.stringify({ views: ["daily_sales"] }));
},
async (db) => {
expect(await collect(db.listMaterializedViews())).toEqual([
"daily_sales",
]);
},
);
});
it("creates a view and decodes the schema it resolved to", async () => {
await withMockDatabase(
(req, res) => {
@@ -1725,14 +1708,13 @@ async function collect<T>(items: AsyncIterable<T>): Promise<T[]> {
}
describe("lazy resource listings", () => {
for (const kind of ["views", "materializedViews", "jobs"] as const) {
for (const kind of ["views", "jobs"] as const) {
const listing = (
db: Connection,
options: { pageToken?: string; pageLimit?: number } = {},
) => {
if (kind === "jobs") return db.listJobs(options);
if (kind === "views") return db.listViews(["team"], options);
return db.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");
}
nodejs/lancedb/connection.ts
+5 -4
View File
@@ -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<void>;
/**
* 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.
*/
python/python/lancedb/db.py
+12 -13
View File
@@ -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 = []
python/python/tests/test_listing.py
+4 -6
View File
@@ -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"]}
python/python/tests/test_materialized_views.py
+36 -100
View File
@@ -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
rust/lancedb/src/materialized_view.rs
+9 -8
View File
@@ -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<str>,
@@ -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;
rust/lancedb/src/remote/db.rs
+30 -169
View File
@@ -47,6 +47,7 @@ use super::client::{
Sender,
};
use super::sql::SqlClient;
use super::sql::materialized_view::quote_identifier;
use super::table::RemoteTable;
use super::util::parse_server_version;
use super::{ARROW_STREAM_CONTENT_TYPE, extract_job_id};
@@ -929,36 +930,26 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
name: &str,
namespace_path: &[String],
) -> Result<Job> {
let identifier = build_table_identifier(name, namespace_path)?;
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<S: HttpSend> Database for RemoteDatabase<S> {
namespace_path: &[String],
options: ListingOptions,
) -> Result<ListingPage<String>> {
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::<Vec<_>>()
.await
.unwrap(),
["adults"]
);
}
#[tokio::test]
async fn test_retries() {
// We'll record the request_id here, to check it matches the one in the error.
@@ -3912,7 +3778,7 @@ mod tests {
assert!(items.try_next().await.unwrap().is_none());
assert!(items.try_next().await.unwrap().is_none());
}
for resource in ["secret", "view", "function", "materialized_view", "jobs"] {
for resource in ["secret", "view", "function", "jobs"] {
let count = Arc::new(AtomicUsize::new(0));
let conn = Connection::new_with_handler({
let count = count.clone();
@@ -3927,11 +3793,7 @@ mod tests {
.unwrap();
assert_eq!(body, serde_json::json!({"page_token": token, "limit": 2}));
} else {
let ns = if resource == "materialized_view" {
"$"
} 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,
}
}
rust/lancedb/src/remote/sql.rs
+1
View File
@@ -35,6 +35,7 @@ use crate::remote::client::{ClientConfig, TlsConfig};
use crate::remote::retry::ResolvedRetryConfig;
use crate::sql::{Query, QueryDescription, QueryHandle, QueryStatus};
pub(super) mod materialized_view;
mod stream;
use stream::{StreamingQuery, StreamingQueryHandle, Submission};
rust/lancedb/src/remote/sql/materialized_view.rs
+129
View File
@@ -0,0 +1,129 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
use arrow_array::{Array, RecordBatch, StringArray};
use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
use futures::TryStreamExt;
use super::SqlClient;
use crate::database::ExecuteQueryRequest;
use crate::error::{Error, Result};
use crate::listing::{ListingOptions, ListingPage};
use crate::materialized_view::{MaterializedViewDefinition, MaterializedViewInfo};
use crate::utils::{validate_namespace, validate_table_name};
pub(in crate::remote) fn quote_identifier(name: &str) -> String {
format!("\"{}\"", name.replace('"', "\"\""))
}
fn strings(batch: &RecordBatch, name: &str) -> Result<Vec<String>> {
let values = batch
.column_by_name(name)
.and_then(|column| column.as_any().downcast_ref::<StringArray>())
.ok_or_else(|| Error::Runtime {
message: format!("materialized-view SQL result is missing UTF-8 column '{name}'"),
})?;
if values.null_count() != 0 {
return Err(Error::Runtime {
message: format!("materialized-view SQL result contains null '{name}'"),
});
}
Ok(values
.iter()
.map(|value| value.unwrap().to_owned())
.collect())
}
impl SqlClient {
pub(in crate::remote) async fn describe_materialized_view(
&self,
name: &str,
namespace: &[String],
) -> Result<MaterializedViewInfo> {
validate_table_name(name)?;
validate_namespace(namespace)?;
let query = self
.execute(ExecuteQueryRequest {
query: format!("DESCRIBE MATERIALIZED VIEW {}", quote_identifier(name)),
default_namespace_path: namespace.to_vec(),
parameters: None,
})
.await?;
let mut reader = query.reader().await?;
let mut definitions = Vec::new();
while let Some(batch) = reader.try_next().await? {
definitions.extend(strings(&batch, "query")?);
}
let [definition_sql] =
<[String; 1]>::try_from(definitions).map_err(|_| Error::Runtime {
message: "DESCRIBE MATERIALIZED VIEW must return exactly one definition".into(),
})?;
let parsed_definition = MaterializedViewDefinition::from_sql(&definition_sql).ok();
Ok(MaterializedViewInfo {
definition_sql,
parsed_definition,
})
}
pub(in crate::remote) async fn list_materialized_views(
&self,
namespace: &[String],
options: ListingOptions,
) -> Result<ListingPage<String>> {
validate_namespace(namespace)?;
let limit = options.page_limit.unwrap_or(100);
if limit == 0 || limit > i32::MAX as u32 {
return Err(Error::InvalidInput {
message: "page limit must be between 1 and i32::MAX".into(),
});
}
let invalid_token = || Error::InvalidInput {
message: "invalid materialized-view page token for this database or namespace".into(),
};
let after = if let Some(token) = options.page_token.filter(|token| !token.is_empty()) {
let decoded = URL_SAFE_NO_PAD.decode(token).map_err(|_| invalid_token())?;
let (database, prefix, path, last): (String, Option<String>, Vec<String>, String) =
serde_json::from_slice(&decoded).map_err(|_| invalid_token())?;
if database != self.inner.database
|| prefix != self.inner.database_prefix
|| path != namespace
{
return Err(invalid_token());
}
format!(" WHERE name > '{}'", last.replace('\'', "''"))
} else {
String::new()
};
let query = self
.execute(ExecuteQueryRequest {
query: format!(
"SHOW MATERIALIZED VIEWS{after} LIMIT {}",
u64::from(limit) + 1
),
default_namespace_path: namespace.to_vec(),
parameters: None,
})
.await?;
let mut reader = query.reader().await?;
let mut items = Vec::new();
while let Some(batch) = reader.try_next().await? {
items.extend(strings(&batch, "name")?);
}
let page_token = if items.len() > limit as usize {
items.truncate(limit as usize);
let cursor = serde_json::to_vec(&(
&self.inner.database,
&self.inner.database_prefix,
namespace,
items.last().unwrap(),
))
.map_err(|error| Error::Runtime {
message: error.to_string(),
})?;
Some(URL_SAFE_NO_PAD.encode(cursor))
} else {
None
};
Ok(ListingPage { items, page_token })
}
}
rust/lancedb/src/remote/sql_test.rs
+164 -2
View File
@@ -1028,7 +1028,21 @@ async fn remote_refresh_omits_unknown_metrics_and_forwards_full() {
let (socket, _) = listener.accept().await?;
Ok::<_, std::io::Error>(Some((socket, listener)))
});
let service = TestSqlService::default();
let result = RecordBatch::try_from_iter(vec![
(
"name",
Arc::new(StringArray::from(vec!["view"])) as ArrayRef,
),
(
"query",
Arc::new(StringArray::from(vec!["SELECT * FROM source"])) as ArrayRef,
),
])
.unwrap();
let service = TestSqlService {
result,
..Default::default()
};
let statements = service.statements.clone();
let headers = service.headers.clone();
let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
@@ -1057,7 +1071,6 @@ async fn remote_refresh_omits_unknown_metrics_and_forwards_full() {
.unwrap_or_default();
captured.lock().unwrap().push(body["version"].clone());
let text = match request.url().path() {
"/v1/materialized_view/ns$view/describe" => r#"{"query": "SELECT * FROM source"}"#,
"/v1/table/ns$view/describe/" => r#"{"version": 42, "schema": {"fields": []}}"#,
other => panic!("unexpected REST request: {other}"),
};
@@ -1111,6 +1124,7 @@ async fn remote_refresh_omits_unknown_metrics_and_forwards_full() {
assert_eq!(
*statements.lock().unwrap(),
[
"DESCRIBE MATERIALIZED VIEW \"view\"",
"REFRESH MATERIALIZED VIEW \"view\"",
"REFRESH MATERIALIZED VIEW \"view\" WITH (full = true)",
]
@@ -1994,3 +2008,151 @@ async fn malformed_adaptive_decision_never_resubmits_the_statement() {
assert_eq!(polls.load(Ordering::SeqCst), 0);
}
}
#[tokio::test]
async fn remote_drop_materialized_view_uses_a_query_job() {
for defer in [false, true] {
let service = TestSqlService {
adaptive: true,
defer,
..Default::default()
};
let statements = service.statements.clone();
let headers = service.headers.clone();
let (client, shutdown) = start_sql_service(service, ClientConfig::default()).await;
let db = RemoteDatabase::new_mock(|request| -> http::Response<String> {
panic!("unexpected REST request: {}", request.url().path());
})
.with_sql_client(client.clone());
let job = db
.drop_materialized_view_async("daily_sales", &["team".into(), "reports".into()])
.await
.unwrap();
assert_eq!(job.job_type().as_deref(), Some("query"));
let id = Uuid::parse_str(job.id().unwrap()).unwrap();
job.wait().await.unwrap();
job.wait().await.unwrap();
assert_eq!(
client.describe(id).await.unwrap().status,
QueryStatus::Finished
);
assert_eq!(
*statements.lock().unwrap(),
["DROP MATERIALIZED VIEW \"daily_sales\""]
);
assert!(
headers
.lock()
.unwrap()
.iter()
.all(|header| header.namespace_path == "team$reports")
);
shutdown.send(()).unwrap();
}
}
#[tokio::test]
async fn remote_mv_listing_uses_scoped_sql_keyset_pagination() {
for defer in [false, true] {
let result = RecordBatch::try_from_iter(vec![(
"name",
Arc::new(StringArray::from(vec!["a", "b'c", "z"])) as ArrayRef,
)])
.unwrap();
let service = TestSqlService {
adaptive: true,
defer,
result,
..Default::default()
};
let statements = service.statements.clone();
let headers = service.headers.clone();
let (client, shutdown) = start_sql_service(service, ClientConfig::default()).await;
let db = RemoteDatabase::new_mock(|request| -> http::Response<String> {
panic!("unexpected REST request: {}", request.url().path());
})
.with_sql_client(client);
let namespace = vec!["team".to_owned(), "reports".to_owned()];
let first = db
.list_materialized_views(
&namespace,
crate::listing::ListingOptions::default().page_limit(2),
)
.await
.unwrap();
assert_eq!(first.items, ["a", "b'c"]);
let token = first.page_token.unwrap();
db.list_materialized_views(
&namespace,
crate::listing::ListingOptions::default()
.page_limit(2)
.page_token(&token),
)
.await
.unwrap();
assert_eq!(
*statements.lock().unwrap(),
[
"SHOW MATERIALIZED VIEWS LIMIT 3",
"SHOW MATERIALIZED VIEWS WHERE name > 'b''c' LIMIT 3"
]
);
assert!(
headers
.lock()
.unwrap()
.iter()
.all(|header| header.namespace_path == "team$reports")
);
for (path, token) in [(vec![], token), (namespace, "invalid".into())] {
assert!(matches!(
db.list_materialized_views(
&path,
crate::listing::ListingOptions::default().page_token(token)
)
.await,
Err(Error::InvalidInput { .. })
));
}
assert_eq!(statements.lock().unwrap().len(), 2);
shutdown.send(()).unwrap();
}
}
#[tokio::test]
async fn remote_describe_materialized_view_preserves_the_sql_definition() {
for defer in [false, true] {
let definition = "SELECT s.x, d.label FROM source s JOIN dim d ON s.id = d.id";
let result = RecordBatch::try_from_iter(vec![(
"query",
Arc::new(StringArray::from(vec![definition])) as ArrayRef,
)])
.unwrap();
let service = TestSqlService {
adaptive: true,
defer,
result,
..Default::default()
};
let statements = service.statements.clone();
let (client, shutdown) = start_sql_service(service, ClientConfig::default()).await;
let table = RemoteTable::new_with_sql_client(
client_with_handler(|request| -> http::Response<String> {
panic!("unexpected REST request: {}", request.url().path());
}),
"daily_sales".into(),
vec!["analytics".into()],
"analytics$daily_sales".into(),
Default::default(),
Some(client),
);
let info = table.materialized_view_info().await.unwrap();
assert_eq!(info.definition_sql, definition);
assert!(info.parsed_definition.is_none());
assert_eq!(
*statements.lock().unwrap(),
["DESCRIBE MATERIALIZED VIEW \"daily_sales\""]
);
shutdown.send(()).unwrap();
}
}
rust/lancedb/src/remote/table.rs
+16 -40
View File
@@ -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<S: HttpSend> BaseTable for RemoteTable<S> {
&self.identifier
}
async fn materialized_view_info(&self) -> Result<MaterializedViewInfo> {
#[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<String> {
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"));
}