Merge main into gatekeeper/fix-4393-1

python/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.40.0-beta.12"
+version = "0.40.0-beta.13"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -8,7 +8,7 @@ license.workspace = true
 repository.workspace = true
 keywords.workspace = true
 categories.workspace = true
-rust-version = "1.91.0"
+rust-version = "1.94.0"
 
 [lib]
 name = "_lancedb"

python/python/lancedb/__init__.py

@@ -46,7 +46,6 @@ from .materialized_view import (
     AsyncMaterializedView,
     MaterializedView,
     MaterializedViewDefinition,
-    vector_dedup as vector_dedup,
 )
 from .view import ViewDescription as ViewDescription
 from .table import AsyncTable, Table
@@ -62,7 +61,6 @@ from .namespace import (
 from .catalog import (
     AsyncCatalog,
     Catalog,
-    ListDatabasesResponse,
     connect_catalog,
     connect_catalog_async,
 )
@@ -578,7 +576,6 @@ async def connect_async(
 __all__ = [
     "Catalog",
     "AsyncCatalog",
-    "ListDatabasesResponse",
     "connect_catalog",
     "connect_catalog_async",
     "AsyncMaterializedView",

python/python/lancedb/_lancedb.pyi

@@ -255,8 +255,6 @@ 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,
@@ -266,8 +264,6 @@ 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(
@@ -500,10 +496,10 @@ class Table:
         limit: Optional[int] = None,
     ) -> FunctionErrors: ...
     async def refresh_materialized_view(
-        self, full: bool = False, source_version: Optional[int] = None
+        self, source_version: Optional[int] = None
     ) -> RefreshMaterializedViewResult: ...
     async def refresh_materialized_view_async(
-        self, full: bool = False, source_version: Optional[int] = None
+        self, source_version: Optional[int] = None
     ) -> Job: ...
     async def materialized_view_definition(self) -> str: ...
     async def add_columns_with_schema(self, schema: pa.Schema) -> AddColumnsResult: ...
@@ -946,9 +942,15 @@ class Catalog:
     async def drop_database(
         self, name: str, *, ignore_missing: bool = False
     ) -> None: ...
-    async def list_databases(
-        self, *, limit: Optional[int] = None, page_token: Optional[str] = None
-    ) -> tuple[list[str], Optional[str]]: ...
+    def list_databases(
+        self, *, page_token: Optional[str] = None, page_limit: Optional[int] = None
+    ) -> DatabaseNames: ...
+
+class DatabaseNames:
+    def num_page_results(self) -> int: ...
+    def page_token(self) -> Optional[str]: ...
+    def __aiter__(self) -> DatabaseNames: ...
+    async def __anext__(self) -> str: ...
 
 async def connect_catalog(
     endpoint: str,

python/python/lancedb/catalog.py

@@ -3,9 +3,8 @@
 
 """Remote catalogs manage databases through a server's root namespace."""
 
-from dataclasses import dataclass
 from datetime import timedelta
-from typing import Any, Optional, Union
+from typing import Any, AsyncIterator, Iterator, Optional, Union
 
 from . import _lancedb
 from .background_loop import LOOP
@@ -14,12 +13,66 @@ from .remote import ClientConfig, OAuthConfig
 from .remote.db import RemoteDBConnection
 
 
-@dataclass
-class ListDatabasesResponse:
-    """A page of database names and an optional continuation token."""
+class AsyncDatabaseNames(AsyncIterator[str]):
+    """An async iterator of database names with pagination state.
 
-    databases: list[str]
-    page_token: Optional[str] = None
+    Returned by [list_databases][lancedb.catalog.AsyncCatalog.list_databases].
+    """
+
+    def __init__(self, inner: _lancedb.DatabaseNames):
+        self._inner = inner
+
+    def __aiter__(self) -> "AsyncDatabaseNames":
+        return self
+
+    async def __anext__(self) -> str:
+        return await self._inner.__anext__()
+
+    def num_page_results(self) -> int:
+        """Return the number of names available without another REST request."""
+        return self._inner.num_page_results()
+
+    def page_token(self) -> Optional[str]:
+        """Return the token for the next REST request.
+
+        Before the first request, this is the supplied starting token (``None``
+        starts at the beginning). After the final page is fetched, it is ``None``,
+        even if names remain cached. Drain the cache before saving a token to
+        avoid skipping those names when resuming. A failed request terminates
+        iteration and leaves its token available for resuming a new iterator.
+        """
+        return self._inner.page_token()
+
+
+class DatabaseNames(Iterator[str]):
+    """A synchronous iterator of database names with pagination state.
+
+    Returned by [Catalog.list_databases][lancedb.catalog.Catalog.list_databases].
+    """
+
+    def __init__(self, inner: AsyncDatabaseNames):
+        self._inner = inner
+
+    def __iter__(self) -> "DatabaseNames":
+        return self
+
+    def __next__(self) -> str:
+        try:
+            return LOOP.run(self._inner.__anext__())
+        except StopAsyncIteration:
+            raise StopIteration from None
+
+    def num_page_results(self) -> int:
+        """Return the number of names available without another REST request."""
+        return self._inner.num_page_results()
+
+    def page_token(self) -> Optional[str]:
+        """Return the token for the next REST request.
+
+        See [page_token][lancedb.catalog.AsyncDatabaseNames.page_token]
+        for initial, cached, final-page, and error behavior.
+        """
+        return self._inner.page_token()
 
 
 class AsyncCatalog:
@@ -50,14 +103,25 @@ class AsyncCatalog:
         """Connect to an existing database by its logical name."""
         return AsyncConnection(await self._inner.connect_database(name))
 
-    async def list_databases(
-        self, *, limit: Optional[int] = None, page_token: Optional[str] = None
-    ) -> ListDatabasesResponse:
-        """List a page of databases. Pass the returned token for the next page."""
-        names, token = await self._inner.list_databases(
-            limit=limit, page_token=page_token
+    def list_databases(
+        self, *, page_token: Optional[str] = None, page_limit: Optional[int] = None
+    ) -> AsyncDatabaseNames:
+        """Iterate lazily over all database names using ``async for``.
+
+        ``page_token`` resumes from a saved token; ``None`` starts at the beginning.
+        ``page_limit`` sets the maximum names per REST response (not the total);
+        ``None`` uses the server default. Request errors are raised during iteration.
+
+        Examples
+        --------
+        ```python
+        async for name in catalog.list_databases():
+            print(name)
+        ```
+        """
+        return AsyncDatabaseNames(
+            self._inner.list_databases(page_token=page_token, page_limit=page_limit)
         )
-        return ListDatabasesResponse(names, token)
 
     async def drop_database(self, name: str, *, ignore_missing: bool = False) -> None:
         """Drop an empty database. A nonempty database is an error."""
@@ -73,7 +137,8 @@ class Catalog:
     ```python
     catalog = lancedb.connect_catalog("https://my-server.example", api_key="secret")
     db = catalog.create_database("analytics", exist_ok=True)
-    page = catalog.list_databases(limit=20)
+    for name in catalog.list_databases():
+        print(name)
     ```
     """
 
@@ -118,10 +183,17 @@ class Catalog:
         )
 
     def list_databases(
-        self, *, limit: Optional[int] = None, page_token: Optional[str] = None
-    ) -> ListDatabasesResponse:
-        """List a page of databases. Pass the returned token for the next page."""
-        return LOOP.run(self._inner.list_databases(limit=limit, page_token=page_token))
+        self, *, page_token: Optional[str] = None, page_limit: Optional[int] = None
+    ) -> DatabaseNames:
+        """Iterate lazily over database names, fetching more results as needed.
+
+        ``page_token`` resumes from a saved token; ``None`` starts at the beginning.
+        ``page_limit`` sets the maximum names per REST response (not the total);
+        ``None`` uses the server default. Request errors are raised during iteration.
+        """
+        return DatabaseNames(
+            self._inner.list_databases(page_token=page_token, page_limit=page_limit)
+        )
 
     def drop_database(self, name: str, *, ignore_missing: bool = False) -> None:
         """Drop an empty database. A nonempty database is an error."""

python/python/lancedb/db.py

@@ -57,7 +57,6 @@ from .materialized_view import (
     AsyncMaterializedView,
     MaterializedView,
     MaterializedViewSource,
-    VectorDedupSource,
     SelectArg,
     normalize_select,
 )
@@ -613,21 +612,12 @@ class DBConnection(EnforceOverrides):
         table. The view is a normal table: it can be queried, indexed and
         searched, and it appears in ``table_names``.
 
-        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. Native dedup sources also support physical
-        row IDs interpreted against their pinned snapshot.
-
         Parameters
         ----------
         name: str
             The name of the view.
-        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.
+        source: str
+            The name of the source table.
         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
@@ -1654,10 +1644,7 @@ class LanceDBConnection(DBConnection):
         Examples
         --------
         >>> import lancedb
-        >>> db = lancedb.connect(
-        ...     "./.lancedb",
-        ...     storage_options={"new_table_enable_stable_row_ids": "true"},
-        ... )
+        >>> db = lancedb.connect("./.lancedb")
         >>> data = [{"name": "ada", "age": 36}, {"name": "kid", "age": 7}]
         >>> table = db.create_table("people", data)
         >>> view = db.create_materialized_view(
@@ -2625,19 +2612,7 @@ class AsyncConnection(object):
         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)
-        }
+        return source, {}
 
     async def create_materialized_view(
         self,

python/python/lancedb/materialized_view.py

@@ -6,11 +6,11 @@ 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
 
+import json
+from dataclasses import dataclass
+
 from .background_loop import LOOP
 from .job import AsyncJob, Job, _typed_job
 
@@ -21,7 +21,6 @@ if TYPE_CHECKING:
     from .table import AsyncTable, LanceTable
 
 DEFINITION_META_KEY = b"mv.definition"
-_MAX_FLOAT32 = float.fromhex("0x1.fffffep+127")
 
 SelectArg = Union[
     str,
@@ -31,104 +30,12 @@ SelectArg = Union[
 ]
 
 
-@dataclass(frozen=True)
-class VectorDedupSource:
-    """A declarative source of retained original rows from an indexed snapshot.
-
-    Construct with [vector_dedup][lancedb.vector_dedup] and pass it as the
-    ``source`` of ``create_materialized_view`` or
-    ``create_materialized_view_async``. An omitted ``dataset_version`` is
-    captured once when creation is submitted, not when this object is made.
-    Refreshes keep that captured version. The source table is never modified.
-    """
-
-    source: str
-    column: str
-    distance_threshold: float
-    dataset_version: Optional[int] = None
-
-    def __post_init__(self):
-        for name in ("source", "column"):
-            value = getattr(self, name)
-            if not isinstance(value, str) or not value:
-                raise ValueError(f"{name} must be a non-empty string")
-        if self.dataset_version is not None and (
-            type(self.dataset_version) is not int
-            or not 0 < self.dataset_version < 2**64
-        ):
-            raise ValueError("dataset_version must be a positive uint64 integer")
-        value = self.distance_threshold
-        if isinstance(value, bool) or not isinstance(value, (int, float)):
-            raise TypeError("distance_threshold must be a number")
-        try:
-            value = float(value)
-        except OverflowError as exc:
-            raise ValueError("distance_threshold must fit a finite float32") from exc
-        if not math.isfinite(value) or abs(value) > _MAX_FLOAT32:
-            raise ValueError("distance_threshold must fit a finite float32")
-        object.__setattr__(self, "distance_threshold", value)
-
-    def _native_source_json(self, version: int) -> str:
-        return json.dumps(
-            {
-                "kind": "Dedup",
-                "dataset_version": version,
-                "column": self.column,
-                "distance_threshold": repr(self.distance_threshold),
-            },
-            allow_nan=False,
-        )
+MaterializedViewSource = str
 
 
-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
+DEFINITION_FORMAT = 2
 """The newest stored layout this version reads: ``{"format": N, "query": "<SQL>"}``,
-format 2 being a query with ``GROUP BY``, format 3 native pairs and format 4
-native dedup results. A ``kind`` key beside it is for
+format 2 being a query with ``GROUP BY``. A ``kind`` key beside it is for
 readers older than the format number."""
 
 
@@ -272,38 +179,31 @@ class AsyncMaterializedView:
         return _definition_from_json(raw)
 
     async def refresh(
-        self, *, full: bool = False, source_version: Optional[int] = None
+        self, *, source_version: Optional[int] = None
     ) -> "RefreshMaterializedViewResult":
         """Recompute the view from its source.
 
-        The refresh is incremental when the source's changes can be
-        reconciled into the view -- rows added, changed or removed since the
-        last one -- and otherwise rebuilds. ``full=True`` forces a rebuild;
+        Every refresh rebuilds the complete view. For local views,
         ``source_version`` refreshes to that source version instead of the
-        latest.
-
-        Concurrent refreshes of one view do not duplicate its rows. Two that
-        plan the same source rows conflict on commit, and the loser raises
-        rather than writing them a second time.
+        latest; remote SQL refreshes do not support source-version pinning.
         """
         return await self._table._inner.refresh_materialized_view(
-            full=full, source_version=source_version
+            source_version=source_version
         )
 
     async def refresh_async(
-        self, *, full: bool = False, source_version: Optional[int] = None
+        self, *, source_version: Optional[int] = None
     ) -> "AsyncJob[RefreshMaterializedViewResult]":
         """Submit a refresh and return its job without waiting.
 
-        The job may already be complete for a local view. On LanceDB Cloud
-        and Enterprise, its ``id`` is the server job identifier returned by
-        the refresh endpoint.
+        The job may already be complete for a local view. Remote refreshes are
+        submitted through the SQL service.
         """
         from ._lancedb import RefreshMaterializedViewResult
 
         return _typed_job(
             await self._table._inner.refresh_materialized_view_async(
-                full=full, source_version=source_version
+                source_version=source_version
             ),
             RefreshMaterializedViewResult.from_json,
         )
@@ -335,22 +235,18 @@ class MaterializedView:
         return LOOP.run(self._async.definition())
 
     def refresh(
-        self, *, full: bool = False, source_version: Optional[int] = None
+        self, *, source_version: Optional[int] = None
     ) -> "RefreshMaterializedViewResult":
         """Recompute the view from its source. See
         [AsyncMaterializedView.refresh][lancedb.materialized_view.AsyncMaterializedView.refresh]."""
-        return LOOP.run(self._async.refresh(full=full, source_version=source_version))
+        return LOOP.run(self._async.refresh(source_version=source_version))
 
     def refresh_async(
-        self, *, full: bool = False, source_version: Optional[int] = None
+        self, *, source_version: Optional[int] = None
     ) -> "Job[RefreshMaterializedViewResult]":
         """Submit a refresh and return its job without waiting.
 
         See
         [AsyncMaterializedView.refresh_async][lancedb.materialized_view.AsyncMaterializedView.refresh_async].
         """
-        return Job(
-            LOOP.run(
-                self._async.refresh_async(full=full, source_version=source_version)
-            )
-        )
+        return Job(LOOP.run(self._async.refresh_async(source_version=source_version)))

python/python/lancedb/remote/db.py

@@ -608,6 +608,9 @@ class RemoteDBConnection(DBConnection):
             One of "error", "drop", "fill".
         fill_value: float
             The value to use when filling vectors. Only used if on_bad_vectors="fill".
+        embedding_functions: list of EmbeddingFunctionConfig, optional
+            The embedding functions to store in the table schema and use to
+            generate vectors when creating the table or adding data.
 
         Returns
         -------

python/python/lancedb/remote/table.py

@@ -64,6 +64,7 @@ from lancedb.table import _normalize_progress
 from ..query import (
     AnalyzePlanDistributedMetrics,
     DocumentGranularity,
+    FullTextQuery,
     LanceQueryBuilder,
     LanceTakeQueryBuilder,
     LanceVectorQueryBuilder,
@@ -78,6 +79,7 @@ from ..table import (
     Tags,
 )
 from ..types import BaseTokenizerType
+from ..util import infer_vector_column_name
 
 
 class RemoteTable(Table):
@@ -731,6 +733,12 @@ class RemoteTable(Table):
             - If the table has multiple vector columns then the *vector_column_name*
             needs to be specified. Otherwise, an error is raised.
 
+        query_type: str, default "auto"
+            The type of search: "auto", "vector", "fts", or "hybrid".
+            With "auto", a string query uses the embedding function for the
+            inferred or specified vector column if available, otherwise it uses
+            full-text search.
+
         fast_search: bool, optional
             Skip a flat search of unindexed data. This may improve
             search performance but search results will not include unindexed data.
@@ -750,6 +758,21 @@ class RemoteTable(Table):
             - and also the "_distance" column which is the distance between the query
             vector and the returned vector.
         """
+        if isinstance(query, FullTextQuery):
+            query_type = "fts"
+        elif (
+            query_type == "auto"
+            and isinstance(query, str)
+            and not self.embedding_functions
+        ):
+            query_type = "fts"
+        vector_column_name = infer_vector_column_name(
+            schema=self.schema,
+            query_type=query_type,
+            query=query,
+            vector_column_name=vector_column_name,
+        )
+
         return LanceQueryBuilder.create(
             self,
             query,

python/python/lancedb/table.py

@@ -5445,11 +5445,16 @@ class AsyncTable:
     async def get_lsm_write_spec(self) -> Optional["LsmWriteSpec"]:
         """Read the LsmWriteSpec currently installed on this table.
 
-        Returns ``None`` when the MemWAL LSM write path is not enabled (no
-        spec has been set, or it was removed with `unset_lsm_write_spec`).
-        The returned spec mirrors what was passed to `set_lsm_write_spec`,
-        except that ``maintained_indexes`` always reports the concrete list
-        resolved when the spec was set — ``None`` never round-trips.
+        Returns ``None`` when the LSM write path is not enabled at all — no
+        spec has been set, or one was removed with `unset_lsm_write_spec`.
+        That is a different answer from a spec whose ``maintained_indexes``
+        is ``None``, which is an installed spec selecting indexes
+        automatically.
+
+        The spec read back is the one that was installed, selection
+        included: ``None`` maintains every supported index the table has now
+        or gains later, ``[]`` maintains none, and a non-empty list maintains
+        exactly those. All three round-trip.
         """
         return await self._inner.get_lsm_write_spec()
 
@@ -6925,9 +6930,11 @@ class AsyncTable:
         """
         versions = await self._inner.list_versions()
         for v in versions:
-            ts_nanos = v["timestamp"]
-            v["timestamp"] = datetime.fromtimestamp(ts_nanos // 1e9) + timedelta(
-                microseconds=(ts_nanos % 1e9) // 1e3
+            # Use integer math: float division on ~1e18 nanosecond
+            # values loses sub-millisecond precision.
+            seconds, nanos = divmod(v["timestamp"], 1_000_000_000)
+            v["timestamp"] = datetime.fromtimestamp(seconds) + timedelta(
+                microseconds=nanos // 1000
             )
 
         return versions

python/python/tests/test_catalog.py

@@ -55,7 +55,7 @@ def test_catalog_sync_scope_and_serialization(catalog_server):
             (200, {}),
             (200, {"tables": []}),
             (200, {"tables": []}),
-            (200, {"namespaces": ["team/search"], "page_token": "next"}),
+            (200, {"namespaces": ["team/search"]}),
             (204, None),
         ]
     )
@@ -85,13 +85,12 @@ def test_catalog_sync_scope_and_serialization(catalog_server):
         with pytest.raises(ValueError, match="sql_host_override must use"):
             connection.execute_query_async("SELECT 1")
     assert restored.table_names() == []
-    page = catalog.list_databases(limit=1, page_token="a/b")
-    assert page == lancedb.ListDatabasesResponse(["team/search"], "next")
+    assert list(catalog.list_databases()) == ["team/search"]
     catalog.drop_database("team/search", ignore_missing=True)
     assert requests[0][0] == "/v1/namespace/team%2Fsearch/create"
     assert requests[0][2] == {"mode": "ExistOk"}
     assert requests[1][0] == "/v1/namespace/team%2Fsearch/describe"
-    assert requests[4][0] == "/v1/namespace/$/list?limit=1&page_token=a%2Fb"
+    assert requests[4][0] == "/v1/namespace/$/list"
     assert requests[5][2] == {"mode": "Skip", "behavior": "Restrict"}
     for i, (_, headers, _) in enumerate(requests):
         headers = {key.lower(): value for key, value in headers.items()}
@@ -138,3 +137,129 @@ async def test_catalog_async_and_errors(catalog_server):
 def test_catalog_requires_remote_endpoint(endpoint):
     with pytest.raises(ValueError, match="endpoint"):
         lancedb.connect_catalog(endpoint)
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.asyncio
+@pytest.mark.parametrize("page_token,page_limit", [(None, None), ("start/token", 2)])
+async def test_database_names_iteration(
+    catalog_server, asynchronous, page_token, page_limit
+):
+    endpoint, requests, responses = catalog_server
+    responses.extend(
+        [
+            (200, {"namespaces": ["first", "second"], "page_token": "a/b"}),
+            (200, {"namespaces": [], "page_token": "next"}),
+            (200, {"namespaces": ["last", "final"], "page_token": ""}),
+        ]
+    )
+    catalog = (
+        await lancedb.connect_catalog_async(endpoint)
+        if asynchronous
+        else lancedb.connect_catalog(endpoint)
+    )
+    names = catalog.list_databases(page_token=page_token, page_limit=page_limit)
+    assert requests == []
+    assert names.num_page_results() == 0
+    assert names.page_token() == page_token
+    for cached, expected in [(1, "first"), (0, "second")]:
+        assert (await names.__anext__() if asynchronous else next(names)) == expected
+        assert names.num_page_results() == cached
+        assert names.page_token() == "a/b"
+        assert len(requests) == 1
+    assert (await names.__anext__() if asynchronous else next(names)) == "last"
+    assert names.num_page_results() == 1
+    assert names.page_token() is None
+    assert (await names.__anext__() if asynchronous else next(names)) == "final"
+    assert names.num_page_results() == 0
+    assert names.page_token() is None
+    for _ in range(2):
+        with pytest.raises(StopAsyncIteration if asynchronous else StopIteration):
+            if asynchronous:
+                await names.__anext__()
+            else:
+                next(names)
+    from urllib.parse import parse_qs, urlsplit
+
+    for request, token in zip(requests, [page_token, "a/b", "next"]):
+        url = urlsplit(request[0])
+        assert url.path == "/v1/namespace/$/list"
+        expected = {}
+        if token is not None:
+            expected["page_token"] = [token]
+        if page_limit is not None:
+            expected["limit"] = [str(page_limit)]
+        assert parse_qs(url.query) == expected
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize("fail", [False, True])
+@pytest.mark.asyncio
+async def test_database_names_empty_or_error(catalog_server, asynchronous, fail):
+    endpoint, requests, responses = catalog_server
+    responses.append(
+        (401, {"error": "unauthorized"}) if fail else (200, {"namespaces": []})
+    )
+    catalog = (
+        await lancedb.connect_catalog_async(endpoint)
+        if asynchronous
+        else lancedb.connect_catalog(endpoint)
+    )
+    names = catalog.list_databases()
+    if fail:
+        with pytest.raises(HttpError):
+            if asynchronous:
+                await names.__anext__()
+            else:
+                next(names)
+    assert ([name async for name in names] if asynchronous else list(names)) == []
+    assert len(requests) == 1
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.asyncio
+async def test_database_names_resume(catalog_server, asynchronous):
+    endpoint, requests, responses = catalog_server
+    responses.extend(
+        [
+            (200, {"namespaces": ["first"], "page_token": "resume/token"}),
+            (200, {"namespaces": ["second"]}),
+        ]
+    )
+    catalog = (
+        await lancedb.connect_catalog_async(endpoint)
+        if asynchronous
+        else lancedb.connect_catalog(endpoint)
+    )
+    names = catalog.list_databases(page_limit=1)
+    assert (await names.__anext__() if asynchronous else next(names)) == "first"
+    assert names.num_page_results() == 0
+    resumed = catalog.list_databases(page_token=names.page_token(), page_limit=1)
+    assert resumed.page_token() == "resume/token"
+    assert len(requests) == 1
+    assert ([name async for name in resumed] if asynchronous else list(resumed)) == [
+        "second"
+    ]
+    assert resumed.page_token() is None
+    assert resumed.num_page_results() == 0
+    assert requests[1][0] == "/v1/namespace/$/list?limit=1&page_token=resume%2Ftoken"
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize("page_limit", [0, 2**32 - 1])
+@pytest.mark.asyncio
+async def test_database_names_invalid_limit(catalog_server, asynchronous, page_limit):
+    endpoint, requests, _ = catalog_server
+    catalog = (
+        await lancedb.connect_catalog_async(endpoint)
+        if asynchronous
+        else lancedb.connect_catalog(endpoint)
+    )
+    names = catalog.list_databases(page_limit=page_limit)
+    with pytest.raises(ValueError, match="limit"):
+        if asynchronous:
+            await names.__anext__()
+        else:
+            next(names)
+    assert names.num_page_results() == 0
+    assert requests == []

python/python/tests/test_materialized_views.py

@@ -97,24 +97,6 @@ def test_remote_list_uses_namespace_route():
     assert requests == ["/v1/namespace/$/materialized_view/list"]
 
 
-def test_remote_create_async_returns_server_job():
-    with mock_remote_materialized_view_create() as (host, requests):
-        db = lancedb.connect(
-            "db://dev",
-            api_key="fake",
-            host_override=host,
-            client_config={"retry_config": {"retries": 0}},
-        )
-        job = db.create_materialized_view_async("adults", "people", where="age >= 18")
-        assert job.id == "mv-create-123"
-    assert requests == [
-        (
-            "/v1/materialized_view/adults/create",
-            {"query": 'SELECT * FROM "people" WHERE age >= 18', "with_no_data": False},
-        )
-    ]
-
-
 def test_remote_drop_async_returns_server_job():
     with mock_remote_materialized_view_create() as (host, requests):
         db = lancedb.connect(
@@ -221,29 +203,29 @@ def test_definition_round_trips(tmp_path):
     )
 
 
-def test_incremental_refresh_after_append(tmp_path):
+def test_full_refresh_after_append(tmp_path):
     db = make_db(tmp_path)
     view = db.create_materialized_view("copy", "people", with_no_data=True)
     view.refresh()
 
     db.open_table("people").add([{"name": "alan", "age": 41}])
     result = view.refresh()
-    assert result.mode == "incremental"
-    assert result.rows_written == 1
+    assert result.mode == "rebuild"
+    assert result.rows_written == 4
     assert view.table.count_rows() == 4
 
-    assert view.refresh().mode == "no_op"
+    assert view.refresh().mode == "rebuild"
 
 
-def test_incremental_refresh_after_update(tmp_path):
+def test_full_refresh_after_update(tmp_path):
     db = make_db(tmp_path)
     view = db.create_materialized_view("copy", "people", with_no_data=True)
     view.refresh()
 
     db.open_table("people").update(where="name = 'kid'", values={"age": 8})
     result = view.refresh()
-    assert result.mode == "incremental"
-    assert result.rows_written == 1
+    assert result.mode == "rebuild"
+    assert result.rows_written == 3
     rows = view.table.search().to_list()
     assert sorted(row["age"] for row in rows) == [8, 36, 85]
 
@@ -331,8 +313,8 @@ async def test_async_create_and_refresh_jobs(tmp_path):
 
 
 @pytest.mark.asyncio
-async def test_async_incremental(tmp_path):
-    db = await lancedb.connect_async(tmp_path, storage_options=STABLE_ROW_IDS)
+async def test_async_refresh_rebuilds(tmp_path):
+    db = await lancedb.connect_async(tmp_path)
     await db.create_table("people", [{"name": "ada", "age": 36}])
     view = await db.create_materialized_view("copy", "people", with_no_data=True)
     await view.refresh()
@@ -340,15 +322,15 @@ async def test_async_incremental(tmp_path):
     table = await db.open_table("people")
     await table.add([{"name": "alan", "age": 41}])
     result = await view.refresh()
-    assert result.mode == "incremental"
-    assert result.rows_written == 1
+    assert result.mode == "rebuild"
+    assert result.rows_written == 2
 
 
-def test_source_requires_stable_row_ids(tmp_path):
+def test_source_does_not_require_stable_row_ids(tmp_path):
     db = lancedb.connect(tmp_path)
     db.create_table("plain", [{"x": 1}])
-    with pytest.raises(Exception, match="stable row ids"):
-        db.create_materialized_view("v", "plain")
+    view = db.create_materialized_view("v", "plain")
+    assert view.table.count_rows() == 1
 
 
 def test_bare_select_names_are_quoted(tmp_path):
@@ -373,7 +355,7 @@ def test_scalar_select_is_one_column(tmp_path):
     view = db.create_materialized_view("just_name", "people", select="name")
     view.refresh()
     rows = view.table.search().to_list()
-    assert set(rows[0]) - {"__source_row_id"} == {"name"}
+    assert set(rows[0]) == {"name"}
     assert sorted(row["name"] for row in rows) == ["ada", "grace", "kid"]
 
 
@@ -384,7 +366,7 @@ async def test_async_scalar_select_is_one_column(tmp_path):
     view = await db.create_materialized_view("just_name", "people", select="name")
     await view.refresh()
     rows = await view.table.query().to_list()
-    assert set(rows[0]) - {"__source_row_id"} == {"name"}
+    assert set(rows[0]) == {"name"}
 
 
 def test_limit_above_i64_max_is_refused(tmp_path):
@@ -520,166 +502,10 @@ 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": 5, "query": query},
+        {"format": 3, "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_remote_db.py

@@ -19,7 +19,7 @@ import lancedb
 import numpy as np
 from lancedb.conftest import MockNonNormTextEmbeddingFunction, MockTextEmbeddingFunction
 from lancedb.embeddings import EmbeddingFunctionConfig, EmbeddingFunctionRegistry
-from lancedb.query import AsyncQuery, ColumnOrdering
+from lancedb.query import AsyncQuery, ColumnOrdering, LanceVectorQueryBuilder
 from lancedb.remote import ClientConfig
 from lancedb.remote.errors import HttpError, RetryError
 import pytest
@@ -597,6 +597,106 @@ def test_remote_permutation_is_picklable():
         ]
 
 
+@pytest.mark.parametrize("with_schema", [True, False])
+def test_create_table_embedding_functions(with_schema):
+    func = MockNonNormTextEmbeddingFunction.create()
+    config = EmbeddingFunctionConfig(
+        source_column="text", vector_column="vector", function=func
+    )
+    schema = pa.schema(
+        [pa.field("text", pa.string()), pa.field("vector", pa.list_(pa.float32(), 10))]
+    )
+    received = {}
+
+    def handler(request):
+        if request.path == "/v1/table/test/describe/":
+            # Echo the metadata actually sent by create_table, as the server does.
+            metadata = received["create"].schema.metadata or {}
+            send_json(
+                request,
+                {
+                    "version": 1,
+                    "schema": {
+                        "fields": [
+                            {
+                                "name": "text",
+                                "type": {"type": "string"},
+                                "nullable": True,
+                            },
+                            {
+                                "name": "vector",
+                                "type": {
+                                    "type": "fixed_size_list",
+                                    "fields": [
+                                        {
+                                            "name": "item",
+                                            "type": {"type": "float"},
+                                            "nullable": True,
+                                        }
+                                    ],
+                                    "length": 10,
+                                },
+                                "nullable": True,
+                            },
+                        ],
+                        "metadata": {
+                            k.decode(): v.decode() for k, v in metadata.items()
+                        },
+                    },
+                },
+            )
+        elif request.path in (
+            "/v1/table/test/create/?mode=create",
+            "/v1/table/test/insert/",
+        ):
+            if request.headers.get("Transfer-Encoding") == "chunked":
+                body = bytearray()
+                while True:
+                    size = int(request.rfile.readline(), 16)
+                    if size == 0:
+                        request.rfile.readline()
+                        break
+                    body.extend(request.rfile.read(size))
+                    request.rfile.read(2)
+            else:
+                body = request.rfile.read(int(request.headers["Content-Length"]))
+            operation = "create" if "/create/" in request.path else "insert"
+            received[operation] = pa.ipc.open_stream(body).read_all()
+            send_json(request, {})
+        else:
+            request.send_response(404)
+            request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        table = db.create_table(
+            "test",
+            schema=schema if with_schema else None,
+            data=None if with_schema else [{"text": "hello world"}],
+            embedding_functions=[config],
+        )
+        metadata = pa.schema(
+            [],
+            metadata=EmbeddingFunctionRegistry.get_instance().get_table_metadata(
+                [config]
+            ),
+        ).metadata
+        assert received["create"].schema.metadata == metadata
+        assert table.schema.metadata == metadata
+        assert table.schema.field("vector").type == pa.list_(pa.float32(), 10)
+        assert table.embedding_functions["vector"].source_column == "text"
+        if with_schema:
+            assert received["create"].num_rows == 0
+        else:
+            assert received["create"]["vector"].to_pylist() == [
+                func.compute_source_embeddings(["hello world"])[0].tolist()
+            ]
+
+        table.add([{"text": "goodbye world"}])
+        assert received["insert"]["vector"].to_pylist() == [
+            func.compute_source_embeddings(["goodbye world"])[0].tolist()
+        ]
+
+
 def test_create_table_exist_ok():
     def handler(request):
         if request.path == "/v1/table/test/create/?mode=exist_ok":
@@ -1660,14 +1760,51 @@ def test_checkpoint_lsm_sync():
 
 
 @contextlib.contextmanager
-def query_test_table(query_handler, *, server_version=Version("0.1.0")):
+def query_test_table(
+    query_handler,
+    *,
+    server_version=Version("0.1.0"),
+    vector_columns=None,
+    embedding_functions=None,
+):
+    if vector_columns is None:
+        vector_columns = {"vector": 3}
+    schema = {
+        "fields": [
+            {"name": "text", "type": {"type": "string"}, "nullable": False},
+            *[
+                {
+                    "name": name,
+                    "type": {
+                        "type": "fixed_size_list",
+                        "fields": [
+                            {
+                                "name": "item",
+                                "type": {"type": "float"},
+                                "nullable": True,
+                            }
+                        ],
+                        "length": dim,
+                    },
+                    "nullable": False,
+                }
+                for name, dim in vector_columns.items()
+            ],
+        ]
+    }
+    if embedding_functions:
+        metadata = EmbeddingFunctionRegistry.get_instance().get_table_metadata(
+            embedding_functions
+        )
+        schema["metadata"] = {key: value.decode() for key, value in metadata.items()}
+
     def handler(request):
         if request.path == "/v1/table/test/describe/":
             request.send_response(200)
             request.send_header("Content-Type", "application/json")
             request.send_header("phalanx-version", str(server_version))
             request.end_headers()
-            request.wfile.write(b'{"version": 1, "schema": {"fields": []}}')
+            request.wfile.write(json.dumps({"version": 1, "schema": schema}).encode())
         elif request.path == "/v1/table/test/query/":
             content_len = int(request.headers.get("Content-Length"))
             body = request.rfile.read(content_len)
@@ -1718,6 +1855,7 @@ def test_query_sync_minimal():
             "upper_bound": None,
             "ef": None,
             "vector": [1.0, 2.0, 3.0],
+            "vector_column": "vector",
             "nprobes": 20,
             "minimum_nprobes": 20,
             "maximum_nprobes": 20,
@@ -1732,7 +1870,84 @@ def test_query_sync_minimal():
         assert data == expected
 
 
-def test_query_sync_empty_query():
+@pytest.mark.parametrize("query_type", ["auto", "vector", "hybrid"])
+@pytest.mark.parametrize("vector_column_name", [None, "embedding"])
+def test_query_sync_text_embedding(query_type, vector_column_name):
+    embedding_func = MockTextEmbeddingFunction.create()
+    config = EmbeddingFunctionConfig(
+        source_column="text", vector_column="embedding", function=embedding_func
+    )
+    query = "quick brown fox"
+    expected_vector = embedding_func.compute_query_embeddings(query)[0]
+    queries = []
+
+    def handler(body):
+        queries.append(body)
+        if "full_text_query" in body:
+            assert query_type == "hybrid"
+            assert body["full_text_query"]["query"] == query
+            return pa.table({"id": [1], "_rowid": [1], "_score": [1.0]})
+        assert body["vector_column"] == "embedding"
+        assert body["vector"] == pytest.approx(expected_vector)
+        return pa.table({"id": [1], "_rowid": [1], "_distance": [0.0]})
+
+    with query_test_table(
+        handler,
+        vector_columns={
+            "embedding": embedding_func.ndims(),
+            **({"other": embedding_func.ndims()} if vector_column_name else {}),
+        },
+        embedding_functions=[config],
+    ) as table:
+        assert list(table.embedding_functions) == ["embedding"]
+        builder = table.search(
+            query, query_type=query_type, vector_column_name=vector_column_name
+        )
+        if query_type != "hybrid":
+            assert isinstance(builder, LanceVectorQueryBuilder)
+        results = builder.to_list()
+        assert results[0]["id"] == 1
+        if query_type != "hybrid":
+            assert results[0]["_distance"] == 0.0
+    assert len(queries) == (2 if query_type == "hybrid" else 1)
+
+
+@pytest.mark.parametrize("query_type", ["auto", "vector", "hybrid"])
+def test_query_sync_ambiguous_vector_columns(query_type):
+    config = EmbeddingFunctionConfig(
+        source_column="text",
+        vector_column="v1",
+        function=MockTextEmbeddingFunction.create(),
+    )
+    queries = []
+
+    def handler(body):
+        queries.append(body)
+        return pa.table({"id": [1]})
+
+    with query_test_table(
+        handler,
+        vector_columns={"v1": 10, "v2": 10},
+        embedding_functions=[config],
+    ) as table:
+        with pytest.raises(ValueError, match=r"Candidates: \['v1', 'v2'\]"):
+            table.search("puppy", query_type=query_type).to_list()
+    assert queries == []
+
+
+@pytest.mark.parametrize("vector_column, dim", [("v1", 3), ("v2", 4)])
+def test_query_sync_infers_vector_column_by_dimension(vector_column, dim):
+    def handler(body):
+        assert body["vector_column"] == vector_column
+        assert body["vector"] == [1.0] * dim
+        return pa.table({"id": [1]})
+
+    with query_test_table(handler, vector_columns={"v1": 3, "v2": 4}) as table:
+        assert table.search([1.0] * dim).to_list() == [{"id": 1}]
+
+
+@pytest.mark.parametrize("vector_columns", [{}, {"v1": 3, "v2": 3}])
+def test_query_sync_empty_query(vector_columns):
     def handler(body):
         assert body == {
             "k": 10,
@@ -1745,7 +1960,7 @@ def test_query_sync_empty_query():
 
         return pa.table({"id": [1, 2, 3]})
 
-    with query_test_table(handler) as table:
+    with query_test_table(handler, vector_columns=vector_columns) as table:
         data = table.search(None).where("true").select(["id"]).limit(10).to_list()
         expected = [{"id": 1}, {"id": 2}, {"id": 3}]
         assert data == expected
@@ -1995,7 +2210,9 @@ def test_query_sync_batch_queries(server_version):
         assert results == [{"id": 1, "query_index": 0}, {"id": 1, "query_index": 1}]
 
 
-def test_query_sync_fts():
+@pytest.mark.parametrize("vector_columns", [{}, {"vector": 3}, {"v1": 3, "v2": 3}])
+@pytest.mark.parametrize("query_type", ["auto", "fts"])
+def test_query_sync_fts(vector_columns, query_type):
     def handler(body):
         assert body == {
             "full_text_query": {
@@ -2010,8 +2227,8 @@ def test_query_sync_fts():
 
         return pa.table({"id": [1, 2, 3]})
 
-    with query_test_table(handler) as table:
-        (table.search("puppy", query_type="fts").to_list())
+    with query_test_table(handler, vector_columns=vector_columns) as table:
+        (table.search("puppy", query_type=query_type).to_list())
 
     def handler(body):
         assert body == {
@@ -2038,9 +2255,11 @@ def test_query_sync_fts():
 
         return pa.table({"id": [1, 2, 3]})
 
-    with query_test_table(handler) as table:
+    with query_test_table(handler, vector_columns=vector_columns) as table:
         (
-            table.search("puppy", query_type="fts", fts_columns=["name", "description"])
+            table.search(
+                "puppy", query_type=query_type, fts_columns=["name", "description"]
+            )
             .with_row_id(True)
             .limit(42)
             .to_list()
@@ -2078,7 +2297,9 @@ def test_query_sync_fts_document_granularity():
             }
         )
 
-    with query_test_table(handler, server_version=Version("0.6.0")) as table:
+    with query_test_table(
+        handler, server_version=Version("0.6.0"), vector_columns={}
+    ) as table:
         result = table.search(
             MatchQuery(
                 "alpha",
@@ -2113,6 +2334,7 @@ def test_query_sync_hybrid():
                 "prefilter": True,
                 "refine_factor": None,
                 "vector": [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
+                "vector_column": "vector",
                 "nprobes": 20,
                 "minimum_nprobes": 20,
                 "maximum_nprobes": 20,
@@ -2124,15 +2346,14 @@ def test_query_sync_hybrid():
             }
             return pa.table({"_rowid": [1, 2, 3], "_distance": [0.1, 0.2, 0.3]})
 
-    with query_test_table(handler) as table:
-        embedding_func = MockTextEmbeddingFunction()
-        embedding_config = MagicMock()
-        embedding_config.function = embedding_func
-
-        embedding_funcs = MagicMock()
-        embedding_funcs.get = MagicMock(return_value=embedding_config)
-        table.embedding_functions = embedding_funcs
-
+    config = EmbeddingFunctionConfig(
+        source_column="text",
+        vector_column="vector",
+        function=MockTextEmbeddingFunction.create(),
+    )
+    with query_test_table(
+        handler, vector_columns={"vector": 10}, embedding_functions=[config]
+    ) as table:
         (table.search("puppy", query_type="hybrid").limit(42).to_list())
 
 

python/python/tests/test_table.py

@@ -993,6 +993,63 @@ def test_polars(mem_db: DBConnection):
     assert len(filtered_result) == 2
 
 
+@pytest.mark.asyncio
+async def test_list_versions_timestamp_precision():
+    # 2026-10-03T01:03:49.274Z in nanoseconds. Float math turns .274000 into
+    # .273999, so make sure the conversion is exact.
+    ts_nanos = 1790989429274000000
+
+    class FakeInner:
+        async def list_versions(self):
+            return [{"version": 1, "timestamp": ts_nanos, "metadata": {}}]
+
+    table = table_module.AsyncTable(FakeInner())
+    versions = await table.list_versions()
+
+    expected = datetime.fromtimestamp(ts_nanos // 1_000_000_000)
+    assert versions[0]["timestamp"] == expected + timedelta(microseconds=274000)
+    assert versions[0]["timestamp"].microsecond == 274000
+
+
+def test_list_versions_timestamp_precision_sync(mem_db: DBConnection):
+    # The sync LanceTable.list_versions delegates to AsyncTable.list_versions,
+    # which is the path the parity report hit.
+    ts_nanos = 1790989429274000000
+
+    class FakeInner:
+        async def list_versions(self):
+            return [{"version": 1, "timestamp": ts_nanos, "metadata": {}}]
+
+    table = mem_db.create_table("ts_precision", data=[{"id": 1}])
+    table._table = table_module.AsyncTable(FakeInner())
+    versions = table.list_versions()
+
+    expected = datetime.fromtimestamp(ts_nanos // 1_000_000_000)
+    assert versions[0]["timestamp"] == expected + timedelta(microseconds=274000)
+    assert versions[0]["timestamp"].microsecond == 274000
+
+
+def test_list_versions_timestamp_precision_remote():
+    # RemoteTable.list_versions also delegates to AsyncTable.list_versions.
+    from lancedb.remote.table import RemoteTable
+
+    ts_nanos = 1790989429274000000
+
+    class FakeInner:
+        def name(self):
+            return "ts_precision"
+
+        async def list_versions(self):
+            return [{"version": 1, "timestamp": ts_nanos, "metadata": {}}]
+
+    table = RemoteTable(table_module.AsyncTable(FakeInner()), "dev")
+    versions = table.list_versions()
+
+    expected = datetime.fromtimestamp(ts_nanos // 1_000_000_000)
+    assert versions[0]["timestamp"] == expected + timedelta(microseconds=274000)
+    assert versions[0]["timestamp"].microsecond == 274000
+
+
 def test_versioning(mem_db: DBConnection):
     table = mem_db.create_table(
         "test",

python/src/catalog.rs

@@ -1,12 +1,13 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors
 
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
-use lancedb::catalog::{
-    CatalogConnection, CreateDatabaseRequest, DropDatabaseRequest, ListDatabasesRequest,
-};
-use pyo3::exceptions::PyValueError;
+use futures::{StreamExt, future::poll_fn};
+
+use lancedb::catalog::{CatalogConnection, CreateDatabaseRequest, DropDatabaseRequest};
+use pyo3::exceptions::{PyStopAsyncIteration, PyValueError};
 use pyo3::{Bound, PyAny, PyRef, PyResult, Python, pyclass, pyfunction, pymethods};
 
 use crate::connection::{Connection, PyClientConfig};
@@ -67,19 +68,48 @@ impl Catalog {
         })
     }
 
-    #[pyo3(signature = (*, limit=None, page_token=None))]
-    fn list_databases<'py>(
-        self_: PyRef<'py, Self>,
-        limit: Option<u32>,
-        page_token: Option<String>,
-    ) -> PyResult<Bound<'py, PyAny>> {
+    #[pyo3(signature = (*, page_token=None, page_limit=None))]
+    fn list_databases(&self, page_token: Option<String>, page_limit: Option<u32>) -> DatabaseNames {
+        DatabaseNames {
+            inner: Arc::new(Mutex::new(
+                self.inner.list_databases(page_token, page_limit),
+            )),
+            next_lock: Arc::new(tokio::sync::Mutex::new(())),
+        }
+    }
+}
+
+#[pyclass]
+pub struct DatabaseNames {
+    inner: Arc<Mutex<lancedb::catalog::DatabaseNames>>,
+    next_lock: Arc<tokio::sync::Mutex<()>>,
+}
+
+#[pymethods]
+impl DatabaseNames {
+    fn num_page_results(&self, py: Python<'_>) -> usize {
+        py.detach(|| self.inner.lock().unwrap().num_page_results())
+    }
+
+    fn page_token(&self, py: Python<'_>) -> Option<String> {
+        py.detach(|| self.inner.lock().unwrap().page_token().map(str::to_owned))
+    }
+
+    fn __aiter__(self_: PyRef<'_, Self>) -> PyRef<'_, Self> {
+        self_
+    }
+
+    fn __anext__(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
         let inner = self_.inner.clone();
+        let next_lock = self_.next_lock.clone();
         future_into_py(self_.py(), async move {
-            let mut request = ListDatabasesRequest::default();
-            request.limit = limit;
-            request.page_token = page_token;
-            let response = inner.list_databases(request).await.infer_error()?;
-            Ok((response.databases, response.page_token))
+            // Serialize advances while allowing synchronous state inspection
+            // between polls, including while a REST request is pending.
+            let _guard = next_lock.lock().await;
+            poll_fn(|cx| inner.lock().unwrap().poll_next_unpin(cx))
+                .await
+                .ok_or_else(|| PyStopAsyncIteration::new_err(""))?
+                .infer_error()
         })
     }
 }

python/src/connection.rs

@@ -407,8 +407,7 @@ impl Connection {
         })
     }
 
-    #[allow(clippy::too_many_arguments)]
-    #[pyo3(signature = (name, source, projections=None, filter=None, limit=None, with_no_data=false, *, vector_source_json=None))]
+    #[pyo3(signature = (name, source, projections=None, filter=None, limit=None, with_no_data=false))]
     pub fn create_materialized_view(
         self_: PyRef<'_, Self>,
         name: String,
@@ -417,17 +416,10 @@ 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);
             }
@@ -443,8 +435,7 @@ impl Connection {
         })
     }
 
-    #[allow(clippy::too_many_arguments)]
-    #[pyo3(signature = (name, source, projections=None, filter=None, limit=None, with_no_data=false, *, vector_source_json=None))]
+    #[pyo3(signature = (name, source, projections=None, filter=None, limit=None, with_no_data=false))]
     pub fn create_materialized_view_async(
         self_: PyRef<'_, Self>,
         name: String,
@@ -453,17 +444,10 @@ 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/lib.rs

@@ -64,6 +64,7 @@ pub fn _lancedb(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
     env_logger::init_from_env(env);
     m.add_class::<Connection>()?;
     m.add_class::<catalog::Catalog>()?;
+    m.add_class::<catalog::DatabaseNames>()?;
     m.add_function(wrap_pyfunction!(catalog::connect_catalog, m)?)?;
     m.add_class::<Session>()?;
     m.add_class::<Table>()?;

python/src/table.rs

@@ -267,7 +267,7 @@ fn fmt_maintained(maintained: &Option<Vec<String>>) -> String {
 /// Constructed via the `bucket(...)`, `identity(...)`, or `unsharded()`
 /// classmethods, then optionally chain `with_maintained_indexes(...)` and
 /// `with_writer_config_defaults(...)`. A fresh spec maintains every index the
-/// MemWAL supports, resolved on install.
+/// table has, including ones created later.
 #[pyclass(module = "lancedb._lancedb", from_py_object)]
 #[derive(Clone, Debug)]
 pub struct LsmWriteSpec {
@@ -307,9 +307,9 @@ impl LsmWriteSpec {
         }
     }
 
-    /// Set which indexes the MemWAL maintains. `None` (the default)
-    /// resolves every supported index on install; a list is verbatim,
-    /// and an empty list maintains nothing.
+    /// Set which indexes the MemWAL maintains. `None` (the default) is
+    /// every index the table has, including ones created later; a list is
+    /// verbatim, and an empty list maintains nothing.
     #[pyo3(signature = (indexes))]
     pub fn with_maintained_indexes(&self, indexes: Option<Vec<String>>) -> Self {
         Self {
@@ -592,8 +592,6 @@ impl From<lancedb::RefreshMaterializedViewResult> for RefreshMaterializedViewRes
     fn from(result: lancedb::RefreshMaterializedViewResult) -> Self {
         let mode = match result.mode {
             lancedb::RefreshMode::Rebuild => "rebuild",
-            lancedb::RefreshMode::Incremental => "incremental",
-            lancedb::RefreshMode::NoOp => "no_op",
         };
         Self {
             mode: mode.to_string(),
@@ -1823,10 +1821,9 @@ impl Table {
         })
     }
 
-    #[pyo3(signature = (full=false, source_version=None))]
+    #[pyo3(signature = (source_version=None))]
     pub fn refresh_materialized_view(
         self_: PyRef<'_, Self>,
-        full: bool,
         source_version: Option<u64>,
     ) -> PyResult<Bound<'_, PyAny>> {
         let inner = self_.inner_ref()?.clone();
@@ -1834,7 +1831,7 @@ impl Table {
             let view = lancedb::MaterializedView::from_table(inner)
                 .await
                 .infer_error()?;
-            let mut builder = view.refresh().full(full);
+            let mut builder = view.refresh();
             if let Some(version) = source_version {
                 builder = builder.source_version(version);
             }
@@ -1843,10 +1840,9 @@ impl Table {
         })
     }
 
-    #[pyo3(signature = (full=false, source_version=None))]
+    #[pyo3(signature = (source_version=None))]
     pub fn refresh_materialized_view_async(
         self_: PyRef<'_, Self>,
-        full: bool,
         source_version: Option<u64>,
     ) -> PyResult<Bound<'_, PyAny>> {
         let inner = self_.inner_ref()?.clone();
@@ -1854,7 +1850,7 @@ impl Table {
             let view = lancedb::MaterializedView::from_table(inner)
                 .await
                 .infer_error()?;
-            let mut builder = view.refresh().full(full);
+            let mut builder = view.refresh();
             if let Some(version) = source_version {
                 builder = builder.source_version(version);
             }
@@ -1869,11 +1865,9 @@ impl Table {
             let view = lancedb::MaterializedView::from_table(inner)
                 .await
                 .infer_error()?;
-            view.definition().to_json().map_err(|err| {
-                PyRuntimeError::new_err(format!(
-                    "failed to serialize materialized-view definition: {err}"
-                ))
-            })
+            Ok(lancedb::materialized_view::definition_metadata_from_sql(
+                view.definition_sql(),
+            ))
         })
     }