feat(python): declare indexed dedup sources for materialized views

docs/src/python/python.md

@@ -198,6 +198,33 @@ listing a storage directory.
 
 ::: lancedb.materialized_view.MaterializedViewDefinition
 
+::: lancedb.vector_dedup
+
+::: lancedb.materialized_view.VectorDedupSource
+
+For an existing indexed source, declare and materialize a dedup result without
+writing SQL. Creation captures the source version once and waits for the complete
+result; remote connections use the service's MV jobs.
+
+```python
+from lancedb import vector_dedup
+
+view = db.create_materialized_view(
+    "images_clean",
+    vector_dedup("images", column="phash", distance_threshold=4),
+)
+cleaned = view.table
+```
+
+To submit without waiting, call `db.create_materialized_view_async(...)` with the
+same source expression and wait on the returned job. To declare only, pass
+`with_no_data=True`, then use `view.refresh_async().wait()` when ready. Pass
+`dataset_version=...` to `vector_dedup` for an explicit snapshot. The default
+policy retains direct representatives (A-B-C without A-C keeps A and C), preserves
+the source, and materializes its original columns. Only within-segment/partition
+pairs are covered, using the index's distance representation.
+
+
 ## Views
 
 ::: lancedb.view.ViewDescription

python/python/lancedb/__init__.py

@@ -46,6 +46,7 @@ from .materialized_view import (
     AsyncMaterializedView,
     MaterializedView,
     MaterializedViewDefinition,
+    vector_dedup as vector_dedup,
 )
 from .view import ViewDescription as ViewDescription
 from .table import AsyncTable, Table

python/python/lancedb/_lancedb.pyi

@@ -247,6 +247,8 @@ class Connection(object):
         filter: Optional[str] = None,
         limit: Optional[int] = None,
         with_no_data: bool = False,
+        *,
+        vector_source_json: Optional[str] = None,
     ) -> Table: ...
     async def create_materialized_view_async(
         self,
@@ -256,6 +258,8 @@ class Connection(object):
         filter: Optional[str] = None,
         limit: Optional[int] = None,
         with_no_data: bool = False,
+        *,
+        vector_source_json: Optional[str] = None,
     ) -> Job: ...
     async def list_materialized_views(self) -> List[str]: ...
     async def drop_materialized_view(

python/python/lancedb/db.py

@@ -56,6 +56,8 @@ from .sql import QueryDescription
 from .materialized_view import (
     AsyncMaterializedView,
     MaterializedView,
+    MaterializedViewSource,
+    VectorDedupSource,
     SelectArg,
     normalize_select,
 )
@@ -555,7 +557,7 @@ class DBConnection(EnforceOverrides):
     def create_materialized_view(
         self,
         name: str,
-        source: str,
+        source: MaterializedViewSource,
         *,
         select: SelectArg = None,
         where: Optional[str] = None,
@@ -569,17 +571,21 @@ class DBConnection(EnforceOverrides):
         table. The view is a normal table: it can be queried, indexed and
         searched, and it appears in ``table_names``.
 
-        The source table must have stable row ids (create it with the
+        Ordinary source tables must have stable row ids (create them with the
         ``new_table_enable_stable_row_ids`` storage option): they keep the
         view's provenance valid across source compactions, and cannot be
-        enabled after a table exists.
+        enabled after a table exists. Native dedup sources also support physical
+        row IDs interpreted against their pinned snapshot.
 
         Parameters
         ----------
         name: str
             The name of the view.
-        source: str
-            The name of the source table, in this database.
+        source: str or VectorDedupSource
+            The name of the source table, or an indexed dedup source declared
+            with [vector_dedup][lancedb.vector_dedup]. Dedup sources capture a
+            fixed snapshot and preserve the source table. They accept no
+            additional select, where or limit options.
         select: list or dict, optional
             The view's columns: column names, ``(alias, SQL expression)``
             pairs, or a dict of the same. Omitting it selects every source
@@ -602,7 +608,7 @@ class DBConnection(EnforceOverrides):
     def create_materialized_view_async(
         self,
         name: str,
-        source: str,
+        source: MaterializedViewSource,
         *,
         select: SelectArg = None,
         where: Optional[str] = None,
@@ -1558,7 +1564,7 @@ class LanceDBConnection(DBConnection):
     def create_materialized_view(
         self,
         name: str,
-        source: str,
+        source: MaterializedViewSource,
         *,
         select: SelectArg = None,
         where: Optional[str] = None,
@@ -1603,7 +1609,7 @@ class LanceDBConnection(DBConnection):
     def create_materialized_view_async(
         self,
         name: str,
-        source: str,
+        source: MaterializedViewSource,
         *,
         select: SelectArg = None,
         where: Optional[str] = None,
@@ -2498,10 +2504,31 @@ class AsyncConnection(object):
             await tbl.checkout(version)
         return tbl
 
+    async def _materialized_view_source(
+        self,
+        source: MaterializedViewSource,
+        select: SelectArg,
+        where: Optional[str],
+        limit: Optional[int],
+    ) -> Tuple[str, Dict[str, str]]:
+        if not isinstance(source, VectorDedupSource):
+            return source, {}
+        if select is not None or where is not None or limit is not None:
+            raise ValueError(
+                "vector_dedup cannot be combined with select, where or limit"
+            )
+        version = source.dataset_version
+        if version is None:
+            table = await self.open_table(source.source)
+            version = await table.version()
+        return source.source, {
+            "vector_source_json": source._native_source_json(version)
+        }
+
     async def create_materialized_view(
         self,
         name: str,
-        source: str,
+        source: MaterializedViewSource,
         *,
         select: SelectArg = None,
         where: Optional[str] = None,
@@ -2512,6 +2539,9 @@ class AsyncConnection(object):
         See
         [DBConnection.create_materialized_view][lancedb.DBConnection.create_materialized_view].
         """
+        source, native_options = await self._materialized_view_source(
+            source, select, where, limit
+        )
         inner = await self._inner.create_materialized_view(
             name,
             source,
@@ -2519,13 +2549,14 @@ class AsyncConnection(object):
             filter=where,
             limit=limit,
             with_no_data=with_no_data,
+            **native_options,
         )
         return AsyncMaterializedView(AsyncTable(inner))
 
     async def create_materialized_view_async(
         self,
         name: str,
-        source: str,
+        source: MaterializedViewSource,
         *,
         select: SelectArg = None,
         where: Optional[str] = None,
@@ -2536,6 +2567,9 @@ class AsyncConnection(object):
 
         Wait for the returned job before opening or querying the view.
         """
+        source, native_options = await self._materialized_view_source(
+            source, select, where, limit
+        )
         inner = await self._inner.create_materialized_view_async(
             name,
             source,
@@ -2543,6 +2577,7 @@ class AsyncConnection(object):
             filter=where,
             limit=limit,
             with_no_data=with_no_data,
+            **native_options,
         )
         return AsyncJob(inner)
 

python/python/lancedb/materialized_view.py

@@ -7,6 +7,7 @@ maintained by refresh. See ``DBConnection.create_materialized_view``."""
 from __future__ import annotations
 
 import json
+import math
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Dict, List, Optional, Sequence, Tuple, Union
 
@@ -20,6 +21,7 @@ if TYPE_CHECKING:
     from .table import AsyncTable, LanceTable
 
 DEFINITION_META_KEY = b"mv.definition"
+_MAX_FLOAT32 = float.fromhex("0x1.fffffep+127")
 
 SelectArg = Union[
     str,
@@ -29,6 +31,100 @@ SelectArg = Union[
 ]
 
 
+@dataclass(frozen=True)
+class VectorDedupSource:
+    """A declarative source of retained original rows from an indexed snapshot.
+
+    Construct with [vector_dedup][lancedb.vector_dedup] and pass it as the
+    ``source`` of ``create_materialized_view`` or
+    ``create_materialized_view_async``. An omitted ``dataset_version`` is
+    captured once when creation is submitted, not when this object is made.
+    Refreshes keep that captured version. The source table is never modified.
+    """
+
+    source: str
+    column: str
+    distance_threshold: float
+    dataset_version: Optional[int] = None
+
+    def __post_init__(self):
+        for name in ("source", "column"):
+            value = getattr(self, name)
+            if not isinstance(value, str) or not value:
+                raise ValueError(f"{name} must be a non-empty string")
+        if self.dataset_version is not None and (
+            type(self.dataset_version) is not int
+            or not 0 < self.dataset_version < 2**64
+        ):
+            raise ValueError("dataset_version must be a positive uint64 integer")
+        value = self.distance_threshold
+        if isinstance(value, bool) or not isinstance(value, (int, float)):
+            raise TypeError("distance_threshold must be a number")
+        try:
+            value = float(value)
+        except OverflowError as exc:
+            raise ValueError("distance_threshold must fit a finite float32") from exc
+        if not math.isfinite(value) or abs(value) > _MAX_FLOAT32:
+            raise ValueError("distance_threshold must fit a finite float32")
+        object.__setattr__(self, "distance_threshold", value)
+
+    def _native_source_json(self, version: int) -> str:
+        return json.dumps(
+            {
+                "kind": "Dedup",
+                "dataset_version": version,
+                "column": self.column,
+                "distance_threshold": repr(self.distance_threshold),
+            },
+            allow_nan=False,
+        )
+
+
+def vector_dedup(
+    source: str,
+    *,
+    column: str,
+    distance_threshold: float,
+    dataset_version: Optional[int] = None,
+) -> VectorDedupSource:
+    """Declare an indexed dedup source for a materialized view.
+
+    The default policy keeps the smallest unassigned snapshot row ID and removes
+    only its direct qualifying neighbors: an A-B-C chain without an A-C edge
+    retains A and C. Rows with no pair, including null vectors, survive.
+    Only same-segment/same-IVF-partition pairs are considered; quantized distances
+    use reconstructed index vectors. The index must cover the selected snapshot.
+
+    ``source`` is an exact table name in the connection's root namespace, just
+    as for ``create_materialized_view``. ``dataset_version=None`` captures the
+    source's version when creation is submitted. Pass an explicit positive
+    version to reproduce a previous snapshot. The result contains original
+    source columns and rows. Do not combine this source with ``select``,
+    ``where`` or ``limit``; query the resulting table for further transforms.
+
+    Creation follows the usual MV contract: by default it waits for the result.
+    Use ``with_no_data=True`` to declare only, or
+    ``create_materialized_view_async`` to get a job handle. On Cloud/Enterprise,
+    execution uses server-side MV registry jobs; local connections run in process.
+
+    Examples
+    --------
+    Given a connected ``db`` and an indexed ``images.phash`` column::
+
+        from lancedb import vector_dedup
+
+        view = db.create_materialized_view(
+            "images_clean",
+            vector_dedup("images", column="phash", distance_threshold=4),
+        )
+        cleaned = view.table
+    """
+    return VectorDedupSource(source, column, distance_threshold, dataset_version)
+
+
+MaterializedViewSource = Union[str, VectorDedupSource]
+
+
 DEFINITION_FORMAT = 4
 """The newest stored layout this version reads: ``{"format": N, "query": "<SQL>"}``,
 format 2 being a query with ``GROUP BY``, format 3 native pairs and format 4

python/python/lancedb/remote/db.py

@@ -39,7 +39,7 @@ from ..functions import FunctionVersion, UdfDefinition
 from ..job import AsyncJob, Job
 from ..sql import Query as SqlQuery
 from ..sql import QueryDescription
-from ..materialized_view import MaterializedView, SelectArg
+from ..materialized_view import MaterializedView, MaterializedViewSource, SelectArg
 from ..secrets import EnvVarSecret, SecretInfo
 from ..view import ViewDescription
 
@@ -713,7 +713,7 @@ class RemoteDBConnection(DBConnection):
     def create_materialized_view(
         self,
         name: str,
-        source: str,
+        source: MaterializedViewSource,
         *,
         select: SelectArg = None,
         where: Optional[str] = None,
@@ -745,7 +745,7 @@ class RemoteDBConnection(DBConnection):
     def create_materialized_view_async(
         self,
         name: str,
-        source: str,
+        source: MaterializedViewSource,
         *,
         select: SelectArg = None,
         where: Optional[str] = None,

python/python/tests/test_materialized_views.py

@@ -531,3 +531,155 @@ def test_stored_queries_and_legacy_layouts_are_read():
     ):
         with pytest.raises(NotImplementedError, match="cannot refresh"):
             read(newer)
+
+
+def indexed_dedup_source(db, name="images"):
+    import pyarrow as pa
+    from lancedb.index import IvfFlat
+
+    vectors = pa.FixedSizeListArray.from_arrays(
+        pa.array(
+            [v for i in range(256) for v in (float(i if i < 3 else i * 10), 0.0)],
+            type=pa.float32(),
+        ),
+        2,
+    )
+    table = db.create_table(name, pa.table({"id": range(256), "vector": vectors}))
+    table.create_index("vector", config=IvfFlat(num_partitions=1))
+    return table
+
+
+@pytest.mark.parametrize("stable", [False, True])
+def test_vector_dedup_source_captures_snapshot_and_preserves_source(tmp_path, stable):
+    db = lancedb.connect(
+        tmp_path,
+        storage_options={"new_table_enable_stable_row_ids": str(stable).lower()},
+    )
+    source = indexed_dedup_source(db, "images.with.dot")
+    version = source.version
+    expression = lancedb.vector_dedup(
+        source.name, column="vector", distance_threshold=1
+    )
+    view = db.create_materialized_view("clean", expression)
+    assert "vector_dedup(" in view.definition.query
+    assert f", {version}, 'vector', 1.0)" in view.definition.query
+    assert view.table.count_rows() == 255
+    assert [r["id"] for r in view.table.search().where("id < 3").to_list()] == [0, 2]
+    assert source.version == version
+    assert source.count_rows() == 256
+
+    source.delete("id = 0")
+    view.refresh_async().wait()
+    assert view.table.search().where("id = 0").to_list() == [
+        {"id": 0, "vector": [0.0, 0.0]}
+    ]
+    assert expression.dataset_version is None  # reusable immutable declaration
+
+
+def test_vector_dedup_create_job_and_explicit_snapshot(tmp_path):
+    db = lancedb.connect(tmp_path)
+    source = indexed_dedup_source(db)
+    version = source.version
+    source.delete("id = 0")
+    job = db.create_materialized_view_async(
+        "clean",
+        lancedb.vector_dedup(
+            "images", column="vector", distance_threshold=1, dataset_version=version
+        ),
+    )
+    assert job.wait() is None
+    view = db.open_materialized_view("clean")
+    assert view.table.count_rows() == 255
+    assert len(view.table.search().where("id = 0").to_list()) == 1
+
+
+@pytest.mark.asyncio
+async def test_async_vector_dedup_declaration_and_creation(tmp_path):
+    sync_db = lancedb.connect(tmp_path)
+    source = indexed_dedup_source(sync_db)
+    db = await lancedb.connect_async(tmp_path)
+    expression = lancedb.vector_dedup("images", column="vector", distance_threshold=1)
+    view = await db.create_materialized_view("clean", expression, with_no_data=True)
+    assert await view.table.count_rows() == 0
+    # The snapshot is captured at declaration even if refresh comes later.
+    source.delete("id = 0")
+    job = await view.refresh_async()
+    await job.wait()
+    assert await view.table.count_rows() == 255
+    another = await db.create_materialized_view_async(
+        "later", expression, with_no_data=True
+    )
+    await another.wait()
+    later = await db.open_materialized_view("later")
+    assert (await later.definition()).query != (await view.definition()).query
+    assert await later.table.count_rows() == 0
+
+
+@pytest.mark.parametrize(
+    "options", [{"select": ["id"]}, {"select": []}, {"where": "id > 0"}, {"limit": 1}]
+)
+def test_vector_dedup_rejects_extra_clauses_before_opening_source(tmp_path, options):
+    db = lancedb.connect(tmp_path)
+    expression = lancedb.vector_dedup("missing", column="vector", distance_threshold=1)
+    with pytest.raises(ValueError, match="cannot be combined"):
+        db.create_materialized_view("bad", expression, **options)
+    assert db.list_tables().tables == []
+
+
+@pytest.mark.parametrize("version", [0, -1, True, 1.5, "2", 2**64])
+def test_vector_dedup_rejects_invalid_snapshot(version):
+    with pytest.raises(ValueError, match="dataset_version"):
+        lancedb.vector_dedup(
+            "images", column="vector", distance_threshold=1, dataset_version=version
+        )
+
+
+@pytest.mark.parametrize(
+    "threshold", [float("nan"), float("inf"), -float("inf"), 1e100, 10**1000, True, "4"]
+)
+def test_vector_dedup_rejects_invalid_threshold(threshold):
+    with pytest.raises((ValueError, TypeError), match="distance_threshold"):
+        lancedb.vector_dedup("images", column="vector", distance_threshold=threshold)
+
+
+def test_vector_dedup_descriptor_is_frozen():
+    from dataclasses import FrozenInstanceError
+
+    expression = lancedb.vector_dedup("images", column="vector", distance_threshold=1)
+    with pytest.raises(FrozenInstanceError):
+        expression.column = "other"
+
+
+@pytest.mark.parametrize("with_no_data", [False, True])
+def test_remote_vector_dedup_uses_mv_endpoint_without_sql_client(with_no_data):
+    with mock_remote_materialized_view_create() as (host, requests):
+        db = lancedb.connect(
+            "db://dev",
+            api_key="fake",
+            host_override=host,
+            sql_host_override="grpc://127.0.0.1:1",
+            client_config={"retry_config": {"retries": 0}},
+        )
+        job = db.create_materialized_view_async(
+            "clean",
+            lancedb.vector_dedup(
+                "images.with'dot",
+                column="vector'field",
+                distance_threshold=4,
+                dataset_version=7,
+            ),
+            with_no_data=with_no_data,
+        )
+        assert job.id == "mv-create-123"
+    assert requests == [
+        (
+            "/v1/materialized_view/clean/create",
+            {
+                "query": (
+                    "SELECT * FROM vector_dedup('`images.with''dot`', "
+                    "7, 'vector''field', 4.0)"
+                ),
+                "with_no_data": with_no_data,
+            },
+        )
+    ]

python/src/connection.rs

@@ -407,7 +407,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, *, vector_source_json=None))]
     pub fn create_materialized_view(
         self_: PyRef<'_, Self>,
         name: String,
@@ -416,10 +417,17 @@ impl Connection {
         filter: Option<String>,
         limit: Option<u64>,
         with_no_data: bool,
+        vector_source_json: Option<String>,
     ) -> 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(source) = vector_source_json {
+                builder = builder.vector_source(
+                    serde_json::from_str(&source)
+                        .map_err(|e| PyValueError::new_err(e.to_string()))?,
+                );
+            }
             if let Some(projections) = projections {
                 builder = builder.select(projections);
             }
@@ -435,7 +443,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, *, vector_source_json=None))]
     pub fn create_materialized_view_async(
         self_: PyRef<'_, Self>,
         name: String,
@@ -444,10 +453,17 @@ impl Connection {
         filter: Option<String>,
         limit: Option<u64>,
         with_no_data: bool,
+        vector_source_json: Option<String>,
     ) -> 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(source) = vector_source_json {
+                builder = builder.vector_source(
+                    serde_json::from_str(&source)
+                        .map_err(|e| PyValueError::new_err(e.to_string()))?,
+                );
+            }
             if let Some(projections) = projections {
                 builder = builder.select(projections);
             }

rust/lancedb/src/materialized_view.rs

@@ -1792,6 +1792,7 @@ pub struct CreateMaterializedViewBuilder {
     namespace: Vec<String>,
     source: String,
     source_namespace: Vec<String>,
+    vector_source: Option<VectorSource>,
     projections: Vec<(String, String)>,
     filter: Option<String>,
     limit: Option<u64>,
@@ -1806,6 +1807,7 @@ impl CreateMaterializedViewBuilder {
             namespace: Vec::new(),
             source,
             source_namespace: Vec::new(),
+            vector_source: None,
             projections: Vec::new(),
             filter: None,
             limit: None,
@@ -1826,6 +1828,38 @@ impl CreateMaterializedViewBuilder {
         self
     }
 
+    /// Read native pairs or retained original rows from a fixed indexed snapshot.
+    /// This source selects its full output schema and cannot be combined with
+    /// projections, filters or limits. Remote creation uses the ordinary MV
+    /// endpoint and its registry job; it does not execute a SQL query client-side.
+    pub fn vector_source(mut self, source: VectorSource) -> Self {
+        self.vector_source = Some(source);
+        self
+    }
+
+    fn native_definition(&self) -> Result<Option<MaterializedViewDefinition>> {
+        let Some(source) = &self.vector_source else {
+            return Ok(None);
+        };
+        source.config()?;
+        if !self.projections.is_empty() || self.filter.is_some() || self.limit.is_some() {
+            return Err(Error::InvalidInput {
+                message: "native vector sources cannot be combined with select, where or limit"
+                    .into(),
+            });
+        }
+        Ok(Some(MaterializedViewDefinition {
+            vector_source: Some(Box::new(source.clone())),
+            source_table: self.source.clone(),
+            source_namespace: self.source_namespace.clone(),
+            projections: vec![ViewProjection::star()],
+            filter: None,
+            lateral: None,
+            group_by: Vec::new(),
+            limit: None,
+        }))
+    }
+
     /// The view's columns, as `(name, SQL expression)` pairs. Not calling
     /// this selects every source column, expanded at creation time.
     pub fn select(
@@ -1858,7 +1892,10 @@ impl CreateMaterializedViewBuilder {
         self
     }
 
-    fn query(&self) -> String {
+    fn query(&self) -> Result<String> {
+        if let Some(definition) = self.native_definition()? {
+            return Ok(definition.to_sql());
+        }
         fn quote(name: &str) -> String {
             format!("\"{}\"", name.replace('"', "\"\""))
         }
@@ -1887,12 +1924,13 @@ impl CreateMaterializedViewBuilder {
         if let Some(limit) = self.limit {
             query.push_str(&format!(" LIMIT {limit}"));
         }
-        query
+        Ok(query)
     }
 
     /// Submit creation and initial population, returning a [`Job`] that
-    /// settles when the view is ready. The source must keep stable row ids --
-    /// they hold provenance across compaction, and cannot be enabled later.
+    /// settles when the view is ready. Ordinary sources must keep stable row
+    /// ids for provenance across compaction. Native vector sources also support
+    /// physical row IDs because their definitions pin the dataset version.
     pub async fn execute_async(self) -> Result<Job> {
         if self.connection.uri().starts_with("db://") {
             return self
@@ -1901,7 +1939,7 @@ impl CreateMaterializedViewBuilder {
                 .create_materialized_view_async(CreateMaterializedViewRequest {
                     name: self.name.clone(),
                     namespace_path: self.namespace.clone(),
-                    query: self.query(),
+                    query: self.query()?,
                     with_no_data: self.with_no_data,
                 })
                 .await;
@@ -1929,19 +1967,24 @@ impl CreateMaterializedViewBuilder {
     }
 
     async fn execute_native(self) -> Result<MaterializedView> {
+        let native_definition = self.native_definition()?;
         let source = self
             .connection
             .open_table(&self.source)
             .namespace(self.source_namespace.clone())
             .execute()
             .await?;
-        let prepared = prepare_declaration(
-            &source,
-            (!self.projections.is_empty()).then_some(self.projections.as_slice()),
-            self.filter.as_deref(),
-            self.limit,
-        )
-        .await?;
+        let prepared = if let Some(definition) = native_definition {
+            prepare_definition(&source, definition).await?
+        } else {
+            prepare_declaration(
+                &source,
+                (!self.projections.is_empty()).then_some(self.projections.as_slice()),
+                self.filter.as_deref(),
+                self.limit,
+            )
+            .await?
+        };
         let view = prepared.create_in(&self.namespace, &self.name).await?;
         if !self.with_no_data {
             view.refresh().execute().await?;
@@ -2384,13 +2427,49 @@ mod tests {
             .select([("double\"age", "age * 2")])
             .only_if("age >= 18")
             .limit(10)
-            .query();
+            .query()
+            .unwrap();
         assert_eq!(
             query,
             "SELECT age * 2 AS \"double\"\"age\" FROM \"raw data\".\"odd\"\"source\" WHERE age >= 18 LIMIT 10"
         );
     }
 
+    #[tokio::test]
+    async fn native_vector_builder_preserves_source_and_rejects_extra_clauses() {
+        let conn = people_db().await;
+        let source = VectorSource {
+            kind: VectorSourceKind::Dedup,
+            dataset_version: 7,
+            column: "vector'field".into(),
+            distance_threshold: "4.0".into(),
+        };
+        let builder = || {
+            conn.create_materialized_view("clean", "odd.source'name")
+                .source_namespace(vec!["raw space".into()])
+                .vector_source(source.clone())
+        };
+        let query = builder().query().unwrap();
+        let definition = MaterializedViewDefinition::from_sql(&query).unwrap();
+        assert_eq!(definition.source_table, "odd.source'name");
+        assert_eq!(definition.source_namespace, ["raw space"]);
+        assert_eq!(definition.vector_source.as_deref(), Some(&source));
+        assert!(definition.selects_star());
+        assert!(builder().select([("id", "id")]).query().is_err());
+        assert!(builder().only_if("true").query().is_err());
+        assert!(builder().limit(1).query().is_err());
+        let invalid_source = VectorSource {
+            dataset_version: 0,
+            ..source
+        };
+        assert!(
+            conn.create_materialized_view("bad", "people")
+                .vector_source(invalid_source)
+                .query()
+                .is_err()
+        );
+    }
+
     /// No projection selects every source column, expanded now: the schema
     /// captured at creation is the definition.
     #[tokio::test]