fix: accept defining queries in Python materialized views (#4980)

Python materialized-view creation now accepts a complete defining SQL
query across sync, async, remote, and namespace connections, including
creation jobs:

```python
view = db.create_materialized_view(
    "events_by_kind",
    query="SELECT kind, count(*) AS n FROM events GROUP BY kind",
)
```

`query` is mutually exclusive with `source`, `select`, `where`, and
`limit`. Existing source-based calls remain supported.

Adds Rust `Connection::create_materialized_view_from_query`: embedded
connections plan through `MaterializedViewDefinition::from_sql`, while
remote connections submit the full query for server-side parsing.
Includes regression coverage for grouping, UNNEST, query forwarding,
creation jobs, and argument validation.

python/src/connection.rs

@@ -26,6 +26,7 @@ use lancedb::{
     database::{CreateTableMode, Database, ReadConsistency},
     function::{GpuClass as LanceGpuClass, RefreshColumnResult},
     graph::PropertyGraphDefinition,
+    materialized_view::CreateMaterializedViewBuilder,
 };
 use pyo3::{
     Bound, FromPyObject, Py, PyAny, PyRef, PyResult, Python,
@@ -34,6 +35,27 @@ use pyo3::{
     types::{PyAnyMethods, PyDict, PyDictMethods, PyList},
 };
 
+fn materialized_view_builder(
+    connection: &LanceConnection,
+    name: String,
+    source: Option<String>,
+    query: Option<String>,
+    has_clauses: bool,
+) -> PyResult<CreateMaterializedViewBuilder> {
+    match (source, query) {
+        (None, Some(query)) if !has_clauses => {
+            Ok(connection.create_materialized_view_from_query(name, query))
+        }
+        (Some(source), None) => Ok(connection.create_materialized_view(name, source)),
+        (None, None) => Err(PyValueError::new_err(
+            "either source or query must be provided",
+        )),
+        _ => Err(PyValueError::new_err(
+            "query cannot be combined with source, select, where or limit",
+        )),
+    }
+}
+
 #[pyclass]
 pub struct Connection {
     inner: Option<LanceConnection>,
@@ -483,21 +505,29 @@ impl Connection {
     }
 
     #[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))]
+    #[pyo3(signature = (name, source=None, projections=None, filter=None, limit=None, with_no_data=false, auto_refresh_enabled=None, auto_refresh_interval_seconds=None, *, query=None))]
     pub fn create_materialized_view(
         self_: PyRef<'_, Self>,
         name: String,
-        source: String,
+        source: Option<String>,
         projections: Option<Vec<(String, String)>>,
         filter: Option<String>,
         limit: Option<u64>,
         with_no_data: bool,
         auto_refresh_enabled: Option<bool>,
         auto_refresh_interval_seconds: Option<f64>,
+        query: Option<String>,
     ) -> PyResult<Bound<'_, PyAny>> {
         let inner = self_.get_inner()?.clone();
+        let builder = materialized_view_builder(
+            &inner,
+            name,
+            source,
+            query,
+            projections.is_some() || filter.is_some() || limit.is_some(),
+        )?;
         future_into_py(self_.py(), async move {
-            let mut builder = inner.create_materialized_view(name, source);
+            let mut builder = builder;
             if let Some(enabled) = auto_refresh_enabled {
                 builder = builder.auto_refresh_enabled(enabled);
             }
@@ -520,21 +550,29 @@ impl Connection {
     }
 
     #[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))]
+    #[pyo3(signature = (name, source=None, projections=None, filter=None, limit=None, with_no_data=false, auto_refresh_enabled=None, auto_refresh_interval_seconds=None, *, query=None))]
     pub fn create_materialized_view_async(
         self_: PyRef<'_, Self>,
         name: String,
-        source: String,
+        source: Option<String>,
         projections: Option<Vec<(String, String)>>,
         filter: Option<String>,
         limit: Option<u64>,
         with_no_data: bool,
         auto_refresh_enabled: Option<bool>,
         auto_refresh_interval_seconds: Option<f64>,
+        query: Option<String>,
     ) -> PyResult<Bound<'_, PyAny>> {
         let inner = self_.get_inner()?.clone();
+        let builder = materialized_view_builder(
+            &inner,
+            name,
+            source,
+            query,
+            projections.is_some() || filter.is_some() || limit.is_some(),
+        )?;
         future_into_py(self_.py(), async move {
-            let mut builder = inner.create_materialized_view(name, source);
+            let mut builder = builder;
             if let Some(enabled) = auto_refresh_enabled {
                 builder = builder.auto_refresh_enabled(enabled);
             }