From f9680806c982d30a85fd70c0542a8dc2b20ee5d9 Mon Sep 17 00:00:00 2001 From: Jack Ye Date: Wed, 21 Oct 2026 23:47:09 -0700 Subject: [PATCH] feat(python): expose materialized view auto-refresh creation options (#4621) Expose optional `auto_refresh_enabled` and `auto_refresh_interval_seconds` when creating materialized views from Python and Rust. Omitted intervals remain unset: the server determines when to refresh. ```python view = db.create_materialized_view( "recent_orders", "orders", auto_refresh_enabled=True, auto_refresh_interval_seconds=300, ) ``` ## Architecture and workflow - Python synchronous, asynchronous, job-returning and namespace wrappers forward both optional values through PyO3 to the Rust creation builder. Both default to `None`; no client interval is selected. - Remote creation calls `POST /v1/materialized_view/{id}/create` with the defining query and only the supplied refresh properties. A `202 Accepted` response must contain a job ID. Existing job tracking waits for the server's ordinary SQL CREATE job before synchronous creation opens the view. - The server validates and authorizes the CREATE, then uses the existing SQL worker to populate the view and persist its refresh settings. The Lance agent subsequently uses those settings to submit REFRESH SQL jobs. - Explicit positive finite fractional seconds are preserved. An interval alone does not enable refresh. Embedded connections retain manual refresh, accept explicit disabled state, and reject requests for automatic scheduling rather than ignore them. Includes API documentation and coverage for omission, false, fractional/invalid intervals, namespace forwarding, server job IDs and invalid acceptance responses. Requires the server REST-create endpoint to be deployed before SDK release. Existing remote `with_no_data=True` remains unsupported. --- docs/src/python/python.md | 16 ++ python/python/lancedb/_lancedb.pyi | 4 + python/python/lancedb/db.py | 27 ++++ python/python/lancedb/namespace.py | 16 ++ python/python/lancedb/remote/db.py | 8 + .../python/tests/test_materialized_views.py | 105 ++++++++++++- python/src/connection.rs | 22 ++- rust/lancedb/src/materialized_view.rs | 44 +++++- rust/lancedb/src/remote/db.rs | 140 ++++++++++++++---- 9 files changed, 349 insertions(+), 33 deletions(-) diff --git a/docs/src/python/python.md b/docs/src/python/python.md index c8a82851f..f4c697497 100644 --- a/docs/src/python/python.md +++ b/docs/src/python/python.md @@ -377,6 +377,22 @@ listing a storage directory. ## Materialized Views (Synchronous) +On Cloud and Enterprise, creation accepts `auto_refresh_enabled` and +`auto_refresh_interval_seconds`: + +```python +view = db.create_materialized_view( + "recent_orders", + "orders", + auto_refresh_enabled=True, + auto_refresh_interval_seconds=300, +) +``` + +Omit `auto_refresh_interval_seconds` to let the server determine when to refresh. +These options also apply to asynchronous creation and creation jobs. Embedded +connections support manual refresh and reject automatic scheduling requests. + ::: lancedb.materialized_view.MaterializedView ::: lancedb.materialized_view.MaterializedViewDefinition diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 96bc7752b..96eef064c 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -270,6 +270,8 @@ class Connection(object): filter: Optional[str] = None, limit: Optional[int] = None, with_no_data: bool = False, + auto_refresh_enabled: Optional[bool] = None, + auto_refresh_interval_seconds: Optional[float] = None, ) -> Table: ... async def create_materialized_view_async( self, @@ -279,6 +281,8 @@ class Connection(object): filter: Optional[str] = None, limit: Optional[int] = None, with_no_data: bool = False, + auto_refresh_enabled: Optional[bool] = None, + auto_refresh_interval_seconds: Optional[float] = None, ) -> Job: ... def list_materialized_views( self, *, page_token: Optional[str] = None, page_limit: Optional[int] = None diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index 47d24b762..5def89b01 100644 --- a/python/python/lancedb/db.py +++ b/python/python/lancedb/db.py @@ -607,6 +607,8 @@ class DBConnection(EnforceOverrides): where: Optional[str] = None, limit: Optional[int] = None, with_no_data: bool = False, + auto_refresh_enabled: Optional[bool] = None, + auto_refresh_interval_seconds: Optional[float] = None, ) -> MaterializedView: """Define a materialized view named ``name`` over the table ``source``. @@ -631,6 +633,13 @@ class DBConnection(EnforceOverrides): Cap the view at this many rows, in materialization order. with_no_data: bool, default False Skip the initial refresh and leave the backing table empty. + auto_refresh_enabled: bool, optional + Enable or disable automatic refresh on Cloud and Enterprise. + When omitted, the server determines whether automatic refresh is enabled. + Embedded databases do not support automatic refresh. + auto_refresh_interval_seconds: float, optional + Positive, finite refresh interval in seconds on Cloud and Enterprise. + When omitted, the server determines when to refresh. Returns ------- @@ -649,6 +658,8 @@ class DBConnection(EnforceOverrides): where: Optional[str] = None, limit: Optional[int] = None, with_no_data: bool = False, + auto_refresh_enabled: Optional[bool] = None, + auto_refresh_interval_seconds: Optional[float] = None, ) -> Job[None]: """Submit materialized-view creation and return its job. @@ -1706,6 +1717,8 @@ class LanceDBConnection(DBConnection): where: Optional[str] = None, limit: Optional[int] = None, with_no_data: bool = False, + auto_refresh_enabled: Optional[bool] = None, + auto_refresh_interval_seconds: Optional[float] = None, ) -> MaterializedView: """Define a materialized view named ``name`` over the table ``source``. See @@ -1734,6 +1747,8 @@ class LanceDBConnection(DBConnection): where=where, limit=limit, with_no_data=with_no_data, + auto_refresh_enabled=auto_refresh_enabled, + auto_refresh_interval_seconds=auto_refresh_interval_seconds, ) ) return MaterializedView(self.open_table(name)) @@ -1748,6 +1763,8 @@ class LanceDBConnection(DBConnection): where: Optional[str] = None, limit: Optional[int] = None, with_no_data: bool = False, + auto_refresh_enabled: Optional[bool] = None, + auto_refresh_interval_seconds: Optional[float] = None, ) -> Job[None]: job = LOOP.run( self._conn.create_materialized_view_async( @@ -1757,6 +1774,8 @@ class LanceDBConnection(DBConnection): where=where, limit=limit, with_no_data=with_no_data, + auto_refresh_enabled=auto_refresh_enabled, + auto_refresh_interval_seconds=auto_refresh_interval_seconds, ) ) return Job(job) @@ -2767,6 +2786,8 @@ class AsyncConnection(object): where: Optional[str] = None, limit: Optional[int] = None, with_no_data: bool = False, + auto_refresh_enabled: Optional[bool] = None, + auto_refresh_interval_seconds: Optional[float] = None, ) -> AsyncMaterializedView: """Define a materialized view named ``name`` over the table ``source``. See @@ -2782,6 +2803,8 @@ class AsyncConnection(object): filter=where, limit=limit, with_no_data=with_no_data, + auto_refresh_enabled=auto_refresh_enabled, + auto_refresh_interval_seconds=auto_refresh_interval_seconds, **native_options, ) return AsyncMaterializedView(AsyncTable(inner)) @@ -2795,6 +2818,8 @@ class AsyncConnection(object): where: Optional[str] = None, limit: Optional[int] = None, with_no_data: bool = False, + auto_refresh_enabled: Optional[bool] = None, + auto_refresh_interval_seconds: Optional[float] = None, ) -> AsyncJob[None]: """Submit materialized-view creation and return its job. @@ -2810,6 +2835,8 @@ class AsyncConnection(object): filter=where, limit=limit, with_no_data=with_no_data, + auto_refresh_enabled=auto_refresh_enabled, + auto_refresh_interval_seconds=auto_refresh_interval_seconds, **native_options, ) return AsyncJob(inner) diff --git a/python/python/lancedb/namespace.py b/python/python/lancedb/namespace.py index 26c62925f..f8675ebb5 100644 --- a/python/python/lancedb/namespace.py +++ b/python/python/lancedb/namespace.py @@ -665,6 +665,8 @@ class LanceNamespaceDBConnection(DBConnection): where: Optional[str] = None, limit: Optional[int] = None, with_no_data: bool = False, + auto_refresh_enabled: Optional[bool] = None, + auto_refresh_interval_seconds: Optional[float] = None, ) -> "MaterializedView": """Define a materialized view over a table in the root namespace. See @@ -680,6 +682,8 @@ class LanceNamespaceDBConnection(DBConnection): where=where, limit=limit, with_no_data=with_no_data, + auto_refresh_enabled=auto_refresh_enabled, + auto_refresh_interval_seconds=auto_refresh_interval_seconds, ) ).name ) @@ -695,6 +699,8 @@ class LanceNamespaceDBConnection(DBConnection): where: Optional[str] = None, limit: Optional[int] = None, with_no_data: bool = False, + auto_refresh_enabled: Optional[bool] = None, + auto_refresh_interval_seconds: Optional[float] = None, ) -> Job[None]: job = LOOP.run( self._inner.create_materialized_view_async( @@ -704,6 +710,8 @@ class LanceNamespaceDBConnection(DBConnection): where=where, limit=limit, with_no_data=with_no_data, + auto_refresh_enabled=auto_refresh_enabled, + auto_refresh_interval_seconds=auto_refresh_interval_seconds, ) ) return Job(job) @@ -1288,6 +1296,8 @@ class AsyncLanceNamespaceDBConnection: where: Optional[str] = None, limit: Optional[int] = None, with_no_data: bool = False, + auto_refresh_enabled: Optional[bool] = None, + auto_refresh_interval_seconds: Optional[float] = None, ) -> "AsyncMaterializedView": """Define a materialized view over a table in the root namespace.""" view = await self._inner.create_materialized_view( @@ -1297,6 +1307,8 @@ class AsyncLanceNamespaceDBConnection: where=where, limit=limit, with_no_data=with_no_data, + auto_refresh_enabled=auto_refresh_enabled, + auto_refresh_interval_seconds=auto_refresh_interval_seconds, ) # Reopen through the namespace so the view's table carries the # namespace client and pushdown configuration a bare inner table lacks. @@ -1311,6 +1323,8 @@ class AsyncLanceNamespaceDBConnection: where: Optional[str] = None, limit: Optional[int] = None, with_no_data: bool = False, + auto_refresh_enabled: Optional[bool] = None, + auto_refresh_interval_seconds: Optional[float] = None, ) -> AsyncJob[None]: """Submit materialized-view creation and return its job.""" return await self._inner.create_materialized_view_async( @@ -1320,6 +1334,8 @@ class AsyncLanceNamespaceDBConnection: where=where, limit=limit, with_no_data=with_no_data, + auto_refresh_enabled=auto_refresh_enabled, + auto_refresh_interval_seconds=auto_refresh_interval_seconds, ) async def open_materialized_view(self, name: str) -> "AsyncMaterializedView": diff --git a/python/python/lancedb/remote/db.py b/python/python/lancedb/remote/db.py index ac6b8b99d..89c7f8a78 100644 --- a/python/python/lancedb/remote/db.py +++ b/python/python/lancedb/remote/db.py @@ -723,6 +723,8 @@ class RemoteDBConnection(DBConnection): where: Optional[str] = None, limit: Optional[int] = None, with_no_data: bool = False, + auto_refresh_enabled: Optional[bool] = None, + auto_refresh_interval_seconds: Optional[float] = None, ) -> MaterializedView: from .table import RemoteTable @@ -734,6 +736,8 @@ class RemoteDBConnection(DBConnection): where=where, limit=limit, with_no_data=with_no_data, + auto_refresh_enabled=auto_refresh_enabled, + auto_refresh_interval_seconds=auto_refresh_interval_seconds, ) ) return MaterializedView( @@ -755,6 +759,8 @@ class RemoteDBConnection(DBConnection): where: Optional[str] = None, limit: Optional[int] = None, with_no_data: bool = False, + auto_refresh_enabled: Optional[bool] = None, + auto_refresh_interval_seconds: Optional[float] = None, ) -> Job[None]: job = LOOP.run( self._conn.create_materialized_view_async( @@ -764,6 +770,8 @@ class RemoteDBConnection(DBConnection): where=where, limit=limit, with_no_data=with_no_data, + auto_refresh_enabled=auto_refresh_enabled, + auto_refresh_interval_seconds=auto_refresh_interval_seconds, ) ) return Job(job) diff --git a/python/python/tests/test_materialized_views.py b/python/python/tests/test_materialized_views.py index bd1b2a0bb..46972a15c 100644 --- a/python/python/tests/test_materialized_views.py +++ b/python/python/tests/test_materialized_views.py @@ -129,8 +129,21 @@ def test_sync_remote_create_uses_public_async_connection(): where=None, limit=None, with_no_data=False, + auto_refresh_enabled=None, + auto_refresh_interval_seconds=None, ): - calls.append((name, source, select, where, limit, with_no_data)) + calls.append( + ( + name, + source, + select, + where, + limit, + with_no_data, + auto_refresh_enabled, + auto_refresh_interval_seconds, + ) + ) return StubAsyncMaterializedView() async def drop_materialized_view(self, name, *, namespace_path=None): @@ -148,9 +161,11 @@ def test_sync_remote_create_uses_public_async_connection(): where="age >= 18", limit=10, with_no_data=True, + auto_refresh_enabled=True, + auto_refresh_interval_seconds=0.5, ) assert view.name == "adults" - assert calls == [("adults", "people", ["name"], "age >= 18", 10, True)] + assert calls == [("adults", "people", ["name"], "age >= 18", 10, True, True, 0.5)] db.drop_materialized_view("adults", namespace_path=["analytics"]) assert calls[-1] == ("drop", "adults", ["analytics"]) @@ -550,3 +565,89 @@ def test_local_materialized_view_listing_pages_and_resume(tmp_path): assert sorted([first, *resumed]) == ["a_view", "z_view"] assert resumed.page_token() is None assert list(resumed) == [] + + +@pytest.mark.parametrize( + "options", + [ + {}, + {"auto_refresh_enabled": True}, + {"auto_refresh_enabled": False, "auto_refresh_interval_seconds": 0.5}, + ], +) +def test_remote_create_forwards_refresh_options_without_interval_default(options): + with mock_remote_materialized_view_create() as (host, requests): + db = lancedb.connect("db://dev", api_key="fake", host_override=host) + job = db.create_materialized_view_async("adults", "people", **options) + assert job.id == "mv-create-123" + assert requests == [ + ( + "/v1/materialized_view/adults/create", + { + "query": 'SELECT * FROM "people"', + **options, + }, + ) + ] + + +@pytest.mark.asyncio +async def test_async_remote_create_keeps_interval_unspecified(): + with mock_remote_materialized_view_create() as (host, requests): + db = await lancedb.connect_async("db://dev", api_key="fake", host_override=host) + job = await db.create_materialized_view_async( + "adults", + "people", + auto_refresh_enabled=True, + ) + assert job.id == "mv-create-123" + assert requests[0][1] == { + "query": 'SELECT * FROM "people"', + "auto_refresh_enabled": True, + } + + +@pytest.mark.parametrize( + "options", + [ + {"auto_refresh_enabled": True}, + {"auto_refresh_interval_seconds": 300}, + ], +) +def test_embedded_create_refuses_automatic_refresh_before_creating_view( + tmp_path, options +): + db = make_db(tmp_path) + with pytest.raises(NotImplementedError, match="Cloud or Enterprise"): + db.create_materialized_view("adults", "people", **options) + assert "adults" not in db.table_names() + + +@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_create() 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 == [] + + +def test_namespace_forwards_automatic_refresh_request(tmp_path): + db = lancedb.connect_namespace("dir", {"root": str(tmp_path)}) + with pytest.raises(NotImplementedError, match="Cloud or Enterprise"): + db.create_materialized_view("mv", "source", auto_refresh_enabled=True) + + +@pytest.mark.asyncio +async def test_async_namespace_forwards_automatic_refresh_request(tmp_path): + db = lancedb.connect_namespace_async("dir", {"root": str(tmp_path)}) + with pytest.raises(NotImplementedError, match="Cloud or Enterprise"): + await db.create_materialized_view("mv", "source", auto_refresh_enabled=True) + + +def test_embedded_explicitly_disabled_refresh_is_supported(tmp_path): + db = make_db(tmp_path) + view = db.create_materialized_view("mv", "people", auto_refresh_enabled=False) + assert view.table.count_rows() == 3 diff --git a/python/src/connection.rs b/python/src/connection.rs index 9ac081b9d..7e20f5f99 100644 --- a/python/src/connection.rs +++ b/python/src/connection.rs @@ -412,7 +412,8 @@ impl Connection { }) } - #[pyo3(signature = (name, source, projections=None, filter=None, limit=None, with_no_data=false))] + #[allow(clippy::too_many_arguments)] + #[pyo3(signature = (name, source, projections=None, filter=None, limit=None, with_no_data=false, auto_refresh_enabled=None, auto_refresh_interval_seconds=None))] pub fn create_materialized_view( self_: PyRef<'_, Self>, name: String, @@ -421,10 +422,18 @@ impl Connection { filter: Option, limit: Option, with_no_data: bool, + auto_refresh_enabled: Option, + auto_refresh_interval_seconds: Option, ) -> PyResult> { let inner = self_.get_inner()?.clone(); future_into_py(self_.py(), async move { let mut builder = inner.create_materialized_view(name, source); + if let Some(enabled) = auto_refresh_enabled { + builder = builder.auto_refresh_enabled(enabled); + } + if let Some(seconds) = auto_refresh_interval_seconds { + builder = builder.auto_refresh_interval_seconds(seconds); + } if let Some(projections) = projections { builder = builder.select(projections); } @@ -440,7 +449,8 @@ impl Connection { }) } - #[pyo3(signature = (name, source, projections=None, filter=None, limit=None, with_no_data=false))] + #[allow(clippy::too_many_arguments)] + #[pyo3(signature = (name, source, projections=None, filter=None, limit=None, with_no_data=false, auto_refresh_enabled=None, auto_refresh_interval_seconds=None))] pub fn create_materialized_view_async( self_: PyRef<'_, Self>, name: String, @@ -449,10 +459,18 @@ impl Connection { filter: Option, limit: Option, with_no_data: bool, + auto_refresh_enabled: Option, + auto_refresh_interval_seconds: Option, ) -> PyResult> { let inner = self_.get_inner()?.clone(); future_into_py(self_.py(), async move { let mut builder = inner.create_materialized_view(name, source); + if let Some(enabled) = auto_refresh_enabled { + builder = builder.auto_refresh_enabled(enabled); + } + if let Some(seconds) = auto_refresh_interval_seconds { + builder = builder.auto_refresh_interval_seconds(seconds); + } if let Some(projections) = projections { builder = builder.select(projections); } diff --git a/rust/lancedb/src/materialized_view.rs b/rust/lancedb/src/materialized_view.rs index 95e3baac7..ecdd1783b 100644 --- a/rust/lancedb/src/materialized_view.rs +++ b/rust/lancedb/src/materialized_view.rs @@ -280,7 +280,7 @@ pub struct MaterializedViewInfo { /// The backend-independent request used to create a remote materialized view. #[doc(hidden)] -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq)] pub struct CreateMaterializedViewRequest { /// Name of the new view. pub name: String, @@ -290,6 +290,10 @@ pub struct CreateMaterializedViewRequest { pub query: String, /// Whether to skip the initial population job. pub with_no_data: bool, + /// Whether the server should refresh the view automatically. + pub auto_refresh_enabled: Option, + /// Refresh interval in seconds. When absent, the server determines when to refresh. + pub auto_refresh_interval_seconds: Option, } /// Prefix of the internal columns holding source columns a computed column @@ -1570,6 +1574,8 @@ pub struct CreateMaterializedViewBuilder { filter: Option, limit: Option, with_no_data: bool, + auto_refresh_enabled: Option, + auto_refresh_interval_seconds: Option, } fn source_sql_identifier( @@ -1603,6 +1609,8 @@ impl CreateMaterializedViewBuilder { filter: None, limit: None, with_no_data: false, + auto_refresh_enabled: None, + auto_refresh_interval_seconds: None, } } @@ -1651,6 +1659,32 @@ impl CreateMaterializedViewBuilder { self } + /// Enable or disable automatic refresh on Cloud and Enterprise. + /// When omitted, the server determines whether automatic refresh is enabled. + /// Embedded databases do not support automatic refresh. + pub fn auto_refresh_enabled(mut self, enabled: bool) -> Self { + self.auto_refresh_enabled = Some(enabled); + self + } + + /// Set a positive, finite refresh interval in seconds on Cloud and Enterprise. + /// When omitted, the server determines when to refresh. + /// + /// ```no_run + /// # #![recursion_limit = "256"] + /// # async fn example(db: &lancedb::Connection) -> lancedb::Result<()> { + /// db.create_materialized_view("recent_orders", "orders") + /// .auto_refresh_enabled(true) + /// .auto_refresh_interval_seconds(300.0) + /// .execute().await?; + /// # Ok(()) + /// # } + /// ``` + pub fn auto_refresh_interval_seconds(mut self, seconds: f64) -> Self { + self.auto_refresh_interval_seconds = Some(seconds); + self + } + fn query(&self) -> Result { fn quote(name: &str) -> String { format!("\"{}\"", name.replace('"', "\"\"")) @@ -1693,6 +1727,8 @@ impl CreateMaterializedViewBuilder { namespace_path: self.namespace.clone(), query: self.query()?, with_no_data: self.with_no_data, + auto_refresh_enabled: self.auto_refresh_enabled, + auto_refresh_interval_seconds: self.auto_refresh_interval_seconds, }) .await; } @@ -1719,6 +1755,12 @@ impl CreateMaterializedViewBuilder { } async fn execute_native(self) -> Result { + if self.auto_refresh_enabled == Some(true) || self.auto_refresh_interval_seconds.is_some() { + return Err(Error::NotSupported { + message: "automatic materialized-view refresh requires Cloud or Enterprise" + .to_string(), + }); + } let source = self .connection .open_table(&self.source) diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs index a0f4aa953..4a8cbf32e 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/rust/lancedb/src/remote/db.rs @@ -51,10 +51,6 @@ use super::table::RemoteTable; use super::util::parse_server_version; use super::{ARROW_STREAM_CONTENT_TYPE, extract_job_id}; -fn quote_sql_identifier(identifier: &str) -> String { - format!("\"{}\"", identifier.replace('"', "\"\"")) -} - // Request structure for the remote clone table API #[derive(serde::Serialize)] struct RemoteCloneTableRequest { @@ -67,6 +63,15 @@ struct RemoteCloneTableRequest { is_shallow: Option, } +#[derive(serde::Serialize)] +struct RemoteCreateMaterializedViewRequest { + query: String, + #[serde(skip_serializing_if = "Option::is_none")] + auto_refresh_enabled: Option, + #[serde(skip_serializing_if = "Option::is_none")] + auto_refresh_interval_seconds: Option, +} + // the versions of the server that we support // for any new feature that we need to change the SDK behavior, we should bump the server version, // and add a feature flag as method of `ServerVersion` here. @@ -877,16 +882,43 @@ impl Database for RemoteDatabase { .to_string(), }); } - let client = self.sql_client.clone().ok_or_else(|| Error::NotSupported { - message: "SQL is unavailable for this remote database client".to_string(), + if let Some(seconds) = request.auto_refresh_interval_seconds + && (!seconds.is_finite() || seconds <= 0.0) + { + return Err(Error::InvalidInput { + message: "auto_refresh_interval_seconds must be positive, finite seconds".into(), + }); + } + let identifier = build_table_identifier(&request.name, &request.namespace_path)?; + let body = RemoteCreateMaterializedViewRequest { + query: request.query, + auto_refresh_enabled: request.auto_refresh_enabled, + auto_refresh_interval_seconds: request.auto_refresh_interval_seconds, + }; + let request = self + .client + .post(&format!("/v1/materialized_view/{identifier}/create")) + .json(&body); + 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())?; + if status != StatusCode::ACCEPTED { + return Err(Error::Http { + source: "materialized-view creation must return 202 Accepted".into(), + request_id, + status_code: Some(status), + }); + } + let job_id = extract_job_id(&body).ok_or_else(|| Error::Http { + source: "materialized-view create response did not contain a valid job_id".into(), + request_id, + status_code: Some(status), })?; - let namespace = request.namespace_path; - let statement = format!( - "CREATE MATERIALIZED VIEW {} AS {}", - quote_sql_identifier(&request.name), - request.query - ); - Ok(client.submit_as_job(statement, namespace, || async { Ok(()) })) + Ok(Job::new(Box::new(RemoteJob::new( + self.client.clone(), + job_id, + )))) } async fn drop_materialized_view_async( @@ -1964,21 +1996,73 @@ mod tests { } #[tokio::test] - async fn test_create_materialized_view_requires_sql_client() { - let db = super::RemoteDatabase::new_mock(|request| -> http::Response { - panic!("unexpected REST request: {}", request.url().path()) - }); - let error = db - .create_materialized_view_async(CreateMaterializedViewRequest { - name: "adults".into(), - namespace_path: vec!["analytics".into()], - query: "SELECT age AS \"age\" FROM \"raw\".\"people\" WHERE age >= 18 LIMIT 10" - .into(), - with_no_data: false, - }) - .await - .unwrap_err(); - assert!(error.to_string().contains("SQL is unavailable")); + async fn test_create_materialized_view_uses_rest_and_preserves_optional_settings() { + for (enabled, interval) in [(None, None), (Some(true), None), (Some(false), Some(0.5))] { + let db = super::RemoteDatabase::new_mock(move |request| { + assert_eq!(request.method(), "POST"); + assert_eq!( + request.url().path(), + "/v1/materialized_view/analytics$adults/create" + ); + let body: serde_json::Value = + serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap(); + assert_eq!(body["query"], "SELECT * FROM people"); + assert_eq!( + body.get("auto_refresh_enabled"), + enabled.map(serde_json::Value::from).as_ref() + ); + assert_eq!( + body.get("auto_refresh_interval_seconds"), + interval.map(serde_json::Value::from).as_ref() + ); + http::Response::builder() + .status(202) + .body(serde_json::json!({"job_id": "j1-mv-create"}).to_string()) + .unwrap() + }); + let job = db + .create_materialized_view_async(CreateMaterializedViewRequest { + name: "adults".into(), + namespace_path: vec!["analytics".into()], + query: "SELECT * FROM people".into(), + with_no_data: false, + auto_refresh_enabled: enabled, + auto_refresh_interval_seconds: interval, + }) + .await + .unwrap(); + assert_eq!(job.id(), Some("j1-mv-create")); + } + } + + #[tokio::test] + async fn test_create_materialized_view_rejects_invalid_acceptance() { + for (status, body) in [ + (200, r#"{"job_id":"j1"}"#), + (202, "{}"), + (202, r#"{"job_id":""}"#), + ] { + let db = super::RemoteDatabase::new_mock(move |_| { + http::Response::builder().status(status).body(body).unwrap() + }); + let error = 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: None, + }) + .await + .err() + .unwrap(); + assert!(error.to_string().contains(if status == 200 { + "202 Accepted" + } else { + "job_id" + })); + } } #[tokio::test]