Merge main into gatekeeper/fix-4260-1

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,9 +31,104 @@ SelectArg = Union[
 ]
 
 
-DEFINITION_FORMAT = 2
+@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``. A ``kind`` key beside it is for
+format 2 being a query with ``GROUP BY``, format 3 native pairs and format 4
+native dedup results. A ``kind`` key beside it is for
 readers older than the format number."""
 
 

python/python/lancedb/namespace.py

@@ -68,6 +68,7 @@ from lance_namespace import (
 from lancedb.materialized_view import (
     AsyncMaterializedView,
     MaterializedView,
+    MaterializedViewSource,
     SelectArg,
 )
 from lancedb.table import AsyncTable, LanceTable, Table
@@ -632,7 +633,7 @@ class LanceNamespaceDBConnection(DBConnection):
     def create_materialized_view(
         self,
         name: str,
-        source: str,
+        source: MaterializedViewSource,
         *,
         select: "SelectArg" = None,
         where: Optional[str] = None,
@@ -662,7 +663,7 @@ class LanceNamespaceDBConnection(DBConnection):
     def create_materialized_view_async(
         self,
         name: str,
-        source: str,
+        source: MaterializedViewSource,
         *,
         select: "SelectArg" = None,
         where: Optional[str] = None,
@@ -1242,7 +1243,7 @@ class AsyncLanceNamespaceDBConnection:
     async def create_materialized_view(
         self,
         name: str,
-        source: str,
+        source: MaterializedViewSource,
         *,
         select: "SelectArg" = None,
         where: Optional[str] = None,
@@ -1265,7 +1266,7 @@ class AsyncLanceNamespaceDBConnection:
     async def create_materialized_view_async(
         self,
         name: str,
-        source: str,
+        source: MaterializedViewSource,
         *,
         select: "SelectArg" = None,
         where: Optional[str] = None,

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/lancedb/rerankers/cross_encoder.py

@@ -31,6 +31,12 @@ class CrossEncoderReranker(Reranker):
     trust_remote_code : bool, default True
         If True, will trust the remote code to be safe. If False, will not trust
         the remote code and will not run it
+    batch_size : int, default 32
+        The number of query-passage pairs scored per forward pass.
+    **kwargs
+        Additional keyword arguments passed to
+        ``sentence_transformers.CrossEncoder``, for example ``max_length`` or
+        ``model_kwargs={"torch_dtype": "bfloat16"}``.
     """
 
     def __init__(
@@ -40,6 +46,8 @@ class CrossEncoderReranker(Reranker):
         device: Union[str, None] = None,
         return_score="relevance",
         trust_remote_code: bool = True,
+        batch_size: int = 32,
+        **kwargs,
     ):
         super().__init__(return_score)
         torch = attempt_import_or_raise("torch")
@@ -47,6 +55,8 @@ class CrossEncoderReranker(Reranker):
         self.column = column
         self.device = device
         self.trust_remote_code = trust_remote_code
+        self.batch_size = batch_size
+        self.kwargs = kwargs
         if self.device is None:
             self.device = "cuda" if torch.cuda.is_available() else "cpu"
 
@@ -61,6 +71,7 @@ class CrossEncoderReranker(Reranker):
             self.model_name,
             device=self.device,
             trust_remote_code=self.trust_remote_code,
+            **self.kwargs,
         )
 
         return cross_encoder
@@ -71,7 +82,7 @@ class CrossEncoderReranker(Reranker):
             return result_set
         passages = result_set[self.column].to_pylist()
         cross_inp = [[query, passage] for passage in passages]
-        cross_scores = self.model.predict(cross_inp)
+        cross_scores = self.model.predict(cross_inp, batch_size=self.batch_size)
         result_set = result_set.append_column(
             "_relevance_score", pa.array(cross_scores, type=pa.float32())
         )

python/python/tests/test_materialized_views.py

@@ -520,10 +520,166 @@ def test_stored_queries_and_legacy_layouts_are_read():
         "SELECT * FROM people"
     )
 
+    for fmt, function in [(3, "vector_duplicate_pairs"), (4, "vector_dedup")]:
+        native = f"SELECT * FROM {function}('images', 3, 'phash', 4)"
+        assert read({"format": fmt, "query": native}).query == native
+
     # A newer writer's layout is reported, never guessed at.
     for newer in (
-        {"format": 3, "query": query},
+        {"format": 5, "query": query},
         {"kind": "select_v3", "source_table": "people"},
     ):
         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/python/tests/test_rerankers.py

@@ -481,6 +481,14 @@ def test_cross_encoder_reranker(tmp_path):
     _run_test_reranker(reranker, table, "single player experience", None, schema)
 
 
+def test_cross_encoder_reranker_kwargs(tmp_path):
+    pytest.importorskip("sentence_transformers")
+    reranker = CrossEncoderReranker(batch_size=4, max_length=64)
+    assert reranker.model.max_length == 64
+    table, schema = get_test_table(tmp_path)
+    _run_test_reranker(reranker, table, "single player experience", None, schema)
+
+
 def test_colbert_reranker(tmp_path):
     pytest.importorskip("rerankers")
     reranker = ColbertReranker()

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);
             }

python/vendor/lance-arrow/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lance-arrow"
-version = "13.0.0-beta.14"
+version = "13.0.0-beta.15"
 edition = "2024"
 authors = ["Lance Devs <dev@lance.org>"]
 license = "Apache-2.0"

python/vendor/lance-arrow/README.lancedb.md

@@ -1,5 +1,5 @@
-This directory contains `rust/lance-arrow` from Lance tag `v13.0.0-beta.14`
-(commit `b80ac1d9f3c24b47bd8e297d03053ea35d216a9f`), licensed under Apache-2.0.
+This directory contains `rust/lance-arrow` from Lance tag `v13.0.0-beta.15`
+(commit `db53dcf367088b32ae3a7461407d43593759ae3e`), licensed under Apache-2.0.
 
 LanceDB carries a local fix in `src/lib.rs` for sliced child validity bitmaps
 until a Lance release includes the fix. The patch is selected in the root