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

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

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

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

---------

Co-authored-by: Xuanwo <github@xuanwo.io>
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
+5 -4
View File
@@ -436,8 +436,8 @@ abstract dropMaterializedView(name, namespacePath?): Promise<void>
Drop the materialized view named `name`. Drop the materialized view named `name`.
The view may become unavailable before physical cleanup finishes. Use Remote drops wait for the SQL command to remove the catalog binding.
[dropMaterializedViewAsync](Connection.md#dropmaterializedviewasync) to retain and wait for the cleanup job. Physical cleanup continues in the background.
Rejects a table that exists but is not a materialized view. 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> abstract dropMaterializedViewAsync(name, namespacePath?): Promise<Job>
``` ```
Start dropping the materialized view named `name` and return its cleanup Start dropping the materialized view named `name` and return its job.
job without waiting for completion. Remote jobs have type `query` and use the SQL query ID as `id`. Waiting
confirms catalog removal; physical cleanup continues in the background.
Rejects a table that exists but is not a materialized view. Rejects a table that exists but is not a materialized view.
+6 -4
View File
@@ -72,10 +72,12 @@ SQL-backed job handles follow the query lifecycle described below.
### SQL-backed job handles ### SQL-backed job handles
Remote materialized-view CREATE and REFRESH use the SQL query backend. Python's Remote materialized-view CREATE, REFRESH, and DROP use the SQL query backend.
`create_materialized_view_async()` and `mv.refresh_async()` still return job Python's `create_materialized_view_async()`, `mv.refresh_async()`, and
handles with `status()`, `wait()` and `cancel()`. Synchronous creation and refresh `drop_materialized_view_async()` return job handles with `status()`, `wait()`,
wait on the same SQL execution before returning. and `cancel()`. Synchronous creation, refresh, and drop wait on the same SQL
execution before returning. A completed DROP means the catalog binding has been
removed; physical cleanup continues in the background.
The existing `job.job_type` identifies query-backed handles as `"query"` from The existing `job.job_type` identifies query-backed handles as `"query"` from
submission, and `job.id` contains the connection-scoped SQL query UUID string. submission, and `job.id` contains the connection-scoped SQL query UUID string.
+3 -25
View File
@@ -481,23 +481,6 @@ describe("remote connection", () => {
expect(insertRequests).toBe(2); 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 () => { it("creates a view and decodes the schema it resolved to", async () => {
await withMockDatabase( await withMockDatabase(
(req, res) => { (req, res) => {
@@ -1725,14 +1708,13 @@ async function collect<T>(items: AsyncIterable<T>): Promise<T[]> {
} }
describe("lazy resource listings", () => { describe("lazy resource listings", () => {
for (const kind of ["views", "materializedViews", "jobs"] as const) { for (const kind of ["views", "jobs"] as const) {
const listing = ( const listing = (
db: Connection, db: Connection,
options: { pageToken?: string; pageLimit?: number } = {}, options: { pageToken?: string; pageLimit?: number } = {},
) => { ) => {
if (kind === "jobs") return db.listJobs(options); if (kind === "jobs") return db.listJobs(options);
if (kind === "views") return db.listViews(["team"], options); return db.listViews(["team"], options);
return db.listMaterializedViews(options);
}; };
const page = (names: string[], token?: string) => { const page = (names: string[], token?: string) => {
const items = const items =
@@ -1807,11 +1789,7 @@ describe("lazy resource listings", () => {
limit: 2, limit: 2,
}); });
} else { } else {
expect(url.pathname).toBe( expect(url.pathname).toBe("/v1/namespace/team/view/list");
kind === "views"
? "/v1/namespace/team/view/list"
: "/v1/namespace/$/materialized_view/list",
);
expect(url.searchParams.get("page_token")).toBe(token); expect(url.searchParams.get("page_token")).toBe(token);
expect(url.searchParams.get("limit")).toBe("2"); expect(url.searchParams.get("limit")).toBe("2");
} }
+5 -4
View File
@@ -308,8 +308,8 @@ export abstract class Connection {
/** /**
* Drop the materialized view named `name`. * Drop the materialized view named `name`.
* *
* The view may become unavailable before physical cleanup finishes. Use * Remote drops wait for the SQL command to remove the catalog binding.
* {@link dropMaterializedViewAsync} to retain and wait for the cleanup job. * Physical cleanup continues in the background.
* *
* Rejects a table that exists but is not a materialized view. * Rejects a table that exists but is not a materialized view.
*/ */
@@ -319,8 +319,9 @@ export abstract class Connection {
): Promise<void>; ): Promise<void>;
/** /**
* Start dropping the materialized view named `name` and return its cleanup * Start dropping the materialized view named `name` and return its job.
* job without waiting for completion. * Remote jobs have type `query` and use the SQL query ID as `id`. Waiting
* confirms catalog removal; physical cleanup continues in the background.
* *
* Rejects a table that exists but is not a materialized view. * Rejects a table that exists but is not a materialized view.
*/ */
+12 -13
View File
@@ -721,9 +721,8 @@ class DBConnection(EnforceOverrides):
) -> None: ) -> None:
"""Drop a materialized view. """Drop a materialized view.
The view may become unavailable before physical cleanup finishes. Use Remote drops wait for the SQL command to remove the catalog binding.
:meth:`drop_materialized_view_async` to retain and wait for the cleanup Physical cleanup continues in the background.
job.
""" """
raise NotImplementedError( raise NotImplementedError(
"materialized views are not supported on this connection type" "materialized views are not supported on this connection type"
@@ -732,11 +731,11 @@ class DBConnection(EnforceOverrides):
def drop_materialized_view_async( def drop_materialized_view_async(
self, name: str, namespace_path: Optional[List[str]] = None self, name: str, namespace_path: Optional[List[str]] = None
) -> Job[None]: ) -> Job[None]:
"""Start dropping a materialized view and return its cleanup job. """Start dropping a materialized view and return its job.
The job may already be complete for a local database. On LanceDB Cloud The job may already be complete for a local database. Remote jobs have
and Enterprise, its ``id`` is the server job identifier from the type ``query`` and use the SQL query ID as ``id``. Waiting confirms
``202 Accepted`` drop response. catalog removal; physical cleanup continues in the background.
""" """
raise NotImplementedError( raise NotImplementedError(
"materialized views are not supported on this connection type" "materialized views are not supported on this connection type"
@@ -2946,9 +2945,8 @@ class AsyncConnection(object):
) -> None: ) -> None:
"""Drop a materialized view. """Drop a materialized view.
The view may become unavailable before physical cleanup finishes. Use Remote drops wait for the SQL command to remove the catalog binding.
:meth:`drop_materialized_view_async` to retain and wait for the cleanup Physical cleanup continues in the background.
job.
""" """
if namespace_path is None: if namespace_path is None:
namespace_path = [] namespace_path = []
@@ -2960,10 +2958,11 @@ class AsyncConnection(object):
*, *,
namespace_path: Optional[List[str]] = None, namespace_path: Optional[List[str]] = None,
) -> AsyncJob[None]: ) -> AsyncJob[None]:
"""Start dropping a materialized view and return its cleanup job. """Start dropping a materialized view and return its job.
Await :meth:`AsyncJob.wait` before assuming physical cleanup has For remote connections, the job has type ``query`` and its ID identifies
finished. the SQL query. Await :meth:`AsyncJob.wait` for catalog removal; physical
cleanup continues in the background.
""" """
if namespace_path is None: if namespace_path is None:
namespace_path = [] namespace_path = []
+4 -6
View File
@@ -66,7 +66,7 @@ def response_page(kind, names, token):
] ]
else: else:
items = names items = names
key = "views" if kind == "materialized_views" else kind key = kind
return {key: items, "page_token": token} return {key: items, "page_token": token}
@@ -85,7 +85,7 @@ async def advance(items, asynchronous):
return await items.__anext__() if asynchronous else next(items) return await items.__anext__() if asynchronous else next(items)
KINDS = ["secrets", "views", "jobs", "functions", "materialized_views"] KINDS = ["secrets", "views", "jobs", "functions"]
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -236,9 +236,7 @@ async def test_listing_lazy_pages_and_resume(listing_server, kind, asynchronous)
) )
db = await connection(endpoint, asynchronous) db = await connection(endpoint, asynchronous)
method = getattr(db, "list_" + kind) method = getattr(db, "list_" + kind)
kwargs = ( kwargs = {} if kind == "jobs" else {"namespace_path": ["team"]}
{} if kind in ("jobs", "materialized_views") else {"namespace_path": ["team"]}
)
items = method(page_token="start/token", page_limit=2, **kwargs) items = method(page_token="start/token", page_limit=2, **kwargs)
assert requests == [] assert requests == []
assert items.page_token() == "start/token" 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 path.path == "/v1/jobs/list"
assert body == {"page_token": token, "limit": 2} assert body == {"page_token": token, "limit": 2}
else: else:
namespace = "$" if kind == "materialized_views" else "team" namespace = "team"
resource = kind.removesuffix("s") resource = kind.removesuffix("s")
assert path.path == f"/v1/namespace/{namespace}/{resource}/list" assert path.path == f"/v1/namespace/{namespace}/{resource}/list"
expected = {"page_token": [token], "limit": ["2"]} expected = {"page_token": [token], "limit": ["2"]}
+36 -100
View File
@@ -1,6 +1,7 @@
# SPDX-License-Identifier: Apache-2.0 # SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors # SPDX-FileCopyrightText: Copyright The LanceDB Authors
import contextlib import contextlib
import http.server import http.server
import json import json
@@ -30,89 +31,6 @@ def make_db(tmp_path):
return db 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 @contextlib.contextmanager
def mock_remote_materialized_view_refresh(): def mock_remote_materialized_view_refresh():
requests = [] requests = []
@@ -123,11 +41,8 @@ def mock_remote_materialized_view_refresh():
def do_POST(self): def do_POST(self):
requests.append(self.path) requests.append(self.path)
if self.path == "/v1/materialized_view/view/describe": assert self.path == "/v1/table/view/describe/"
response = {"query": "SELECT * FROM source"} response = {"version": 42, "schema": {"fields": []}}
else:
assert self.path == "/v1/table/view/describe/"
response = {"version": 42, "schema": {"fields": []}}
encoded = json.dumps(response).encode() encoded = json.dumps(response).encode()
self.send_response(200) self.send_response(200)
self.send_header("Content-Type", "application/json") 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): def do_get(self, context, ticket):
self.commands.append(ticket.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. # SQL completion carries no refresh statistics.
return flight.RecordBatchStream(pa.table({"status": ["ok"]})) 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, " "RefreshMaterializedViewResult(mode=None, rows_written=None, "
"source_version=None, version=42)" "source_version=None, version=42)"
) )
(command,) = server.commands describes = [
assert b'REFRESH MATERIALIZED VIEW "view"' in command command
assert (b"WITH (full = true)" in command) is full for command in server.commands
if b"DESCRIBE MATERIALIZED VIEW" in command
]
refreshes = [
command
for command in server.commands
if b"REFRESH MATERIALIZED VIEW" in command
]
assert len(describes) == 2
assert len(refreshes) == 1
assert all(
b'DESCRIBE MATERIALIZED VIEW "view"' in command for command in describes
)
(refresh,) = refreshes
assert b'REFRESH MATERIALIZED VIEW "view"' in refresh
assert (b"WITH (full = true)" in refresh) is full
assert not any("count_rows" in path for path in requests) 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(): def test_stored_queries_and_legacy_layouts_are_read():
import json
import pyarrow as pa import pyarrow as pa
from lancedb.materialized_view import _definition_from_schema 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")]) @pytest.mark.parametrize("seconds", [0, -1, float("nan"), float("inf")])
def test_remote_create_rejects_invalid_interval_without_request(seconds): def test_remote_create_rejects_invalid_interval_without_request(seconds):
with mock_remote_materialized_view_drop() as (host, requests): db = lancedb.connect("db://dev", api_key="fake", host_override="http://127.0.0.1:1")
db = lancedb.connect("db://dev", api_key="fake", host_override=host) with pytest.raises(ValueError, match="positive, finite"):
with pytest.raises(ValueError, match="positive, finite"): db.create_materialized_view_async(
db.create_materialized_view_async( "mv", "source", auto_refresh_interval_seconds=seconds
"adults", "people", auto_refresh_interval_seconds=seconds )
)
assert requests == []
@pytest.mark.experimental_functions @pytest.mark.experimental_functions
+9 -8
View File
@@ -1987,8 +1987,8 @@ impl Connection {
/// Drop a materialized view. /// Drop a materialized view.
/// ///
/// The view may become unavailable before its physical data is removed. /// The view may become unavailable before its physical data is removed.
/// Use [`Connection::drop_materialized_view_async`] to retain the cleanup /// Remote drops wait for the SQL command to remove the catalog binding;
/// job and wait for it explicitly. /// physical cleanup proceeds in the background.
pub async fn drop_materialized_view( pub async fn drop_materialized_view(
&self, &self,
name: impl AsRef<str>, name: impl AsRef<str>,
@@ -1999,8 +1999,9 @@ impl Connection {
return self return self
.database() .database()
.drop_materialized_view_async(name, namespace_path) .drop_materialized_view_async(name, namespace_path)
.await .await?
.map(|_| ()); .wait()
.await;
} }
let table = self let table = self
.open_table(name) .open_table(name)
@@ -2011,12 +2012,12 @@ impl Connection {
self.drop_table(name, namespace_path).await 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 /// This validates that the named resource is a materialized view rather
/// than an ordinary table. Call [`Job::wait`] before assuming physical /// than an ordinary table. For remote connections the job has type `query`
/// cleanup has finished. When the backend performs cleanup inline, the /// and its ID identifies the SQL query. Waiting completes once the catalog
/// returned job is already finished and has no job ID. /// binding is removed; physical cleanup continues in the background.
/// ///
/// ```no_run /// ```no_run
/// # use lancedb::Connection; /// # use lancedb::Connection;
+30 -169
View File
@@ -47,6 +47,7 @@ use super::client::{
Sender, Sender,
}; };
use super::sql::SqlClient; use super::sql::SqlClient;
use super::sql::materialized_view::quote_identifier;
use super::table::RemoteTable; use super::table::RemoteTable;
use super::util::parse_server_version; use super::util::parse_server_version;
use super::{ARROW_STREAM_CONTENT_TYPE, extract_job_id}; use super::{ARROW_STREAM_CONTENT_TYPE, extract_job_id};
@@ -929,36 +930,26 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
name: &str, name: &str,
namespace_path: &[String], namespace_path: &[String],
) -> Result<Job> { ) -> Result<Job> {
let identifier = build_table_identifier(name, namespace_path)?; validate_table_name(name)?;
let request = self validate_namespace(namespace_path)?;
.client let client = self
.post(&format!("/v1/materialized_view/{identifier}/drop")); .sql_client
let (request_id, response) = self.client.send(request).await?; .as_ref()
let response = self.client.check_response(&request_id, response).await?; .ok_or_else(|| Error::NotSupported {
let status = response.status(); message: "SQL is unavailable for this remote database client".into(),
let body = response.text().await.err_to_http(request_id.clone())?; })?;
let job = match status { let cache = self.table_cache.clone();
StatusCode::OK => Job::new_done(), let key = build_cache_key(name, namespace_path);
StatusCode::ACCEPTED => { client
let job_id = extract_job_id(&body).ok_or_else(|| Error::Http { .submit_as_job(
source: "materialized-view drop response did not contain a valid job_id".into(), format!("DROP MATERIALIZED VIEW {}", quote_identifier(name)),
request_id, namespace_path.to_vec(),
status_code: Some(status), move || async move {
})?; cache.remove(&key).await;
Job::new(Box::new(RemoteJob::new(self.client.clone(), job_id))) Ok(())
} },
_ => { )
return Err(Error::Http { .await
source: "materialized-view drop must return 200 OK or 202 Accepted".into(),
request_id,
status_code: Some(status),
});
}
};
self.table_cache
.remove(&build_cache_key(name, namespace_path))
.await;
Ok(job)
} }
async fn list_materialized_views( async fn list_materialized_views(
@@ -966,23 +957,13 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
namespace_path: &[String], namespace_path: &[String],
options: ListingOptions, options: ListingOptions,
) -> Result<ListingPage<String>> { ) -> Result<ListingPage<String>> {
let namespace_id = build_namespace_identifier(namespace_path)?; self.sql_client
let path = format!("/v1/namespace/{namespace_id}/materialized_view/list"); .as_ref()
let mut req = self.client.get(&path); .ok_or_else(|| Error::NotSupported {
if let Some(token) = options.page_token { message: "SQL is unavailable for this remote database client".into(),
req = req.query(&[("page_token", token)]); })?
} .list_materialized_views(namespace_path, options)
if let Some(limit) = options.page_limit { .await
req = req.query(&[("limit", limit)]);
}
let (request_id, response) = self.client.send(req).await?;
let response = self.client.check_response(&request_id, response).await?;
let response: RemoteListViewsResponse = response.json().await.err_to_http(request_id)?;
let items = response.views;
Ok(ListingPage {
items,
page_token: response.page_token,
})
} }
async fn create_function_async( 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] #[tokio::test]
async fn test_drop_function_async_returns_job() { async fn test_drop_function_async_returns_job() {
let db = super::RemoteDatabase::new_mock(|request| { 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] #[tokio::test]
async fn test_retries() { async fn test_retries() {
// We'll record the request_id here, to check it matches the one in the error. // 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());
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 count = Arc::new(AtomicUsize::new(0));
let conn = Connection::new_with_handler({ let conn = Connection::new_with_handler({
let count = count.clone(); let count = count.clone();
@@ -3927,11 +3793,7 @@ mod tests {
.unwrap(); .unwrap();
assert_eq!(body, serde_json::json!({"page_token": token, "limit": 2})); assert_eq!(body, serde_json::json!({"page_token": token, "limit": 2}));
} else { } else {
let ns = if resource == "materialized_view" { let ns = "team";
"$"
} else {
"team"
};
assert_eq!( assert_eq!(
request.url().path(), request.url().path(),
format!("/v1/namespace/{ns}/{resource}/list") format!("/v1/namespace/{ns}/{resource}/list")
@@ -3971,7 +3833,6 @@ mod tests {
"secret" => check(conn.list_secrets(&namespace, options), count).await, "secret" => check(conn.list_secrets(&namespace, options), count).await,
"view" => check(conn.list_views(&namespace, options), count).await, "view" => check(conn.list_views(&namespace, options), count).await,
"function" => check(conn.list_functions(&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, _ => check(conn.list_jobs(options), count).await,
} }
} }
+1
View File
@@ -35,6 +35,7 @@ use crate::remote::client::{ClientConfig, TlsConfig};
use crate::remote::retry::ResolvedRetryConfig; use crate::remote::retry::ResolvedRetryConfig;
use crate::sql::{Query, QueryDescription, QueryHandle, QueryStatus}; use crate::sql::{Query, QueryDescription, QueryHandle, QueryStatus};
pub(super) mod materialized_view;
mod stream; mod stream;
use stream::{StreamingQuery, StreamingQueryHandle, Submission}; use stream::{StreamingQuery, StreamingQueryHandle, Submission};
@@ -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 })
}
}
+164 -2
View File
@@ -1028,7 +1028,21 @@ async fn remote_refresh_omits_unknown_metrics_and_forwards_full() {
let (socket, _) = listener.accept().await?; let (socket, _) = listener.accept().await?;
Ok::<_, std::io::Error>(Some((socket, listener))) 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 statements = service.statements.clone();
let headers = service.headers.clone(); let headers = service.headers.clone();
let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel(); 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(); .unwrap_or_default();
captured.lock().unwrap().push(body["version"].clone()); captured.lock().unwrap().push(body["version"].clone());
let text = match request.url().path() { 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": []}}"#, "/v1/table/ns$view/describe/" => r#"{"version": 42, "schema": {"fields": []}}"#,
other => panic!("unexpected REST request: {other}"), other => panic!("unexpected REST request: {other}"),
}; };
@@ -1111,6 +1124,7 @@ async fn remote_refresh_omits_unknown_metrics_and_forwards_full() {
assert_eq!( assert_eq!(
*statements.lock().unwrap(), *statements.lock().unwrap(),
[ [
"DESCRIBE MATERIALIZED VIEW \"view\"",
"REFRESH MATERIALIZED VIEW \"view\"", "REFRESH MATERIALIZED VIEW \"view\"",
"REFRESH MATERIALIZED VIEW \"view\" WITH (full = true)", "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); 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();
}
}
+16 -40
View File
@@ -18,9 +18,7 @@ use crate::index::IndexStatistics;
use crate::index::scalar::FtsQuery; use crate::index::scalar::FtsQuery;
use crate::index::waiter::{MAX_WAIT, wait_for_index}; use crate::index::waiter::{MAX_WAIT, wait_for_index};
use crate::job::Job; use crate::job::Job;
use crate::materialized_view::{ use crate::materialized_view::{MaterializedViewInfo, RefreshMaterializedViewResult};
MaterializedViewDefinition, MaterializedViewInfo, RefreshMaterializedViewResult,
};
use crate::query::wal_fusion::PkFusionMemory; // WAL-PK-FUSION: delete. use crate::query::wal_fusion::PkFusionMemory; // WAL-PK-FUSION: delete.
use crate::query::{QueryFilter, QueryRequest, Select, VectorQueryRequest}; use crate::query::{QueryFilter, QueryRequest, Select, VectorQueryRequest};
use crate::remote::job::RemoteJob; use crate::remote::job::RemoteJob;
@@ -2201,24 +2199,13 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
&self.identifier &self.identifier
} }
async fn materialized_view_info(&self) -> Result<MaterializedViewInfo> { async fn materialized_view_info(&self) -> Result<MaterializedViewInfo> {
#[derive(Deserialize)] self.sql_client
struct DescribeMaterializedViewResponse { .as_ref()
query: String, .ok_or_else(|| Error::NotSupported {
} message: "SQL is unavailable for this remote table client".into(),
})?
let request = self.client.post(&format!( .describe_materialized_view(&self.name, &self.namespace)
"/v1/materialized_view/{}/describe", .await
self.identifier
));
let (request_id, response) = self.send(request, true).await?;
let response = self.check_table_response(&request_id, response).await?;
let response: DescribeMaterializedViewResponse =
response.json().await.err_to_http(request_id)?;
let parsed_definition = MaterializedViewDefinition::from_sql(&response.query).ok();
Ok(MaterializedViewInfo {
definition_sql: response.query,
parsed_definition,
})
} }
async fn refresh_materialized_view_async( async fn refresh_materialized_view_async(
@@ -14610,26 +14597,15 @@ mod tests {
} }
#[tokio::test] #[tokio::test]
async fn test_materialized_view_describe_and_refresh_requires_sql() { async fn test_materialized_view_requires_sql() {
const QUERY: &str = let table = Table::new_with_handler("my_table", |request| -> http::Response<String> {
"SELECT s.x, d.label FROM analytics.source s JOIN analytics.dim d ON s.id = d.id"; panic!("unexpected REST request: {}", request.url().path())
let table = Table::new_with_handler("my_table", |request| match request.url().path() {
"/v1/materialized_view/my_table/describe" => http::Response::builder()
.status(200)
.body(
json!({
"name": "my_table",
"query": QUERY
})
.to_string(),
)
.unwrap(),
path => panic!("unexpected request: {path}"),
}); });
let view = crate::MaterializedView::from_table(table).await.unwrap(); let error = table
assert_eq!(view.definition_sql(), QUERY); .base_table()
assert!(view.definition().is_err()); .materialized_view_info()
let error = view.refresh().execute().await.unwrap_err(); .await
.unwrap_err();
assert!(error.to_string().contains("SQL is unavailable")); assert!(error.to_string().contains("SQL is unavailable"));
} }