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

@@ -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

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

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)

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":

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)

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

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<String>,
         limit: Option<u64>,
         with_no_data: bool,
+        auto_refresh_enabled: Option<bool>,
+        auto_refresh_interval_seconds: Option<f64>,
     ) -> PyResult<Bound<'_, PyAny>> {
         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<String>,
         limit: Option<u64>,
         with_no_data: bool,
+        auto_refresh_enabled: Option<bool>,
+        auto_refresh_interval_seconds: Option<f64>,
     ) -> PyResult<Bound<'_, PyAny>> {
         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);
             }

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<bool>,
+    /// Refresh interval in seconds. When absent, the server determines when to refresh.
+    pub auto_refresh_interval_seconds: Option<f64>,
 }
 
 /// Prefix of the internal columns holding source columns a computed column
@@ -1570,6 +1574,8 @@ pub struct CreateMaterializedViewBuilder {
     filter: Option<String>,
     limit: Option<u64>,
     with_no_data: bool,
+    auto_refresh_enabled: Option<bool>,
+    auto_refresh_interval_seconds: Option<f64>,
 }
 
 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<String> {
         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<MaterializedView> {
+        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)

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<bool>,
 }
 
+#[derive(serde::Serialize)]
+struct RemoteCreateMaterializedViewRequest {
+    query: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    auto_refresh_enabled: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    auto_refresh_interval_seconds: Option<f64>,
+}
+
 // 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<S: HttpSend> Database for RemoteDatabase<S> {
                     .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<String> {
-            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]