Merge main into gatekeeper/fix-4284-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(
@@ -514,6 +518,9 @@ class Table:
     async def set_lsm_write_spec(self, spec: LsmWriteSpec) -> None: ...
     async def unset_lsm_write_spec(self) -> None: ...
     async def get_lsm_write_spec(self) -> Optional[LsmWriteSpec]: ...
+    # WAL-PK-FUSION: delete both.
+    def hybrid_pk_fusion_learned(self) -> bool: ...
+    def note_hybrid_pk_fusion(self) -> None: ...
     async def checkpoint_lsm(self) -> None: ...
     async def flush_lsm(self) -> None: ...
     async def compact_lsm(self) -> None: ...

python/python/lancedb/_wal_hybrid.py

@@ -0,0 +1,282 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""WAL-PK-FUSION: hybrid search's primary-key fallback for MemWAL tables.
+
+Hybrid search joins its vector and full-text legs on `_rowid`, which a MemWAL
+table cannot supply: the fresh tier has no stable row id, so the server rejects
+`with_row_id` before it plans. When that happens the legs are joined on a
+surrogate `_rowid` built from the table's primary key instead.
+
+This is a stopgap until MemWAL supports `_rowid`, and everything it needs lives
+here so it can be removed in one pass:
+
+1. Delete this file and `tests/test_wal_hybrid.py`.
+2. Run `grep -rn WAL-PK-FUSION python/ rust/` and follow the note at each
+   marker; the Rust side has its own copy in
+   `rust/lancedb/src/query/wal_fusion.rs`.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Awaitable, Callable, List, Optional, Tuple, TypeVar
+
+import pyarrow as pa
+
+from ._blob import blob_v2_projection_sources
+
+if TYPE_CHECKING:
+    from .types import QueryProjection
+
+T = TypeVar("T")
+
+
+# Lance records the unenforced primary key as field metadata, which survives the
+# `describe` round trip — so a remote table can resolve the key without opening
+# the dataset. Keys are bytes on a pyarrow field.
+_PK_MARKER = b"lance-schema:unenforced-primary-key"
+_PK_POSITION_MARKER = b"lance-schema:unenforced-primary-key:position"
+_TRUTHY = {"1", "true", "on", "yes", "y"}
+
+
+def _pk_columns(schema: pa.Schema) -> List[str]:
+    """The unenforced primary key columns of `schema`, in key order.
+
+    Mirrors Lance's ordering: fields with an explicit 1-based position first, in
+    position order, then fields carrying only the legacy boolean marker, in
+    declaration order.
+    """
+    keyed = []
+    for index, field in enumerate(schema):
+        metadata = field.metadata or {}
+        position = metadata.get(_PK_POSITION_MARKER)
+        if position is not None:
+            try:
+                position = int(position)
+            except ValueError:
+                position = None
+        if position is None:
+            legacy = metadata.get(_PK_MARKER)
+            if legacy is None or legacy.decode().lower() not in _TRUTHY:
+                continue
+            position = 0
+        keyed.append((position == 0, position, index, field.name))
+    return [name for *_, name in sorted(keyed)]
+
+
+# Both layers raise this same sentence verbatim -- sophon's
+# `LSM_WITH_ROW_ID_UNSUPPORTED` says so explicitly, and the OSS LSM scanner
+# builds it from the same words -- so this substring survives either path and
+# the wrapping each adds. It is a coupling to a message rather than a code,
+# because the refusal arrives as a generic 400; if the server ever rewords it
+# the fallback stops firing and hybrid on a MemWAL table errors visibly again,
+# which is loud rather than silently wrong.
+_WAL_ROW_ID_REFUSAL = "does not support with_row_id"
+
+WAL_ROW_ID_UNSUPPORTED = (
+    "hybrid search on a MemWAL table cannot return _rowid: the fresh tier has "
+    "no stable row id, and the ids the fusion joins on are synthesized from "
+    "the primary key. Set use_lsm(False) to read the base table only (results "
+    "will exclude un-compacted MemWAL data)"
+)
+
+
+def _is_wal_row_id_refusal(error: BaseException) -> bool:
+    """Whether `error` is the server declining `_rowid` because the table is
+    MemWAL-backed, as opposed to any other failure the query might hit."""
+    return _WAL_ROW_ID_REFUSAL in str(error)
+
+
+def _with_pk_columns(columns, pk_columns: List[str]):
+    """Add the key columns a projection is missing.
+
+    Returns the projection to use and the columns added, which the caller drops
+    from the output again so an injected key never reaches a caller who did not
+    select it.
+    """
+    if not columns:
+        # Every non-system column, the key among them.
+        return columns, []
+    # `PyQueryRequest.select` reports a dynamic projection as (name, sql) pairs.
+    if isinstance(columns, (list, tuple)) and isinstance(columns[0], (list, tuple)):
+        columns = dict(columns)
+    missing = [c for c in pk_columns if c not in columns]
+    if isinstance(columns, dict):
+        return {**columns, **{c: c for c in missing}}, missing
+    return list(columns) + missing, missing
+
+
+def _stamp_surrogate_row_ids(
+    tables: List[pa.Table], pk_columns: List[str]
+) -> List[pa.Table]:
+    """Give each table a `_rowid` column derived from its primary key.
+
+    The fusion only ever compares that column for equality — to deduplicate, to
+    group by in a reranker, and to restore the original scores — so on a MemWAL
+    table, where the server has no stable row id to hand out, any injective
+    mapping of the primary key does the same job. Ids are assigned in first-seen
+    order across `tables`, so passing them in fusion order (vector, then FTS)
+    preserves "first occurrence wins" and its tie break.
+
+    Writing it under the `_rowid` name is what lets every reranker, third-party
+    implementations included, run unmodified. The column never reaches the
+    caller.
+    """
+    if not pk_columns:
+        raise ValueError(
+            "hybrid search on a MemWAL table needs an unenforced primary key to "
+            "deduplicate on, and this table declares none"
+        )
+
+    ids = {}
+    stamped = []
+    for table in tables:
+        if table.num_rows == 0:
+            # Both legs empty: the combined schema carries no key column, and no
+            # row that would need one.
+            row_ids = pa.array([], type=pa.uint64())
+        else:
+            missing = [c for c in pk_columns if c not in table.column_names]
+            if missing:
+                raise ValueError(
+                    "hybrid search could not deduplicate: primary key "
+                    f"column(s) {missing} missing from a result set with "
+                    f"{table.num_rows} rows"
+                )
+            keys = zip(*(table.column(c).to_pylist() for c in pk_columns))
+            row_ids = pa.array(
+                [ids.setdefault(key, len(ids)) for key in keys], type=pa.uint64()
+            )
+        stamped.append(table.append_column("_rowid", row_ids))
+    return stamped
+
+
+def pk_fusion_learned(table, use_lsm: Optional[bool]) -> bool:
+    """Whether a previous hybrid query on `table` was refused `_rowid`, so this
+    one should go straight to the primary key.
+
+    Learned from a refusal, never probed, so asking is free.
+    """
+    if table is None or use_lsm is False:
+        return False
+    # Only an unambiguous True forks to the key: a table type that answers the
+    # question loosely keeps the `_rowid` path.
+    return table._hybrid_pk_fusion_learned() is True
+
+
+class PkFusion:
+    """How the legs of a hybrid query on a MemWAL table are joined: on a
+    surrogate `_rowid` built from the primary key."""
+
+    def __init__(self, pk_columns: List[str]):
+        self.pk_columns = pk_columns
+        # Key columns the caller did not select, added to the projection so the
+        # surrogate can be built and dropped again before returning.
+        self.injected: List[str] = []
+
+    @classmethod
+    def for_query(
+        cls,
+        schema: pa.Schema,
+        projection: QueryProjection,
+        *,
+        caller_requested_row_id: bool,
+    ) -> PkFusion:
+        """Resolve the key for a hybrid query, refusing the shapes that need a
+        real `_rowid`."""
+        if caller_requested_row_id:
+            raise NotImplementedError(WAL_ROW_ID_UNSUPPORTED)
+        # A projected blob needs a real row id to fetch through, whatever else
+        # has turned `_with_row_id` on — including `return_score="all"`.
+        if blob_v2_projection_sources(schema, projection):
+            raise NotImplementedError(
+                "hybrid search cannot project a blob column on a MemWAL table: "
+                "fetching blobs needs a real _rowid, and the fresh tier has no "
+                "stable one. Set use_lsm(False) to read the base table only "
+                "(results will exclude un-compacted MemWAL data)"
+            )
+        return cls(_pk_columns(schema))
+
+    def inject(self, columns):
+        """`columns` with the key columns it is missing, to project in place of
+        the `_rowid` a MemWAL table would refuse."""
+        columns, self.injected = _with_pk_columns(columns, self.pk_columns)
+        return columns
+
+    def stamp(
+        self, vector_results: pa.Table, fts_results: pa.Table
+    ) -> Tuple[pa.Table, pa.Table]:
+        """Give both legs a surrogate `_rowid` built from the key."""
+        # Vector first: `merge_results` concatenates in that order and keeps
+        # the first occurrence, so first-seen ids preserve its tie break.
+        vector_results, fts_results = _stamp_surrogate_row_ids(
+            [vector_results, fts_results], self.pk_columns
+        )
+        return vector_results, fts_results
+
+    def strip(self, results: pa.Table) -> pa.Table:
+        """Drop the surrogate and any injected key columns from the reranked
+        results.
+
+        The surrogate is an internal join key, never an answer, and a caller
+        who asked for `_rowid` was refused in `for_query`.
+        """
+        drop = [c for c in ["_rowid", *self.injected] if c in results.column_names]
+        return results.drop(drop) if drop else results
+
+
+def _check_refusal(error: Exception, *, fused: bool, caller_requested_row_id: bool):
+    """Re-raise `error` unless it is a MemWAL `_rowid` refusal worth retrying on
+    the primary key."""
+    if fused or not _is_wal_row_id_refusal(error):
+        raise error
+    # `return_score="all"` turns row ids on for the reranker, not the caller, so
+    # it still falls back; only a caller who asked for them is refused, and with
+    # the reason rather than a bare 400.
+    if caller_requested_row_id:
+        raise NotImplementedError(WAL_ROW_ID_UNSUPPORTED) from error
+
+
+def with_pk_fallback(
+    run: Callable[[], T],
+    table,
+    *,
+    fused: Callable[[], bool],
+    caller_requested_row_id: bool,
+) -> T:
+    """Run a hybrid query, retrying on the primary key if the table refuses
+    `_rowid`.
+
+    `_rowid` is asked for first rather than paying a round trip up front to
+    find out: only a MemWAL table refuses, and the refusal has to be handled
+    anyway -- a write spec installed elsewhere can arrive between any two
+    queries. It costs nothing server-side: it is raised before the query is
+    planned. `fused` reports whether the attempt already used the key.
+    """
+    try:
+        return run()
+    except Exception as e:
+        _check_refusal(
+            e, fused=fused(), caller_requested_row_id=caller_requested_row_id
+        )
+        # Learned, so this retry and later queries go straight to the key.
+        table._note_hybrid_pk_fusion()
+        return run()
+
+
+async def with_pk_fallback_async(
+    run: Callable[[], Awaitable[T]],
+    table,
+    *,
+    fused: Callable[[], bool],
+    caller_requested_row_id: bool,
+) -> T:
+    """The async form of `with_pk_fallback`."""
+    try:
+        return await run()
+    except Exception as e:
+        _check_refusal(
+            e, fused=fused(), caller_requested_row_id=caller_requested_row_id
+        )
+        table._note_hybrid_pk_fusion()
+        return await run()

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/query.py

@@ -41,6 +41,7 @@ from .rerankers.rrf import RRFReranker
 from .rerankers.util import check_reranker_result
 from .schema import is_blob_like_field, schema_has_blob_field
 from .util import flatten_columns
+from . import _wal_hybrid  # WAL-PK-FUSION: delete.
 from ._blob import (
     BLOB_MODE_TO_HANDLING,
     FetchBlobsAsync,
@@ -2207,6 +2208,12 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
         self._phrase_query = None
         self._lower_bound = None
         self._upper_bound = None
+        # WAL-PK-FUSION: delete both, and the `with_row_id` override below.
+        # Set by `_create_query_builders` when the legs are fused on the key.
+        self._pk_fusion: Optional[_wal_hybrid.PkFusion] = None
+        # `_with_row_id` is also turned on by `rerank(return_score="all")` for
+        # the reranker's own use, so it cannot answer "did the caller ask?".
+        self._caller_requested_row_id = False
 
     def _validate_query(self, query, vector=None, text=None):
         if query is not None and (vector is not None or text is not None):
@@ -2228,6 +2235,16 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
 
         return vector_query, text_query
 
+    def with_row_id(self, with_row_id: bool) -> Self:  # WAL-PK-FUSION: delete.
+        """Set whether to return row ids.
+
+        Recorded separately from `_with_row_id`, which `rerank(return_score=
+        "all")` also sets for its own use — on a MemWAL table only a caller who
+        asked is refused, and the reranker still falls back to the primary key.
+        """
+        self._caller_requested_row_id = with_row_id
+        return super().with_row_id(with_row_id)
+
     def phrase_query(self, phrase_query: bool = True) -> LanceHybridQueryBuilder:
         """Set whether to use phrase query.
 
@@ -2248,16 +2265,30 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
         raise NotImplementedError("to_query_object not yet supported on a hybrid query")
 
     def to_arrow(self, *, timeout: Optional[timedelta] = None) -> pa.Table:
+        # WAL-PK-FUSION: without the fallback, this body is `_run_hybrid`'s.
+        return _wal_hybrid.with_pk_fallback(
+            lambda: self._run_hybrid(timeout=timeout),
+            self._table,
+            fused=lambda: self._pk_fusion is not None,
+            caller_requested_row_id=self._caller_requested_row_id,
+        )
+
+    def _run_hybrid(self, *, timeout: Optional[timedelta] = None) -> pa.Table:
         self._create_query_builders()
+        fts_query, vector_query = self._fts_query, self._vector_query
+        # WAL-PK-FUSION: without the fallback, both legs always ask for row ids.
+        if self._pk_fusion is None:
+            fts_query = fts_query.with_row_id(True)
+            vector_query = vector_query.with_row_id(True)
         with ThreadPoolExecutor() as executor:
-            fts_future = executor.submit(
-                self._fts_query.with_row_id(True).to_arrow, timeout=timeout
-            )
-            vector_future = executor.submit(
-                self._vector_query.with_row_id(True).to_arrow, timeout=timeout
-            )
+            fts_future = executor.submit(fts_query.to_arrow, timeout=timeout)
+            vector_future = executor.submit(vector_query.to_arrow, timeout=timeout)
             fts_results = fts_future.result()
             vector_results = vector_future.result()
+        if self._pk_fusion is not None:  # WAL-PK-FUSION: delete.
+            vector_results, fts_results = self._pk_fusion.stamp(
+                vector_results, fts_results
+            )
 
         results = self._combine_hybrid_results(
             fts_results=fts_results,
@@ -2265,13 +2296,15 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
             norm=self._norm,
             fts_query=self._fts_query._query,
             reranker=self._reranker,
-            limit=self._limit,
+            limit=self._limit or DEFAULT_HYBRID_LIMIT,
             with_row_ids=True,
             offset=self._offset,
         )
         return self._finish_hybrid_results(results)
 
     def _finish_hybrid_results(self, results: pa.Table) -> pa.Table:
+        if self._pk_fusion is not None:  # WAL-PK-FUSION: delete.
+            return self._pk_fusion.strip(results)
         if self._user_requested_row_id():
             return results
         if self._blob_auto_row_id_enabled():
@@ -2445,7 +2478,10 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
         self._norm = normalize
         self._reranker = reranker
         if reranker.score == "all":
-            self.with_row_id(True)
+            # WAL-PK-FUSION: without the fallback, `self.with_row_id(True)`.
+            # Not through `with_row_id`, which records caller intent: the
+            # reranker's need for row ids must not refuse a MemWAL table.
+            self._with_row_id = True
 
         return self
 
@@ -2712,20 +2748,31 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
         )
 
         # Apply common configurations
-        if self._limit:
-            # The final offset/limit window is sliced out of the combined,
-            # reranked results, so each sub-query must fetch enough rows to
-            # cover the skipped prefix as well as the window itself.
-            sub_query_limit = self._limit + (self._offset or 0)
-            self._vector_query.limit(sub_query_limit)
-            self._fts_query.limit(sub_query_limit)
-        if self._columns:
-            self._vector_query.select(self._columns)
-            self._fts_query.select(self._columns)
+        # The final offset/limit window is sliced out of the combined,
+        # reranked results, so each sub-query must fetch enough rows to
+        # cover the skipped prefix as well as the window itself.
+        limit = self._limit or DEFAULT_HYBRID_LIMIT
+        sub_query_limit = limit + (self._offset or 0)
+        self._vector_query.limit(sub_query_limit)
+        self._fts_query.limit(sub_query_limit)
+        # WAL-PK-FUSION: without the fallback, select `self._columns` as is.
+        self._pk_fusion = None
+        columns = self._columns
+        if _wal_hybrid.pk_fusion_learned(self._table, self._use_lsm):
+            self._pk_fusion = _wal_hybrid.PkFusion.for_query(
+                self._table.schema,
+                self._columns,
+                caller_requested_row_id=self._caller_requested_row_id,
+            )
+            columns = self._pk_fusion.inject(columns)
+        if columns:
+            self._vector_query.select(columns)
+            self._fts_query.select(columns)
         if self._where:
             self._vector_query.where(self._where, not self._postfilter)
             self._fts_query.where(self._where, not self._postfilter)
-        if self._with_row_id:
+        # WAL-PK-FUSION: without the fallback, `if self._with_row_id:`.
+        if self._with_row_id and self._pk_fusion is None:
             self._vector_query.with_row_id(True)
             self._fts_query.with_row_id(True)
         if self._use_lsm is not None:
@@ -3866,6 +3913,9 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
         self._inner = inner
         self._norm = "score"
         self._reranker = RRFReranker()
+        # WAL-PK-FUSION: delete.
+        # Set by `_create_child_queries` when the legs are fused on the key.
+        self._pk_fusion: Optional[_wal_hybrid.PkFusion] = None
 
     def rerank(
         self, reranker: Reranker = RRFReranker(), normalize: str = "score"
@@ -3897,7 +3947,7 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
 
         return self
 
-    def _create_child_queries(
+    async def _create_child_queries(
         self,
     ) -> Tuple["AsyncFTSQuery", "AsyncVectorQuery", int, int]:
         """Build the sub-queries that make up this hybrid query.
@@ -3906,7 +3956,8 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
         the plans that are reported are the plans that actually run.
 
         Returns the two sub-queries along with the effective limit and offset of
-        the hybrid query itself.
+        the hybrid query itself. The key the legs are fused on, if any, is left
+        on `self._pk_fusion`.
         """
         fts_query = AsyncFTSQuery(self._inner.to_fts_query(), self._table)
         vec_query = AsyncVectorQuery(self._inner.to_vector_query(), self._table)
@@ -3924,8 +3975,27 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
             limit = DEFAULT_HYBRID_LIMIT
         offset = fts_req.offset or vec_req.offset or 0
 
-        fts_query.with_row_id()
-        vec_query.with_row_id()
+        # WAL-PK-FUSION: without the fallback, keep only the two `with_row_id`
+        # calls.
+        self._pk_fusion = None
+        if _wal_hybrid.pk_fusion_learned(self._table, fts_req.use_lsm):
+            self._pk_fusion = _wal_hybrid.PkFusion.for_query(
+                await self._table.schema(),
+                _query_request_projection(fts_req),
+                caller_requested_row_id=self._user_requested_row_id(),
+            )
+        if self._pk_fusion is None:
+            fts_query.with_row_id()
+            vec_query.with_row_id()
+        else:
+            # Ask for the key columns instead, and never for `_rowid`.
+            # `select` carries the whole projection; `select_source_columns`
+            # keeps only plain column references, so rebuilding from it would
+            # drop computed ones.
+            columns = self._pk_fusion.inject(fts_req.select)
+            if self._pk_fusion.injected:
+                fts_query.select(columns)
+                vec_query.select(columns)
 
         # offset() pushes the offset down into both sub-queries, which would make
         # each of them skip its own first `offset` rows. The window has to be
@@ -3944,7 +4014,23 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
         max_batch_length: Optional[int] = None,
         timeout: Optional[timedelta] = None,
     ) -> AsyncRecordBatchReader:
-        fts_query, vec_query, limit, offset = self._create_child_queries()
+        # WAL-PK-FUSION: without the fallback, this body is `_run_hybrid`'s.
+        return await _wal_hybrid.with_pk_fallback_async(
+            lambda: self._run_hybrid(
+                max_batch_length=max_batch_length, timeout=timeout
+            ),
+            self._table,
+            fused=lambda: self._pk_fusion is not None,
+            caller_requested_row_id=self._user_requested_row_id(),
+        )
+
+    async def _run_hybrid(
+        self,
+        *,
+        max_batch_length: Optional[int] = None,
+        timeout: Optional[timedelta] = None,
+    ) -> AsyncRecordBatchReader:
+        fts_query, vec_query, limit, offset = await self._create_child_queries()
 
         req = fts_query._inner.to_query_request()
         blob_auto_row_id = False
@@ -3968,6 +4054,10 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
             fts_query.to_arrow(timeout=timeout),
             vec_query.to_arrow(timeout=timeout),
         )
+        if self._pk_fusion is not None:  # WAL-PK-FUSION: delete.
+            vector_results, fts_results = self._pk_fusion.stamp(
+                vector_results, fts_results
+            )
 
         result = LanceHybridQueryBuilder._combine_hybrid_results(
             fts_results=fts_results,
@@ -3979,7 +4069,9 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
             with_row_ids=True,
             offset=offset,
         )
-        if (
+        if self._pk_fusion is not None:  # WAL-PK-FUSION: delete this branch.
+            result = self._pk_fusion.strip(result)
+        elif (
             not self._user_requested_row_id()
             and not blob_auto_row_id
             and "_rowid" in result.column_names
@@ -4028,7 +4120,7 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
         plan : str
         """  # noqa: E501
 
-        fts_query, vec_query, _, _ = self._create_child_queries()
+        fts_query, vec_query, _, _ = await self._create_child_queries()
         vector_plan = await vec_query.explain_plan(verbose)
         fts_plan = await fts_query.explain_plan(verbose)
         # Indent sub-plans under the reranker
@@ -4057,7 +4149,7 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
         -------
         plan : str
         """
-        fts_query, vec_query, _, _ = self._create_child_queries()
+        fts_query, vec_query, _, _ = await self._create_child_queries()
 
         results = ["Vector Search Query:"]
         results.append(await vec_query.analyze_plan(distributed_metrics))

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/remote/table.py

@@ -1036,6 +1036,14 @@ class RemoteTable(Table):
         """Read the installed LsmWriteSpec, or ``None``."""
         return LOOP.run(self._table.get_lsm_write_spec())
 
+    # WAL-PK-FUSION: delete both hooks.
+    def _hybrid_pk_fusion_learned(self) -> bool:
+        # `self._table` is the async table, which owns the Rust handle.
+        return self._table._hybrid_pk_fusion_learned()
+
+    def _note_hybrid_pk_fusion(self) -> None:
+        self._table._note_hybrid_pk_fusion()
+
     def checkpoint_lsm(self) -> None:
         """Synchronous version of
         [`AsyncTable.checkpoint_lsm`][lancedb.AsyncTable.checkpoint_lsm]."""

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

@@ -107,8 +107,24 @@ class RRFReranker(Reranker):
                     `search().with_row_id(True)`"
             )
 
-        combined = pa.concat_tables(vector_results, **self._concat_tables_args)
-        empty_table = pa.Table.from_arrays([], names=[])
-        reranked = self.rerank_hybrid(query, combined, empty_table)
+        # RRF ranks each result list on its own, so a document's rank is its
+        # position within the list it came from, not within the concatenation.
+        rrf_score_map = defaultdict(float)
+        for result in vector_results:
+            for i, result_id in enumerate(result["_rowid"].to_pylist(), 1):
+                rrf_score_map[result_id] += 1 / (i + self.K)
 
-        return reranked
+        combined = pa.concat_tables(vector_results, **self._concat_tables_args)
+        combined = self._deduplicate(combined)
+        relevance_scores = [
+            rrf_score_map[row_id] for row_id in combined["_rowid"].to_pylist()
+        ]
+        combined = combined.append_column(
+            "_relevance_score", pa.array(relevance_scores, type=pa.float32())
+        )
+        combined = combined.sort_by([("_relevance_score", "descending")])
+
+        if self.score == "relevance":
+            combined = self._keep_relevance_score(combined)
+
+        return combined

python/python/lancedb/table.py

@@ -134,6 +134,34 @@ _MODEL_BACKED_TOKENIZER_ERRORS = (
 )
 
 
+def _optimize_cleanup_since_ms(
+    cleanup_older_than: Optional[timedelta], retrain: bool
+) -> Optional[int]:
+    # Called directly by both the sync and async optimize so stacklevel=3
+    # names the user's call site rather than the background event loop.
+    cleanup_since_ms: Optional[int] = None
+    if cleanup_older_than is not None:
+        cleanup_since_ms = round(cleanup_older_than.total_seconds() * 1000)
+        if cleanup_since_ms <= 0:
+            warnings.warn(
+                "optimize(cleanup_older_than=0) removes every version except "
+                "the latest. Any concurrent reader or writer still using an "
+                "older version will fail. Use a longer cleanup_older_than "
+                "unless no other process is working on this table.",
+                UserWarning,
+                stacklevel=3,
+            )
+
+    if retrain:
+        warnings.warn(
+            "The 'retrain' parameter is deprecated and will be removed in a "
+            "future version.",
+            DeprecationWarning,
+            stacklevel=3,
+        )
+    return cleanup_since_ms
+
+
 def _add_unique_note(exception: BaseException, note: str) -> None:
     existing_notes = getattr(exception, "__notes__", ()) or ()
     message = (
@@ -1940,8 +1968,10 @@ class Table(ABC):
         One missing or deleted row ID rejects the entire batch; no partial
         results are returned.
 
-        Convenience for small payloads. For large values use
-        :meth:`fetch_blob_files`.
+        Remote servers limit each request to 1024 row IDs and 64 MiB of blob
+        bytes. The client splits requests automatically and reads an individual
+        larger blob through the Range route. This method still materializes all
+        bytes in memory; for large values use :meth:`fetch_blob_files`.
         """
 
     @abstractmethod
@@ -2229,6 +2259,13 @@ class Table(ABC):
             All files belonging to versions older than this will be removed.  Set
             to 0 days to remove all versions except the latest.  The latest version
             is never removed.
+
+            .. warning::
+
+                Setting this to 0 deletes the data files of every older
+                version, so any other reader or writer still using an older
+                version of the table will fail. Only set it to 0 if no other
+                process is working on this dataset.
         delete_unverified: bool, default False
             Files leftover from a failed transaction may appear to be part of an
             in-progress operation (e.g. appending new data) and these files will not
@@ -2659,6 +2696,18 @@ class Table(ABC):
         to check if the table is already using the new path style.
         """
 
+    # WAL-PK-FUSION: delete both hooks, here and on AsyncTable and RemoteTable.
+    def _hybrid_pk_fusion_learned(self) -> bool:
+        """Whether a hybrid query here has already been refused ``_rowid``.
+
+        Learned from a refusal, never probed, so asking is free. ``False`` for
+        table types that never refuse.
+        """
+        return False
+
+    def _note_hybrid_pk_fusion(self) -> None:
+        """Remember a ``_rowid`` refusal, so later hybrid queries skip it."""
+
 
 class LanceTable(Table):
     """
@@ -4427,6 +4476,13 @@ class LanceTable(Table):
             All files belonging to versions older than this will be removed.  Set
             to 0 days to remove all versions except the latest.  The latest version
             is never removed.
+
+            .. warning::
+
+                Setting this to 0 deletes the data files of every older
+                version, so any other reader or writer still using an older
+                version of the table will fail. Only set it to 0 if no other
+                process is working on this dataset.
         delete_unverified: bool, default False
             Files leftover from a failed transaction may appear to be part of an
             in-progress operation (e.g. appending new data) and these files will not
@@ -4451,10 +4507,9 @@ class LanceTable(Table):
         modification operations.
         """
         LOOP.run(
-            self._table.optimize(
-                cleanup_older_than=cleanup_older_than,
-                delete_unverified=delete_unverified,
-                retrain=retrain,
+            self._table._do_optimize(
+                _optimize_cleanup_since_ms(cleanup_older_than, retrain),
+                delete_unverified,
             )
         )
 
@@ -5310,6 +5365,15 @@ class AsyncTable:
         """
         return await self._inner.get_lsm_write_spec()
 
+    # WAL-PK-FUSION: delete both hooks.
+    def _hybrid_pk_fusion_learned(self) -> bool:
+        """See [`Table._hybrid_pk_fusion_learned`][lancedb.table.Table]."""
+        return self._inner.hybrid_pk_fusion_learned()
+
+    def _note_hybrid_pk_fusion(self) -> None:
+        """See [`Table._note_hybrid_pk_fusion`][lancedb.table.Table]."""
+        self._inner.note_hybrid_pk_fusion()
+
     async def checkpoint_lsm(self) -> None:
         """Converge this table's LSM write path into its base table.
 
@@ -6957,6 +7021,13 @@ class AsyncTable:
             All files belonging to versions older than this will be removed.  Set
             to 0 days to remove all versions except the latest.  The latest version
             is never removed.
+
+            .. warning::
+
+                Setting this to 0 deletes the data files of every older
+                version, so any other reader or writer still using an older
+                version of the table will fail. Only set it to 0 if no other
+                process is working on this dataset.
         delete_unverified: bool, default False
             Files leftover from a failed transaction may appear to be part of an
             in-progress operation (e.g. appending new data) and these files will not
@@ -6980,20 +7051,14 @@ class AsyncTable:
         you have added or modified 100,000 or more records or run more than 20 data
         modification operations.
         """
-        cleanup_since_ms: Optional[int] = None
-        if cleanup_older_than is not None:
-            cleanup_since_ms = round(cleanup_older_than.total_seconds() * 1000)
-
-        if retrain:
-            import warnings
-
-            warnings.warn(
-                "The 'retrain' parameter is deprecated and will be removed in a "
-                "future version.",
-                DeprecationWarning,
-                stacklevel=2,
-            )
+        return await self._do_optimize(
+            _optimize_cleanup_since_ms(cleanup_older_than, retrain),
+            delete_unverified,
+        )
 
+    async def _do_optimize(
+        self, cleanup_since_ms: Optional[int], delete_unverified: bool
+    ) -> OptimizeStats:
         return await self._inner.optimize(
             cleanup_since_ms=cleanup_since_ms,
             delete_unverified=delete_unverified,

python/python/lancedb/util.py

@@ -5,6 +5,7 @@
 import binascii
 import functools
 import importlib
+import math
 import os
 import pathlib
 import warnings
@@ -351,6 +352,13 @@ def _(value: int):
 
 @value_to_sql.register(float)
 def _(value: float):
+    # str() gives "nan" / "inf", which SQL would read as column names.
+    if math.isnan(value):
+        return "CAST('NaN' AS DOUBLE)"
+    if math.isinf(value):
+        return (
+            "CAST('Infinity' AS DOUBLE)" if value > 0 else "CAST('-Infinity' AS DOUBLE)"
+        )
     return str(value)
 
 

python/python/tests/test_hybrid_query.py

@@ -315,6 +315,33 @@ def test_hybrid_query_offset(sync_table: Table):
     assert offset_result["_rowid"].to_pylist() == full["_rowid"].to_pylist()[2:]
 
 
+def test_hybrid_query_default_limit(sync_table: Table):
+    # The vector and FTS legs match disjoint rows, so fusing their results
+    # yields more rows than the default limit.
+    new_rows = []
+    for i in range(20):
+        new_rows.append({"text": "close_vec", "vector": [0.1, 0.1]})
+        new_rows.append({"text": "dog", "vector": [50.0 + i, 50.0 + i]})
+    sync_table.add(new_rows)
+
+    def query():
+        return (
+            sync_table.search(query_type="hybrid")
+            .vector([0.1, 0.1])
+            .text("dog")
+            .with_row_id(True)
+        )
+
+    # Like the async hybrid query, the default limit is 10.
+    result = query().to_arrow()
+    assert len(result) == 10
+
+    # The offset window is taken after the default limit is applied.
+    full = query().limit(15).to_arrow()
+    offset_result = query().offset(5).to_arrow()
+    assert offset_result["_rowid"].to_pylist() == full["_rowid"].to_pylist()[5:]
+
+
 def test_hybrid_query_minimum_nprobes_zero_raises(sync_table: Table):
     # minimum_nprobes(0) must raise the same validation error a plain vector
     # query raises, not silently no-op because 0 is falsy.

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

@@ -395,6 +395,38 @@ def test_mrr_multivector_rewards_consensus():
     assert result["_rowid"].to_pylist()[0] == 2
 
 
+def test_rrf_multivector_ranks_each_result_list():
+    # RRF scores a document by its rank *within each* result list. The top hit
+    # of the second list must score the same as the top hit of the first list,
+    # not as if it were ranked after every row of the first list.
+    reranker = RRFReranker(K=60)
+
+    def ranking(row_ids):
+        return pa.table({"_rowid": pa.array(row_ids, type=pa.int64())})
+
+    rs1 = ranking([1, 2, 3])
+    rs2 = ranking([4, 5, 6])
+
+    result = reranker.rerank_multivector([rs1, rs2])
+    scores = dict(
+        zip(result["_rowid"].to_pylist(), result["_relevance_score"].to_pylist())
+    )
+
+    assert scores[1] == pytest.approx(1 / 61)
+    assert scores[4] == pytest.approx(1 / 61)
+    assert scores[5] == pytest.approx(1 / 62)
+    # the second-ranked hit of the second list beats the third hit of the first
+    assert scores[5] > scores[3]
+
+    # A document found by both lists sums its reciprocal ranks from each list.
+    result = reranker.rerank_multivector([ranking([1, 2]), ranking([2, 3])])
+    scores = dict(
+        zip(result["_rowid"].to_pylist(), result["_relevance_score"].to_pylist())
+    )
+    assert scores[2] == pytest.approx(1 / 62 + 1 / 61)
+    assert result["_rowid"].to_pylist() == [2, 1, 3]
+
+
 def test_rrf_reranker_distance():
     data = pa.table(
         {
@@ -481,6 +513,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/python/tests/test_table.py

@@ -4307,7 +4307,8 @@ async def test_optimize(mem_db_async: AsyncConnection):
     assert stats.prune.bytes_removed == 0
     assert stats.prune.old_versions_removed == 0
 
-    stats = await table.optimize(cleanup_older_than=timedelta(seconds=0))
+    with pytest.warns(UserWarning, match="concurrent"):
+        stats = await table.optimize(cleanup_older_than=timedelta(seconds=0))
     assert stats.prune.bytes_removed > 0
     assert stats.prune.old_versions_removed == 3
 
@@ -4335,12 +4336,35 @@ async def test_optimize_delete_unverified(tmp_db_async: AsyncConnection, tmp_pat
 
     stats = await table.optimize(delete_unverified=False)
     assert stats.prune.old_versions_removed == 0
-    stats = await table.optimize(
-        cleanup_older_than=timedelta(seconds=0), delete_unverified=True
-    )
+    with pytest.warns(UserWarning, match="concurrent"):
+        stats = await table.optimize(
+            cleanup_older_than=timedelta(seconds=0), delete_unverified=True
+        )
     assert stats.prune.old_versions_removed == 2
 
 
+@pytest.mark.asyncio
+async def test_optimize_warns_on_zero_cleanup(mem_db_async: AsyncConnection):
+    table = await mem_db_async.create_table("test", data=[{"x": [1]}])
+    with pytest.warns(UserWarning, match="concurrent"):
+        await table.optimize(cleanup_older_than=timedelta(0))
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("error")
+        await table.optimize(cleanup_older_than=timedelta(days=1))
+        await table.optimize()
+
+
+def test_optimize_warns_on_zero_cleanup_sync(mem_db: DBConnection):
+    table = mem_db.create_table("test", data=[{"x": [1]}])
+    with warnings.catch_warnings(record=True) as seen:
+        warnings.simplefilter("default")
+        table.optimize(cleanup_older_than=timedelta(0))
+        table.optimize(cleanup_older_than=timedelta(0))
+    assert [w.filename for w in seen] == [__file__, __file__]
+    assert all("concurrent" in str(w.message) for w in seen)
+
+
 def test_replace_field_metadata(tmp_path):
     db = lancedb.connect(tmp_path)
     table = db.create_table("my_table", data=[{"x": 0}])

python/python/tests/test_util.py

@@ -213,6 +213,35 @@ def test_value_to_sql_numpy_scalars():
     assert value_to_sql(np.bool_(False)) == "FALSE"
 
 
+def test_value_to_sql_nan_and_infinity(tmp_path):
+    # str(float("nan")) is "nan", which SQL reads as a column named nan, so
+    # table.update(values={"x": float("nan")}) failed with "No field named nan".
+    import math
+
+    import numpy as np
+
+    db = lancedb.connect(tmp_path)
+    table = db.create_table(
+        "test",
+        pa.table(
+            {
+                "id": [1, 2, 3, 4],
+                "f64": [1.0, 2.0, 3.0, 4.0],
+                "f32": pa.array([1.0, 2.0, 3.0, 4.0], pa.float32()),
+            }
+        ),
+    )
+    table.update(where="id = 1", values={"f64": float("nan"), "f32": np.nan})
+    table.update(where="id = 2", values={"f64": math.inf, "f32": np.float32("inf")})
+    table.update(where="id = 3", values={"f64": -math.inf, "f32": -math.inf})
+
+    rows = {row["id"]: row for row in table.to_arrow().to_pylist()}
+    assert math.isnan(rows[1]["f64"]) and math.isnan(rows[1]["f32"])
+    assert rows[2]["f64"] == math.inf and rows[2]["f32"] == math.inf
+    assert rows[3]["f64"] == -math.inf and rows[3]["f32"] == -math.inf
+    assert rows[4]["f64"] == 4.0 and rows[4]["f32"] == 4.0
+
+
 def test_append_vector_columns():
     registry = EmbeddingFunctionRegistry.get_instance()
     registry.register("test")(MockTextEmbeddingFunction)

python/python/tests/test_wal_hybrid.py

@@ -0,0 +1,219 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""WAL-PK-FUSION: delete this file with `lancedb/_wal_hybrid.py`."""
+
+from unittest import mock
+
+import lancedb
+import pyarrow as pa
+import pytest
+from lancedb._wal_hybrid import WAL_ROW_ID_UNSUPPORTED, with_pk_fallback
+from lancedb.query import LanceHybridQueryBuilder
+from lancedb.rerankers import RRFReranker
+
+
+def _wal_table(schema: pa.Schema) -> mock.Mock:
+    """A MemWAL-backed table a previous query has already been refused
+    `_rowid` on, so the fusion goes straight to the primary key."""
+    table = mock.Mock()
+    table._hybrid_pk_fusion_learned.return_value = True
+    table.schema = schema
+    return table
+
+
+_PK_SCHEMA = pa.schema(
+    [
+        pa.field(
+            "id",
+            pa.string(),
+            metadata={b"lance-schema:unenforced-primary-key:position": b"1"},
+        ),
+        pa.field("text", pa.string()),
+    ]
+)
+
+
+def _wal_hybrid(**select):
+    builder = (
+        LanceHybridQueryBuilder(_wal_table(_PK_SCHEMA)).vector([0.1, 0.2]).text("puppy")
+    )
+    if "columns" in select:
+        builder = builder.select(select["columns"])
+    return builder
+
+
+def test_wal_hybrid_projects_the_key_instead_of_asking_for_row_ids():
+    builder = _wal_hybrid(columns=["text"])
+    builder._create_query_builders()
+
+    assert builder._pk_fusion.pk_columns == ["id"]
+    assert builder._pk_fusion.injected == ["id"]
+    for leg in (builder._vector_query, builder._fts_query):
+        assert leg._columns == ["text", "id"]
+        # A MemWAL table rejects with_row_id before it plans.
+        assert leg._with_row_id is not True
+
+
+def test_wal_hybrid_selecting_the_key_injects_nothing():
+    builder = _wal_hybrid(columns=["text", "id"])
+    builder._create_query_builders()
+    assert builder._pk_fusion.injected == []
+
+
+def test_wal_hybrid_without_a_projection_leaves_it_alone():
+    builder = _wal_hybrid()
+    builder._create_query_builders()
+    # Select-all already returns the key.
+    assert builder._pk_fusion.injected == []
+    assert builder._vector_query._columns is None
+
+
+def test_base_table_hybrid_still_joins_on_row_ids():
+    table = mock.Mock()
+    table._hybrid_pk_fusion_learned.return_value = False
+    builder = LanceHybridQueryBuilder(table).vector([0.1, 0.2]).text("puppy")
+    builder._create_query_builders()
+
+    # `to_arrow` is what turns row ids on for this path; all that matters here
+    # is that no key was resolved and no projection was touched.
+    assert builder._pk_fusion is None
+    assert builder._vector_query._columns is None
+
+
+def test_wal_hybrid_with_use_lsm_false_joins_on_row_ids():
+    builder = _wal_hybrid().use_lsm(False)
+    builder._create_query_builders()
+    assert builder._pk_fusion is None
+
+
+def test_wal_hybrid_refuses_an_explicit_with_row_id():
+    builder = _wal_hybrid().with_row_id(True)
+    with pytest.raises(NotImplementedError, match="use_lsm"):
+        builder._create_query_builders()
+
+
+def test_wal_hybrid_return_score_all_is_not_refused():
+    """`return_score="all"` turns row ids on for the reranker, not the caller."""
+    builder = _wal_hybrid().rerank(RRFReranker(return_score="all"))
+    builder._create_query_builders()
+    assert builder._pk_fusion.pk_columns == ["id"]
+
+
+def test_wal_hybrid_fuses_on_the_key_and_drops_it_again():
+    builder = _wal_hybrid(columns=["text"])
+    builder._create_query_builders()
+
+    vector_results = pa.table(
+        {"text": ["a", "b"], "id": ["a", "b"], "_distance": [0.1, 0.2]}
+    )
+    fts_results = pa.table({"text": ["b", "c"], "id": ["b", "c"], "_score": [1.0, 2.0]})
+    vector_results, fts_results = builder._pk_fusion.stamp(vector_results, fts_results)
+
+    combined = LanceHybridQueryBuilder._combine_hybrid_results(
+        fts_results=fts_results,
+        vector_results=vector_results,
+        norm="score",
+        fts_query="puppy",
+        reranker=RRFReranker(),
+        limit=10,
+        with_row_ids=True,
+    )
+    results = builder._finish_hybrid_results(combined)
+
+    assert "_rowid" not in results.column_names, "the surrogate is internal"
+    assert "id" not in results.column_names, "the caller selected only `text`"
+    assert sorted(results.column("text").to_pylist()) == ["a", "b", "c"]
+
+
+@pytest.mark.parametrize(
+    "row_id_first", [True, False], ids=["row_id_first", "rerank_first"]
+)
+def test_wal_hybrid_refuses_row_id_even_with_return_score_all(row_id_first: bool):
+    """`return_score="all"` sets the same flag the caller's request does, so
+    asking for both must still be refused — in either call order."""
+    builder = _wal_hybrid()
+    reranker = RRFReranker(return_score="all")
+    if row_id_first:
+        builder = builder.with_row_id(True).rerank(reranker)
+    else:
+        builder = builder.rerank(reranker).with_row_id(True)
+
+    with pytest.raises(NotImplementedError, match="use_lsm"):
+        builder._create_query_builders()
+
+
+def test_wal_hybrid_refuses_a_blob_projection_with_return_score_all():
+    """A projected blob needs a real row id to fetch through, which the
+    reranker turning `_with_row_id` on does not provide."""
+    schema = pa.schema(
+        [
+            pa.field(
+                "id",
+                pa.string(),
+                metadata={b"lance-schema:unenforced-primary-key:position": b"1"},
+            ),
+            pa.field("text", pa.string()),
+            lancedb.blob("blob"),
+        ]
+    )
+    builder = (
+        LanceHybridQueryBuilder(_wal_table(schema))
+        .vector([0.1, 0.2])
+        .text("puppy")
+        .select(["blob"])
+        .rerank(RRFReranker(return_score="all"))
+    )
+
+    with pytest.raises(NotImplementedError, match="blob"):
+        builder._create_query_builders()
+
+
+_REFUSAL = RuntimeError(
+    "Bad request: the MemWAL LSM scanner does not support with_row_id"
+)
+
+
+def test_fallback_retries_on_the_key_and_remembers_it():
+    table = mock.Mock()
+    attempts = []
+
+    def run():
+        attempts.append(len(attempts))
+        if len(attempts) == 1:
+            raise _REFUSAL
+        return "fused"
+
+    result = with_pk_fallback(
+        run, table, fused=lambda: False, caller_requested_row_id=False
+    )
+
+    assert result == "fused"
+    assert len(attempts) == 2
+    table._note_hybrid_pk_fusion.assert_called_once()
+
+
+def test_fallback_explains_the_refusal_to_a_caller_who_asked_for_row_ids():
+    def run():
+        raise _REFUSAL
+
+    with pytest.raises(NotImplementedError, match="use_lsm") as info:
+        with_pk_fallback(
+            run, mock.Mock(), fused=lambda: False, caller_requested_row_id=True
+        )
+    assert str(info.value) == WAL_ROW_ID_UNSUPPORTED
+
+
+@pytest.mark.parametrize(
+    "error, fused",
+    [(ValueError("unrelated"), False), (_REFUSAL, True)],
+    ids=["other_error", "already_fused"],
+)
+def test_fallback_reraises_what_it_cannot_retry(error, fused):
+    def run():
+        raise error
+
+    table = mock.Mock()
+    with pytest.raises(type(error)):
+        with_pk_fallback(run, table, fused=lambda: fused, caller_requested_row_id=False)
+    table._note_hybrid_pk_fusion.assert_not_called()

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/src/table.rs

@@ -1653,6 +1653,21 @@ impl Table {
         })
     }
 
+    /// Whether a hybrid query on this table has already been refused `_rowid`.
+    /// Learned from a previous refusal, never probed, so this is free.
+    ///
+    /// WAL-PK-FUSION: delete this and `note_hybrid_pk_fusion`.
+    pub fn hybrid_pk_fusion_learned(self_: PyRef<'_, Self>) -> PyResult<bool> {
+        Ok(self_.inner_ref()?.base_table().hybrid_pk_fusion_learned())
+    }
+
+    /// Remember that this table refused `_rowid`, so later hybrid queries skip
+    /// straight to the primary-key fusion.
+    pub fn note_hybrid_pk_fusion(self_: PyRef<'_, Self>) -> PyResult<()> {
+        self_.inner_ref()?.base_table().note_hybrid_pk_fusion();
+        Ok(())
+    }
+
     /// Converge the table's LSM write path into its base table.
     ///
     /// Best-effort: with writes flowing, new rows may land after the last