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]